Build a Multi-Tenant Event Delivery Engine

Introduction

30 Second Summary

Every time you check out online, update a subscription, or receive a shipping notification, a webhook fires behind the scenes to keep systems in sync. When that webhook fails silently, orders vanish, payments double, and customers lose trust.

In this project, you will build SignalForge, a production-grade event delivery engine that accepts events via a REST API and reliably delivers them as HTTP callbacks to registered endpoints. You will guarantee zero event loss through retries, dead-letter handling, circuit breakers, and a full observability dashboard.

What You'll Build

A multi-tenant event delivery engine powered by Java 21, Spring Boot, PostgreSQL, Redis, and Apache Kafka. This is the same infrastructure that companies like Svix ($490/month) and Hookdeck sell as SaaS.

By the end of this project, you'll have:

• A multi-tenant REST API for tenant registration, endpoint management, and event ingestion with idempotency guarantees.
• A delivery pipeline using PostgreSQL SKIP LOCKED for contention-free work claiming, exponential backoff with jitter, per-endpoint circuit breakers, and dead-letter handling.
• A distributed architecture with separated API and Worker services communicating through Kafka, powered by Java 21 Virtual Threads for 10,000+ concurrent deliveries, with a full Prometheus and Grafana observability dashboard.
• Secret Mission: add HMAC-SHA256 request signing so endpoints can verify event authenticity, and build a delivery replay API that re-enqueues failed events on demand.

Are there any prerequisites?

This project assumes you have basic Java and Spring Boot CRUD experience. Familiarity with REST APIs and relational databases will help you focus on the distributed systems concepts instead of the basics.

You will need IntelliJ IDEA installed. Step 1 walks you through installing Java 21, Maven, Docker Desktop, and Git.

Before We Start

Before We Start...

Set Up Your Development Environment

SignalForge is a multi-service backend application built with Java, Spring Boot, and PostgreSQL.

Before you can write any code, you need three tools on your machine: the Java runtime to compile and run your application, a build tool to manage dependencies, and a container engine to run your database and infrastructure locally.

In this step, get ready to:

• Install Java 21 (Eclipse Temurin).
• Install Apache Maven.
• Install Docker Desktop.

Install Java 21 (Eclipse Temurin)

Java 21 is a Long-Term Support (LTS) release. It includes Virtual Threads, a feature you will rely on later to handle thousands of concurrent HTTP deliveries without running out of threads.

What is Eclipse Temurin?

Eclipse Temurin is a free, open-source distribution of Java maintained by the Adoptium project. It is the same OpenJDK codebase that Oracle distributes, but with no licensing fees or usage restrictions.

macOS

• Open https://adoptium.net/temurin/releases/ in your browser.
• Select macOS as the operating system, aarch64 as the architecture (or x64 if you have an older Intel Mac), and 21 - LTS as the version.
• Download the .pkg installer.
• Open the downloaded .pkg file and follow the installation wizard. Accept all the defaults.

Windows

• Open https://adoptium.net/temurin/releases/ in your browser.
• Select Windows as the operating system, x64 as the architecture, and 21 - LTS as the version.
• Download the .msi installer.
• Open the downloaded .msi file and follow the installation wizard.
• On the Custom Setup screen, make sure Add to PATH and Set JAVA_HOME variable are both enabled.
• Click Install and accept any prompts.

Now verify that Java is available from your terminal.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
• Check your Java installation by running this command:

java -version

✔️ I see a version number

You should see output that starts with openjdk version "21.0 followed by a patch number. Java 21 is ready to go.

ⓧ Command not found

This means Java was not added to your system PATH during installation.

• Close your terminal completely and open a new one. Newly installed PATH entries only take effect in new terminal windows.
• If it still does not work on Windows, re-run the Temurin .msi installer. On the Custom Setup screen, enable Add to PATH and Set JAVA_HOME variable.
• On macOS, re-run the .pkg installer. The Temurin macOS installer automatically sets up JAVA_HOME and PATH entries.

Still stuck? Help me fix my Java 21 installation. I ran java -version and got 'command not found'.

Install Apache Maven

Maven is the build tool that compiles your Java code, downloads third-party libraries, and packages your application into a runnable JAR. Every Spring Boot project uses a pom.xml file to declare its dependencies, and Maven reads that file to fetch everything you need.

macOS

The simplest way to install Maven on macOS is through Homebrew.

• Install Maven by running this command in your terminal:

brew install maven

Don't have Homebrew?

If you see brew: command not found, you need to install Homebrew first. Visit https://brew.sh and follow the one-line install command on the homepage. Then retry the Maven install above.

Windows

Maven on Windows requires a manual download and PATH setup.

• Open https://maven.apache.org/download.cgi in your browser.
• Under Apache Maven 3.9.15, download the Binary zip archive link (apache-maven-3.9.15-bin.zip).
• Extract the zip to C:\Program Files\Apache\maven. You should end up with a folder like C:\Program Files\Apache\maven\apache-maven-3.9.15.
• Press the Windows key and type environment variables.
• Click Edit the system environment variables.
• Click Environment Variables at the bottom of the dialog.
• Under System variables, find the Path variable and click Edit.
• Click New and add C:\Program Files\Apache\maven\apache-maven-3.9.15\bin.
• Click OK on all three dialogs to save your changes.

• Open a new terminal window (PATH changes only apply to new windows).
• Verify your Maven installation by running this command:

mvn -version

✔️ I see a version number

You should see output starting with Apache Maven 3.9 and a line showing Java version: 21. This confirms Maven is installed and pointing at your Java 21 JDK.

ⓧ Command not found

• Make sure you opened a brand new terminal window after editing your PATH. Old windows still use the previous PATH.
• On Windows, double-check that the path you added ends with \bin. The Maven executable lives inside the bin subdirectory, not in the root of the extracted folder.
• On macOS, try running brew doctor to check for Homebrew issues, then rerun brew install maven.

Still stuck? Help me fix my Maven installation. I ran mvn -version and got 'command not found'.

Why does Maven show a Java version?

Maven runs on the JDK, so it reports which Java it is using. If you see Java version: 21 in the output, Maven has correctly detected the Temurin JDK you just installed.

Install Docker Desktop

Docker lets you run services like PostgreSQL and Redis inside isolated containers on your machine. Instead of installing each database natively, you define all your infrastructure in a single docker-compose.yml file and start everything with one command.

macOS

• Open https://www.docker.com/products/docker-desktop/ in your browser.
• Click Download for Mac. Choose Apple Chip if you have an M-series Mac, or Intel Chip for older machines.
• Open the downloaded .dmg file and drag the Docker icon into your Applications folder.
• Open Docker Desktop from your Applications folder. It will ask for your system password to install its networking components. Enter your password to proceed.
• Wait for the Docker engine to start. The Docker whale icon in your menu bar will stop animating once it is ready.

Windows

• Open https://www.docker.com/products/docker-desktop/ in your browser.
• Click Download for Windows.
• Run the downloaded installer. The installer will ask to enable the WSL 2 backend. Make sure this checkbox stays checked.
• Click Ok and wait for the installation to complete. You may need to restart your computer.
• After restarting, Docker Desktop should launch automatically. If not, press the Windows key, type Docker Desktop, and press Enter.
• Wait for the Docker engine to finish starting. The status bar at the bottom left of the Docker Desktop window will show Engine running when it is ready.

What is WSL 2?

WSL 2 (Windows Subsystem for Linux 2) is a lightweight Linux kernel built into Windows. Docker Desktop uses it to run Linux containers natively on your machine instead of a heavier virtual machine.

• Verify your Docker installation by running this command in your terminal:

docker --version

✔️ I see a version number

You should see output like Docker version 27.x.x (or a newer version). Docker is ready.

ⓧ Command not found or error

• Make sure Docker Desktop is running. Look for the whale icon in your menu bar (macOS) or system tray (Windows).
• On Windows, if you see a WSL 2 error, open PowerShell as Administrator and run wsl --install to install or update WSL 2. Restart your computer after this completes.
• Close your terminal and open a new one, then try docker --version again.

Still stuck? Help me fix my Docker Desktop installation. I cannot run docker --version successfully.

Keep Docker Desktop running

Docker Desktop must be running in the background for the rest of this project. You will use it to run your PostgreSQL database, Redis, and eventually Kafka. If you restart your computer, remember to launch Docker Desktop before continuing.

Your development environment is ready. You have Java 21 for compiling and running the application, Maven for managing builds and dependencies, and Docker for running infrastructure. Next up, you will scaffold the Spring Boot project and spin up your PostgreSQL database.

Scaffold the Spring Boot Project and PostgreSQL

Your development environment is ready to go. The next piece is the actual application and the database that backs it.

Without a database, the event delivery engine has nowhere to store tenants, endpoints, or delivery attempts. In this step, you'll generate a Spring Boot project and connect it to a PostgreSQL database running inside Docker.

In this step, get ready to:

• Generate a Spring Boot project with the right dependencies.
• Set up a PostgreSQL database with Docker.
• Configure the application to connect to the database and verify it works.

Generate the Spring Boot project from Spring Initializr

Spring Initializr is a web tool that generates a pre-configured project structure for you. It wires up the build file, directory layout, and dependencies so you can start coding immediately.

• Open your browser and go to start.spring.io.
• Under Project, make sure Maven is selected.
• Under Language, make sure Java is selected.
• Under Spring Boot, select 3.5.14.
• Set Group to com.signalforge.
• Set Artifact to signalforge-engine.
• Set Packaging to Jar.
• Set Java to 21.

Don't see version 3.5.14?

Spring Initializr only shows a few recent versions. If 3.5.14 isn't listed, select the highest 3.5.x version available. Avoid any version labeled SNAPSHOT or M1.

• Click Add Dependencies on the right side of the page.
• Search for Spring Web and select it.
• Search for Spring Data JPA and select it.
• Search for PostgreSQL Driver and select it.
• Search for Flyway Migration and select it.
• Search for Spring Boot Actuator and select it.
• Search for Validation and select it.

What are these dependencies?

• Spring Web lets you build REST APIs that accept and respond to HTTP requests.
• Spring Data JPA maps your Java objects to database tables automatically.
• PostgreSQL Driver is the connector that lets Java talk to a PostgreSQL database.
• Flyway Migration manages database schema changes through versioned SQL scripts.
• Spring Boot Actuator adds health checks and monitoring endpoints to your app.
• Validation lets you enforce rules on incoming API data using annotations.

• Click Generate at the bottom of the page to download the project as a zip file.
• Find the downloaded signalforge-engine.zip in your Downloads folder.
• Extract the zip file to your Desktop.

Now open the extracted project in IntelliJ IDEA.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type IntelliJ and press Enter to open it.
• Click File in the top menu bar.
• Click Open.
• Navigate to your Desktop and select the signalforge-engine folder.
• Click OK.

Wait for Maven to finish syncing

When IntelliJ opens the project, it detects the pom.xml file and starts downloading all the project dependencies. You'll see a progress bar at the bottom of the IntelliJ window.

This can take a minute or two on the first run. Wait for the progress bar to finish before moving on.

IntelliJ not recognizing the project?

Make sure you opened the signalforge-engine folder itself, not a parent folder or the zip file. The folder should contain a pom.xml at its root.

If IntelliJ asks you to import the project as a Maven project, click Trust Project to proceed.

help me troubleshoot my IntelliJ project setup for SignalForge

Create the Docker and application configuration

Instead of installing PostgreSQL directly on your machine, you'll run it inside a Docker container. This keeps your database isolated and easy to tear down.

• Right-click on the project root (signalforge-engine) in IntelliJ's Project panel on the left.
• Click New, then File.
• Name the file docker-compose.yml and press Enter.
• Paste the following configuration into the new file:

# Docker Compose configuration for SignalForge infrastructure
version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: signalforge
      POSTGRES_USER: signalforge
      POSTGRES_PASSWORD: signalforge
    ports:
      - '5432:5432'
    volumes:
      # Persist database data across container restarts
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

What does this configuration do?

• image: postgres:16 pulls the official PostgreSQL 16 image from Docker Hub.
• The three POSTGRES_ environment variables create a database named signalforge with matching credentials on first startup.
• ports: '5432:5432' maps the container's PostgreSQL port to your machine so your app can connect to localhost:5432.
• The postgres_data volume persists your data across container restarts. Without it, stopping the container would delete all your data.

• Save docker-compose.yml.

Next, you need to tell your Spring Boot app how to connect to this database. Spring Boot reads its configuration from a file called application.yml.

• In IntelliJ's Project panel, expand src > main > resources.
• Right-click application.properties and click Refactor, then Rename.
• Rename it to application.yml and press Enter.
• Open the newly renamed application.yml and paste the following configuration:

spring:
  # PostgreSQL connection settings
  datasource:
    url: jdbc:postgresql://localhost:5432/signalforge
    username: signalforge
    password: signalforge
  # JPA and Hibernate settings
  jpa:
    hibernate:
      ddl-auto: validate
    open-in-view: false
  # Flyway database migration settings
  flyway:
    enabled: true
    locations: classpath:db/migration

What does this configuration do?

• datasource tells Spring Boot how to connect to the PostgreSQL container you defined in docker-compose.yml.
• ddl-auto: validate means Hibernate checks that your Java entities match the database schema on startup. It never creates or modifies tables on its own.
• open-in-view: false disables a Spring default that holds database connections open during HTTP responses. Turning it off prevents resource waste.
• The flyway block tells Flyway to look for versioned SQL migration scripts in the db/migration folder. You'll create your first migration in the next step.

• Save application.yml.

✔️ Awesome, I've got everything!

Double check you've saved both files before moving on.

ⓧ I'd like to double check the full code

Here is the full docker-compose.yml:

version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: signalforge
      POSTGRES_USER: signalforge
      POSTGRES_PASSWORD: signalforge
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

Here is the full application.yml:

spring:
  datasource:
    url: jdbc:postgresql://localhost:5432/signalforge
    username: signalforge
    password: signalforge
  jpa:
    hibernate:
      ddl-auto: validate
    open-in-view: false
  flyway:
    enabled: true
    locations: classpath:db/migration

Start PostgreSQL and verify the connection

Your configuration files are in place. Now you'll start the PostgreSQL container and confirm your Spring Boot app can connect to it.

• Open IntelliJ's built-in terminal by clicking the Terminal tab at the bottom of the window, or press Alt+F12 (Windows) / Option+F12 (macOS).
• Start the PostgreSQL container by running this command:

docker compose up -d

• Verify the container is running by running:

docker compose ps

You should see a postgres container with a status of running and port 5432 mapped.

Container not starting?

• Make sure Docker Desktop is running. Check your system tray (Windows) or menu bar (macOS) for the Docker whale icon.
• If you see a port conflict error, another process is already using port 5432. Stop any existing PostgreSQL installations or change the host port in docker-compose.yml.
• Make sure your terminal is in the signalforge-engine folder where docker-compose.yml lives. IntelliJ's built-in terminal opens here by default.

help me troubleshoot my Docker Compose setup for SignalForge

With PostgreSQL running, it's time to start the Spring Boot application and confirm the connection.

• In IntelliJ's Project panel, expand src > main > java > com.signalforge.
• Open SignalforgeEngineApplication.java.
• Click the green play button in the gutter next to the main method, then click Run.

What should I see in the console?

IntelliJ opens a Run panel at the bottom of the window. You should see the Spring Boot banner followed by log lines. Look for a line that says Started SignalforgeEngineApplication. This confirms the app started successfully and connected to PostgreSQL.

• Confirm the health endpoint is working by running this command in IntelliJ's terminal:

curl http://localhost:8080/actuator/health

You should see a JSON response:

{"status":"UP"}

What is the actuator health endpoint?

Spring Boot Actuator exposes a /actuator/health endpoint that reports the application's status. When it returns UP, it means the app is running and its database connection is healthy.

App not starting or health check failing?

• If you see a Connection refused error in the console, the PostgreSQL container might not be running. Run docker compose ps to check.
• Check that your application.yml indentation is correct. YAML is sensitive to spaces. Each nested level should be indented by exactly 2 spaces.
• If the curl command returns Connection refused, the app hasn't finished starting. Wait a few seconds and try again.

help me fix my Spring Boot application startup

Your Spring Boot project is scaffolded and connected to a running PostgreSQL database. Next up, you'll design the data model and create your first Flyway migration to build the tables that store tenants, endpoints, events, and delivery attempts.

Design the Data Model and Flyway Migrations

Your Spring Boot app connects to PostgreSQL and reports healthy, but the database is completely empty. Before SignalForge can accept events or deliver webhooks, it needs tables to store tenants, their endpoints, incoming events, and every delivery attempt.

In this step, you'll define the complete data model as a Flyway migration and then map each table to a JPA entity. Flyway runs your SQL migration automatically on startup, so the schema is versioned, reproducible, and never out of sync with your code.

In this step, get ready to:

• Create a Flyway SQL migration with four tables for the delivery pipeline.
• Add performance indexes and constraints for the delivery worker.
• Map the database tables to JPA entities and Java record DTOs.

Create the Flyway migration file

Flyway manages database schema changes through numbered SQL scripts. Your application.yml already tells Flyway to look in classpath:db/migration for migration files. You need to create that directory and your first migration script.

• In IntelliJ's Project sidebar, expand src/main/resources.
• Right-click on resources, select New, then Directory.
• Type db/migration and press Enter. IntelliJ creates both directories in one step.
• Right-click on the new migration folder, select New, then File.
• Name the file V1__init_schema.sql. The V1 prefix tells Flyway this is version 1, and the double underscore separates the version from the description.

The data model has four tables that form a pipeline: tenants own endpoints, events arrive for tenants, and delivery attempts track each HTTP callback to each endpoint.

• Add the tenants and endpoints tables to V1__init_schema.sql by copying the following SQL:

-- Each tenant is an API consumer who registers endpoints
CREATE TABLE tenants (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  name VARCHAR(255) NOT NULL,
  api_key VARCHAR(255) NOT NULL UNIQUE,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Endpoints are HTTP URLs where a tenant wants events delivered
CREATE TABLE endpoints (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id UUID NOT NULL REFERENCES tenants(id),
  url TEXT NOT NULL,
  status VARCHAR(20) NOT NULL DEFAULT 'ACTIVE',
  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

What does this SQL do?

The tenants table stores each API consumer. The api_key column has a UNIQUE constraint so no two tenants share the same key.

The endpoints table holds the webhook URLs a tenant registers. The tenant_id foreign key enforces that every endpoint belongs to a real tenant. The status column defaults to ACTIVE but can be set to DISABLED to pause deliveries.

• Below the endpoints table, add the events and delivery_attempts tables by copying this SQL:

-- Events are the payloads tenants send to SignalForge for delivery
CREATE TABLE events (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id UUID NOT NULL REFERENCES tenants(id),
  event_type VARCHAR(255) NOT NULL,
  payload JSONB NOT NULL,
  idempotency_key VARCHAR(255),
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  CONSTRAINT uq_events_idempotency UNIQUE (tenant_id, idempotency_key)
);

-- Each delivery attempt tracks one HTTP callback to one endpoint
CREATE TABLE delivery_attempts (
  id BIGSERIAL PRIMARY KEY,
  event_id UUID NOT NULL REFERENCES events(id),
  endpoint_id UUID NOT NULL REFERENCES endpoints(id),
  status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
  attempt_number INT NOT NULL DEFAULT 0,
  http_status INT,
  response_body_snippet TEXT,
  error_message TEXT,
  next_retry_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  version BIGINT NOT NULL DEFAULT 0
);

What does this SQL do?

The events table stores every event a tenant sends to SignalForge. The payload column uses PostgreSQL's JSONB type so you can store arbitrary event data. The UNIQUE (tenant_id, idempotency_key) constraint prevents duplicate event ingestion for the same tenant.

The delivery_attempts table is the heart of the delivery pipeline. Each row represents one attempt to POST an event to an endpoint. The status column tracks the lifecycle: PENDING, DELIVERING, DELIVERED, FAILED, or DEAD. The version column enables optimistic locking as a safety net against concurrent updates.

Add the delivery worker index

The delivery worker will repeatedly query for rows that are ready to deliver. Without an index, that query scans the entire delivery_attempts table on every tick. A partial index narrows the scan to only the rows that matter.

• At the bottom of V1__init_schema.sql, add this partial index:

-- Only index rows the delivery worker cares about (PENDING or FAILED, ready to retry)
CREATE INDEX idx_delivery_ready ON delivery_attempts (next_retry_at, id)
  WHERE status IN ('PENDING', 'FAILED') AND next_retry_at <= now();

Why a partial index?

A regular index covers every row in the table. A partial index (the WHERE clause) only indexes rows that match the filter. Since most delivery attempts eventually reach DELIVERED or DEAD status, the partial index stays small and fast.

When the delivery worker runs its SELECT ... FOR UPDATE SKIP LOCKED query in a later step, PostgreSQL hits this index instead of scanning the heap. This is the same pattern production job queues use to handle millions of rows efficiently.

• Save V1__init_schema.sql.

✔️ Awesome, I've got everything!

Double check you've saved your file. The full migration runs in one shot when you restart the app.

ⓧ I'd like to double check the full code

CREATE TABLE tenants (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  name VARCHAR(255) NOT NULL,
  api_key VARCHAR(255) NOT NULL UNIQUE,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE endpoints (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id UUID NOT NULL REFERENCES tenants(id),
  url TEXT NOT NULL,
  status VARCHAR(20) NOT NULL DEFAULT 'ACTIVE',
  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE events (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id UUID NOT NULL REFERENCES tenants(id),
  event_type VARCHAR(255) NOT NULL,
  payload JSONB NOT NULL,
  idempotency_key VARCHAR(255),
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  CONSTRAINT uq_events_idempotency UNIQUE (tenant_id, idempotency_key)
);

CREATE TABLE delivery_attempts (
  id BIGSERIAL PRIMARY KEY,
  event_id UUID NOT NULL REFERENCES events(id),
  endpoint_id UUID NOT NULL REFERENCES endpoints(id),
  status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
  attempt_number INT NOT NULL DEFAULT 0,
  http_status INT,
  response_body_snippet TEXT,
  error_message TEXT,
  next_retry_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  version BIGINT NOT NULL DEFAULT 0
);

CREATE INDEX idx_delivery_ready ON delivery_attempts (next_retry_at, id)
  WHERE status IN ('PENDING', 'FAILED') AND next_retry_at <= now();

• Verify Flyway applies the migration by restarting your app from IntelliJ. Click the green Run button in the toolbar.
• Check the console output. You should see a log line containing Successfully applied 1 migration from Flyway, followed by the app starting on port 8080.

App fails to start with a migration error?

• Check that your file is named exactly V1__init_schema.sql with a capital V, the number 1, and two underscores. Flyway is strict about this naming convention.
• Confirm the file is inside src/main/resources/db/migration/. If IntelliJ created the directory outside resources, drag it to the correct location.
• If Flyway already ran a partial migration (e.g. you fixed a typo), you may need to drop and recreate the database. Stop the app, run docker compose down -v then docker compose up -d to start fresh.

Why is my Flyway migration failing on startup?

Create the JPA entities

Your application.yml uses ddl-auto: validate, which means Hibernate checks that your Java entity classes match the database schema on startup. If a field is missing or a column name is wrong, the app refuses to start. You need entity classes that map exactly to the tables you just created.

• In IntelliJ's Project sidebar, expand src/main/java/com/signalforge.
• Right-click on the signalforge package, select New, then Package.
• Name the package entity and press Enter.

• Right-click on the entity package, select New, then Java Class.
• Name it Tenant and paste the following code:

package com.signalforge.entity;

import jakarta.persistence.*;
import java.time.OffsetDateTime;
import java.util.UUID;

@Entity
@Table(name = "tenants")
public class Tenant {

    // Database-generated UUID primary key
    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    @Column(nullable = false)
    private String name;

    // Unique API key for tenant authentication
    @Column(name = "api_key", nullable = false, unique = true)
    private String apiKey;

    // Timestamp with timezone, set once on creation
    @Column(name = "created_at", nullable = false, updatable = false)
    private OffsetDateTime createdAt;

    @PrePersist
    protected void onCreate() {
        this.createdAt = OffsetDateTime.now();
    }

    // Getters and setters
    public UUID getId() { return id; }
    public void setId(UUID id) { this.id = id; }

    public String getName() { return name; }
    public void setName(String name) { this.name = name; }

    public String getApiKey() { return apiKey; }
    public void setApiKey(String apiKey) { this.apiKey = apiKey; }

    public OffsetDateTime getCreatedAt() { return createdAt; }
}

What does this entity do?

• @Entity and @Table(name = "tenants") tell Hibernate this class maps to the tenants table.
• GenerationType.UUID lets the JPA provider generate UUIDs for new rows, matching the gen_random_uuid() default in your SQL.
• @PrePersist runs onCreate() just before Hibernate inserts the row, setting createdAt to the current time.

• Create another Java class in the entity package named Endpoint and paste the following code:

package com.signalforge.entity;

import jakarta.persistence.*;
import java.time.OffsetDateTime;
import java.util.UUID;

@Entity
@Table(name = "endpoints")
public class Endpoint {

    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    // Foreign key to the owning tenant
    @Column(name = "tenant_id", nullable = false)
    private UUID tenantId;

    @Column(nullable = false)
    private String url;

    // ACTIVE or DISABLED, controls whether the worker delivers to this endpoint
    @Column(nullable = false, length = 20)
    private String status = "ACTIVE";

    @Column(name = "created_at", nullable = false, updatable = false)
    private OffsetDateTime createdAt;

    @PrePersist
    protected void onCreate() {
        this.createdAt = OffsetDateTime.now();
    }

    // Getters and setters
    public UUID getId() { return id; }
    public void setId(UUID id) { this.id = id; }

    public UUID getTenantId() { return tenantId; }
    public void setTenantId(UUID tenantId) { this.tenantId = tenantId; }

    public String getUrl() { return url; }
    public void setUrl(String url) { this.url = url; }

    public String getStatus() { return status; }
    public void setStatus(String status) { this.status = status; }

    public OffsetDateTime getCreatedAt() { return createdAt; }
}

What does this entity do?

The Endpoint entity maps to the endpoints table. Notice that tenantId is stored as a plain UUID rather than a JPA @ManyToOne relationship. This keeps the entity simple and avoids lazy-loading pitfalls. The foreign key integrity is enforced at the database level by the migration SQL.

• Create another Java class in the entity package named Event and paste the following code:

package com.signalforge.entity;

import jakarta.persistence.*;
import java.time.OffsetDateTime;
import java.util.UUID;

@Entity
@Table(name = "events")
public class Event {

    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    @Column(name = "tenant_id", nullable = false)
    private UUID tenantId;

    @Column(name = "event_type", nullable = false)
    private String eventType;

    // Maps to PostgreSQL JSONB column, stored as a plain String in Java
    @Column(nullable = false, columnDefinition = "jsonb")
    private String payload;

    // Optional key for idempotent event ingestion
    @Column(name = "idempotency_key")
    private String idempotencyKey;

    @Column(name = "created_at", nullable = false, updatable = false)
    private OffsetDateTime createdAt;

    @PrePersist
    protected void onCreate() {
        this.createdAt = OffsetDateTime.now();
    }

    // Getters and setters
    public UUID getId() { return id; }
    public void setId(UUID id) { this.id = id; }

    public UUID getTenantId() { return tenantId; }
    public void setTenantId(UUID tenantId) { this.tenantId = tenantId; }

    public String getEventType() { return eventType; }
    public void setEventType(String eventType) { this.eventType = eventType; }

    public String getPayload() { return payload; }
    public void setPayload(String payload) { this.payload = payload; }

    public String getIdempotencyKey() { return idempotencyKey; }
    public void setIdempotencyKey(String idempotencyKey) { this.idempotencyKey = idempotencyKey; }

    public OffsetDateTime getCreatedAt() { return createdAt; }
}

How does JSONB mapping work here?

The columnDefinition = "jsonb" annotation tells Hibernate this column uses PostgreSQL's JSONB type. In Java, the payload is stored as a plain String. The PostgreSQL JDBC driver handles the conversion between Java strings and JSONB columns automatically.

• Create the final entity class in the entity package named DeliveryAttempt and paste the following code:

package com.signalforge.entity;

import jakarta.persistence.*;
import java.time.OffsetDateTime;
import java.util.UUID;

@Entity
@Table(name = "delivery_attempts")
public class DeliveryAttempt {

    // BIGSERIAL auto-incrementing primary key
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(name = "event_id", nullable = false)
    private UUID eventId;

    @Column(name = "endpoint_id", nullable = false)
    private UUID endpointId;

    // Lifecycle status: PENDING, DELIVERING, DELIVERED, FAILED, DEAD
    @Column(nullable = false, length = 20)
    private String status = "PENDING";

    @Column(name = "attempt_number", nullable = false)
    private int attemptNumber = 0;

    // HTTP status code from the endpoint response (nullable for pending attempts)
    @Column(name = "http_status")
    private Integer httpStatus;

    @Column(name = "response_body_snippet")
    private String responseBodySnippet;

    @Column(name = "error_message")
    private String errorMessage;

    // When the delivery worker should next attempt this delivery
    @Column(name = "next_retry_at", nullable = false)
    private OffsetDateTime nextRetryAt;

    @Column(name = "created_at", nullable = false, updatable = false)
    private OffsetDateTime createdAt;

    @Column(name = "updated_at", nullable = false)
    private OffsetDateTime updatedAt;

    // Optimistic locking: Hibernate checks this value before updating
    @Version
    private Long version;

    @PrePersist
    protected void onCreate() {
        OffsetDateTime now = OffsetDateTime.now();
        this.createdAt = now;
        this.updatedAt = now;
        if (this.nextRetryAt == null) {
            this.nextRetryAt = now;
        }
    }

    @PreUpdate
    protected void onUpdate() {
        this.updatedAt = OffsetDateTime.now();
    }

    // Getters and setters
    public Long getId() { return id; }
    public void setId(Long id) { this.id = id; }

    public UUID getEventId() { return eventId; }
    public void setEventId(UUID eventId) { this.eventId = eventId; }

    public UUID getEndpointId() { return endpointId; }
    public void setEndpointId(UUID endpointId) { this.endpointId = endpointId; }

    public String getStatus() { return status; }
    public void setStatus(String status) { this.status = status; }

    public int getAttemptNumber() { return attemptNumber; }
    public void setAttemptNumber(int attemptNumber) { this.attemptNumber = attemptNumber; }

    public Integer getHttpStatus() { return httpStatus; }
    public void setHttpStatus(Integer httpStatus) { this.httpStatus = httpStatus; }

    public String getResponseBodySnippet() { return responseBodySnippet; }
    public void setResponseBodySnippet(String responseBodySnippet) { this.responseBodySnippet = responseBodySnippet; }

    public String getErrorMessage() { return errorMessage; }
    public void setErrorMessage(String errorMessage) { this.errorMessage = errorMessage; }

    public OffsetDateTime getNextRetryAt() { return nextRetryAt; }
    public void setNextRetryAt(OffsetDateTime nextRetryAt) { this.nextRetryAt = nextRetryAt; }

    public OffsetDateTime getCreatedAt() { return createdAt; }

    public OffsetDateTime getUpdatedAt() { return updatedAt; }

    public Long getVersion() { return version; }
}

What is @Version and why does DeliveryAttempt need it?

The @Version annotation enables optimistic locking. Every time Hibernate updates a DeliveryAttempt row, it adds a WHERE version = ? check to the UPDATE statement and increments the version. If another transaction changed the row first, the update finds zero matching rows and Hibernate throws an OptimisticLockException.

The primary concurrency control for the delivery worker is SELECT FOR UPDATE SKIP LOCKED (coming in Step 6). The @Version field acts as a safety net in case a code path bypasses the SKIP LOCKED claim and tries to update a row that was already modified by another worker.

Why use Long instead of long for the version field?

Spring Data JPA uses the @Version property to detect whether an entity is new. A boxed Long starts as null for unsaved entities. A primitive long defaults to 0, which JPA treats as an existing entity. Using the boxed type ensures Hibernate correctly distinguishes new rows from existing ones.

You also need Java record DTOs for the API request and response shapes. Records are ideal for DTOs because they are immutable and auto-generate constructors, getters, equals(), and toString().

• Create a new package named dto inside com.signalforge.
• Create a Java class named CreateTenantRequest in the dto package and paste the following code:

package com.signalforge.dto;

// Java record: auto-generates constructor, getters, equals, hashCode, toString
public record CreateTenantRequest(String name) {}

• Create another record named CreateEndpointRequest in the dto package:

package com.signalforge.dto;

// Request body for registering a new webhook endpoint URL
public record CreateEndpointRequest(String url) {}

• Create a record named IngestEventRequest in the dto package:

package com.signalforge.dto;

// Request body for ingesting a new event into SignalForge
public record IngestEventRequest(String eventType, String payload) {}

What are these records for?

These records define the shape of incoming API requests. Each record maps to the JSON body that a client sends to your REST endpoints. You will add validation annotations and response DTOs in the next step when you build the controllers.

• Save all files by pressing Cmd+S (macOS) or Ctrl+S (Windows).
• Restart the app from IntelliJ by clicking the green Run button.
• Check the console. You should see the app start successfully with no Hibernate validation errors. Flyway shows Schema `public` is up to date. No migration necessary. because the migration already ran on the previous restart.

Hibernate validation error on startup?

• A SchemaManagementException means a column name in your entity does not match the database. Double-check every @Column(name = "...") annotation against the SQL migration.
• Make sure the package declaration at the top of each entity file is com.signalforge.entity. If the entity is in a different package, Spring Boot's component scan may not find it.
• If you see Missing column: version for DeliveryAttempt, confirm your SQL migration includes the version BIGINT NOT NULL DEFAULT 0 column in the delivery_attempts table.

Why is Hibernate throwing a schema validation error for my entities?

Your data model is locked in. Four tables, a partial index for the delivery worker, and JPA entities that Hibernate has validated against the real schema. Next up, you'll build the REST API that lets tenants register and manage their endpoints.

Build the Tenant and Endpoint REST API

Your data model and Flyway migrations are in place. The tables exist, and your JPA entities map to them. But right now, there is no way for the outside world to interact with SignalForge.

In this step, you will build the REST API layer that lets tenants register themselves, manage their webhook endpoints, and eventually receive events. You will also add a global exception handler so every error returns a clean, consistent RFC 9457 Problem Detail response.

In this step, get ready to:

• Build tenant registration and lookup endpoints.
• Build endpoint management endpoints with URL validation.
• Add a global exception handler returning RFC 9457 Problem Detail responses.

Create the tenant registration and lookup API

The tenant API is the entry point to SignalForge. A tenant registers once and receives an API key. Every other operation in the system is scoped to a tenant.

• In IntelliJ, create a new package called repository inside com.signalforge.
• Create a new Java interface called TenantRepository.java inside the repository package with the following code:

package com.signalforge.repository;

import com.signalforge.entity.Tenant;
import org.springframework.data.jpa.repository.JpaRepository;

import java.util.UUID;

// Spring Data JPA auto-generates CRUD methods for the Tenant entity
public interface TenantRepository extends JpaRepository<Tenant, UUID> {
}

This interface is all Spring Data JPA needs to generate the full suite of CRUD operations at runtime.

• Create a new package called exception inside com.signalforge.
• Create NotFoundException.java inside the exception package with the following code:

package com.signalforge.exception;

// Thrown when a requested resource (tenant, endpoint, event) does not exist
public class NotFoundException extends RuntimeException {

    public NotFoundException(String message) {
        super(message);
    }
}

You will wire this exception into the global error handler later in this step.

• Create a new package called service inside com.signalforge.
• Create TenantService.java inside the service package with the following code:

package com.signalforge.service;

import com.signalforge.entity.Tenant;
import com.signalforge.exception.NotFoundException;
import com.signalforge.repository.TenantRepository;
import org.springframework.stereotype.Service;

import java.util.UUID;

@Service
public class TenantService {

    private final TenantRepository tenantRepository;

    public TenantService(TenantRepository tenantRepository) {
        this.tenantRepository = tenantRepository;
    }

    // Register a new tenant with a randomly generated API key
    public Tenant createTenant(String name) {
        Tenant tenant = new Tenant();
        tenant.setId(UUID.randomUUID());
        tenant.setName(name);
        tenant.setApiKey(UUID.randomUUID().toString());
        return tenantRepository.save(tenant);
    }

    // Retrieve a tenant by ID or throw if not found
    public Tenant getTenant(UUID id) {
        return tenantRepository.findById(id)
                .orElseThrow(() -> new NotFoundException("Tenant not found: " + id));
    }
}

What does this code do?

The TenantService is the business logic layer between the controller and the database. It handles two operations.

• createTenant() generates a random UUID for the tenant ID and a separate UUID as the API key. Both are set before persisting.
• getTenant() looks up a tenant by ID and throws a NotFoundException if the row does not exist. This exception will be caught by the global handler you build later.

Now you need a request DTO with validation annotations so the controller can reject bad input before it reaches the service.

• Create a new package called dto inside com.signalforge.
• Create CreateTenantRequest.java inside the dto package with the following code:

package com.signalforge.dto;

import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.Size;

// Request body for registering a new tenant
public record CreateTenantRequest(
        @NotBlank(message = "Tenant name is required")
        @Size(max = 255, message = "Tenant name must be 255 characters or fewer")
        String name
) {
}

What is Jakarta Bean Validation?

Jakarta Bean Validation is a standard for declaring constraints on Java objects using annotations. Spring Boot includes Hibernate Validator as the default implementation.

When you annotate a controller parameter with @Valid, Spring automatically runs these constraints before your method body executes. If validation fails, Spring throws a MethodArgumentNotValidException with details about every violated constraint.

• Create a new package called controller inside com.signalforge.
• Create TenantController.java inside the controller package with the following code:

package com.signalforge.controller;

import com.signalforge.dto.CreateTenantRequest;
import com.signalforge.entity.Tenant;
import com.signalforge.service.TenantService;
import jakarta.validation.Valid;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.UUID;

@RestController
@RequestMapping("/api/v1/tenants")
public class TenantController {

    private final TenantService tenantService;

    public TenantController(TenantService tenantService) {
        this.tenantService = tenantService;
    }

    // POST /api/v1/tenants - register a new tenant
    @PostMapping
    public ResponseEntity<Tenant> createTenant(@Valid @RequestBody CreateTenantRequest request) {
        Tenant tenant = tenantService.createTenant(request.name());
        return ResponseEntity.status(HttpStatus.CREATED).body(tenant);
    }

    // GET /api/v1/tenants/{id} - retrieve tenant details
    @GetMapping("/{id}")
    public ResponseEntity<Tenant> getTenant(@PathVariable UUID id) {
        return ResponseEntity.ok(tenantService.getTenant(id));
    }
}

What does this code do?

• @RequestMapping("/api/v1/tenants") sets the base path for all endpoints in this controller.
• @Valid on the @RequestBody parameter tells Spring to run Bean Validation before the method body executes.
• The POST endpoint returns HTTP 201 Created with the full tenant object (including the generated API key) in the response body.
• The GET endpoint returns 200 OK with the tenant details, or delegates to NotFoundException if the ID does not match any row.

• Save all files.
• Make sure your app is running in IntelliJ (or restart it if it was already running).
• Test the tenant registration endpoint by running this command in your terminal:

curl -s -X POST http://localhost:8080/api/v1/tenants \
  -H "Content-Type: application/json" \
  -d '{"name": "Acme Corp"}' | python3 -m json.tool

You should see a JSON response containing an id, name, apiKey, and createdAt field.

Copy the tenant ID

Copy the id value from the response. You will use it in the next curl commands to test lookup and endpoint registration.

• Test the tenant lookup endpoint by running this command, replacing the UUID with the id from the previous response:

curl -s http://localhost:8080/api/v1/tenants/[[TENANT_ID="your tenant UUID from the response above"]] | python3 -m json.tool

You should see the same tenant object returned.

Seeing a 404 or 500 error?

• Check that your app restarted successfully in IntelliJ. Look for the Started SignalforgeEngineApplication log line.
• Verify that Docker is running and PostgreSQL is up. Run docker compose ps to check.
• Confirm you pasted the correct UUID in the GET request. A malformed UUID returns a 400 error.

Why is my tenant API returning an error?

Create the endpoint management API

Tenants need to register the URLs where they want to receive webhook callbacks. The endpoint API handles this. Each endpoint belongs to a tenant and can be toggled between ACTIVE and DISABLED.

• Create EndpointRepository.java inside the repository package with the following code:

package com.signalforge.repository;

import com.signalforge.entity.Endpoint;
import org.springframework.data.jpa.repository.JpaRepository;

import java.util.List;
import java.util.UUID;

public interface EndpointRepository extends JpaRepository<Endpoint, UUID> {

    // Find all endpoints that belong to a specific tenant
    List<Endpoint> findByTenantId(UUID tenantId);
}

What is a derived query method?

Spring Data JPA reads the method name findByTenantId and automatically generates a SQL query: SELECT * FROM endpoints WHERE tenant_id = ?. You do not need to write any query yourself.

• Create EndpointService.java inside the service package with the following code:

package com.signalforge.service;

import com.signalforge.entity.Endpoint;
import com.signalforge.exception.NotFoundException;
import com.signalforge.repository.EndpointRepository;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.UUID;

@Service
public class EndpointService {

    private final EndpointRepository endpointRepository;
    private final TenantService tenantService;

    public EndpointService(EndpointRepository endpointRepository, TenantService tenantService) {
        this.endpointRepository = endpointRepository;
        this.tenantService = tenantService;
    }

    // Register a new endpoint URL for a tenant
    public Endpoint createEndpoint(UUID tenantId, String url) {
        // Verify the tenant exists before creating an endpoint
        tenantService.getTenant(tenantId);

        Endpoint endpoint = new Endpoint();
        endpoint.setId(UUID.randomUUID());
        endpoint.setTenantId(tenantId);
        endpoint.setUrl(url);
        endpoint.setStatus("ACTIVE");
        return endpointRepository.save(endpoint);
    }

    // List all endpoints for a tenant
    public List<Endpoint> getEndpoints(UUID tenantId) {
        tenantService.getTenant(tenantId);
        return endpointRepository.findByTenantId(tenantId);
    }

    // Toggle an endpoint between ACTIVE and DISABLED
    public Endpoint updateEndpointStatus(UUID tenantId, UUID endpointId, String status) {
        tenantService.getTenant(tenantId);
        Endpoint endpoint = endpointRepository.findById(endpointId)
                .orElseThrow(() -> new NotFoundException("Endpoint not found: " + endpointId));
        endpoint.setStatus(status);
        return endpointRepository.save(endpoint);
    }
}

What does this code do?

• Every method calls tenantService.getTenant(tenantId) first. This ensures the tenant exists before any endpoint operation proceeds. If the tenant does not exist, a NotFoundException is thrown immediately.
• createEndpoint() defaults the status to ACTIVE. New endpoints start receiving deliveries right away.
• updateEndpointStatus() accepts ACTIVE or DISABLED as the new status. Disabled endpoints will be skipped by the delivery worker you build later.

Now create the request DTOs for the endpoint controller. You need one for creating endpoints and one for updating status.

• Create CreateEndpointRequest.java inside the dto package with the following code:

package com.signalforge.dto;

import jakarta.validation.constraints.NotBlank;
import org.hibernate.validator.constraints.URL;

// Request body for registering a new endpoint
public record CreateEndpointRequest(
        @NotBlank(message = "Endpoint URL is required")
        @URL(message = "Must be a valid URL")
        String url
) {
}

What is the @URL annotation?

The @URL annotation comes from Hibernate Validator (org.hibernate.validator.constraints.URL). It validates that the string is a well-formed URL using Java's java.net.URL constructor under the hood.

This means the URL must include a protocol. For example, https://example.com/webhook passes validation, but example.com/webhook without a scheme does not.

• Create UpdateEndpointStatusRequest.java inside the dto package with the following code:

package com.signalforge.dto;

import jakarta.validation.constraints.Pattern;

// Request body for enabling or disabling an endpoint
public record UpdateEndpointStatusRequest(
        @Pattern(regexp = "ACTIVE|DISABLED", message = "Status must be ACTIVE or DISABLED")
        String status
) {
}

• Create EndpointController.java inside the controller package with the following code:

package com.signalforge.controller;

import com.signalforge.dto.CreateEndpointRequest;
import com.signalforge.dto.UpdateEndpointStatusRequest;
import com.signalforge.entity.Endpoint;
import com.signalforge.service.EndpointService;
import jakarta.validation.Valid;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.List;
import java.util.UUID;

@RestController
@RequestMapping("/api/v1/tenants/{tenantId}/endpoints")
public class EndpointController {

    private final EndpointService endpointService;

    public EndpointController(EndpointService endpointService) {
        this.endpointService = endpointService;
    }

    // POST /api/v1/tenants/{tenantId}/endpoints - register a new endpoint
    @PostMapping
    public ResponseEntity<Endpoint> createEndpoint(
            @PathVariable UUID tenantId,
            @Valid @RequestBody CreateEndpointRequest request) {
        Endpoint endpoint = endpointService.createEndpoint(tenantId, request.url());
        return ResponseEntity.status(HttpStatus.CREATED).body(endpoint);
    }

    // GET /api/v1/tenants/{tenantId}/endpoints - list all endpoints
    @GetMapping
    public ResponseEntity<List<Endpoint>> getEndpoints(@PathVariable UUID tenantId) {
        return ResponseEntity.ok(endpointService.getEndpoints(tenantId));
    }

    // PATCH /api/v1/tenants/{tenantId}/endpoints/{endpointId} - enable or disable
    @PatchMapping("/{endpointId}")
    public ResponseEntity<Endpoint> updateEndpointStatus(
            @PathVariable UUID tenantId,
            @PathVariable UUID endpointId,
            @Valid @RequestBody UpdateEndpointStatusRequest request) {
        Endpoint endpoint = endpointService.updateEndpointStatus(tenantId, endpointId, request.status());
        return ResponseEntity.ok(endpoint);
    }
}

What does this code do?

• The base path /api/v1/tenants/{tenantId}/endpoints nests endpoints under their tenant. Every request automatically includes the tenantId as a path variable.
• The POST endpoint uses @Valid to enforce both @NotBlank and @URL validation on the incoming URL before the service method runs.
• The PATCH endpoint accepts a status value constrained to ACTIVE or DISABLED by the @Pattern annotation on the DTO.

• Save all files and restart your app in IntelliJ.
• Test endpoint registration by running this command, replacing the tenant ID with yours:

curl -s -X POST http://localhost:8080/api/v1/tenants/[[TENANT_ID="your tenant UUID"]]/endpoints \
  -H "Content-Type: application/json" \
  -d '{"url": "https://example.com/webhook"}' | python3 -m json.tool

You should see a JSON response with the new endpoint, including its id, url, and status set to ACTIVE.

• Test listing endpoints by running:

curl -s http://localhost:8080/api/v1/tenants/[[TENANT_ID="your tenant UUID"]]/endpoints | python3 -m json.tool

You should see a JSON array containing the endpoint you just created.

Getting a validation error on the URL?

• Make sure your URL includes the protocol. https://example.com/webhook works. example.com/webhook without https:// does not.
• Check that the Content-Type header is set to application/json in your curl command.

Why is my endpoint registration failing?

Add the global exception handler

Right now, unhandled exceptions produce Spring Boot's default error response. That response format varies depending on the error type, which makes it hard for API consumers to parse errors consistently.

RFC 9457 defines a standard Problem Detail format for HTTP error responses. Spring Framework has built-in support for this via the ProblemDetail class.

• Create GlobalExceptionHandler.java inside the com.signalforge.exception package with the following code:

package com.signalforge.exception;

import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.http.HttpStatus;
import org.springframework.http.ProblemDetail;
import org.springframework.web.bind.MethodArgumentNotValidException;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.RestControllerAdvice;

import java.net.URI;
import java.util.stream.Collectors;

@RestControllerAdvice
public class GlobalExceptionHandler {

    // Handle 404 - resource not found
    @ExceptionHandler(NotFoundException.class)
    public ProblemDetail handleNotFound(NotFoundException ex) {
        ProblemDetail problem = ProblemDetail.forStatusAndDetail(
                HttpStatus.NOT_FOUND, ex.getMessage());
        problem.setTitle("Resource Not Found");
        problem.setType(URI.create("https://signalforge.dev/errors/not-found"));
        return problem;
    }

    // Handle 400 - validation failures from @Valid
    @ExceptionHandler(MethodArgumentNotValidException.class)
    public ProblemDetail handleValidation(MethodArgumentNotValidException ex) {
        String detail = ex.getBindingResult().getFieldErrors().stream()
                .map(e -> e.getField() + ": " + e.getDefaultMessage())
                .collect(Collectors.joining(", "));
        ProblemDetail problem = ProblemDetail.forStatusAndDetail(
                HttpStatus.BAD_REQUEST, detail);
        problem.setTitle("Validation Failed");
        problem.setType(URI.create("https://signalforge.dev/errors/validation"));
        return problem;
    }

    // Handle 409 - duplicate idempotency key or unique constraint violation
    @ExceptionHandler(DataIntegrityViolationException.class)
    public ProblemDetail handleConflict(DataIntegrityViolationException ex) {
        ProblemDetail problem = ProblemDetail.forStatusAndDetail(
                HttpStatus.CONFLICT,
                "A resource with the given identifier already exists");
        problem.setTitle("Conflict");
        problem.setType(URI.create("https://signalforge.dev/errors/conflict"));
        return problem;
    }
}

What does this code do?

The @RestControllerAdvice annotation tells Spring to intercept exceptions thrown by any controller and route them through the handler methods here. Each method returns a ProblemDetail object, which Spring serializes as application/problem+json.

• handleNotFound() catches your custom NotFoundException and returns a 404 with the exception message.
• handleValidation() catches MethodArgumentNotValidException (thrown when @Valid fails), collects all field errors, and returns a 400 with a human-readable detail string.
• handleConflict() catches DataIntegrityViolationException (thrown when a unique constraint is violated, like a duplicate idempotency key) and returns a 409.

• Save the file and restart your app.
• Test the 404 handler by requesting a tenant that does not exist:

curl -s http://localhost:8080/api/v1/tenants/00000000-0000-0000-0000-000000000000 | python3 -m json.tool

You should see a Problem Detail response with "title": "Resource Not Found" and status 404.

• Test the validation handler by sending an empty name:

curl -s -X POST http://localhost:8080/api/v1/tenants \
  -H "Content-Type: application/json" \
  -d '{"name": ""}' | python3 -m json.tool

You should see a Problem Detail response with "title": "Validation Failed" and status 400.

Still seeing Spring's default error format?

• Make sure GlobalExceptionHandler is in a package that Spring component scanning picks up. It should be inside com.signalforge.exception (a sub-package of your main application class).
• Verify the @RestControllerAdvice annotation is present on the class. Without it, Spring does not register the handler.
• Restart your app after adding the file. Hot-reload may not pick up new classes in all configurations.

Why are my error responses not using Problem Detail format?

Your REST API is fully wired. Tenants can register, manage their endpoints, and every error returns a clean Problem Detail response. Next up, you will build the event ingestion layer that accepts events and writes them to the database using the Transactional Outbox pattern.

Build Event Ingestion with the Transactional Outbox

Your tenants can register and manage their endpoints. Now you need to give them a way to send events into the system.

The critical challenge is guaranteeing that every event reaches every active endpoint. If the app crashes after saving an event but before scheduling its deliveries, that event is lost forever. In this step, you'll solve this with the Transactional Outbox pattern, which writes both the event and its delivery tasks in a single database transaction.

In this step, get ready to:

• Create an event ingestion REST endpoint.
• Implement the Transactional Outbox to guarantee delivery scheduling.
• Add idempotency protection so duplicate event submissions are handled gracefully.

Create the event ingestion endpoint

The event ingestion endpoint accepts an event type and a JSON payload from the tenant. It also reads an optional Idempotency-Key header that prevents duplicate processing.

• Create a new file CreateEventRequest.java inside your src/main/java/com/signalforge/dto directory.
• Copy the following code into the file:

package com.signalforge.dto;

import com.fasterxml.jackson.databind.JsonNode;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;

// Request body for ingesting a new event
public record CreateEventRequest(
    @NotBlank String eventType,
    @NotNull JsonNode payload
) {}

What does this code do?

This is a Java record that defines the shape of the JSON body your API expects. JsonNode from Jackson lets the payload field accept any valid JSON structure without needing a fixed schema.

• Save CreateEventRequest.java.

Now you need the controller that uses this DTO to accept incoming events.

• Create a new file EventController.java inside src/main/java/com/signalforge/controller.
• Copy the following code into the file:

package com.signalforge.controller;

import com.signalforge.dto.CreateEventRequest;
import com.signalforge.entity.Event;
import com.signalforge.service.EventService;
import jakarta.validation.Valid;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.UUID;

@RestController
@RequestMapping("/api/v1/tenants/{tenantId}/events")
public class EventController {

    private final EventService eventService;

    public EventController(EventService eventService) {
        this.eventService = eventService;
    }

    // Accept a new event and schedule it for delivery
    @PostMapping
    public ResponseEntity<Event> ingestEvent(
            @PathVariable UUID tenantId,
            @RequestHeader(value = "Idempotency-Key", required = false) String idempotencyKey,
            @Valid @RequestBody CreateEventRequest request) {

        // Delegate to the service layer for transactional event processing
        Event event = eventService.ingestEvent(tenantId, request, idempotencyKey);
        return ResponseEntity.status(HttpStatus.CREATED).body(event);
    }
}

What does this code do?

The @RequestHeader annotation reads the Idempotency-Key HTTP header from the request. Setting required = false makes it optional, so requests without the header still work.

Right now the controller always returns 201 Created. You'll update this to return 200 OK for duplicate events once you add idempotency handling later in this step.

• Save EventController.java.

The controller won't compile yet because EventService doesn't exist. You'll create that next.

Implement the Transactional Outbox in EventService

What is the Transactional Outbox?

The Transactional Outbox pattern writes business data and its downstream side effects in a single database transaction. If the transaction succeeds, both the event and all of its delivery tasks are guaranteed to exist. If it fails, nothing is written.

Without this pattern, you'd save the event in one operation and create delivery tasks in another. A crash between those two operations means the event exists but has no deliveries scheduled. That's a lost event.

Before building the service, you need repository interfaces so Spring Data JPA can manage your Event and DeliveryAttempt entities.

• Create a new file EventRepository.java inside src/main/java/com/signalforge/repository.
• Copy the following code into the file:

package com.signalforge.repository;

import com.signalforge.entity.Event;
import org.springframework.data.jpa.repository.JpaRepository;

import java.util.Optional;
import java.util.UUID;

public interface EventRepository extends JpaRepository<Event, UUID> {

    // Used for idempotency checks when a duplicate key is submitted
    Optional<Event> findByTenantIdAndIdempotencyKey(UUID tenantId, String idempotencyKey);
}

• Save EventRepository.java.
• Create a new file DeliveryAttemptRepository.java in the same repository package.
• Copy the following code into the file:

package com.signalforge.repository;

import com.signalforge.entity.DeliveryAttempt;
import org.springframework.data.jpa.repository.JpaRepository;

public interface DeliveryAttemptRepository extends JpaRepository<DeliveryAttempt, Long> {
}

• Save DeliveryAttemptRepository.java.

The delivery worker also needs to know which endpoints are active for a given tenant. Your EndpointRepository already exists, but it needs a new query method.

• Open EndpointRepository.java in the repository package.
• Add this query method inside the interface body:

// Find all endpoints for a tenant filtered by status (ACTIVE or DISABLED)
List<Endpoint> findByTenantIdAndStatus(UUID tenantId, String status);

What does this query method do?

Spring Data JPA generates the SQL query automatically from the method name. findByTenantIdAndStatus becomes SELECT * FROM endpoints WHERE tenant_id = ? AND status = ?. No native SQL needed.

• Save EndpointRepository.java.

Now you have all the data access pieces in place. Time to build the service that ties them together.

• Create a new file EventService.java inside src/main/java/com/signalforge/service.
• Copy the following code into the file:

package com.signalforge.service;

import com.signalforge.dto.CreateEventRequest;
import com.signalforge.entity.DeliveryAttempt;
import com.signalforge.entity.Endpoint;
import com.signalforge.entity.Event;
import com.signalforge.repository.DeliveryAttemptRepository;
import com.signalforge.repository.EndpointRepository;
import com.signalforge.repository.EventRepository;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.List;
import java.util.UUID;

@Service
public class EventService {

    private final EventRepository eventRepository;
    private final EndpointRepository endpointRepository;
    private final DeliveryAttemptRepository deliveryAttemptRepository;

    public EventService(EventRepository eventRepository,
                        EndpointRepository endpointRepository,
                        DeliveryAttemptRepository deliveryAttemptRepository) {
        this.eventRepository = eventRepository;
        this.endpointRepository = endpointRepository;
        this.deliveryAttemptRepository = deliveryAttemptRepository;
    }

    // Single @Transactional method: if anything fails, nothing is written
    @Transactional
    public Event ingestEvent(UUID tenantId, CreateEventRequest request, String idempotencyKey) {
        // Step 1: Save the event to the events table
        Event event = new Event();
        event.setTenantId(tenantId);
        event.setEventType(request.eventType());
        event.setPayload(request.payload().toString());
        event.setIdempotencyKey(idempotencyKey);
        event = eventRepository.save(event);

        // Step 2: Fan out - create a DeliveryAttempt for each ACTIVE endpoint
        List<Endpoint> activeEndpoints =
            endpointRepository.findByTenantIdAndStatus(tenantId, "ACTIVE");
        Instant now = Instant.now();

        for (Endpoint endpoint : activeEndpoints) {
            DeliveryAttempt attempt = new DeliveryAttempt();
            attempt.setEventId(event.getId());
            attempt.setEndpointId(endpoint.getId());
            attempt.setStatus("PENDING");
            attempt.setAttemptNumber(0);
            attempt.setNextRetryAt(now);
            deliveryAttemptRepository.save(attempt);
        }

        return event;
    }
}

Why is this the Transactional Outbox?

The @Transactional annotation wraps the entire method in a single database transaction. The event INSERT and all the delivery_attempt INSERTs either all succeed together or all fail together.

Each DeliveryAttempt starts with status=PENDING and nextRetryAt=now(), which means the delivery worker (built in the next step) can pick them up immediately.

• Save EventService.java.
• Run the application from IntelliJ by clicking the green play button.
• Test the endpoint by sending a POST request. Open your terminal and run this command:

curl -X POST http://localhost:8080/api/v1/tenants/[[TENANT_ID="your tenant UUID from Step 4"]]/events \
  -H "Content-Type: application/json" \
  -d '{"eventType": "order.completed", "payload": {"orderId": 123, "amount": 49.99}}'

You should see a 201 Created response with the event JSON, including a generated id and createdAt timestamp.

Seeing a 404 or 500 error?

• A 404 means the tenant UUID in the URL doesn't exist in your database. Use the UUID you received when you created a tenant in Step 4.
• A 500 with a column-mapping error means your Event entity field names don't match the column names in V1__init_schema.sql. Check that eventType maps to event_type and idempotencyKey maps to idempotency_key.
• Make sure you have at least one ACTIVE endpoint registered for this tenant. Without active endpoints, the event is saved but no delivery attempts are created.

Help me debug my EventService ingestion

To verify the Transactional Outbox is working, check that delivery_attempt rows were created alongside the event. Run this query against your database:

docker exec -it signalforge-engine-postgres-1 psql -U signalforge -d signalforge \
  -c "SELECT da.id, da.event_id, da.endpoint_id, da.status, da.next_retry_at FROM delivery_attempts da ORDER BY da.id DESC LIMIT 5;"

You should see one PENDING row for each active endpoint that your tenant has registered.

Add idempotent event ingestion

Network retries, load balancers, and client bugs can all cause the same event to be submitted more than once. Without protection, each submission creates duplicate delivery attempts. Idempotency means processing a duplicate request produces the same result as the original, with no extra side effects.

How does idempotency work here?

Your events table already has a unique constraint on (tenant_id, idempotency_key) from the Flyway migration in Step 3. If a client sends the same idempotency key twice, the database rejects the duplicate INSERT.

The service checks for an existing event first. If one is found, it returns the existing event and skips the fan-out entirely. The unique constraint acts as a safety net for race conditions where two requests arrive at the exact same time.

You need to update EventService to check for duplicates and return a result that tells the controller whether the event was new or already existed.

• Open EventService.java.
• Replace the entire contents of the file with the updated version below:

package com.signalforge.service;

import com.signalforge.dto.CreateEventRequest;
import com.signalforge.entity.DeliveryAttempt;
import com.signalforge.entity.Endpoint;
import com.signalforge.entity.Event;
import com.signalforge.repository.DeliveryAttemptRepository;
import com.signalforge.repository.EndpointRepository;
import com.signalforge.repository.EventRepository;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.List;
import java.util.Optional;
import java.util.UUID;

@Service
public class EventService {

    private final EventRepository eventRepository;
    private final EndpointRepository endpointRepository;
    private final DeliveryAttemptRepository deliveryAttemptRepository;

    public EventService(EventRepository eventRepository,
                        EndpointRepository endpointRepository,
                        DeliveryAttemptRepository deliveryAttemptRepository) {
        this.eventRepository = eventRepository;
        this.endpointRepository = endpointRepository;
        this.deliveryAttemptRepository = deliveryAttemptRepository;
    }

    @Transactional
    public IngestResult ingestEvent(UUID tenantId, CreateEventRequest request, String idempotencyKey) {
        // Idempotency check: if this key was already used, return the existing event
        if (idempotencyKey != null) {
            Optional<Event> existing =
                eventRepository.findByTenantIdAndIdempotencyKey(tenantId, idempotencyKey);
            if (existing.isPresent()) {
                return new IngestResult(existing.get(), true);
            }
        }

        // Save the event to the events table
        Event event = new Event();
        event.setTenantId(tenantId);
        event.setEventType(request.eventType());
        event.setPayload(request.payload().toString());
        event.setIdempotencyKey(idempotencyKey);
        event = eventRepository.save(event);

        // Transactional Outbox: create a DeliveryAttempt for each ACTIVE endpoint
        List<Endpoint> activeEndpoints =
            endpointRepository.findByTenantIdAndStatus(tenantId, "ACTIVE");
        Instant now = Instant.now();

        for (Endpoint endpoint : activeEndpoints) {
            DeliveryAttempt attempt = new DeliveryAttempt();
            attempt.setEventId(event.getId());
            attempt.setEndpointId(endpoint.getId());
            attempt.setStatus("PENDING");
            attempt.setAttemptNumber(0);
            attempt.setNextRetryAt(now);
            deliveryAttemptRepository.save(attempt);
        }

        return new IngestResult(event, false);
    }

    // Result wrapper that signals whether the event was a duplicate
    public record IngestResult(Event event, boolean duplicate) {}
}

What changed?

• The method now returns an IngestResult record instead of a bare Event. The duplicate flag tells the controller whether to return 201 or 200.
• Before saving, the method queries for an existing event with the same tenantId and idempotencyKey. If found, it returns immediately without creating any delivery attempts.
• The unique constraint on (tenant_id, idempotency_key) in the database acts as a safety net for the rare race condition where two identical requests slip past the check simultaneously. Your GlobalExceptionHandler already catches DataIntegrityViolationException and returns a 409 conflict for that edge case.

• Save EventService.java.

Now update the controller to use the new return type and respond with the correct HTTP status code.

• Open EventController.java.
• Find the ingestEvent method and find this line:

        // Delegate to the service layer for transactional event processing
        Event event = eventService.ingestEvent(tenantId, request, idempotencyKey);
        return ResponseEntity.status(HttpStatus.CREATED).body(event);

• Replace those lines with this updated version:

        // Delegate to the service layer for transactional event processing
        EventService.IngestResult result = eventService.ingestEvent(tenantId, request, idempotencyKey);

        // 200 for duplicate (idempotent acceptance), 201 for new event
        HttpStatus status = result.duplicate() ? HttpStatus.OK : HttpStatus.CREATED;
        return ResponseEntity.status(status).body(result.event());

• Save EventController.java.
• Restart the application from IntelliJ.
• Test idempotency by sending the same event twice with an Idempotency-Key header. Run this command:

curl -v -X POST http://localhost:8080/api/v1/tenants/[[TENANT_ID="your tenant UUID from Step 4"]]/events \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: test-key-001" \
  -d '{"eventType": "order.completed", "payload": {"orderId": 456}}'

• Run the exact same command a second time.

The first request should return 201 Created. The second request should return 200 OK with the same event body. The -v flag on curl shows the response status code in the output.

Second request returning 201 instead of 200?

• Make sure both requests include the same Idempotency-Key header value AND the same tenant UUID in the URL path.
• Verify your EventRepository has the findByTenantIdAndIdempotencyKey method. Without it, the check-first query can't find the existing event.
• Confirm that the IngestResult record is returned with duplicate = true for existing events. Check that the controller reads result.duplicate() to set the HTTP status.

Help me debug my idempotency handling

✔️ Awesome, I've got everything!

Double check you've saved all your files before moving on.

ⓧ I'd like to double check the full code

Here is the full EventService.java:

package com.signalforge.service;

import com.signalforge.dto.CreateEventRequest;
import com.signalforge.entity.DeliveryAttempt;
import com.signalforge.entity.Endpoint;
import com.signalforge.entity.Event;
import com.signalforge.repository.DeliveryAttemptRepository;
import com.signalforge.repository.EndpointRepository;
import com.signalforge.repository.EventRepository;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.List;
import java.util.Optional;
import java.util.UUID;

@Service
public class EventService {

    private final EventRepository eventRepository;
    private final EndpointRepository endpointRepository;
    private final DeliveryAttemptRepository deliveryAttemptRepository;

    public EventService(EventRepository eventRepository,
                        EndpointRepository endpointRepository,
                        DeliveryAttemptRepository deliveryAttemptRepository) {
        this.eventRepository = eventRepository;
        this.endpointRepository = endpointRepository;
        this.deliveryAttemptRepository = deliveryAttemptRepository;
    }

    @Transactional
    public IngestResult ingestEvent(UUID tenantId, CreateEventRequest request, String idempotencyKey) {
        // Idempotency check: if this key was already used, return the existing event
        if (idempotencyKey != null) {
            Optional<Event> existing =
                eventRepository.findByTenantIdAndIdempotencyKey(tenantId, idempotencyKey);
            if (existing.isPresent()) {
                return new IngestResult(existing.get(), true);
            }
        }

        // Save the event to the events table
        Event event = new Event();
        event.setTenantId(tenantId);
        event.setEventType(request.eventType());
        event.setPayload(request.payload().toString());
        event.setIdempotencyKey(idempotencyKey);
        event = eventRepository.save(event);

        // Transactional Outbox: create a DeliveryAttempt for each ACTIVE endpoint
        List<Endpoint> activeEndpoints =
            endpointRepository.findByTenantIdAndStatus(tenantId, "ACTIVE");
        Instant now = Instant.now();

        for (Endpoint endpoint : activeEndpoints) {
            DeliveryAttempt attempt = new DeliveryAttempt();
            attempt.setEventId(event.getId());
            attempt.setEndpointId(endpoint.getId());
            attempt.setStatus("PENDING");
            attempt.setAttemptNumber(0);
            attempt.setNextRetryAt(now);
            deliveryAttemptRepository.save(attempt);
        }

        return new IngestResult(event, false);
    }

    // Result wrapper that signals whether the event was a duplicate
    public record IngestResult(Event event, boolean duplicate) {}
}

Here is the full EventController.java:

package com.signalforge.controller;

import com.signalforge.dto.CreateEventRequest;
import com.signalforge.entity.Event;
import com.signalforge.service.EventService;
import jakarta.validation.Valid;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.UUID;

@RestController
@RequestMapping("/api/v1/tenants/{tenantId}/events")
public class EventController {

    private final EventService eventService;

    public EventController(EventService eventService) {
        this.eventService = eventService;
    }

    // Accept a new event and schedule it for delivery
    @PostMapping
    public ResponseEntity<Event> ingestEvent(
            @PathVariable UUID tenantId,
            @RequestHeader(value = "Idempotency-Key", required = false) String idempotencyKey,
            @Valid @RequestBody CreateEventRequest request) {

        // Delegate to the service layer for transactional event processing
        EventService.IngestResult result = eventService.ingestEvent(tenantId, request, idempotencyKey);

        // 200 for duplicate (idempotent acceptance), 201 for new event
        HttpStatus status = result.duplicate() ? HttpStatus.OK : HttpStatus.CREATED;
        return ResponseEntity.status(status).body(result.event());
    }
}

Here is the full EventRepository.java:

package com.signalforge.repository;

import com.signalforge.entity.Event;
import org.springframework.data.jpa.repository.JpaRepository;

import java.util.Optional;
import java.util.UUID;

public interface EventRepository extends JpaRepository<Event, UUID> {

    // Used for idempotency checks when a duplicate key is submitted
    Optional<Event> findByTenantIdAndIdempotencyKey(UUID tenantId, String idempotencyKey);
}

Here is the full DeliveryAttemptRepository.java:

package com.signalforge.repository;

import com.signalforge.entity.DeliveryAttempt;
import org.springframework.data.jpa.repository.JpaRepository;

public interface DeliveryAttemptRepository extends JpaRepository<DeliveryAttempt, Long> {
}

Your event ingestion pipeline is live, and the Transactional Outbox guarantees that no event goes undelivered. Next up, you'll build the delivery worker that picks up those PENDING rows and actually sends HTTP callbacks to the registered endpoints.

Build the Delivery Worker with SKIP LOCKED

Your Transactional Outbox is working. When an event arrives, EventService writes it to the database alongside PENDING delivery_attempts rows for every active endpoint. But those rows are just sitting there. Nothing is picking them up and actually delivering them.

In this step, you'll build the delivery worker that continuously polls for pending work, claims it safely using PostgreSQL's SELECT FOR UPDATE SKIP LOCKED, and fires HTTP callbacks to customer endpoints.

In this step, get ready to:

• Create a delivery worker that claims pending attempts using SELECT FOR UPDATE SKIP LOCKED.
• Deliver events via HTTP POST using RestClient with configurable timeouts.
• Add structured logging as your V1 audit trail.

Create the delivery worker with SKIP LOCKED

When multiple worker instances poll the same table, you need a way to guarantee that no two workers claim the same row. SKIP LOCKED solves this at the database engine level. It atomically locks available rows and silently skips any rows already locked by another transaction.

What is SKIP LOCKED?

A normal SELECT FOR UPDATE blocks when it encounters a locked row. It waits until the lock is released before proceeding.

Adding SKIP LOCKED tells PostgreSQL to skip over locked rows entirely. This turns a regular database table into a concurrent work queue with zero contention between workers.

First, you need to enable Spring's task scheduling so the @Scheduled annotation works.

• In IntelliJ IDEA, open your main application class at src/main/java/com/signalforge/SignalforgeEngineApplication.java.
• Find the @SpringBootApplication annotation. It looks like this:

@SpringBootApplication
public class SignalforgeEngineApplication {

• Add @EnableScheduling directly above the class declaration, so it now looks like this:

import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
public class SignalforgeEngineApplication {

Next, you need native SQL queries that claim and lock delivery attempt rows. The standard JPA @Lock annotation does not reliably produce SKIP LOCKED across all Hibernate versions, so you will write the SQL directly.

• Open DeliveryAttemptRepository.java in the repository package.
• Add these two native query methods to the interface:

import org.springframework.data.jpa.repository.Modifying;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;

// Atomically claim a batch of ready delivery attempts
// SKIP LOCKED ensures no two workers claim the same row
@Query(value = "SELECT id FROM delivery_attempts "
        + "WHERE status IN ('PENDING', 'FAILED') "
        + "AND next_retry_at <= now() "
        + "ORDER BY next_retry_at, id "
        + "FOR UPDATE SKIP LOCKED LIMIT :batchSize",
        nativeQuery = true)
List<Long> claimNextBatch(@Param("batchSize") int batchSize);

// Mark claimed rows as DELIVERING so no other worker touches them
@Modifying
@Query(value = "UPDATE delivery_attempts "
        + "SET status = 'DELIVERING', updated_at = now() "
        + "WHERE id IN (:ids)",
        nativeQuery = true)
void updateStatusToDelivering(@Param("ids") List<Long> ids);

What does this code do?

• claimNextBatch() selects up to batchSize rows that are PENDING or FAILED and whose next_retry_at has passed. The FOR UPDATE SKIP LOCKED clause locks these rows and silently skips any rows already locked by another worker.
• updateStatusToDelivering() flips the claimed rows to DELIVERING status. This prevents them from being picked up again by a future poll cycle.
• The ORDER BY next_retry_at, id ensures oldest-first processing, which hits the idx_delivery_ready partial index you created in your Flyway migration.

Now create the worker class itself. This is the component that runs on a timer and uses the repository queries to claim work.

• Create a new package called worker inside src/main/java/com/signalforge/.
• Create a new file called DeliveryWorker.java inside the worker package and paste the following code:

package com.signalforge.worker;

import com.signalforge.entity.DeliveryAttempt;
import com.signalforge.repository.DeliveryAttemptRepository;
import com.signalforge.repository.EndpointRepository;
import com.signalforge.repository.EventRepository;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestClient;

import java.util.List;

@Component
public class DeliveryWorker {

    private static final int BATCH_SIZE = 20;

    private final DeliveryAttemptRepository deliveryAttemptRepository;
    private final EndpointRepository endpointRepository;
    private final EventRepository eventRepository;
    private final RestClient restClient;

    public DeliveryWorker(DeliveryAttemptRepository deliveryAttemptRepository,
                          EndpointRepository endpointRepository,
                          EventRepository eventRepository,
                          RestClient restClient) {
        this.deliveryAttemptRepository = deliveryAttemptRepository;
        this.endpointRepository = endpointRepository;
        this.eventRepository = eventRepository;
        this.restClient = restClient;
    }

    // Runs every 500ms to poll for pending delivery work
    @Scheduled(fixedDelay = 500)
    public void poll() {
        List<DeliveryAttempt> claimed = claimBatch();
        for (DeliveryAttempt attempt : claimed) {
            deliver(attempt);
        }
    }

    // Claim and lock a batch of ready delivery attempts in one transaction
    @Transactional
    public List<DeliveryAttempt> claimBatch() {
        List<Long> ids = deliveryAttemptRepository.claimNextBatch(BATCH_SIZE);
        if (ids.isEmpty()) {
            return List.of();
        }
        // Mark claimed rows as DELIVERING so future polls skip them
        deliveryAttemptRepository.updateStatusToDelivering(ids);
        return deliveryAttemptRepository.findAllById(ids);
    }

    // Placeholder: HTTP delivery logic comes next
    private void deliver(DeliveryAttempt attempt) {
        // TODO: implement HTTP delivery in the next substep
    }
}

What does this code do?

• @Scheduled(fixedDelay = 500) tells Spring to run poll() every 500 milliseconds. The delay starts after the previous execution finishes, so long batches do not cause overlap.
• claimBatch() runs inside a single @Transactional boundary. It selects IDs with SKIP LOCKED, updates them to DELIVERING, and returns the full entities. The claim transaction stays fast (under 10ms) because no heavy work happens inside it.
• The actual HTTP delivery happens outside the claim transaction in deliver(). This keeps the database lock held for the shortest possible time.

• Save SignalforgeEngineApplication.java, DeliveryAttemptRepository.java, and DeliveryWorker.java.

The app will not compile yet because the RestClient bean does not exist. You will create it in the next substep.

Why a native query instead of @Lock?

Spring Data JPA's @Lock(PESSIMISTIC_WRITE) combined with a @QueryHint timeout of -2 can generate SKIP LOCKED SQL. However, this depends on the exact Hibernate dialect and version. A native query with the full FOR UPDATE SKIP LOCKED clause gives you full control over the generated SQL. No surprises.

Deliver events with RestClient

The worker can claim work, but it still needs to actually deliver it. Each delivery is an HTTP POST to the customer's endpoint URL carrying the event payload as the body.

Spring Boot's RestClient is the modern synchronous HTTP client introduced in Spring Boot 3.2. You will configure it with strict timeouts so a slow endpoint cannot stall your entire worker.

• Create a new package called config inside src/main/java/com/signalforge/.
• Create a new file called RestClientConfig.java inside the config package and paste the following code:

package com.signalforge.config;

import org.springframework.boot.http.client.ClientHttpRequestFactoryBuilder;
import org.springframework.boot.http.client.ClientHttpRequestFactorySettings;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.client.ClientHttpRequestFactory;
import org.springframework.web.client.RestClient;

import java.time.Duration;

@Configuration
public class RestClientConfig {

    @Bean
    public RestClient restClient() {
        // 5s connect timeout, 15s read timeout
        // Prevents a slow endpoint from stalling the entire worker
        ClientHttpRequestFactorySettings settings = ClientHttpRequestFactorySettings.defaults()
                .withConnectTimeout(Duration.ofSeconds(5))
                .withReadTimeout(Duration.ofSeconds(15));

        ClientHttpRequestFactory factory = ClientHttpRequestFactoryBuilder
                .detect()
                .build(settings);

        return RestClient.builder()
                .requestFactory(factory)
                .build();
    }
}

What does this code do?

• ClientHttpRequestFactorySettings.defaults() creates a settings object with Spring Boot's default values. You override the connect and read timeouts.
• ClientHttpRequestFactoryBuilder.detect().build(settings) automatically picks the best available HTTP client library on the classpath and applies your timeout settings.
• The 5-second connect timeout catches endpoints that are unreachable. The 15-second read timeout catches endpoints that accept the connection but never respond.

• Save RestClientConfig.java.

Now implement the deliver() method. This is where the worker looks up the endpoint URL, sends the HTTP POST, and records the result.

• In DeliveryWorker.java, find the placeholder deliver() method:

// Placeholder: HTTP delivery logic comes next
private void deliver(DeliveryAttempt attempt) {
    // TODO: implement HTTP delivery in the next substep
}

• Replace that entire method with the following delivery implementation. Also add the required imports at the top of the file:

// Add these imports at the top of DeliveryWorker.java
import com.signalforge.entity.Endpoint;
import com.signalforge.entity.Event;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestClientResponseException;
import java.time.Instant;

private void deliver(DeliveryAttempt attempt) {
    // Look up the target endpoint URL and the event payload
    Endpoint endpoint = endpointRepository.findById(attempt.getEndpointId()).orElseThrow();
    Event event = eventRepository.findById(attempt.getEventId()).orElseThrow();

    int httpStatus = 0;

    try {
        // POST the event payload to the customer's endpoint
        ResponseEntity<Void> response = restClient.post()
                .uri(endpoint.getUrl())
                .contentType(MediaType.APPLICATION_JSON)
                .body(event.getPayload())
                .retrieve()
                .toBodilessEntity();

        httpStatus = response.getStatusCode().value();
        attempt.setHttpStatus(httpStatus);
        attempt.setStatus("DELIVERED");

    } catch (RestClientResponseException ex) {
        // Non-2xx HTTP response (4xx, 5xx)
        httpStatus = ex.getStatusCode().value();
        attempt.setHttpStatus(httpStatus);
        markFailed(attempt, ex.getMessage());

    } catch (Exception ex) {
        // Network error, timeout, DNS failure, etc.
        markFailed(attempt, ex.getMessage());
    }

    attempt.setUpdatedAt(Instant.now());
    deliveryAttemptRepository.save(attempt);
}

private void markFailed(DeliveryAttempt attempt, String errorMessage) {
    attempt.setStatus("FAILED");
    // Truncate error message to fit the database column
    attempt.setErrorMessage(errorMessage != null
            ? errorMessage.substring(0, Math.min(errorMessage.length(), 255))
            : "Unknown error");
    attempt.setAttemptNumber(attempt.getAttemptNumber() + 1);
    // Simple 30-second retry delay for now (replaced with exponential backoff in Step 7)
    attempt.setNextRetryAt(Instant.now().plusSeconds(30));
}

What does this code do?

• The restClient.post() chain builds and sends an HTTP POST with the event's JSON payload to the customer endpoint. toBodilessEntity() discards the response body but keeps the status code.
• RestClient throws a RestClientResponseException for any non-2xx HTTP response. The second catch block handles network-level failures like timeouts or DNS errors.
• markFailed() increments the attempt number and schedules a retry 30 seconds from now. You will replace this with proper exponential backoff in the next step.

• Save DeliveryWorker.java.
• Make sure Docker Compose is running, then start the app from IntelliJ.
• Check the IntelliJ console. You should see the app start without errors. The worker is now polling every 500ms, but it will not do anything visible until there are PENDING delivery attempts in the database.

Seeing a compilation error?

• Check that RestClientConfig.java is saved and the @Bean method returns a RestClient. Without this bean, Spring cannot inject the RestClient into DeliveryWorker.
• Verify that the import for ClientHttpRequestFactorySettings is from org.springframework.boot.http.client. This class was introduced in Spring Boot 3.4.
• If you see No qualifying bean of type 'RestClient' at startup, the config class may not be in a package that Spring scans. It must be under com.signalforge.

Still stuck? Ask for help.

Log every delivery attempt

Every delivery attempt needs a structured log line. This is your V1 audit trail. When something goes wrong in production, you need to be able to trace which events were delivered, where, and how long it took.

• In DeliveryWorker.java, add the SLF4J Logger import and field at the top of the class. Find this line:

@Component
public class DeliveryWorker {

    private static final int BATCH_SIZE = 20;

• Add the Logger field and import so it looks like this:

// Add these imports at the top of the file
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Component
public class DeliveryWorker {

    private static final Logger log = LoggerFactory.getLogger(DeliveryWorker.class);
    private static final int BATCH_SIZE = 20;

Now add timing and a structured log statement to the deliver() method.

• In the deliver() method, find these two lines near the beginning:

Endpoint endpoint = endpointRepository.findById(attempt.getEndpointId()).orElseThrow();
Event event = eventRepository.findById(attempt.getEventId()).orElseThrow();

• Add a timing variable directly after those two lines:

Endpoint endpoint = endpointRepository.findById(attempt.getEndpointId()).orElseThrow();
Event event = eventRepository.findById(attempt.getEventId()).orElseThrow();

// Track how long the delivery takes
long startMs = System.currentTimeMillis();

• Then find the deliveryAttemptRepository.save(attempt) line at the end of deliver(). Add the structured log statement directly after it:

    attempt.setUpdatedAt(Instant.now());
    deliveryAttemptRepository.save(attempt);

    // Structured audit log: one line per delivery attempt
    long durationMs = System.currentTimeMillis() - startMs;
    log.info("Delivery: event_id={} endpoint_id={} attempt={} http_status={} duration_ms={}",
            attempt.getEventId(), attempt.getEndpointId(),
            attempt.getAttemptNumber(), httpStatus, durationMs);
}

Why structured log fields?

Each log line includes five key fields: event_id, endpoint_id, attempt, http_status, and duration_ms. This format makes log lines searchable with simple text tools.

If a customer reports a missing event, you can grep for its ID and instantly see every delivery attempt, the HTTP status, and how long each one took. In Step 12, you will replace this with proper Micrometer metrics and JSON log output.

• Save DeliveryWorker.java.
• Restart the app from IntelliJ.

To test the full delivery flow, you need an endpoint that accepts HTTP POST requests. Open a browser and go to https://webhook.site to get a free test URL.

• Copy the unique URL from webhook.site (it starts with https://webhook.site/...).
• Use curl or your API client to create a tenant, register an endpoint with that webhook URL, and ingest a test event by running these commands in order:

# Create a tenant
curl -s -X POST http://localhost:8080/api/v1/tenants \
  -H "Content-Type: application/json" \
  -d '{"name": "Test Tenant"}'

Copy the tenant id from the response and use it in the next two commands.

# Register an endpoint (replace TENANT_ID and YOUR_WEBHOOK_URL)
curl -s -X POST http://localhost:8080/api/v1/tenants/[[TENANT_ID="your tenant UUID"]]/endpoints \
  -H "Content-Type: application/json" \
  -d '{"url": "[[WEBHOOK_URL="your webhook.site URL"]]"}'

# Ingest an event
curl -s -X POST http://localhost:8080/api/v1/tenants/[[TENANT_ID="your tenant UUID"]]/events \
  -H "Content-Type: application/json" \
  -d '{"eventType": "order.created", "payload": {"orderId": 42, "amount": 99.99}}'

• Switch to IntelliJ and check the console output. Within 500ms you should see a log line like:

Delivery: event_id=abc-123 endpoint_id=def-456 attempt=0 http_status=200 duration_ms=312

• Switch to webhook.site in your browser. You should see the event payload appear as a received request.

Not seeing a delivery log line?

• Make sure @EnableScheduling is on your main application class. Without it, Spring never runs the @Scheduled method.
• Verify that the endpoint you registered has status ACTIVE. A DISABLED endpoint will not receive delivery attempts.
• Check that next_retry_at on your delivery_attempts rows is in the past. If EventService sets it to Instant.now(), the worker should pick it up immediately.

Still stuck? Ask for help.

✔️ Awesome, I've got everything!

Great. Double check you have saved all your files.

ⓧ I'd like to double check the full code

Here is the complete DeliveryWorker.java as it should look at the end of this step:

package com.signalforge.worker;

import com.signalforge.entity.DeliveryAttempt;
import com.signalforge.entity.Endpoint;
import com.signalforge.entity.Event;
import com.signalforge.repository.DeliveryAttemptRepository;
import com.signalforge.repository.EndpointRepository;
import com.signalforge.repository.EventRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestClient;
import org.springframework.web.client.RestClientResponseException;

import java.time.Instant;
import java.util.List;

@Component
public class DeliveryWorker {

    private static final Logger log = LoggerFactory.getLogger(DeliveryWorker.class);
    private static final int BATCH_SIZE = 20;

    private final DeliveryAttemptRepository deliveryAttemptRepository;
    private final EndpointRepository endpointRepository;
    private final EventRepository eventRepository;
    private final RestClient restClient;

    public DeliveryWorker(DeliveryAttemptRepository deliveryAttemptRepository,
                          EndpointRepository endpointRepository,
                          EventRepository eventRepository,
                          RestClient restClient) {
        this.deliveryAttemptRepository = deliveryAttemptRepository;
        this.endpointRepository = endpointRepository;
        this.eventRepository = eventRepository;
        this.restClient = restClient;
    }

    // Runs every 500ms to poll for pending delivery work
    @Scheduled(fixedDelay = 500)
    public void poll() {
        List<DeliveryAttempt> claimed = claimBatch();
        for (DeliveryAttempt attempt : claimed) {
            deliver(attempt);
        }
    }

    // Claim and lock a batch of ready delivery attempts in one transaction
    @Transactional
    public List<DeliveryAttempt> claimBatch() {
        List<Long> ids = deliveryAttemptRepository.claimNextBatch(BATCH_SIZE);
        if (ids.isEmpty()) {
            return List.of();
        }
        // Mark claimed rows as DELIVERING so future polls skip them
        deliveryAttemptRepository.updateStatusToDelivering(ids);
        return deliveryAttemptRepository.findAllById(ids);
    }

    private void deliver(DeliveryAttempt attempt) {
        // Look up the target endpoint URL and the event payload
        Endpoint endpoint = endpointRepository.findById(attempt.getEndpointId()).orElseThrow();
        Event event = eventRepository.findById(attempt.getEventId()).orElseThrow();

        // Track how long the delivery takes
        long startMs = System.currentTimeMillis();
        int httpStatus = 0;

        try {
            // POST the event payload to the customer's endpoint
            ResponseEntity<Void> response = restClient.post()
                    .uri(endpoint.getUrl())
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(event.getPayload())
                    .retrieve()
                    .toBodilessEntity();

            httpStatus = response.getStatusCode().value();
            attempt.setHttpStatus(httpStatus);
            attempt.setStatus("DELIVERED");

        } catch (RestClientResponseException ex) {
            // Non-2xx HTTP response (4xx, 5xx)
            httpStatus = ex.getStatusCode().value();
            attempt.setHttpStatus(httpStatus);
            markFailed(attempt, ex.getMessage());

        } catch (Exception ex) {
            // Network error, timeout, DNS failure, etc.
            markFailed(attempt, ex.getMessage());
        }

        attempt.setUpdatedAt(Instant.now());
        deliveryAttemptRepository.save(attempt);

        // Structured audit log: one line per delivery attempt
        long durationMs = System.currentTimeMillis() - startMs;
        log.info("Delivery: event_id={} endpoint_id={} attempt={} http_status={} duration_ms={}",
                attempt.getEventId(), attempt.getEndpointId(),
                attempt.getAttemptNumber(), httpStatus, durationMs);
    }

    private void markFailed(DeliveryAttempt attempt, String errorMessage) {
        attempt.setStatus("FAILED");
        // Truncate error message to fit the database column
        attempt.setErrorMessage(errorMessage != null
                ? errorMessage.substring(0, Math.min(errorMessage.length(), 255))
                : "Unknown error");
        attempt.setAttemptNumber(attempt.getAttemptNumber() + 1);
        // Simple 30-second retry delay for now (replaced with exponential backoff in Step 7)
        attempt.setNextRetryAt(Instant.now().plusSeconds(30));
    }
}

And here is the complete RestClientConfig.java:

package com.signalforge.config;

import org.springframework.boot.http.client.ClientHttpRequestFactoryBuilder;
import org.springframework.boot.http.client.ClientHttpRequestFactorySettings;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.client.ClientHttpRequestFactory;
import org.springframework.web.client.RestClient;

import java.time.Duration;

@Configuration
public class RestClientConfig {

    @Bean
    public RestClient restClient() {
        // 5s connect timeout, 15s read timeout
        // Prevents a slow endpoint from stalling the entire worker
        ClientHttpRequestFactorySettings settings = ClientHttpRequestFactorySettings.defaults()
                .withConnectTimeout(Duration.ofSeconds(5))
                .withReadTimeout(Duration.ofSeconds(15));

        ClientHttpRequestFactory factory = ClientHttpRequestFactoryBuilder
                .detect()
                .build(settings);

        return RestClient.builder()
                .requestFactory(factory)
                .build();
    }
}

Your delivery worker is live. Events flow from ingestion through the Transactional Outbox, into the SKIP LOCKED claim, and out as HTTP callbacks. Next up, you will replace the flat 30-second retry delay with proper exponential backoff and jitter so failed deliveries get progressively spaced retries instead of hammering the endpoint.

Implement Retry with Exponential Backoff and Jitter

Your delivery worker can claim events and fire HTTP callbacks. But right now, a single failed delivery is a dead end.

In production, endpoints go down for minutes, hours, or even days. A robust delivery engine does not give up on the first failure. Instead, it retries with increasing delays, backs off when things look bad, and eventually dead-letters events that will never succeed. In this step, you will build that retry engine.

In this step, get ready to:

• Create a RetryPolicy utility with exponential backoff and jitter.
• Update the DeliveryWorker to retry failures and dead-letter exhausted attempts.
• Build a DeliveryStatsController so tenants can see what is failing and why.

Create the RetryPolicy utility class

When a delivery fails, you do not want to retry immediately. Hammering a struggling endpoint with instant retries just makes things worse.

Exponential backoff solves this by spacing retries further apart with each attempt. The formula is base_delay * 2^(attempt - 1), which produces a schedule like 5 seconds, 10 seconds, 20 seconds, 40 seconds, and so on. Svix, a commercial webhook delivery service at $490/month, uses 11 attempts over 5 days. Your engine will match that.

• In IntelliJ, right-click the com.signalforge.worker package in the **Project** panel.
• Select **New**, then **Java Class**.
• Name it RetryPolicy and press **Enter**.
• Replace the generated contents with the following code:

package com.signalforge.worker;

import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.concurrent.ThreadLocalRandom;

/**
 * Calculates retry delays using exponential backoff with jitter.
 * Matches industry standard: 11 attempts over ~5 days.
 */
@Component
public class RetryPolicy {

    // Maximum delivery attempts before dead-lettering
    public static final int MAX_ATTEMPTS = 11;

    // Base delay for the first retry
    private static final long BASE_DELAY_SECONDS = 5;

    // Cap at 1 hour to prevent absurdly long waits
    private static final long MAX_DELAY_SECONDS = 3600;

    /**
     * Calculates the delay before the next retry attempt.
     * Formula: min(base * 2^(attempt-1), maxDelay) + jitter(0..20%)
     */
    public Duration calculateNextRetryDelay(int attemptNumber) {
        // Exponential growth: 5s, 10s, 20s, 40s, 80s, 160s, 320s, 640s, 1280s, 2560s, 3600s
        long exponentialDelay = (long) (BASE_DELAY_SECONDS * Math.pow(2, attemptNumber - 1));

        // Cap at 1 hour
        long cappedDelay = Math.min(exponentialDelay, MAX_DELAY_SECONDS);

        // Add random jitter: 0-20% of the delay to prevent thundering herd
        long jitter = (long) (cappedDelay * 0.2 * ThreadLocalRandom.current().nextDouble());

        return Duration.ofSeconds(cappedDelay + jitter);
    }
}

What does this code do?

MAX_ATTEMPTS = 11 sets the cutoff. After 11 failed attempts, the delivery is dead-lettered instead of retried again.

The formula base * 2^(attempt-1) doubles the delay each time. Attempt 1 waits 5 seconds. Attempt 5 waits 80 seconds. Attempt 10 waits 2,560 seconds (about 42 minutes).

Math.min() caps the delay at 3,600 seconds (1 hour) so retries never stretch into multi-hour gaps.

The jitter line adds a random offset of 0 to 20% of the delay. This is critical for preventing the thundering herd problem, where hundreds of retries fire at the exact same second after an endpoint recovers.

• Save RetryPolicy.java.

You will not see a visible change yet. The next substep wires RetryPolicy into your delivery worker, where you can observe it in action.

Seeing a compilation error?

• Check that the file is in the com.signalforge.worker package. The package declaration at the top must match the folder path.
• Verify that ThreadLocalRandom is imported from java.util.concurrent.ThreadLocalRandom. IntelliJ may auto-import the wrong class.
• Still stuck? Help me fix a compilation error in my RetryPolicy class.

Why add jitter instead of using clean exponential delays?

Imagine 500 events all fail at the same moment because an endpoint goes down. Without jitter, all 500 retries land at exactly second 5, then second 10, then second 20. Each wave hammers the recovering endpoint with the full load.

Jitter spreads those retries across a window (e.g., 5s to 6s instead of all at 5s). The endpoint gets a manageable trickle instead of a wall of requests. This is called the thundering herd problem, and jitter is the standard fix.

Update the DeliveryWorker with retry and dead-letter logic

Right now, your DeliveryWorker sets failed deliveries to FAILED with no plan for what comes next. You need two changes: inject the new RetryPolicy, and replace the failure handling with logic that either schedules a retry or dead-letters the attempt.

• Open DeliveryWorker.java in the com.signalforge.worker package.
• Add RetryPolicy as a constructor parameter and store it as a private final field, just like your existing dependencies. Spring injects it automatically because RetryPolicy is a @Component.

Next, you need to replace the block of code that handles failed deliveries. This is the section inside your delivery loop where you currently set status = FAILED on a non-2xx response or exception.

• Find the section in your delivery loop that handles a failed attempt.
• Replace it with the following retry and dead-letter logic:

// Retry or dead-letter the failed delivery attempt
if (attempt.getAttemptNumber() < RetryPolicy.MAX_ATTEMPTS) {
    // Schedule a retry with exponential backoff + jitter
    attempt.setStatus("FAILED");
    attempt.setAttemptNumber(attempt.getAttemptNumber() + 1);
    Duration delay = retryPolicy.calculateNextRetryDelay(attempt.getAttemptNumber());
    attempt.setNextRetryAt(Instant.now().plus(delay));

    log.info("Delivery failed, scheduling retry. event_id={}, endpoint_id={}, attempt={}, next_retry_at={}",
            attempt.getEvent().getId(),
            attempt.getEndpoint().getId(),
            attempt.getAttemptNumber(),
            attempt.getNextRetryAt());
} else {
    // Max retries exhausted: dead-letter this attempt
    attempt.setStatus("DEAD");

    log.warn("Delivery dead-lettered after {} attempts. event_id={}, endpoint_id={}",
            RetryPolicy.MAX_ATTEMPTS,
            attempt.getEvent().getId(),
            attempt.getEndpoint().getId());
}

deliveryAttemptRepository.save(attempt);

What does this code do?

• The if branch handles retryable failures. It increments the attempt counter, calculates the next delay using your RetryPolicy, and sets next_retry_at so the worker picks it up later.
• The else branch is the dead-letter path. After 11 failed attempts, the delivery is marked DEAD and the worker never picks it up again.
• Structured log lines capture every retry and dead-letter event. This is your audit trail for debugging delivery failures.

• Make sure you have import java.time.Duration; and import java.time.Instant; at the top of DeliveryWorker.java.
• Save DeliveryWorker.java.
• Restart your app in IntelliJ. It should compile and start without errors.

To see the retry logic in action, you need a failing endpoint. If you registered an endpoint in a previous step that points to a URL that is not running (like http://localhost:9999/webhook), ingest an event for that tenant and watch the IntelliJ console. You should see log lines showing the attempt number incrementing and the next_retry_at values growing with each retry.

Not seeing retries in the logs?

• Confirm that your constructor now accepts RetryPolicy retryPolicy and assigns it to a private final field. If Spring cannot inject it, the app will fail to start.
• Check that the delivery_attempts row has status = 'FAILED' and a future next_retry_at timestamp. The worker only picks up rows where next_retry_at <= now(), so the first retry will not appear until the delay elapses.
• Need help? Help me debug why my DeliveryWorker retries are not showing up in the logs.

Build the DeliveryStatsController

Your retry engine works, but tenants have no way to see what is happening with their deliveries. A failing endpoint could silently burn through 11 retries and dead-letter events without anyone noticing.

You will create a REST endpoint that lets tenants query their delivery attempt history. They can filter by status and endpoint to quickly find what is failing and why.

First, your repository needs a query method that supports optional filters. Spring Data JPA does not generate this automatically.

• Open your DeliveryAttemptRepository interface.
• Add the following query method:

// Query delivery attempts for a tenant with optional status and endpoint filters
@Query("SELECT da FROM DeliveryAttempt da JOIN da.event e " +
       "WHERE e.tenant.id = :tenantId " +
       "AND (:status IS NULL OR da.status = :status) " +
       "AND (:endpointId IS NULL OR da.endpoint.id = :endpointId) " +
       "ORDER BY da.createdAt DESC")
List<DeliveryAttempt> findByTenantWithFilters(
        @Param("tenantId") UUID tenantId,
        @Param("status") String status,
        @Param("endpointId") UUID endpointId);

• Save DeliveryAttemptRepository.java.

Now create the controller that exposes this data over HTTP.

• Right-click the com.signalforge.controller package in the **Project** panel.
• Select **New**, then **Java Class**.
• Name it DeliveryStatsController and press **Enter**.
• Replace the generated contents with the following code:

package com.signalforge.controller;

import com.signalforge.entity.DeliveryAttempt;
import com.signalforge.repository.DeliveryAttemptRepository;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.List;
import java.util.UUID;

@RestController
@RequestMapping("/api/v1/tenants/{tenantId}")
public class DeliveryStatsController {

    private final DeliveryAttemptRepository deliveryAttemptRepository;

    // Spring injects the repository automatically
    public DeliveryStatsController(DeliveryAttemptRepository deliveryAttemptRepository) {
        this.deliveryAttemptRepository = deliveryAttemptRepository;
    }

    /**
     * Returns delivery attempt history for a tenant.
     * Supports optional filtering by status and endpoint.
     */
    @GetMapping("/deliveries")
    public ResponseEntity<List<DeliveryAttempt>> getDeliveries(
            @PathVariable UUID tenantId,
            @RequestParam(required = false) String status,
            @RequestParam(required = false) UUID endpointId) {

        // Query with optional filters (null params are ignored by the JPQL query)
        List<DeliveryAttempt> attempts = deliveryAttemptRepository
                .findByTenantWithFilters(tenantId, status, endpointId);

        return ResponseEntity.ok(attempts);
    }
}

What does this code do?

• The @GetMapping("/deliveries") maps to GET /api/v1/tenants/{tenantId}/deliveries. The tenant ID comes from the URL path.
• Both status and endpointId are optional query parameters. When omitted, the JPQL :status IS NULL check evaluates to true and the filter is skipped.
• This gives tenants three query modes: all deliveries, deliveries filtered by status (e.g. ?status=DEAD), or deliveries for a specific endpoint.

• Save DeliveryStatsController.java and DeliveryAttemptRepository.java.
• Restart your app in IntelliJ.

Test the new endpoint by querying delivery attempts for a tenant you created in a previous step. Replace the tenant ID below with your own.

• Send a GET request to fetch all deliveries for your tenant by running:

curl http://localhost:8080/api/v1/tenants/[[TENANT_ID="your-tenant-uuid"]]/deliveries

You should see a JSON array of delivery attempt objects. Each one includes the status, attemptNumber, nextRetryAt, and httpStatus fields.

• Try filtering by status by running:

curl http://localhost:8080/api/v1/tenants/[[TENANT_ID="your-tenant-uuid"]]/deliveries?status=DEAD

If any deliveries have been dead-lettered, they appear here. If none have yet, you get an empty array. Both are correct.

Getting a 404 or 500 on the deliveries endpoint?

• Check that the @RequestMapping path includes {tenantId} in curly braces. A missing brace causes the path to not match.
• If you see a 500 error, check the IntelliJ console for the full stack trace. A common cause is the JPQL query referencing a field name that does not match your entity. The query uses da.event and e.tenant.id. These must match the field names in your DeliveryAttempt and Event entities.
• Still stuck? Help me debug my DeliveryStatsController.

✔️ Awesome, I've got everything!

Make sure you have saved all three files: RetryPolicy.java, DeliveryWorker.java, and DeliveryStatsController.java.

ⓧ I'd like to double check the full code

Here is the full RetryPolicy.java:

package com.signalforge.worker;

import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.concurrent.ThreadLocalRandom;

/**
 * Calculates retry delays using exponential backoff with jitter.
 * Matches industry standard: 11 attempts over ~5 days.
 */
@Component
public class RetryPolicy {

    // Maximum delivery attempts before dead-lettering
    public static final int MAX_ATTEMPTS = 11;

    // Base delay for the first retry
    private static final long BASE_DELAY_SECONDS = 5;

    // Cap at 1 hour to prevent absurdly long waits
    private static final long MAX_DELAY_SECONDS = 3600;

    /**
     * Calculates the delay before the next retry attempt.
     * Formula: min(base * 2^(attempt-1), maxDelay) + jitter(0..20%)
     */
    public Duration calculateNextRetryDelay(int attemptNumber) {
        // Exponential growth: 5s, 10s, 20s, 40s, 80s, 160s, 320s, 640s, 1280s, 2560s, 3600s
        long exponentialDelay = (long) (BASE_DELAY_SECONDS * Math.pow(2, attemptNumber - 1));

        // Cap at 1 hour
        long cappedDelay = Math.min(exponentialDelay, MAX_DELAY_SECONDS);

        // Add random jitter: 0-20% of the delay to prevent thundering herd
        long jitter = (long) (cappedDelay * 0.2 * ThreadLocalRandom.current().nextDouble());

        return Duration.ofSeconds(cappedDelay + jitter);
    }
}

Here is the full DeliveryStatsController.java:

package com.signalforge.controller;

import com.signalforge.entity.DeliveryAttempt;
import com.signalforge.repository.DeliveryAttemptRepository;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.List;
import java.util.UUID;

@RestController
@RequestMapping("/api/v1/tenants/{tenantId}")
public class DeliveryStatsController {

    private final DeliveryAttemptRepository deliveryAttemptRepository;

    // Spring injects the repository automatically
    public DeliveryStatsController(DeliveryAttemptRepository deliveryAttemptRepository) {
        this.deliveryAttemptRepository = deliveryAttemptRepository;
    }

    /**
     * Returns delivery attempt history for a tenant.
     * Supports optional filtering by status and endpoint.
     */
    @GetMapping("/deliveries")
    public ResponseEntity<List<DeliveryAttempt>> getDeliveries(
            @PathVariable UUID tenantId,
            @RequestParam(required = false) String status,
            @RequestParam(required = false) UUID endpointId) {

        // Query with optional filters (null params are ignored by the JPQL query)
        List<DeliveryAttempt> attempts = deliveryAttemptRepository
                .findByTenantWithFilters(tenantId, status, endpointId);

        return ResponseEntity.ok(attempts);
    }
}

For the DeliveryWorker.java changes, confirm that your constructor now accepts RetryPolicy retryPolicy as a parameter, and that your failure handling block matches the retry/dead-letter logic above.

Your delivery engine now handles failures like a production system. Events get 11 chances with increasing delays, and dead-lettered events are permanently flagged for investigation. Next up, you will write integration tests that verify this entire pipeline end-to-end using Testcontainers.

Integration Testing with Testcontainers

Your V1 delivery pipeline is complete. You have event ingestion, a SKIP LOCKED worker, retries with exponential backoff, and dead-letter handling. But how do you know it all works together?

Unit tests verify individual methods, but they can't catch the subtle bugs that surface when your code interacts with a real database. In this step, you'll write integration tests that spin up a real PostgreSQL instance inside Docker and simulate customer endpoints using a mock HTTP server.

In this step, get ready to:

• Add Testcontainers and MockWebServer dependencies to your project.
• Create a reusable base test class that boots a PostgreSQL container for every test.
• Write five integration tests covering the happy path, failure scenarios, idempotency, and disabled endpoints.

Add test dependencies to your project

Integration tests need two things your production code does not: a throwaway database that starts fresh for every test run, and a fake HTTP server that pretends to be a customer endpoint.

What is Testcontainers?

Testcontainers is a Java library that spins up real Docker containers during your tests. Instead of mocking your database or using an in-memory substitute like H2, you get an actual PostgreSQL instance. This guarantees your tests behave exactly like production, especially for PostgreSQL-specific features like JSONB, partial indexes, and FOR UPDATE SKIP LOCKED.

MockWebServer from OkHttp is a scriptable HTTP server. You enqueue responses (like a 200 or 500), then verify the requests your code actually sent.

• Open pom.xml in IntelliJ and find the <dependencies> section.
• Add the following four test dependencies at the end of the <dependencies> block:

<!-- Testcontainers: spins up a real PostgreSQL Docker container for tests -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-testcontainers</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>postgresql</artifactId>
    <scope>test</scope>
</dependency>

<!-- MockWebServer: simulates customer HTTP endpoints in tests -->
<dependency>
    <groupId>com.squareup.okhttp3</groupId>
    <artifactId>mockwebserver</artifactId>
    <version>4.12.0</version>
    <scope>test</scope>
</dependency>

What does this code do?

• spring-boot-testcontainers and junit-jupiter integrate Testcontainers with Spring Boot and JUnit 5. Spring Boot's parent BOM manages the Testcontainers version automatically.
• The postgresql module provides the PostgreSQLContainer class that starts a real PostgreSQL 16 Docker container.
• mockwebserver needs an explicit version (4.12.0) because Spring Boot 3.4+ removed OkHttp from its dependency management.

• Save pom.xml.
• IntelliJ may show a notification to reload Maven changes. Click the reload icon in the Maven tool window or press Ctrl+Shift+O (Windows) or Cmd+Shift+I (macOS) to re-import.
• Verify the dependencies resolved by expanding External Libraries in the Project tool window. You should see entries for testcontainers, postgresql (under org.testcontainers), and mockwebserver.

Dependencies not resolving?

Make sure Docker Desktop is running. Testcontainers requires a running Docker daemon even at dependency resolution time for some IDE features. Also confirm each <dependency> block has <scope>test</scope> so they don't pollute your production classpath.

help me resolve Testcontainers or MockWebServer dependency issues in my Maven pom.xml for Spring Boot 3.5

Create the base integration test class

Every integration test in your project needs a real PostgreSQL instance. Instead of repeating the container setup in each test class, you'll create an abstract base class that all tests extend.

• Right-click the src/test/java/com/signalforge directory in IntelliJ's Project tool window.
• Select New, then Java Class.
• Name it AbstractIntegrationTest and paste the following code:

package com.signalforge;

import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.containers.PostgreSQLContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;

// Starts the full Spring application context on a random port
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
// Activates Testcontainers lifecycle management for JUnit 5
@Testcontainers
public abstract class AbstractIntegrationTest {

    // Starts a real PostgreSQL 16 container shared across all tests in a subclass
    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16-alpine");

    // Injects the container's dynamic connection details into Spring's datasource config
    @DynamicPropertySource
    static void configureProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.datasource.url", postgres::getJdbcUrl);
        registry.add("spring.datasource.username", postgres::getUsername);
        registry.add("spring.datasource.password", postgres::getPassword);
    }
}

What does this code do?

• @SpringBootTest(webEnvironment = RANDOM_PORT) boots your entire application on a random port. This lets you make real HTTP calls against your controllers.
• @Testcontainers tells JUnit 5 to manage the container lifecycle. The container starts before the first test and stops after the last test.
• @Container on the static field marks it as a managed container. Using postgres:16-alpine pins the exact PostgreSQL version to match your production Docker Compose setup.
• @DynamicPropertySource overrides your application.yml datasource properties at runtime. The container assigns a random port, so you cannot hardcode connection details.

Flyway will automatically run your V1__init_schema.sql migration against this fresh container, creating all four tables before the first test executes.

• Save AbstractIntegrationTest.java.
• You won't run this class directly. The next substep creates a test class that extends it.

Seeing import errors on Testcontainers classes?

Confirm that IntelliJ reloaded Maven after you added the dependencies. Open the Maven tool window (usually on the right sidebar) and click the Reload All Maven Projects button. Also verify the file is under src/test/java (green folder icon), not src/main/java.

help me fix Testcontainers import errors in my AbstractIntegrationTest class

Write the delivery integration tests

Now you'll write five integration tests that exercise your entire V1 pipeline end-to-end. Each test calls the REST API, then verifies the database state and HTTP delivery behavior.

• In the same src/test/java/com/signalforge directory, create a new Java class called DeliveryIntegrationTest.
• Paste the following code. This is the full test class. You'll walk through each test method afterward:

package com.signalforge;

import com.signalforge.entity.DeliveryAttempt;
import com.signalforge.entity.Endpoint;
import com.signalforge.entity.Event;
import com.signalforge.entity.Tenant;
import okhttp3.mockwebserver.MockResponse;
import okhttp3.mockwebserver.MockWebServer;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;

import javax.sql.DataSource;
import java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

// Extends the base class to get a real PostgreSQL container
class DeliveryIntegrationTest extends AbstractIntegrationTest {

    @Autowired
    private TestRestTemplate restTemplate;

    @Autowired
    private DataSource dataSource;

    // A mock HTTP server that simulates customer endpoints
    private MockWebServer mockWebServer;

    @BeforeEach
    void setUp() throws IOException {
        // Start a fresh MockWebServer for each test to avoid side effects
        mockWebServer = new MockWebServer();
        mockWebServer.start();
    }

    @AfterEach
    void tearDown() throws IOException {
        mockWebServer.shutdown();
    }

    @Test
    void happyPath_eventIsDeliveredSuccessfully() throws Exception {
        // Enqueue a 200 response so the delivery worker succeeds
        mockWebServer.enqueue(new MockResponse().setResponseCode(200).setBody("OK"));

        // Step 1: Create a tenant
        ResponseEntity<Map> tenantResponse = restTemplate.postForEntity(
                "/api/v1/tenants",
                Map.of("name", "Test Tenant"),
                Map.class
        );
        assertThat(tenantResponse.getStatusCode()).isEqualTo(HttpStatus.CREATED);
        String tenantId = tenantResponse.getBody().get("id").toString();

        // Step 2: Register an endpoint pointing to MockWebServer
        ResponseEntity<Map> endpointResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/endpoints",
                Map.of("url", mockWebServer.url("/webhook").toString()),
                Map.class
        );
        assertThat(endpointResponse.getStatusCode()).isEqualTo(HttpStatus.CREATED);

        // Step 3: Ingest an event
        ResponseEntity<Map> eventResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/events",
                Map.of("eventType", "order.created", "payload", Map.of("orderId", "123")),
                Map.class
        );
        assertThat(eventResponse.getStatusCode()).isEqualTo(HttpStatus.CREATED);

        // Step 4: Wait for the delivery worker to process (runs every 500ms)
        Thread.sleep(2000);

        // Step 5: Verify the delivery attempt reached DELIVERED status
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(
                     "SELECT status FROM delivery_attempts WHERE event_id = '" +
                     eventResponse.getBody().get("id") + "'")) {
            assertThat(rs.next()).isTrue();
            assertThat(rs.getString("status")).isEqualTo("DELIVERED");
        }

        // Verify the mock server received exactly one request
        assertThat(mockWebServer.getRequestCount()).isEqualTo(1);
    }

    @Test
    void endpointReturns500_deliveryIsMarkedFailed() throws Exception {
        // Enqueue a 500 Internal Server Error
        mockWebServer.enqueue(new MockResponse().setResponseCode(500).setBody("Server Error"));

        // Create tenant and endpoint
        String tenantId = createTenantAndEndpoint();

        // Ingest event
        ResponseEntity<Map> eventResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/events",
                Map.of("eventType", "order.failed", "payload", Map.of("orderId", "456")),
                Map.class
        );
        String eventId = eventResponse.getBody().get("id").toString();

        // Wait for the delivery worker to attempt delivery
        Thread.sleep(2000);

        // Verify status is FAILED and next_retry_at is set
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(
                     "SELECT status, next_retry_at FROM delivery_attempts WHERE event_id = '" +
                     eventId + "'")) {
            assertThat(rs.next()).isTrue();
            assertThat(rs.getString("status")).isEqualTo("FAILED");
            assertThat(rs.getTimestamp("next_retry_at")).isNotNull();
        }
    }

    @Test
    void endpointTimesOut_deliveryIsMarkedFailed() throws Exception {
        // Enqueue a response that takes longer than the 15-second read timeout
        mockWebServer.enqueue(new MockResponse()
                .setResponseCode(200)
                .setBody("Delayed")
                .setBodyDelay(20, TimeUnit.SECONDS));

        // Create tenant and endpoint
        String tenantId = createTenantAndEndpoint();

        // Ingest event
        ResponseEntity<Map> eventResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/events",
                Map.of("eventType", "order.timeout", "payload", Map.of("orderId", "789")),
                Map.class
        );
        String eventId = eventResponse.getBody().get("id").toString();

        // Wait for delivery attempt (timeout + worker interval)
        Thread.sleep(22000);

        // Verify status is FAILED and next_retry_at is set
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(
                     "SELECT status, next_retry_at FROM delivery_attempts WHERE event_id = '" +
                     eventId + "'")) {
            assertThat(rs.next()).isTrue();
            assertThat(rs.getString("status")).isEqualTo("FAILED");
            assertThat(rs.getTimestamp("next_retry_at")).isNotNull();
        }
    }

    @Test
    void duplicateIdempotencyKey_returns200AndNoDuplicateDeliveries() throws Exception {
        mockWebServer.enqueue(new MockResponse().setResponseCode(200).setBody("OK"));

        // Create tenant and endpoint
        String tenantId = createTenantAndEndpoint();
        String idempotencyKey = UUID.randomUUID().toString();

        // First ingestion
        HttpHeaders headers = new HttpHeaders();
        headers.set("Idempotency-Key", idempotencyKey);
        headers.set("Content-Type", "application/json");

        HttpEntity<Map<String, Object>> request = new HttpEntity<>(
                Map.of("eventType", "order.created", "payload", Map.of("orderId", "111")),
                headers
        );

        ResponseEntity<Map> firstResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/events", request, Map.class
        );
        assertThat(firstResponse.getStatusCode()).isEqualTo(HttpStatus.CREATED);

        // Second ingestion with the same idempotency key
        ResponseEntity<Map> secondResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/events", request, Map.class
        );
        // Should return 200 (not 201) to signal idempotent acceptance
        assertThat(secondResponse.getStatusCode()).isEqualTo(HttpStatus.OK);

        // Verify only one set of delivery attempts exists
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(
                     "SELECT COUNT(*) AS cnt FROM delivery_attempts da " +
                     "JOIN events e ON da.event_id = e.id " +
                     "WHERE e.tenant_id = '" + tenantId + "'")) {
            assertThat(rs.next()).isTrue();
            // Only one delivery attempt for the single event
            assertThat(rs.getInt("cnt")).isEqualTo(1);
        }
    }

    @Test
    void disabledEndpoint_noDeliveryAttemptCreated() throws Exception {
        // Create tenant
        ResponseEntity<Map> tenantResponse = restTemplate.postForEntity(
                "/api/v1/tenants",
                Map.of("name", "Disabled EP Tenant"),
                Map.class
        );
        String tenantId = tenantResponse.getBody().get("id").toString();

        // Register endpoint
        ResponseEntity<Map> endpointResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/endpoints",
                Map.of("url", mockWebServer.url("/webhook").toString()),
                Map.class
        );
        String endpointId = endpointResponse.getBody().get("id").toString();

        // Disable the endpoint via PATCH
        restTemplate.patchForObject(
                "/api/v1/tenants/" + tenantId + "/endpoints/" + endpointId,
                Map.of("status", "DISABLED"),
                Map.class
        );

        // Ingest an event
        ResponseEntity<Map> eventResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/events",
                Map.of("eventType", "order.created", "payload", Map.of("orderId", "999")),
                Map.class
        );
        String eventId = eventResponse.getBody().get("id").toString();

        // Wait briefly to ensure the worker had time to process
        Thread.sleep(2000);

        // Verify no delivery attempt was created for the disabled endpoint
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(
                     "SELECT COUNT(*) AS cnt FROM delivery_attempts WHERE event_id = '" +
                     eventId + "'")) {
            assertThat(rs.next()).isTrue();
            assertThat(rs.getInt("cnt")).isEqualTo(0);
        }

        // Verify MockWebServer received zero requests
        assertThat(mockWebServer.getRequestCount()).isEqualTo(0);
    }

    // Helper: creates a tenant and registers a MockWebServer endpoint
    private String createTenantAndEndpoint() {
        ResponseEntity<Map> tenantResponse = restTemplate.postForEntity(
                "/api/v1/tenants",
                Map.of("name", "Test Tenant " + UUID.randomUUID()),
                Map.class
        );
        String tenantId = tenantResponse.getBody().get("id").toString();

        restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/endpoints",
                Map.of("url", mockWebServer.url("/webhook").toString()),
                Map.class
        );

        return tenantId;
    }
}

What does this test class do?

Each test method exercises a different scenario from your V1 pipeline. Here is what each one verifies:

• happyPath: Creates a tenant, registers an endpoint, ingests an event, waits for the delivery worker, then asserts the delivery_attempts row reached DELIVERED status and the MockWebServer received exactly one request.
• endpointReturns500: Enqueues a 500 response. Verifies the delivery attempt is marked FAILED and next_retry_at is populated by your RetryPolicy.
• endpointTimesOut: Uses setBodyDelay(20, SECONDS) to exceed the 15-second read timeout. Verifies the same FAILED + next_retry_at behavior.
• duplicateIdempotencyKey: Sends the same event twice with the same Idempotency-Key header. The first call returns 201, the second returns 200. Only one set of delivery attempts exists in the database.
• disabledEndpoint: Creates an endpoint, disables it via PATCH, then ingests an event. Verifies zero delivery_attempts rows exist and MockWebServer received zero requests.

The createTenantAndEndpoint() helper method keeps the test code DRY. Each call creates a unique tenant name to avoid collisions between tests.

• Save DeliveryIntegrationTest.java.
• Make sure Docker Desktop is running.
• Run the tests by right-clicking DeliveryIntegrationTest in the Project tool window and selecting Run 'DeliveryIntegrationTest'.
• You should see all five tests pass with green checkmarks in the Run tool window. The first run takes 30-60 seconds because Testcontainers downloads the PostgreSQL Docker image.

Speed up subsequent test runs

After the first run, the PostgreSQL Docker image is cached locally. Future test runs start in about 5-10 seconds. You can also add testcontainers.reuse.enable=true to a ~/.testcontainers.properties file and .withReuse(true) on the container to reuse containers across runs during development.

Tests failing or container not starting?

• Docker Desktop must be running. Testcontainers communicates with Docker to start the PostgreSQL container. Verify by running docker info in your terminal.
• If the timeout test takes too long, increase the Thread.sleep to 30000 (30 seconds) to give enough buffer for the read timeout plus worker interval.
• If assertions on delivery status fail, check that your DeliveryWorker's @Scheduled(fixedDelay = 500) annotation is present and the worker is being picked up by component scanning.

help me debug why my integration tests are failing in the SignalForge project

✔️ Awesome, I've got everything!

Great. Double check you have saved both AbstractIntegrationTest.java and DeliveryIntegrationTest.java.

ⓧ I'd like to double check the full code

Here is the complete code for both test files.

package com.signalforge;

import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.containers.PostgreSQLContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;

// Starts the full Spring application context on a random port
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
// Activates Testcontainers lifecycle management for JUnit 5
@Testcontainers
public abstract class AbstractIntegrationTest {

    // Starts a real PostgreSQL 16 container shared across all tests in a subclass
    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16-alpine");

    // Injects the container's dynamic connection details into Spring's datasource config
    @DynamicPropertySource
    static void configureProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.datasource.url", postgres::getJdbcUrl);
        registry.add("spring.datasource.username", postgres::getUsername);
        registry.add("spring.datasource.password", postgres::getPassword);
    }
}

package com.signalforge;

import com.signalforge.entity.DeliveryAttempt;
import com.signalforge.entity.Endpoint;
import com.signalforge.entity.Event;
import com.signalforge.entity.Tenant;
import okhttp3.mockwebserver.MockResponse;
import okhttp3.mockwebserver.MockWebServer;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;

import javax.sql.DataSource;
import java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

// Extends the base class to get a real PostgreSQL container
class DeliveryIntegrationTest extends AbstractIntegrationTest {

    @Autowired
    private TestRestTemplate restTemplate;

    @Autowired
    private DataSource dataSource;

    // A mock HTTP server that simulates customer endpoints
    private MockWebServer mockWebServer;

    @BeforeEach
    void setUp() throws IOException {
        // Start a fresh MockWebServer for each test to avoid side effects
        mockWebServer = new MockWebServer();
        mockWebServer.start();
    }

    @AfterEach
    void tearDown() throws IOException {
        mockWebServer.shutdown();
    }

    @Test
    void happyPath_eventIsDeliveredSuccessfully() throws Exception {
        // Enqueue a 200 response so the delivery worker succeeds
        mockWebServer.enqueue(new MockResponse().setResponseCode(200).setBody("OK"));

        // Step 1: Create a tenant
        ResponseEntity<Map> tenantResponse = restTemplate.postForEntity(
                "/api/v1/tenants",
                Map.of("name", "Test Tenant"),
                Map.class
        );
        assertThat(tenantResponse.getStatusCode()).isEqualTo(HttpStatus.CREATED);
        String tenantId = tenantResponse.getBody().get("id").toString();

        // Step 2: Register an endpoint pointing to MockWebServer
        ResponseEntity<Map> endpointResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/endpoints",
                Map.of("url", mockWebServer.url("/webhook").toString()),
                Map.class
        );
        assertThat(endpointResponse.getStatusCode()).isEqualTo(HttpStatus.CREATED);

        // Step 3: Ingest an event
        ResponseEntity<Map> eventResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/events",
                Map.of("eventType", "order.created", "payload", Map.of("orderId", "123")),
                Map.class
        );
        assertThat(eventResponse.getStatusCode()).isEqualTo(HttpStatus.CREATED);

        // Step 4: Wait for the delivery worker to process (runs every 500ms)
        Thread.sleep(2000);

        // Step 5: Verify the delivery attempt reached DELIVERED status
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(
                     "SELECT status FROM delivery_attempts WHERE event_id = '" +
                     eventResponse.getBody().get("id") + "'")) {
            assertThat(rs.next()).isTrue();
            assertThat(rs.getString("status")).isEqualTo("DELIVERED");
        }

        // Verify the mock server received exactly one request
        assertThat(mockWebServer.getRequestCount()).isEqualTo(1);
    }

    @Test
    void endpointReturns500_deliveryIsMarkedFailed() throws Exception {
        // Enqueue a 500 Internal Server Error
        mockWebServer.enqueue(new MockResponse().setResponseCode(500).setBody("Server Error"));

        // Create tenant and endpoint
        String tenantId = createTenantAndEndpoint();

        // Ingest event
        ResponseEntity<Map> eventResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/events",
                Map.of("eventType", "order.failed", "payload", Map.of("orderId", "456")),
                Map.class
        );
        String eventId = eventResponse.getBody().get("id").toString();

        // Wait for the delivery worker to attempt delivery
        Thread.sleep(2000);

        // Verify status is FAILED and next_retry_at is set
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(
                     "SELECT status, next_retry_at FROM delivery_attempts WHERE event_id = '" +
                     eventId + "'")) {
            assertThat(rs.next()).isTrue();
            assertThat(rs.getString("status")).isEqualTo("FAILED");
            assertThat(rs.getTimestamp("next_retry_at")).isNotNull();
        }
    }

    @Test
    void endpointTimesOut_deliveryIsMarkedFailed() throws Exception {
        // Enqueue a response that takes longer than the 15-second read timeout
        mockWebServer.enqueue(new MockResponse()
                .setResponseCode(200)
                .setBody("Delayed")
                .setBodyDelay(20, TimeUnit.SECONDS));

        // Create tenant and endpoint
        String tenantId = createTenantAndEndpoint();

        // Ingest event
        ResponseEntity<Map> eventResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/events",
                Map.of("eventType", "order.timeout", "payload", Map.of("orderId", "789")),
                Map.class
        );
        String eventId = eventResponse.getBody().get("id").toString();

        // Wait for delivery attempt (timeout + worker interval)
        Thread.sleep(22000);

        // Verify status is FAILED and next_retry_at is set
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(
                     "SELECT status, next_retry_at FROM delivery_attempts WHERE event_id = '" +
                     eventId + "'")) {
            assertThat(rs.next()).isTrue();
            assertThat(rs.getString("status")).isEqualTo("FAILED");
            assertThat(rs.getTimestamp("next_retry_at")).isNotNull();
        }
    }

    @Test
    void duplicateIdempotencyKey_returns200AndNoDuplicateDeliveries() throws Exception {
        mockWebServer.enqueue(new MockResponse().setResponseCode(200).setBody("OK"));

        // Create tenant and endpoint
        String tenantId = createTenantAndEndpoint();
        String idempotencyKey = UUID.randomUUID().toString();

        // First ingestion
        HttpHeaders headers = new HttpHeaders();
        headers.set("Idempotency-Key", idempotencyKey);
        headers.set("Content-Type", "application/json");

        HttpEntity<Map<String, Object>> request = new HttpEntity<>(
                Map.of("eventType", "order.created", "payload", Map.of("orderId", "111")),
                headers
        );

        ResponseEntity<Map> firstResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/events", request, Map.class
        );
        assertThat(firstResponse.getStatusCode()).isEqualTo(HttpStatus.CREATED);

        // Second ingestion with the same idempotency key
        ResponseEntity<Map> secondResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/events", request, Map.class
        );
        // Should return 200 (not 201) to signal idempotent acceptance
        assertThat(secondResponse.getStatusCode()).isEqualTo(HttpStatus.OK);

        // Verify only one set of delivery attempts exists
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(
                     "SELECT COUNT(*) AS cnt FROM delivery_attempts da " +
                     "JOIN events e ON da.event_id = e.id " +
                     "WHERE e.tenant_id = '" + tenantId + "'")) {
            assertThat(rs.next()).isTrue();
            // Only one delivery attempt for the single event
            assertThat(rs.getInt("cnt")).isEqualTo(1);
        }
    }

    @Test
    void disabledEndpoint_noDeliveryAttemptCreated() throws Exception {
        // Create tenant
        ResponseEntity<Map> tenantResponse = restTemplate.postForEntity(
                "/api/v1/tenants",
                Map.of("name", "Disabled EP Tenant"),
                Map.class
        );
        String tenantId = tenantResponse.getBody().get("id").toString();

        // Register endpoint
        ResponseEntity<Map> endpointResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/endpoints",
                Map.of("url", mockWebServer.url("/webhook").toString()),
                Map.class
        );
        String endpointId = endpointResponse.getBody().get("id").toString();

        // Disable the endpoint via PATCH
        restTemplate.patchForObject(
                "/api/v1/tenants/" + tenantId + "/endpoints/" + endpointId,
                Map.of("status", "DISABLED"),
                Map.class
        );

        // Ingest an event
        ResponseEntity<Map> eventResponse = restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/events",
                Map.of("eventType", "order.created", "payload", Map.of("orderId", "999")),
                Map.class
        );
        String eventId = eventResponse.getBody().get("id").toString();

        // Wait briefly to ensure the worker had time to process
        Thread.sleep(2000);

        // Verify no delivery attempt was created for the disabled endpoint
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(
                     "SELECT COUNT(*) AS cnt FROM delivery_attempts WHERE event_id = '" +
                     eventId + "'")) {
            assertThat(rs.next()).isTrue();
            assertThat(rs.getInt("cnt")).isEqualTo(0);
        }

        // Verify MockWebServer received zero requests
        assertThat(mockWebServer.getRequestCount()).isEqualTo(0);
    }

    // Helper: creates a tenant and registers a MockWebServer endpoint
    private String createTenantAndEndpoint() {
        ResponseEntity<Map> tenantResponse = restTemplate.postForEntity(
                "/api/v1/tenants",
                Map.of("name", "Test Tenant " + UUID.randomUUID()),
                Map.class
        );
        String tenantId = tenantResponse.getBody().get("id").toString();

        restTemplate.postForEntity(
                "/api/v1/tenants/" + tenantId + "/endpoints",
                Map.of("url", mockWebServer.url("/webhook").toString()),
                Map.class
        );

        return tenantId;
    }
}

You now have a comprehensive integration test suite that proves your V1 pipeline works end-to-end. Next up, you'll unlock Java 21 Virtual Threads and watch your delivery throughput explode to 10,000+ concurrent deliveries.

Enable Java 21 Virtual Threads

Your V1 pipeline is complete and battle-tested. You have integration tests proving that ingestion, delivery, retries, and dead-lettering all work correctly. There is one bottleneck hiding in the design, though.

Every delivery attempt makes an outbound HTTP POST that blocks a platform thread for 100ms to 30 seconds. With a default pool of roughly 200 platform threads, your maximum concurrency is 200 simultaneous deliveries. Virtual Threads change this entirely. A single property in Spring Boot unlocks 10,000+ concurrent deliveries with the same blocking-style code you already wrote.

In this step, get ready to:

• Enable Virtual Threads in application.yml.
• Audit and fix synchronized blocks to prevent thread pinning.
• Run a load test with 1,000 events and measure delivery throughput.

Add the Virtual Threads property to application.yml

Each Virtual Thread uses roughly 1KB of memory compared to 1MB for a platform thread. When a Virtual Thread blocks on I/O, the JVM unmounts it from its carrier OS thread so another Virtual Thread can run. Your existing blocking code stays the same.

• In IntelliJ IDEA, open src/main/resources/application.yml.
• Find the existing spring: block. At the end of the block (after the flyway section), add the following threads configuration:

  threads:
    virtual:
      enabled: true

What does this property do?

Setting spring.threads.virtual.enabled: true tells Spring Boot to run every HTTP request handler and @Scheduled task on a Virtual Thread instead of a platform thread.

Your DeliveryWorker makes outbound HTTP calls that block for up to 15 seconds each. Without Virtual Threads, each blocked call ties up a platform thread. With Virtual Threads, the JVM unmounts the blocked thread and reuses the carrier for other work.

• Save application.yml.

✔️ Awesome, I've got everything!

Double check you have saved application.yml before moving on.

ⓧ I'd like to double check the full code

Here is the complete application.yml at this point:

spring:
  datasource:
    url: jdbc:postgresql://localhost:5432/signalforge
    username: signalforge
    password: signalforge
  jpa:
    hibernate:
      ddl-auto: validate
    open-in-view: false
  flyway:
    enabled: true
    locations: classpath:db/migration
  threads:
    virtual:
      enabled: true

Audit for synchronized blocks and configure pinning detection

Virtual Threads have one important gotcha on Java 21. When a Virtual Thread enters a synchronized block, it gets pinned to its carrier OS thread. The carrier cannot run any other Virtual Threads until the synchronized block exits.

If the code inside that block performs a blocking operation (like an HTTP call or a database query), the carrier thread is held captive. This defeats the purpose of Virtual Threads entirely.

What is thread pinning?

Pinning happens when a Virtual Thread cannot unmount from its carrier thread. The most common cause on Java 21 is synchronized blocks that contain blocking I/O. The fix is to replace synchronized with ReentrantLock, which is scheduler-aware. When a Virtual Thread calls lock() on a ReentrantLock, it can unmount while waiting.

Java 24 resolves this limitation via JEP 491. Since this project uses Java 21, the ReentrantLock approach is necessary.

• Search your project for synchronized by pressing Cmd+Shift+F (macOS) or Ctrl+Shift+F (Windows) to open Find in Files.
• Type synchronized and review every result.

For each synchronized block that contains blocking I/O (HTTP calls, database queries, file reads), replace it with a ReentrantLock. Here is the pattern:

Before (pinning-prone):

// synchronized pins the Virtual Thread to its carrier
synchronized (this) {
    doBlockingWork();
}

After (Virtual Thread-safe):

import java.util.concurrent.locks.ReentrantLock;

// Declare the lock as a field
private final ReentrantLock lock = new ReentrantLock();

// ReentrantLock allows the Virtual Thread to unmount while waiting
lock.lock();
try {
    doBlockingWork();
} finally {
    lock.unlock();
}

Why the try/finally pattern?

The finally block guarantees the lock is released even if doBlockingWork() throws an exception. Without it, a failed call would leave the lock held forever, blocking all other threads.

Short-lived synchronized blocks are safe

If a synchronized block only guards a quick, non-blocking operation (like reading a field or checking a size), pinning lasts microseconds and has no practical impact. Focus your refactoring on blocks that contain I/O or long-running work.

Next, configure the JVM to detect any pinning you might have missed. This flag prints a stack trace whenever a Virtual Thread is pinned.

• In IntelliJ IDEA, click Run in the top menu bar.
• Click Edit Configurations....
• Select your application's run configuration in the left panel (e.g. SignalforgeEngineApplication).
• Click Modify options, then select Add VM Options.
• In the VM options field that appears, enter:

-Djdk.tracePinnedThreads=full

• Click Apply, then OK.
• Run the application. If you see stack traces mentioning VirtualThread pinning in the console, revisit those code paths and apply the ReentrantLock pattern above.

What does this JVM flag do?

The -Djdk.tracePinnedThreads=full flag tells the JVM to print a full stack trace whenever a Virtual Thread is pinned to its carrier thread. This makes it easy to find the exact synchronized block causing the problem.

Seeing pinning stack traces?

Look at the stack trace for any synchronized block in your own code. Replace it with ReentrantLock using the pattern above. Pinning from third-party libraries (like JDBC drivers) is expected and handled by HikariCP's connection pool.

Why am I still seeing Virtual Thread pinning after replacing synchronized?

Run a load test and measure throughput

Your delivery engine now runs on Virtual Threads with pinning detection active. Time to put it under load and see how it performs.

You will create a simple bash script that creates a test tenant, registers 10 endpoints, and fires 1,000 events through the API. You can run this in Git Bash (included with Git on Windows) or any terminal on macOS/Linux.

• Make sure your application is running in IntelliJ and PostgreSQL is up via Docker Compose.
• Create a file called load-test.sh in your project root by running this command:

touch load-test.sh

• Open load-test.sh in IntelliJ and paste the following script:

#!/bin/bash
# Load test for SignalForge delivery engine
# Creates a tenant, 10 endpoints, and ingests 1,000 events

BASE_URL="http://localhost:8080/api/v1"

# Step 1: Create a test tenant
echo "Creating test tenant..."
TENANT_RESPONSE=$(curl -s -X POST "$BASE_URL/tenants" \
  -H "Content-Type: application/json" \
  -d '{"name": "load-test-tenant"}')

# Extract tenant ID from the response
TENANT_ID=$(echo "$TENANT_RESPONSE" | grep -o '"id":"[^"]*"' | head -1 | cut -d'"' -f4)
echo "Tenant ID: $TENANT_ID"

# Step 2: Register 10 endpoints (using httpbin as a test receiver)
echo "Registering 10 endpoints..."
for i in $(seq 1 10); do
  curl -s -X POST "$BASE_URL/tenants/$TENANT_ID/endpoints" \
    -H "Content-Type: application/json" \
    -d "{\"url\": \"https://httpbin.org/post\"}" > /dev/null
done
echo "Endpoints registered."

# Step 3: Ingest 1,000 events in batches of 50 concurrent requests
echo "Ingesting 1,000 events..."
START_TIME=$(date +%s)

for i in $(seq 1 1000); do
  curl -s -X POST "$BASE_URL/tenants/$TENANT_ID/events" \
    -H "Content-Type: application/json" \
    -d "{\"eventType\": \"load.test\", \"payload\": {\"index\": $i}}" > /dev/null &

  # Wait every 50 requests to avoid overwhelming the local machine
  if (( i % 50 == 0 )); then
    wait
    echo "  Sent $i / 1000 events"
  fi
done

wait
END_TIME=$(date +%s)
DURATION=$((END_TIME - START_TIME))

echo "All 1,000 events ingested in ${DURATION} seconds."
echo "Check /actuator/metrics for delivery throughput."

What does this script do?

• Creates a single test tenant via the REST API.
• Registers 10 endpoints pointing to httpbin.org/post (a free HTTP echo service that accepts POST requests).
• Fires 1,000 events in batches of 50 concurrent curl requests.
• Each event fans out to all 10 endpoints, creating 10,000 total delivery attempts.

• Make the script executable and run it:

chmod +x load-test.sh
./load-test.sh

• Watch the IntelliJ console as the delivery worker processes the 10,000 delivery attempts. You should see structured log output for each attempt with event_id, endpoint_id, attempt_number, and http_status.
• After the script completes, check the ingestion time printed in the terminal. Then check the delivery throughput by hitting the Actuator metrics endpoint:

curl -s http://localhost:8080/actuator/metrics | grep -i delivery

The key metric to note is how quickly the worker drains the queue. With Virtual Threads, the worker can have hundreds of HTTP calls in flight simultaneously instead of being capped at the platform thread pool size.

Why is the throughput improvement dramatic at scale?

Without Virtual Threads, each delivery attempt ties up a platform thread for the full HTTP round-trip (100ms to 15 seconds). A pool of 200 threads means at most 200 concurrent deliveries. With Virtual Threads, blocked threads unmount instantly, letting the JVM run thousands of deliveries on just a handful of carrier threads.

In a local test, the bottleneck shifts to the network and PostgreSQL connection pool rather than thread count. You will see the full impact when you configure HikariCP and add Redis in the next step.

Script not running on Windows?

If you see bash: command not found, open Git Bash instead of PowerShell. Git Bash is included with Git for Windows and supports bash scripts.

Help me run the load test script on Windows

Your delivery engine is now powered by Virtual Threads, processing thousands of concurrent HTTP deliveries without thread pool exhaustion. Next up, you will add Redis for per-endpoint rate limiting and delivery deduplication.

Add Redis for Rate Limiting and Deduplication

Your delivery worker is running on Virtual Threads, ready to handle thousands of concurrent deliveries.

But raw concurrency creates new risks. What happens when your worker hammers a customer endpoint with thousands of requests per second? And what happens when two worker instances both claim the same delivery before SKIP LOCKED prevents it? In this step, you will add Redis for two critical guardrails: a per-endpoint rate limiter and a delivery deduplication service.

In this step, get ready to:

• Add Redis to your Docker Compose stack and build a Token Bucket rate limiter.
• Create a Redis-based deduplication service and wire both services into the delivery worker.
• Configure HikariCP for safe Virtual Thread concurrency.

Add Redis to Docker Compose and build the rate limiter

A Token Bucket rate limiter gives each endpoint a bucket of tokens that refills over time. Every delivery consumes one token. If the bucket is empty, the delivery gets rescheduled instead of hammering the endpoint.

You need a shared data store for the token counts so that all worker instances see the same state. Redis is ideal because its Lua scripting lets you read, check, and update tokens in a single atomic operation.

• Open docker-compose.yml in IntelliJ and add the Redis service below your postgres service. Copy the full updated file below:

version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: signalforge
      POSTGRES_USER: signalforge
      POSTGRES_PASSWORD: signalforge
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

  # In-memory data store for rate limiting and deduplication
  redis:
    image: redis:7
    ports:
      - '6379:6379'

volumes:
  postgres_data:

What does this add?

The new redis service pulls the official redis:7 image and maps port 6379 to your local machine. Redis runs entirely in memory, so it does not need a volume for persistent storage.

• Save docker-compose.yml.
• Start the updated stack by running this command in your terminal from the project root:

docker compose up -d

• Verify both containers are running by running:

docker compose ps

• You should see both postgres and redis listed with a status of Up.

Redis container not starting?

Make sure Docker Desktop is running. If port 6379 is already in use, check for another Redis instance or process on that port.

Help me debug my Redis Docker container

Now connect your Spring Boot project to Redis.

• Open pom.xml and add the Spring Data Redis starter inside the <dependencies> block:

<!-- Redis for rate limiting and delivery deduplication -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

Why no version tag?

The Spring Boot parent BOM manages the version for spring-boot-starter-data-redis automatically. This keeps all Spring Data versions aligned and avoids compatibility issues.

• Open src/main/resources/application.yml and add the Redis connection settings. Place these under the spring: block, after the existing threads: section:

  data:
    redis:
      host: localhost
      port: 6379

What does this configure?

Spring Boot auto-configures a StringRedisTemplate bean when it detects the Redis starter on the classpath. These two properties tell it where to connect. The StringRedisTemplate is what your rate limiter and deduplication services will use to read and write Redis keys.

• Save both pom.xml and application.yml.
• In IntelliJ, click the Maven reload button (the circular arrows icon in the Maven tool window) to download the new dependency.
• Run the application. In the console output, you should see a line containing Started SignalforgeEngineApplication with no Redis connection errors.

Seeing a Redis connection refused error?

Make sure the Redis container is still running with docker compose ps. Confirm the port in application.yml matches 6379. Also check that the data: key is nested under spring: (YAML indentation matters).

Help me fix my Redis connection error in Spring Boot

With Redis running and Spring Boot connected, you can now build the rate limiter. The key design decision is atomicity. If you read the token count in one Redis command and decrement it in another, two workers could both read "1 token left" and both proceed. A Lua script runs atomically inside Redis, preventing this race.

• Create a new file src/main/resources/scripts/token_bucket.lua and paste the following Lua script:

-- Token Bucket rate limiter: atomic check-and-decrement
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local refill_interval = tonumber(ARGV[3])
local now = tonumber(ARGV[4])

-- Fetch current bucket state from Redis hash
local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1])
local last_refill = tonumber(bucket[2])

-- Initialize on first request
if tokens == nil then
    tokens = capacity
    last_refill = now
end

-- Refill tokens based on elapsed time
local time_passed = now - last_refill
local refills = math.floor(time_passed / refill_interval)
if refills > 0 then
    tokens = math.min(capacity, tokens + (refills * refill_rate))
    last_refill = last_refill + (refills * refill_interval)
end

-- Try to consume one token
local allowed = 0
if tokens >= 1 then
    tokens = tokens - 1
    allowed = 1
end

-- Save updated state back to Redis
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', last_refill)
redis.call('EXPIRE', key, refill_interval * 2)

return allowed

What does this Lua script do?

This script runs atomically inside Redis. No other command can interrupt it mid-execution.

• It stores each endpoint's token count and last refill time in a Redis hash.
• On each call, it calculates how many tokens to refill based on elapsed time, capping at the bucket capacity.
• If at least one token is available, it decrements and returns 1 (allowed). Otherwise it returns 0 (denied).
• The EXPIRE call sets a TTL so idle endpoint keys clean themselves up automatically.

• Save token_bucket.lua.

Now create the Java service that loads and executes this script.

• Create a new file src/main/java/com/signalforge/service/RateLimiter.java and paste the following code:

package com.signalforge.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;

import java.time.Instant;
import java.util.List;
import java.util.UUID;

@Service
public class RateLimiter {

    private static final Logger log = LoggerFactory.getLogger(RateLimiter.class);

    // Maximum deliveries per endpoint before throttling
    private static final int CAPACITY = 10;

    // Tokens added per refill interval
    private static final int REFILL_RATE = 10;

    // Refill interval in seconds
    private static final int REFILL_INTERVAL = 1;

    private final StringRedisTemplate redisTemplate;
    private final DefaultRedisScript<Long> tokenBucketScript;

    public RateLimiter(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;

        // Load the Lua script once; Spring caches the SHA1 for EVALSHA
        this.tokenBucketScript = new DefaultRedisScript<>();
        this.tokenBucketScript.setLocation(new ClassPathResource("scripts/token_bucket.lua"));
        this.tokenBucketScript.setResultType(Long.class);
    }

    /**
     * Try to acquire a delivery token for the given endpoint.
     * Returns true if delivery is permitted, false if rate limited.
     */
    public boolean tryAcquire(UUID endpointId) {
        String key = "ratelimit:" + endpointId;
        long now = Instant.now().getEpochSecond();

        Long result = redisTemplate.execute(
                tokenBucketScript,
                List.of(key),
                String.valueOf(CAPACITY),
                String.valueOf(REFILL_RATE),
                String.valueOf(REFILL_INTERVAL),
                String.valueOf(now)
        );

        boolean acquired = result != null && result == 1L;

        if (!acquired) {
            log.info("Rate limit exceeded for endpoint={}", endpointId);
        }

        return acquired;
    }
}

What does this code do?

• DefaultRedisScript loads the Lua file from the classpath once. Spring automatically computes the script's SHA1 and uses EVALSHA for subsequent calls, so the full script is not sent to Redis every time.
• tryAcquire() builds the Redis key as ratelimit:{endpointId} and passes the bucket configuration (capacity of 10, refill rate of 10 per second) plus the current epoch time.
• The result is 1 (allowed) or 0 (denied). If denied, the worker will reschedule the delivery attempt instead of dropping it.

• Save RateLimiter.java.
• Rebuild the project in IntelliJ (click Build then Rebuild Project). You should see BUILD SUCCESSFUL with no compilation errors.

Compilation error on StringRedisTemplate?

Make sure you reloaded Maven after adding the spring-boot-starter-data-redis dependency. The StringRedisTemplate class lives in that starter.

Help me fix my RateLimiter compilation error

Why Token Bucket instead of Fixed Window?

A fixed window counter resets at the boundary (e.g., every minute), so a client can fire the entire quota at 12:00:59 and again at 12:01:00. Token Bucket avoids this boundary problem by refilling tokens smoothly over time.

Token Bucket also allows controlled bursts. If an endpoint has been idle, its bucket is full, so a short burst of deliveries proceeds immediately. This matches real-world webhook traffic patterns well.

Build the deduplication service and update the delivery worker

Even with SKIP LOCKED, edge cases can cause duplicate deliveries. If a worker claims a row, delivers successfully, but crashes before marking it DELIVERED, a recovery process will re-queue that row and another worker will deliver it again.

The fix is a Redis-based deduplication check. Before delivering, the worker writes a key dedup:{endpointId}:{eventId} to Redis with a 24-hour TTL. If the key already exists, the event was already delivered and the worker skips it.

• Create a new file src/main/java/com/signalforge/service/DeduplicationService.java and paste the following code:

package com.signalforge.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.UUID;

@Service
public class DeduplicationService {

    private static final Logger log = LoggerFactory.getLogger(DeduplicationService.class);

    // Deduplication window: 24 hours
    private static final Duration DEDUP_TTL = Duration.ofHours(24);

    private final StringRedisTemplate redisTemplate;

    public DeduplicationService(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     * Check if this endpoint+event combination was already delivered.
     * Returns true if this is a duplicate (skip delivery).
     */
    public boolean isDuplicate(UUID endpointId, UUID eventId) {
        String key = "dedup:" + endpointId + ":" + eventId;

        // SET NX with TTL: returns true if the key was newly set (not a duplicate)
        Boolean wasSet = redisTemplate.opsForValue()
                .setIfAbsent(key, "1", DEDUP_TTL);

        boolean duplicate = wasSet == null || !wasSet;

        if (duplicate) {
            log.info("Duplicate delivery detected: endpoint={}, event={}",
                    endpointId, eventId);
        }

        return duplicate;
    }
}

What does this code do?

• setIfAbsent() maps to Redis SET NX (set if not exists). This is atomic. If two workers race on the same event, only one will succeed in setting the key.
• The 24-hour TTL ensures keys clean themselves up. After 24 hours, the same event could theoretically be re-delivered, but by then any retry window has long expired.
• If setIfAbsent() returns false, the key already existed, meaning another worker already delivered this event. The method returns true (it is a duplicate) so the caller can skip delivery.

• Save DeduplicationService.java.

Both services are ready. Now wire them into the delivery worker so it checks the rate limiter and deduplication before every HTTP delivery.

• Open src/main/java/com/signalforge/worker/DeliveryWorker.java.
• Add the two new fields and update the constructor to inject them. Add these alongside your existing fields:

// Rate limiter prevents overwhelming customer endpoints
private final RateLimiter rateLimiter;

// Deduplication prevents delivering the same event twice
private final DeduplicationService deduplicationService;

• Add RateLimiter rateLimiter and DeduplicationService deduplicationService as parameters in your constructor and assign them to the fields.
• Add the necessary imports at the top of the file:

import com.signalforge.service.RateLimiter;
import com.signalforge.service.DeduplicationService;

Now add the guard checks inside the delivery loop. These go before the HTTP POST call to the customer endpoint.

• In your delivery loop (where you iterate over claimed attempts), add the following checks before the HTTP delivery call:

// Check for duplicate delivery via Redis
if (deduplicationService.isDuplicate(
        attempt.getEndpoint().getId(),
        attempt.getEvent().getId())) {
    log.info("Skipping duplicate delivery: attempt={}", attempt.getId());
    attempt.setStatus(DeliveryStatus.DELIVERED);
    deliveryAttemptRepository.save(attempt);
    continue;
}

// Check rate limit for this endpoint
if (!rateLimiter.tryAcquire(attempt.getEndpoint().getId())) {
    log.info("Rate limited: rescheduling attempt={}", attempt.getId());
    attempt.setStatus(DeliveryStatus.FAILED);
    attempt.setNextRetryAt(Instant.now().plusSeconds(5));
    deliveryAttemptRepository.save(attempt);
    continue;
}

What does this code do?

• The deduplication check runs first. If the event was already delivered to this endpoint, the attempt is marked DELIVERED and skipped. No HTTP call is made.
• The rate limiter check runs second. If the endpoint's token bucket is empty, the attempt is re-scheduled with a short 5-second delay. The worker moves on to the next attempt instead of waiting.
• Both guards use continue to skip to the next attempt in the loop. The HTTP delivery code below these checks only runs for non-duplicate, rate-permitted attempts.

• Save DeliveryWorker.java.
• Rebuild the project. You should see no compilation errors.
• Run the application and ingest a test event via your API. In the console logs, you should see the delivery attempt proceed without any "Rate limit exceeded" or "Duplicate delivery detected" messages.

Seeing a constructor injection error?

Make sure both RateLimiter and DeduplicationService have the @Service annotation. Spring needs that annotation to register them as beans for injection into DeliveryWorker.

Help me fix my Spring dependency injection error in DeliveryWorker

Configure HikariCP for Virtual Thread concurrency

With Virtual Threads enabled, your application can run thousands of concurrent tasks. Each task that touches the database needs a connection from the HikariCP connection pool.

The default pool size is 10. That is too small for thousands of Virtual Threads. You need to size the pool based on what your database can handle, not how many threads you have. A setting of 30 is a solid starting point for a single PostgreSQL instance.

• In application.yml, find the datasource: section. It currently looks like this:

  datasource:
    url: jdbc:postgresql://localhost:5432/signalforge
    username: signalforge
    password: signalforge

• Add the hikari: block below password: signalforge, so the full datasource section looks like this:

  datasource:
    url: jdbc:postgresql://localhost:5432/signalforge
    username: signalforge
    password: signalforge
    hikari:
      maximum-pool-size: 30

Why 30 connections instead of matching the thread count?

With platform threads, you might set the pool size equal to your thread count. With Virtual Threads, thousands of virtual threads share a small number of carrier OS threads.

HikariCP works correctly here because its internal Semaphore.acquire() call unmounts the blocked Virtual Thread from its carrier. The Virtual Thread waits without holding an OS thread. Setting maximum-pool-size: 30 keeps database load manageable while letting thousands of Virtual Threads queue for connections efficiently.

• Save application.yml.
• Restart the application. In the logs, look for the HikariCP initialization line. It should show maximum pool size: 30.

Pool size still showing 10?

Check your YAML indentation. The hikari: key must be nested directly under datasource: at the same level as url:. If it is indented incorrectly, Spring ignores it and falls back to the default of 10.

Help me fix my HikariCP pool size configuration

✔️ Awesome, I've got everything!

Double check you have saved all your files.

ⓧ I'd like to double check the full application.yml

Here is the complete application.yml with all changes from this step:

spring:
  datasource:
    url: jdbc:postgresql://localhost:5432/signalforge
    username: signalforge
    password: signalforge
    hikari:
      maximum-pool-size: 30
  jpa:
    hibernate:
      ddl-auto: validate
    open-in-view: false
  flyway:
    enabled: true
    locations: classpath:db/migration
  threads:
    virtual:
      enabled: true
  data:
    redis:
      host: localhost
      port: 6379

Your delivery pipeline now has two production-grade safety nets. The rate limiter protects customer endpoints from being overwhelmed. The deduplication service prevents the same event from being delivered twice. Next up, you will add per-endpoint circuit breakers so that a single degraded endpoint does not consume all your delivery capacity.

Add Resilience4j Circuit Breakers Per Endpoint

Your delivery worker now has rate limiting and deduplication in place. There is one remaining vulnerability. A single degraded customer endpoint can consume all of your delivery capacity.

If one endpoint is returning 500 errors or timing out on every request, the worker wastes cycles retrying it instead of delivering to healthy endpoints. Circuit breakers solve this by detecting degraded endpoints and temporarily stopping deliveries to them. In this step, you will add per-endpoint circuit breakers using Resilience4j and expose their health state through the API.

In this step, get ready to:

• Create a circuit breaker registry that gives each endpoint its own named circuit breaker.
• Wire the circuit breaker into the delivery worker so it skips degraded endpoints.
• Expose a health endpoint that returns each endpoint's circuit breaker state.

Add the Resilience4j dependency and create a circuit breaker registry

Every endpoint your system delivers to has different reliability characteristics. Some respond in milliseconds. Others go down for hours. A per-endpoint circuit breaker tracks each endpoint's health individually and trips open when failures cross a threshold.

• Add the Resilience4j Spring Boot 3 dependency to your pom.xml by adding this inside the <dependencies> block:

<dependency>
    <groupId>io.github.resilience4j</groupId>
    <artifactId>resilience4j-spring-boot3</artifactId>
    <version>2.4.0</version>
</dependency>

What is Resilience4j?

Resilience4j is a lightweight fault-tolerance library for Java. It provides circuit breakers, rate limiters, retries, and bulkheads as decorators you can wrap around any function call. It replaced the now-deprecated Netflix Hystrix as the standard for Spring Boot applications.

• In IntelliJ, click the Maven reload button (the refresh icon in the Maven tool window) or press Ctrl+Shift+O to reimport dependencies.

Now you will create the registry that manages one circuit breaker per endpoint.

• Create a new package com.signalforge.resilience by right-clicking the com.signalforge package in IntelliJ's Project sidebar, selecting New, then Package.
• Create a new file CircuitBreakerRegistry.java inside the resilience package and paste the following code:

package com.signalforge.resilience;

import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Manages one Resilience4j CircuitBreaker per endpoint.
 * Each endpoint gets its own breaker so a single degraded
 * endpoint cannot consume all delivery capacity.
 */
@Component
public class CircuitBreakerRegistry {

    // Thread-safe map: endpointId -> CircuitBreaker
    private final ConcurrentHashMap<String, CircuitBreaker> breakers = new ConcurrentHashMap<>();

    // Shared configuration applied to every new breaker
    private final CircuitBreakerConfig config;

    public CircuitBreakerRegistry() {
        this.config = CircuitBreakerConfig.custom()
                // Open the circuit when 50% or more of calls fail
                .failureRateThreshold(50)
                // Open the circuit when 80% or more of calls are slow
                .slowCallRateThreshold(80)
                // A call is "slow" if it takes longer than 10 seconds
                .slowCallDurationThreshold(Duration.ofSeconds(10))
                // Stay open for 60 seconds before transitioning to half-open
                .waitDurationInOpenState(Duration.ofSeconds(60))
                // Allow 5 test calls in half-open state
                .permittedNumberOfCallsInHalfOpenState(5)
                // Use a sliding window of the last 10 calls
                .slidingWindowType(CircuitBreakerConfig.SlidingWindowType.COUNT_BASED)
                .slidingWindowSize(10)
                .build();
    }

    /**
     * Returns the circuit breaker for the given endpoint,
     * creating one with the shared config if it doesn't exist yet.
     */
    public CircuitBreaker getOrCreate(String endpointId) {
        return breakers.computeIfAbsent(endpointId,
                id -> CircuitBreaker.of("endpoint-" + id, config));
    }
}

What does this code do?

The CircuitBreakerRegistry stores one Resilience4j CircuitBreaker instance per endpoint in a ConcurrentHashMap. When the delivery worker asks for an endpoint's breaker, computeIfAbsent either returns the existing one or creates a new one atomically.

The shared CircuitBreakerConfig uses a count-based sliding window of 10 calls. If 50% or more fail, or 80% or more are slow (over 10 seconds), the breaker trips open. It stays open for 60 seconds, then transitions to half-open and allows 5 test calls to check if the endpoint has recovered.

Why per-endpoint instead of a global circuit breaker?

A global circuit breaker would stop all deliveries when any endpoint degrades. With per-endpoint breakers, only the failing endpoint gets cut off. The worker continues delivering to healthy endpoints at full speed.

• Save CircuitBreakerRegistry.java.
• Verify the file compiles by running Build > Build Project in IntelliJ. You should see no compilation errors in the Build output panel.

Seeing import errors on Resilience4j classes?

Make sure the resilience4j-spring-boot3 dependency is inside the <dependencies> block of pom.xml, not inside <dependencyManagement>. Then reload Maven by pressing Ctrl+Shift+O in IntelliJ.

Help me fix Resilience4j import errors in my Spring Boot project

Wire the circuit breaker into the delivery worker

The registry exists, but the delivery worker does not use it yet. You need to check the circuit breaker state before each delivery attempt. If the breaker is open, the worker should skip that delivery and re-schedule it for later.

• Open DeliveryWorker.java in IntelliJ.
• Add the CircuitBreakerRegistry as a constructor-injected dependency alongside the existing RateLimiter and DeduplicationService fields.
• Add these imports to the top of the file:

import com.signalforge.resilience.CircuitBreakerRegistry;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;

• Inside the delivery method, after the rate limiter and deduplication checks, add the circuit breaker check by copying this code block. Place it before the HTTP POST call:

// Get or create the circuit breaker for this endpoint
CircuitBreaker cb = circuitBreakerRegistry.getOrCreate(endpointId.toString());

// If the circuit is open, skip delivery and re-schedule
if (cb.getState() == CircuitBreaker.State.OPEN) {
    log.warn("Circuit breaker OPEN for endpoint {}, re-scheduling delivery", endpointId);
    attempt.setStatus("FAILED");
    attempt.setNextRetryAt(Instant.now().plusSeconds(60));
    attempt.setErrorMessage("Circuit breaker open - endpoint degraded");
    deliveryAttemptRepository.save(attempt);
    return;
}

What does this code do?

Before making the HTTP call, the worker checks the circuit breaker's state. If the state is OPEN, the endpoint has been failing too much and the worker skips the delivery entirely. It sets the next retry time to 60 seconds later (matching the waitDurationInOpenState you configured), giving the endpoint time to recover.

If the state is CLOSED or HALF_OPEN, the worker proceeds with the delivery normally.

You also need to tell the circuit breaker the outcome of each delivery so it can track failure rates.

• After the HTTP POST call succeeds (where you set status to DELIVERED), add this line to record the success:

// Record successful delivery to the circuit breaker
cb.onSuccess(durationMs, java.util.concurrent.TimeUnit.MILLISECONDS);

• In the catch block where you handle failures (non-2xx responses or exceptions), add this line to record the failure:

// Record failed delivery to the circuit breaker
cb.onError(durationMs, java.util.concurrent.TimeUnit.MILLISECONDS, ex);

Why record duration and not just success/failure?

The onSuccess and onError methods accept the call duration so the circuit breaker can track slow call rates. If 80% of calls to an endpoint take longer than 10 seconds, the breaker trips open even if none of them technically failed. This catches endpoints that are degraded but not completely down.

• Save DeliveryWorker.java.
• Run Build > Build Project to confirm there are no compilation errors.

Seeing a compilation error on durationMs?

The durationMs variable should already exist in your delivery method from Step 6, where you log delivery duration. If it does not exist, capture the start time before the HTTP call with long start = System.currentTimeMillis() and compute long durationMs = System.currentTimeMillis() - start after the HTTP call returns.

Help me wire the circuit breaker into my DeliveryWorker

Expose circuit breaker health through the API

Tenants need visibility into why their deliveries are being skipped. A health endpoint lets them check whether an endpoint's circuit breaker is open, closed, or half-open, along with the current failure and slow call rates.

• Open DeliveryStatsController.java and add the CircuitBreakerRegistry as a constructor-injected dependency.
• Add this new endpoint method to the controller:

/**
 * Returns the circuit breaker health for a specific endpoint.
 * Tenants use this to check if their endpoint is being
 * throttled due to failures or slow responses.
 */
@GetMapping("/api/v1/tenants/{tenantId}/endpoints/{endpointId}/health")
public ResponseEntity<Map<String, Object>> getEndpointHealth(
        @PathVariable UUID tenantId,
        @PathVariable UUID endpointId) {

    // Get or create the breaker (returns existing if already tracked)
    CircuitBreaker cb = circuitBreakerRegistry.getOrCreate(endpointId.toString());
    CircuitBreaker.Metrics metrics = cb.getMetrics();

    // Build the health response with state and key metrics
    Map<String, Object> health = Map.of(
            "endpointId", endpointId,
            "state", cb.getState().name(),
            "failureRate", metrics.getFailureRate(),
            "slowCallRate", metrics.getSlowCallRate()
    );

    return ResponseEntity.ok(health);
}

What does this endpoint return?

The /health endpoint returns a JSON object with four fields. state is CLOSED (healthy), OPEN (tripped), or HALF_OPEN (testing recovery). failureRate and slowCallRate are percentages computed over the sliding window of the last 10 calls.

• Make sure you add these imports at the top of DeliveryStatsController.java:

import com.signalforge.resilience.CircuitBreakerRegistry;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import java.util.Map;
import java.util.UUID;

• Save DeliveryStatsController.java.
• Run the application from IntelliJ (click the green play button or press Shift+F10).
• Test the health endpoint by running this command in your terminal (replace the UUIDs with a real tenant and endpoint ID from your database):

curl -s http://localhost:8080/api/v1/tenants/[[TENANT_ID="your-tenant-uuid"]]/endpoints/[[ENDPOINT_ID="your-endpoint-uuid"]]/health | python -m json.tool

You should see a JSON response with "state": "CLOSED", "failureRate": -1.0, and "slowCallRate": -1.0. The -1.0 values mean the sliding window does not have enough calls recorded yet to compute a rate.

Getting a 404 on the health endpoint?

Check that the @GetMapping path includes both {tenantId} and {endpointId} path variables. Also verify that DeliveryStatsController has the @RestController annotation at the class level.

Help me debug my circuit breaker health endpoint returning 404

✔️ Awesome, I've got everything!

Double check you have saved all three files: pom.xml, CircuitBreakerRegistry.java, and DeliveryStatsController.java. The updated DeliveryWorker.java should also be saved.

ⓧ I'd like to double check the full code

Here is the full CircuitBreakerRegistry.java file:

package com.signalforge.resilience;

import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Manages one Resilience4j CircuitBreaker per endpoint.
 * Each endpoint gets its own breaker so a single degraded
 * endpoint cannot consume all delivery capacity.
 */
@Component
public class CircuitBreakerRegistry {

    // Thread-safe map: endpointId -> CircuitBreaker
    private final ConcurrentHashMap<String, CircuitBreaker> breakers = new ConcurrentHashMap<>();

    // Shared configuration applied to every new breaker
    private final CircuitBreakerConfig config;

    public CircuitBreakerRegistry() {
        this.config = CircuitBreakerConfig.custom()
                // Open the circuit when 50% or more of calls fail
                .failureRateThreshold(50)
                // Open the circuit when 80% or more of calls are slow
                .slowCallRateThreshold(80)
                // A call is "slow" if it takes longer than 10 seconds
                .slowCallDurationThreshold(Duration.ofSeconds(10))
                // Stay open for 60 seconds before transitioning to half-open
                .waitDurationInOpenState(Duration.ofSeconds(60))
                // Allow 5 test calls in half-open state
                .permittedNumberOfCallsInHalfOpenState(5)
                // Use a sliding window of the last 10 calls
                .slidingWindowType(CircuitBreakerConfig.SlidingWindowType.COUNT_BASED)
                .slidingWindowSize(10)
                .build();
    }

    /**
     * Returns the circuit breaker for the given endpoint,
     * creating one with the shared config if it doesn't exist yet.
     */
    public CircuitBreaker getOrCreate(String endpointId) {
        return breakers.computeIfAbsent(endpointId,
                id -> CircuitBreaker.of("endpoint-" + id, config));
    }
}

Here is the health endpoint method that should be in DeliveryStatsController.java:

/**
 * Returns the circuit breaker health for a specific endpoint.
 * Tenants use this to check if their endpoint is being
 * throttled due to failures or slow responses.
 */
@GetMapping("/api/v1/tenants/{tenantId}/endpoints/{endpointId}/health")
public ResponseEntity<Map<String, Object>> getEndpointHealth(
        @PathVariable UUID tenantId,
        @PathVariable UUID endpointId) {

    // Get or create the breaker (returns existing if already tracked)
    CircuitBreaker cb = circuitBreakerRegistry.getOrCreate(endpointId.toString());
    CircuitBreaker.Metrics metrics = cb.getMetrics();

    // Build the health response with state and key metrics
    Map<String, Object> health = Map.of(
            "endpointId", endpointId,
            "state", cb.getState().name(),
            "failureRate", metrics.getFailureRate(),
            "slowCallRate", metrics.getSlowCallRate()
    );

    return ResponseEntity.ok(health);
}

Your delivery engine now protects itself from degraded endpoints. When an endpoint starts failing, the circuit breaker trips open and the worker stops wasting time on it. Next up, you will add Prometheus and Grafana to see all of this happening in real time on a dashboard.

Observability with Micrometer, Prometheus, and Grafana

Your delivery worker now has circuit breakers, rate limiting, and deduplication protecting every outbound call. But you have no visibility into how the system is actually performing.

How many deliveries are succeeding per second? What is your p99 latency? How deep is the queue right now? Without metrics, you are flying blind. In this step, you will wire up Micrometer to record custom delivery metrics, expose them via Prometheus, and visualize everything in a Grafana dashboard.

In this step, get ready to:

• Add Micrometer and custom delivery metrics (Timer, Counter, Gauge, DistributionSummary).
• Set up Prometheus and Grafana in Docker Compose for metrics collection and dashboarding.
• Instrument the delivery worker with metrics recording and structured MDC logging.

Add Micrometer and custom delivery metrics

Micrometer is a vendor-neutral metrics facade for JVM applications. It works like SLF4J for logging, but for metrics. You write instrumentation code once, and Micrometer translates it to whatever backend you choose.

Spring Boot auto-configures a MeterRegistry when it finds a registry implementation on the classpath. By adding micrometer-registry-prometheus, Spring Boot will expose a /actuator/prometheus endpoint that Prometheus can scrape.

• Add the Micrometer Prometheus registry dependency to your pom.xml inside the <dependencies> section:

<!-- Micrometer Prometheus registry for metrics export -->
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>

What does this dependency do?

The Spring Boot BOM manages the version, so you do not need to specify one. This single dependency enables the /actuator/prometheus endpoint and auto-configures a Prometheus-compatible MeterRegistry.

By default, Spring Boot only exposes the health endpoint over HTTP. You need to explicitly expose the prometheus endpoint so Prometheus can reach it.

• Open src/main/resources/application.yml and add the following management block at the top level of the file:

# Expose Actuator endpoints for Prometheus scraping
management:
  endpoints:
    web:
      exposure:
        include: health, info, metrics, prometheus

What does this configuration do?

The management.endpoints.web.exposure.include property lists which Actuator endpoints are available over HTTP. Adding prometheus to this list tells Spring Boot to serve your metrics in Prometheus text format at /actuator/prometheus.

• Save application.yml.
• Restart the app in IntelliJ and hit http://localhost:8080/actuator/prometheus in your browser. You should see a wall of plain text metrics in Prometheus exposition format.

Getting a 404 on /actuator/prometheus?

• Make sure you added the micrometer-registry-prometheus dependency to pom.xml and reloaded Maven.
• Verify the YAML indentation is correct. management should be at the root level, not nested under another key.
• Why isn't my /actuator/prometheus endpoint showing up?

Now you need custom metrics specific to SignalForge. Spring Boot gives you JVM and HTTP metrics for free, but delivery-specific signals require manual instrumentation.

• Create the file src/main/java/com/signalforge/metrics/DeliveryMetrics.java and add the following code:

package com.signalforge.metrics;

import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.DistributionSummary;
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

@Component
public class DeliveryMetrics {

    private final MeterRegistry registry;

    public DeliveryMetrics(MeterRegistry registry, JdbcTemplate jdbcTemplate) {
        this.registry = registry;

        // Gauge reads the current queue depth from the database on each scrape
        Gauge.builder("signalforge.queue.depth", jdbcTemplate, jt ->
                jt.queryForObject(
                    "SELECT count(*) FROM delivery_attempts "
                    + "WHERE status IN ('PENDING', 'FAILED') "
                    + "AND next_retry_at <= now()",
                    Long.class
                ).doubleValue())
            .description("Delivery attempts ready for claiming")
            .register(registry);
    }

    // Start a timer sample to measure a single delivery attempt
    public Timer.Sample startTimer() {
        return Timer.start(registry);
    }

    // Stop the timer and record the duration with dimensional tags
    public void recordDeliveryDuration(Timer.Sample sample,
                                       String tenantId,
                                       String endpointId,
                                       String status) {
        sample.stop(Timer.builder("signalforge.delivery.duration")
            .tag("tenant_id", tenantId)
            .tag("endpoint_id", endpointId)
            .tag("status", status)
            .description("Duration of individual delivery attempts")
            .register(registry));
    }

    // Count delivery outcomes by final status
    public void recordDeliveryAttempt(String status) {
        Counter.builder("signalforge.delivery.attempts")
            .tag("status", status)
            .description("Total delivery attempts by outcome")
            .register(registry)
            .increment();
    }

    // Record which attempt number a delivery completed on
    public void recordRetryDistribution(int attemptNumber) {
        DistributionSummary.builder("signalforge.retry.distribution")
            .description("Distribution of attempt numbers at delivery time")
            .register(registry)
            .record(attemptNumber);
    }
}

What does this code do?

This class registers four custom metrics with Micrometer's MeterRegistry:

• signalforge.queue.depth (Gauge) queries the database on each Prometheus scrape to report how many delivery attempts are waiting to be claimed.
• signalforge.delivery.duration (Timer) measures how long each HTTP delivery takes, tagged by tenant, endpoint, and status. This is where p99 latency comes from.
• signalforge.delivery.attempts (Counter) increments for every delivery outcome (DELIVERED, FAILED, DEAD). You will use this for success rate calculations.
• signalforge.retry.distribution (DistributionSummary) records the attempt number each time a delivery completes. This tells you whether most events succeed on the first try or need retries.

Why Timer.Sample instead of a static Timer?

The Timer.Sample pattern lets you decide the timer's tags at the end of the operation. You do not know the delivery status (DELIVERED vs FAILED) when the timer starts. The sample captures the start time, and you choose the tags when you call stop().

• Save DeliveryMetrics.java.

Seeing a compile error on JdbcTemplate?

• JdbcTemplate is auto-configured by Spring Boot when you have a datasource. Make sure the spring-boot-starter-data-jpa dependency is in your pom.xml (it has been there since Step 2).
• Check that your import is org.springframework.jdbc.core.JdbcTemplate.
• Help me fix a compile error in DeliveryMetrics.java.

Set up Prometheus and Grafana in Docker Compose

Prometheus is a pull-based monitoring system. It scrapes your /actuator/prometheus endpoint on a schedule and stores the time-series data. Grafana then queries Prometheus to render dashboards.

You need a Prometheus configuration file before you can add it to Docker Compose. This file tells Prometheus where to scrape.

• Create a new file called prometheus.yml in the project root (the same directory as docker-compose.yml) and add the following content:

global:
  scrape_interval: 5s

scrape_configs:
  - job_name: signalforge
    static_configs:
      - targets: ['host.docker.internal:8080']

What does this configuration do?

• scrape_interval: 5s tells Prometheus to pull metrics every 5 seconds.
• host.docker.internal:8080 is a special DNS name that Docker provides. It points from inside the container back to your host machine where the Spring Boot app is running.
• Prometheus defaults to scraping /metrics, but Spring Boot's Actuator exposes metrics at /actuator/prometheus. Prometheus will auto-discover the correct path from the Spring Boot endpoint.

• Save prometheus.yml.

Now add the Prometheus and Grafana services to your Docker Compose stack.

• Open docker-compose.yml and replace its entire contents with the following:

version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: signalforge
      POSTGRES_USER: signalforge
      POSTGRES_PASSWORD: signalforge
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7
    ports:
      - '6379:6379'

  # Prometheus scrapes /actuator/prometheus every 5 seconds
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - '9090:9090'

  # Grafana reads from Prometheus for dashboarding
  grafana:
    image: grafana/grafana:latest
    ports:
      - '3000:3000'
    depends_on:
      - prometheus

volumes:
  postgres_data:

What's new in this Docker Compose file?

• The prometheus service mounts your local prometheus.yml into the container at the path Prometheus expects. It exposes port 9090 for the Prometheus web UI.
• The grafana service depends on Prometheus and exposes port 3000 for the Grafana dashboard UI.

• Save docker-compose.yml.
• Restart your Docker Compose stack by running:

docker compose up -d

• Open http://localhost:9090/targets in your browser. You should see the signalforge target with a green UP state.

Target showing as DOWN in Prometheus?

• Make sure your Spring Boot app is running on port 8080. Prometheus inside Docker uses host.docker.internal to reach your host machine.
• Verify that http://localhost:8080/actuator/prometheus returns metrics in your browser before checking Prometheus.
• On Linux, host.docker.internal may not resolve. Add extra_hosts: ["host.docker.internal:host-gateway"] to the prometheus service in docker-compose.yml.
• Why is my Prometheus target showing DOWN for SignalForge?

Now set up Grafana to read from Prometheus and create your dashboard.

• Open http://localhost:3000 in your browser.
• Log in with the default credentials: username admin, password admin. Grafana will ask you to change the password. You can skip this for local development.
• Click the hamburger menu on the left sidebar.
• Click Connections, then Data sources.
• Click Add data source and select Prometheus.
• Set the Prometheus server URL to http://prometheus:9090. This uses the Docker Compose service name so Grafana can reach Prometheus inside the Docker network.
• Click Save & test. You should see a green success message.

Create a dashboard with panels for your key delivery metrics.

• Click the + icon in the sidebar and select New dashboard.
• Click Add visualization and select the Prometheus data source.
• Create the following panels (one at a time, using Add panel for each):

• P99 Delivery Latency: Use the PromQL query histogram_quantile(0.99, rate(signalforge_delivery_duration_seconds_bucket[5m])). Set visualization type to Time series.
• Delivery Success Rate (%): Use rate(signalforge_delivery_attempts_total{status="DELIVERED"}[5m]) / rate(signalforge_delivery_attempts_total[5m]) * 100. Set visualization type to Stat.
• Queue Depth Over Time: Use signalforge_queue_depth. Set visualization type to Time series.
• Retry Attempt Distribution: Use signalforge_retry_distribution_bucket. Set visualization type to Bar chart.
• Circuit Breaker States: Use resilience4j_circuitbreaker_state. Set visualization type to Table to see per-endpoint states.

• Click the save icon (floppy disk) in the top right corner and name your dashboard SignalForge Delivery Dashboard.

Panels will show data after deliveries

Your panels will be empty until the delivery worker processes events. After you ingest some test events (via the EventController API), the worker will deliver them and the metrics will populate in Grafana.

Instrument the delivery worker with metrics and structured logging

The DeliveryMetrics class defines the meters, but nothing records to them yet. You need to call these methods from DeliveryWorker each time a delivery attempt completes.

You will also add SLF4J MDC fields so every log line includes the tenant ID, event ID, endpoint ID, and attempt number. This makes production debugging trivial.

• Open DeliveryWorker.java. Add the DeliveryMetrics dependency to the constructor and store it as a field:

// Add this import at the top of DeliveryWorker.java
import com.signalforge.metrics.DeliveryMetrics;
import org.slf4j.MDC;

// Add DeliveryMetrics to the constructor alongside existing dependencies
private final DeliveryMetrics deliveryMetrics;

// Update constructor to include DeliveryMetrics
public DeliveryWorker(/* existing params */, DeliveryMetrics deliveryMetrics) {
    // existing assignments
    this.deliveryMetrics = deliveryMetrics;
}

What does this code do?

Spring will auto-inject the DeliveryMetrics bean alongside the existing dependencies (RestClient, RetryPolicy, RateLimiter, etc.). The worker can now record timing and outcome metrics for every delivery attempt.

Now wrap each delivery attempt with timer recording, MDC context, and outcome counting. Find the section in your delivery loop where you make the HTTP POST for each claimed attempt.

• Inside the delivery loop (where you iterate over claimed attempts and make HTTP calls), add the following pattern around each delivery attempt:

// Set MDC context before each delivery attempt
MDC.put("tenantId", attempt.getEvent().getTenant().getId().toString());
MDC.put("eventId", attempt.getEvent().getId().toString());
MDC.put("endpointId", attempt.getEndpoint().getId().toString());
MDC.put("attemptNumber", String.valueOf(attempt.getAttemptNumber()));

try {
    // Start the timer before the HTTP call
    Timer.Sample sample = deliveryMetrics.startTimer();

    // ... your existing HTTP POST delivery logic here ...

    // On success: record metrics
    deliveryMetrics.recordDeliveryDuration(sample,
        attempt.getEvent().getTenant().getId().toString(),
        attempt.getEndpoint().getId().toString(),
        "DELIVERED");
    deliveryMetrics.recordDeliveryAttempt("DELIVERED");
    deliveryMetrics.recordRetryDistribution(attempt.getAttemptNumber());

} catch (Exception e) {
    // On failure: record metrics with failure status
    // (sample timing and counter logic here, using "FAILED" or "DEAD")
    deliveryMetrics.recordDeliveryAttempt(
        attempt.getAttemptNumber() >= maxAttempts ? "DEAD" : "FAILED");

} finally {
    // Always clear MDC to avoid context leaking between deliveries
    MDC.clear();
}

What does this code do?

• MDC.put() stores key-value pairs on the current thread. Every log statement inside the try block automatically includes these fields. This lets you filter logs by tenant, event, or endpoint without manually including them in every log message.
• Timer.Sample captures the start time. When recordDeliveryDuration() calls sample.stop(), it records the elapsed time with tags you specify at stop time.
• MDC.clear() in the finally block prevents context from one delivery leaking into the next. With Virtual Threads, many deliveries share carrier threads, so clearing MDC is essential.

What is MDC?

MDC stands for Mapped Diagnostic Context. It is a per-thread map managed by SLF4J. When you put values into MDC, Logback (the logging framework Spring Boot uses by default) automatically includes them in every log line. In production, this means you can search your logs for all entries related to a specific event or tenant.

Finally, configure JSON-formatted log output so MDC fields appear as structured, searchable JSON fields in production logs.

• Open application.yml and add the following Spring profile section for production logging. Spring Boot 3.4+ has built-in structured logging support:

# Add under a production profile section
---
spring:
  config:
    activate:
      on-profile: production

# Structured JSON logging for production
logging:
  structured:
    format:
      console: ecs

What does this configuration do?

The logging.structured.format.console: ecs property tells Spring Boot to output logs in Elastic Common Schema JSON format. Every MDC key-value pair you set (tenantId, eventId, etc.) is automatically included as a top-level JSON field.

This only activates when you run with the production profile. During local development, you keep the human-readable default log format.

• Save DeliveryWorker.java and application.yml.
• Restart the app in IntelliJ. Ingest a test event via the API and check http://localhost:9090. Search for signalforge_delivery_duration_seconds_count in the Prometheus expression browser. You should see a non-zero value once a delivery completes.

Not seeing custom metrics in Prometheus?

• Custom metrics only appear after the first recording. Ingest at least one event and wait for the delivery worker to process it.
• Prometheus converts dots to underscores in metric names. Your signalforge.delivery.duration metric appears as signalforge_delivery_duration_seconds in PromQL queries.
• My custom SignalForge metrics are not showing up in Prometheus after ingesting events.

Your delivery engine now has full observability. Every delivery attempt is timed, counted, and tagged. Prometheus stores the history, and Grafana makes it visual. Next up, you will run multiple instances of SignalForge to verify the SKIP LOCKED pattern under real contention.

Multi-Instance Deployment and Concurrency Verification

Your observability stack is live. You can see delivery latency, success rates, and queue depth on your Grafana dashboard.

But everything is running on a single instance. In production, you would run multiple copies of your application to handle more load and survive instance failures. The real question is: does your SKIP LOCKED delivery worker actually prevent two instances from claiming the same work? In this step, you will Dockerize your application, deploy two instances behind an Nginx load balancer, and prove that the concurrency guarantees hold.

In this step, get ready to:

• Dockerize the application and run two instances behind Nginx.
• Verify distributed delivery with a 500-event load test.
• Write a concurrency integration test proving SKIP LOCKED prevents double-claiming.

Dockerize the application and configure multi-instance deployment

To run two copies of SignalForge inside Docker Compose, you first need a Dockerfile that packages the Spring Boot app into a container image. You also need an Nginx config that load-balances incoming traffic across both instances.

• Create a file called Dockerfile in the root of your signalforge-engine project folder (the same folder that contains pom.xml) and paste the following code:

# Stage 1: Build the application using Maven and Java 21
FROM maven:3.9-eclipse-temurin-21 AS build
WORKDIR /app

# Copy the POM first to cache dependency downloads
COPY pom.xml .
RUN mvn dependency:go-offline -B

# Copy source code and build the JAR
COPY src ./src
RUN mvn clean package -DskipTests -B

# Stage 2: Create a lightweight runtime image
FROM eclipse-temurin:21-jre
WORKDIR /app

# Copy only the built JAR from the build stage
COPY --from=build /app/target/*.jar app.jar

EXPOSE 8080

# Run with container-aware memory settings
ENTRYPOINT ["java", "-XX:+UseContainerSupport", "-XX:MaxRAMPercentage=75.0", "-jar", "app.jar"]

What does this Dockerfile do?

This is a multi-stage build. Stage 1 uses the full Maven + JDK image to compile your project. Stage 2 copies only the built JAR into a smaller JRE image, keeping the final container lean.

The -XX:+UseContainerSupport flag tells the JVM to respect Docker memory limits instead of reading the host's total RAM. -XX:MaxRAMPercentage=75.0 caps heap usage at 75% of the container's memory allocation, leaving room for off-heap memory.

• Save Dockerfile.
• Confirm the file exists by running:

ls Dockerfile

• You should see Dockerfile in the output.

• Create a file called nginx.conf in the same project root folder and paste the following configuration:

events {
    worker_connections 1024;
}

http {
    # Round-robin load balancing across both SignalForge instances
    upstream signalforge {
        server signalforge-1:8080;
        server signalforge-2:8080;
    }

    server {
        listen 80;

        location / {
            proxy_pass http://signalforge;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        }
    }
}

What does this config do?

The upstream block lists both SignalForge containers by their Docker Compose service names. Nginx uses round-robin by default, sending each incoming request to the next server in the list.

The proxy_set_header directives forward the original client IP to your Spring Boot app. Without them, every request looks like it came from the Nginx container.

• Save nginx.conf.

Now update your Docker Compose file to define the two app instances and the Nginx load balancer.

• Open docker-compose.yml and replace its entire contents with the following:

version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: signalforge
      POSTGRES_USER: signalforge
      POSTGRES_PASSWORD: signalforge
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7
    ports:
      - '6379:6379'

  # First app instance on host port 8080
  signalforge-1:
    build: .
    ports:
      - '8080:8080'
    environment:
      SPRING_DATASOURCE_URL: jdbc:postgresql://postgres:5432/signalforge
      SPRING_DATASOURCE_USERNAME: signalforge
      SPRING_DATASOURCE_PASSWORD: signalforge
      SPRING_DATA_REDIS_HOST: redis
    depends_on:
      - postgres
      - redis

  # Second app instance on host port 8081
  signalforge-2:
    build: .
    ports:
      - '8081:8080'
    environment:
      SPRING_DATASOURCE_URL: jdbc:postgresql://postgres:5432/signalforge
      SPRING_DATASOURCE_USERNAME: signalforge
      SPRING_DATASOURCE_PASSWORD: signalforge
      SPRING_DATA_REDIS_HOST: redis
    depends_on:
      - postgres
      - redis

  # Nginx load balancer in front of both instances
  nginx:
    image: nginx:alpine
    ports:
      - '80:80'
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - signalforge-1
      - signalforge-2

  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - '9090:9090'

  grafana:
    image: grafana/grafana:latest
    ports:
      - '3000:3000'
    depends_on:
      - prometheus

volumes:
  postgres_data:

What changed in docker-compose.yml?

• signalforge-1 and signalforge-2 both use build: . to build from your new Dockerfile. Each one maps a different host port (8080 and 8081) to the container's internal port 8080.
• The environment block overrides your application.yml datasource settings. Inside Docker, containers reach each other by service name, so postgres and redis replace localhost.
• The nginx service mounts your nginx.conf as read-only and exposes port 80 as the single entry point.

• Save docker-compose.yml.

Prometheus also needs to know about both instances so it can scrape metrics from each one.

• Open prometheus.yml and find this line:

      - targets: ['host.docker.internal:8080']

• Replace it with the following, which targets both app containers by their Docker Compose service names:

      - targets: ['signalforge-1:8080', 'signalforge-2:8080']

• Save prometheus.yml.

Build failing or container won't start?

• Make sure Dockerfile, nginx.conf, and docker-compose.yml are all in the project root alongside pom.xml.
• Verify Docker Desktop is running. If containers fail to connect to postgres, check that the SPRING_DATASOURCE_URL environment variable uses postgres as the hostname (the Docker service name), not localhost.
• My Docker build is failing or containers won't connect to the database.

✔️ Awesome, I've got everything!

Make sure you have saved all four files: Dockerfile, nginx.conf, docker-compose.yml, and prometheus.yml.

ⓧ I'd like to double check the full code

version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: signalforge
      POSTGRES_USER: signalforge
      POSTGRES_PASSWORD: signalforge
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7
    ports:
      - '6379:6379'

  signalforge-1:
    build: .
    ports:
      - '8080:8080'
    environment:
      SPRING_DATASOURCE_URL: jdbc:postgresql://postgres:5432/signalforge
      SPRING_DATASOURCE_USERNAME: signalforge
      SPRING_DATASOURCE_PASSWORD: signalforge
      SPRING_DATA_REDIS_HOST: redis
    depends_on:
      - postgres
      - redis

  signalforge-2:
    build: .
    ports:
      - '8081:8080'
    environment:
      SPRING_DATASOURCE_URL: jdbc:postgresql://postgres:5432/signalforge
      SPRING_DATASOURCE_USERNAME: signalforge
      SPRING_DATASOURCE_PASSWORD: signalforge
      SPRING_DATA_REDIS_HOST: redis
    depends_on:
      - postgres
      - redis

  nginx:
    image: nginx:alpine
    ports:
      - '80:80'
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - signalforge-1
      - signalforge-2

  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - '9090:9090'

  grafana:
    image: grafana/grafana:latest
    ports:
      - '3000:3000'
    depends_on:
      - prometheus

volumes:
  postgres_data:

global:
  scrape_interval: 5s
scrape_configs:
  - job_name: signalforge
    static_configs:
      - targets: ['signalforge-1:8080', 'signalforge-2:8080']

Verify distributed delivery under contention

With both instances defined, it is time to bring the entire stack up and prove that work is distributed evenly with no double-delivery.

• If your previous Docker Compose stack is still running, stop it first by running:

docker compose down

• Build and start the full stack by running:

docker compose up -d --build

What does --build do?

The --build flag forces Docker Compose to build the image from your Dockerfile before starting. Without it, Compose reuses a cached image that might not include your latest code changes.

• Wait about 30 seconds for the Maven build and container startup to complete. Verify all containers are running by executing:

docker compose ps

• You should see 7 containers in the running state: postgres, redis, signalforge-1, signalforge-2, nginx, prometheus, and grafana.
• Confirm both app instances are healthy by hitting their actuator endpoints:

curl http://localhost:8080/actuator/health
curl http://localhost:8081/actuator/health

• Both should return {"status":"UP"}.
• Also confirm the Nginx load balancer is forwarding requests by running:

curl http://localhost/actuator/health

• This hits port 80 (Nginx), which forwards to one of the two app instances. You should see {"status":"UP"}.

One or both app containers exiting?

• Check the logs with docker compose logs signalforge-1 to see the Spring Boot startup output. Look for connection refused errors pointing at postgres or redis.
• If the app starts before Postgres is ready, restart just the app containers with docker compose restart signalforge-1 signalforge-2.
• My SignalForge containers keep exiting and I see connection errors in the logs.

Now you need a test tenant and endpoint so you can blast events through the load balancer.

• Create a test tenant through the Nginx load balancer by running:

curl -s -X POST http://localhost/api/v1/tenants \
  -H "Content-Type: application/json" \
  -d '{"name": "load-test-tenant"}'

• Copy the id from the response. You will need it in the next commands.
• Register a test endpoint for that tenant by running this command, replacing the tenant ID placeholder with your actual tenant ID:

curl -s -X POST http://localhost/api/v1/tenants/[[TENANT_ID="your tenant ID from the previous response"]]/endpoints \
  -H "Content-Type: application/json" \
  -d '{"url": "https://httpbin.org/post"}'

Now ingest 500 events to create enough delivery work for both instances to compete over.

• Run the load test by executing this script, replacing the tenant ID placeholder with your tenant ID:

for i in $(seq 1 500); do
  curl -s -X POST http://localhost/api/v1/tenants/[[TENANT_ID="your tenant ID"]]/events \
    -H "Content-Type: application/json" \
    -d "{\"event_type\": \"test.load\", \"payload\": {\"index\": $i}}" &
  # Send 50 at a time to avoid overwhelming curl
  if [ $((i % 50)) -eq 0 ]; then wait; echo "Sent $i events"; fi
done
wait
echo "All 500 events sent"

• Wait about 30 seconds for the delivery workers on both instances to process the events.
• Check the logs from each instance to confirm both are claiming and delivering work:

docker compose logs signalforge-1 | grep "DELIVERED" | wc -l
docker compose logs signalforge-2 | grep "DELIVERED" | wc -l

• Both counts should be greater than zero. The exact split depends on timing, but you should see work distributed across both instances.

The most important check is that no delivery attempt was claimed by both instances.

• Connect to the PostgreSQL container and run a query to check for duplicates:

docker compose exec postgres psql -U signalforge -d signalforge -c "
SELECT id, COUNT(*)
FROM delivery_attempts
WHERE status = 'DELIVERED'
GROUP BY id
HAVING COUNT(*) > 1;"

• This query should return zero rows. If it returns any rows, a delivery attempt was processed more than once, which would mean SKIP LOCKED has a gap in your implementation.

• Confirm Prometheus is scraping both instances. Open http://localhost:9090/targets in your browser.
• You should see two targets listed under the signalforge job: signalforge-1:8080 and signalforge-2:8080, both with state UP.

Write a concurrency integration test

The manual verification proves things work, but a repeatable automated test is what gives you confidence during future refactoring. This test will start 5 threads that simultaneously call the delivery worker's claim method and assert that no delivery attempt ID is claimed by more than one thread.

• Create a new test file called ConcurrencyTest.java in src/test/java/com/signalforge/ and paste the following code:

package com.signalforge;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

/**
 * Proves that SKIP LOCKED prevents double-claiming
 * when multiple threads compete for the same delivery attempts.
 */
public class ConcurrencyTest extends AbstractIntegrationTest {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    private UUID tenantId;
    private UUID endpointId;
    private UUID eventId;

    @BeforeEach
    void seedTestData() {
        // Create a tenant, endpoint, and event for testing
        tenantId = UUID.randomUUID();
        endpointId = UUID.randomUUID();
        eventId = UUID.randomUUID();

        jdbcTemplate.update(
            "INSERT INTO tenants (id, name, api_key, created_at) VALUES (?, ?, ?, now())",
            tenantId, "concurrency-test", UUID.randomUUID().toString()
        );
        jdbcTemplate.update(
            "INSERT INTO endpoints (id, tenant_id, url, status, created_at) VALUES (?, ?, ?, 'ACTIVE', now())",
            endpointId, tenantId, "https://httpbin.org/post"
        );
        jdbcTemplate.update(
            "INSERT INTO events (id, tenant_id, event_type, payload, created_at) VALUES (?, ?, ?, '{}'::jsonb, now())",
            eventId, tenantId, "test.concurrency"
        );

        // Insert 50 delivery attempts all ready for claiming
        for (int i = 0; i < 50; i++) {
            jdbcTemplate.update(
                "INSERT INTO delivery_attempts (event_id, endpoint_id, status, attempt_number, next_retry_at, created_at, updated_at) "
                + "VALUES (?, ?, 'PENDING', 1, now(), now(), now())",
                eventId, endpointId
            );
        }
    }

    @Test
    void skipLockedPreventsDoubleClaiming() throws Exception {
        int threadCount = 5;
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        // Latch ensures all threads start claiming at the same instant
        CountDownLatch startLatch = new CountDownLatch(threadCount);

        List<Future<List<Long>>> futures = new ArrayList<>();

        for (int t = 0; t < threadCount; t++) {
            futures.add(executor.submit(() -> {
                // Signal ready, then wait for all threads
                startLatch.countDown();
                startLatch.await();

                // Each thread claims a batch using the same SKIP LOCKED query
                return jdbcTemplate.queryForList(
                    "WITH claimed AS ("
                    + "  SELECT id FROM delivery_attempts"
                    + "  WHERE status = 'PENDING' AND next_retry_at <= now()"
                    + "  ORDER BY next_retry_at, id"
                    + "  FOR UPDATE SKIP LOCKED"
                    + "  LIMIT 20"
                    + ") "
                    + "UPDATE delivery_attempts SET status = 'DELIVERING'"
                    + " FROM claimed"
                    + " WHERE delivery_attempts.id = claimed.id"
                    + " RETURNING delivery_attempts.id",
                    Long.class
                );
            }));
        }

        // Collect all claimed IDs across all threads
        Set<Long> allClaimedIds = ConcurrentHashMap.newKeySet();
        int totalClaimed = 0;

        for (Future<List<Long>> future : futures) {
            List<Long> claimed = future.get(10, TimeUnit.SECONDS);
            for (Long id : claimed) {
                // If add() returns false, the ID was already claimed by another thread
                assertThat(allClaimedIds.add(id))
                    .as("Delivery attempt %d was claimed by multiple threads", id)
                    .isTrue();
            }
            totalClaimed += claimed.size();
        }

        // Every claimed ID should be unique across all threads
        assertThat(totalClaimed).isEqualTo(allClaimedIds.size());
        // All 50 attempts should be claimed across the 5 threads
        assertThat(totalClaimed).isEqualTo(50);

        executor.shutdown();
    }
}

What does this test prove?

• The CountDownLatch ensures all 5 threads fire their claim query at exactly the same moment, maximizing the chance of contention.
• Each thread runs the same CTE-based SELECT ... FOR UPDATE SKIP LOCKED query that your DeliveryWorker uses. When one thread locks a row, other threads skip it and grab the next available row.
• The assertion checks that every claimed ID is unique across all threads. If any ID appears twice, SKIP LOCKED failed to prevent a double-claim.

• Save ConcurrencyTest.java.
• Run the test from your terminal by executing:

mvn test -Dtest=ConcurrencyTest -pl .

• You should see BUILD SUCCESS with 1 test passed. This confirms that PostgreSQL SKIP LOCKED correctly prevents double-claiming even under heavy thread contention.

Test failing with connection or assertion errors?

• If you see Connection refused, make sure your Testcontainers PostgreSQL container is starting correctly. Check that Docker Desktop is running.
• If the total claimed count is less than 50, your delivery_attempts inserts in seedTestData() may have failed. Verify the INSERT column names match your V1__init_schema.sql exactly.
• My ConcurrencyTest is failing. Help me debug the SKIP LOCKED test.

✔️ Awesome, I've got everything!

Make sure you have saved ConcurrencyTest.java and the test is passing.

ⓧ I'd like to double check the full code

package com.signalforge;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

/**
 * Proves that SKIP LOCKED prevents double-claiming
 * when multiple threads compete for the same delivery attempts.
 */
public class ConcurrencyTest extends AbstractIntegrationTest {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    private UUID tenantId;
    private UUID endpointId;
    private UUID eventId;

    @BeforeEach
    void seedTestData() {
        // Create a tenant, endpoint, and event for testing
        tenantId = UUID.randomUUID();
        endpointId = UUID.randomUUID();
        eventId = UUID.randomUUID();

        jdbcTemplate.update(
            "INSERT INTO tenants (id, name, api_key, created_at) VALUES (?, ?, ?, now())",
            tenantId, "concurrency-test", UUID.randomUUID().toString()
        );
        jdbcTemplate.update(
            "INSERT INTO endpoints (id, tenant_id, url, status, created_at) VALUES (?, ?, ?, 'ACTIVE', now())",
            endpointId, tenantId, "https://httpbin.org/post"
        );
        jdbcTemplate.update(
            "INSERT INTO events (id, tenant_id, event_type, payload, created_at) VALUES (?, ?, ?, '{}'::jsonb, now())",
            eventId, tenantId, "test.concurrency"
        );

        // Insert 50 delivery attempts all ready for claiming
        for (int i = 0; i < 50; i++) {
            jdbcTemplate.update(
                "INSERT INTO delivery_attempts (event_id, endpoint_id, status, attempt_number, next_retry_at, created_at, updated_at) "
                + "VALUES (?, ?, 'PENDING', 1, now(), now(), now())",
                eventId, endpointId
            );
        }
    }

    @Test
    void skipLockedPreventsDoubleClaiming() throws Exception {
        int threadCount = 5;
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        // Latch ensures all threads start claiming at the same instant
        CountDownLatch startLatch = new CountDownLatch(threadCount);

        List<Future<List<Long>>> futures = new ArrayList<>();

        for (int t = 0; t < threadCount; t++) {
            futures.add(executor.submit(() -> {
                // Signal ready, then wait for all threads
                startLatch.countDown();
                startLatch.await();

                // Each thread claims a batch using the same SKIP LOCKED query
                return jdbcTemplate.queryForList(
                    "WITH claimed AS ("
                    + "  SELECT id FROM delivery_attempts"
                    + "  WHERE status = 'PENDING' AND next_retry_at <= now()"
                    + "  ORDER BY next_retry_at, id"
                    + "  FOR UPDATE SKIP LOCKED"
                    + "  LIMIT 20"
                    + ") "
                    + "UPDATE delivery_attempts SET status = 'DELIVERING'"
                    + " FROM claimed"
                    + " WHERE delivery_attempts.id = claimed.id"
                    + " RETURNING delivery_attempts.id",
                    Long.class
                );
            }));
        }

        // Collect all claimed IDs across all threads
        Set<Long> allClaimedIds = ConcurrentHashMap.newKeySet();
        int totalClaimed = 0;

        for (Future<List<Long>> future : futures) {
            List<Long> claimed = future.get(10, TimeUnit.SECONDS);
            for (Long id : claimed) {
                // If add() returns false, the ID was already claimed by another thread
                assertThat(allClaimedIds.add(id))
                    .as("Delivery attempt %d was claimed by multiple threads", id)
                    .isTrue();
            }
            totalClaimed += claimed.size();
        }

        // Every claimed ID should be unique across all threads
        assertThat(totalClaimed).isEqualTo(allClaimedIds.size());
        // All 50 attempts should be claimed across the 5 threads
        assertThat(totalClaimed).isEqualTo(50);

        executor.shutdown();
    }
}

You now have mathematical proof that your delivery worker handles multi-instance contention correctly. Phase 2 is complete. Next, you will split SignalForge into separate API and Worker services and introduce Kafka for event-driven communication.

Split Into API Service and Delivery Worker Service

Your V2 system is battle-tested. Two app instances share the workload behind Nginx, circuit breakers protect degraded endpoints, and your Grafana dashboard tracks every delivery.

But every instance still runs the full monolith. The REST API and the delivery worker compete for the same resources inside a single process. In this step, you will split the monolith into two focused services: an API service that handles HTTP requests, and a Worker service that claims and delivers events.

In this step, get ready to:

• Restructure the Maven project into a multi-module build with a shared common library.
• Create separate application entry points for the API and Worker services.
• Update Docker Compose to run both services as independent containers and verify end-to-end delivery.

Restructure into a multi-module Maven build

A Maven multi-module project uses a parent POM with <packaging>pom</packaging> to coordinate child modules. Each module gets its own pom.xml, its own src/ tree, and its own dependency list.

Why split into modules instead of separate repositories?

Both services share the same JPA entities, DTOs, and repository interfaces. A multi-module build keeps that shared code in a single signalforge-common module that both services depend on.

You get a single mvn install to build everything, while still producing two independent deployable JARs.

• Create the three module directories by running these commands from your project root (signalforge-engine):

mkdir -p signalforge-common/src/main/java/com/signalforge/common
mkdir -p signalforge-common/src/main/resources/db/migration
mkdir -p signalforge-api/src/main/java/com/signalforge/api
mkdir -p signalforge-api/src/main/resources
mkdir -p signalforge-worker/src/main/java/com/signalforge/worker
mkdir -p signalforge-worker/src/main/resources

What does this create?

Three sibling module directories, each with the standard Maven src/main/java layout. The signalforge-common module also gets a db/migration folder because Flyway migrations belong with the shared schema.

• Confirm the directories exist by running:

ls -d signalforge-*/

You should see signalforge-api/, signalforge-common/, and signalforge-worker/ listed.

Now you need a parent POM that ties these modules together. Your existing pom.xml at the project root will become this parent.

• Replace the entire contents of your root pom.xml with the following parent POM:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <!-- Inherit Spring Boot dependency management -->
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.5.14</version>
        <relativePath/>
    </parent>

    <groupId>com.signalforge</groupId>
    <artifactId>signalforge-engine</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <!-- packaging=pom tells Maven this is a parent, not a JAR -->
    <packaging>pom</packaging>
    <name>SignalForge Engine</name>

    <!-- Declare the three child modules -->
    <modules>
        <module>signalforge-common</module>
        <module>signalforge-api</module>
        <module>signalforge-worker</module>
    </modules>

    <properties>
        <java.version>21</java.version>
    </properties>
</project>

What does this POM do?

• spring-boot-starter-parent at version 3.5.14 manages dependency versions for all child modules.
• <packaging>pom</packaging> tells Maven this project coordinates modules. It does not produce a JAR.
• The <modules> block lists the three directories Maven will build in order.

• Create the signalforge-common/pom.xml file with the following content:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <!-- Inherit from the parent POM -->
    <parent>
        <groupId>com.signalforge</groupId>
        <artifactId>signalforge-engine</artifactId>
        <version>0.0.1-SNAPSHOT</version>
    </parent>

    <artifactId>signalforge-common</artifactId>
    <name>SignalForge Common</name>

    <!-- Shared library: entities, DTOs, repositories -->
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-jpa</artifactId>
        </dependency>
        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>postgresql</artifactId>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-validation</artifactId>
        </dependency>
    </dependencies>
</project>

Why no spring-boot-maven-plugin here?

The common module is a plain library JAR, not an executable application. Only the API and Worker modules need the Spring Boot plugin to produce fat JARs.

• Create the signalforge-api/pom.xml file with the following content:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <groupId>com.signalforge</groupId>
        <artifactId>signalforge-engine</artifactId>
        <version>0.0.1-SNAPSHOT</version>
    </parent>

    <artifactId>signalforge-api</artifactId>
    <name>SignalForge API</name>

    <dependencies>
        <!-- Shared entities, DTOs, and repositories -->
        <dependency>
            <groupId>com.signalforge</groupId>
            <artifactId>signalforge-common</artifactId>
            <version>${project.version}</version>
        </dependency>

        <!-- REST API stack -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

        <!-- Database migrations run from the API service -->
        <dependency>
            <groupId>org.flywaydb</groupId>
            <artifactId>flyway-core</artifactId>
        </dependency>
        <dependency>
            <groupId>org.flywaydb</groupId>
            <artifactId>flyway-database-postgresql</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!-- Produces an executable fat JAR -->
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

What does this POM include?

• signalforge-common brings in the shared entities and repositories.
• Flyway is only in the API module because only one service should run database migrations.
• spring-boot-maven-plugin packages this module as an executable JAR with an embedded server.

• Create the signalforge-worker/pom.xml file with the following content:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <groupId>com.signalforge</groupId>
        <artifactId>signalforge-engine</artifactId>
        <version>0.0.1-SNAPSHOT</version>
    </parent>

    <artifactId>signalforge-worker</artifactId>
    <name>SignalForge Worker</name>

    <dependencies>
        <!-- Shared entities, DTOs, and repositories -->
        <dependency>
            <groupId>com.signalforge</groupId>
            <artifactId>signalforge-common</artifactId>
            <version>${project.version}</version>
        </dependency>

        <!-- Minimal web stack for Actuator only -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

        <!-- Redis for rate limiting and deduplication -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

        <!-- Circuit breakers -->
        <dependency>
            <groupId>io.github.resilience4j</groupId>
            <artifactId>resilience4j-spring-boot3</artifactId>
        </dependency>

        <!-- Prometheus metrics -->
        <dependency>
            <groupId>io.micrometer</groupId>
            <artifactId>micrometer-registry-prometheus</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!-- Produces an executable fat JAR -->
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

What does the worker POM include?

• Redis for the RateLimiter and DeduplicationService.
• Resilience4j for per-endpoint circuit breakers.
• Micrometer with Prometheus registry for delivery metrics that Prometheus scrapes.
• No Flyway. The Worker reads the database but never runs migrations.

• Save all four pom.xml files.
• Validate the multi-module structure by running:

mvn validate

You should see all three modules listed in the Reactor Summary with BUILD SUCCESS.

Seeing a BUILD FAILURE?

Check that each child pom.xml is inside its module directory, not at the project root. The <module> names in the parent POM must match the directory names exactly.

help me debug my Maven multi-module validate failure

Now it is time to move your existing source code into the correct modules.

• Move entities to the common module. Copy the following files from src/main/java/com/signalforge/entity/ into signalforge-common/src/main/java/com/signalforge/common/entity/:
• Move all repository interfaces from src/main/java/com/signalforge/repository/ into signalforge-common/src/main/java/com/signalforge/common/repository/.
• Move all DTO classes from src/main/java/com/signalforge/dto/ into signalforge-common/src/main/java/com/signalforge/common/dto/.
• Copy V1__init_schema.sql from src/main/resources/db/migration/ into signalforge-common/src/main/resources/db/migration/.
• Move API-side classes to the API module. Copy the following into signalforge-api/src/main/java/com/signalforge/api/:
• Move worker-side classes to the Worker module. Copy the following into signalforge-worker/src/main/java/com/signalforge/worker/:
• Update the package declarations in every moved file to match the new directory structure.

Use IntelliJ refactoring

In IntelliJ IDEA, use **Refactor > Move** (F6) to move classes between modules. IntelliJ automatically updates the package declaration and fixes import statements across all files.

What about the old src/ directory?

After moving all classes, the original src/ directory at the project root should be empty. You can safely delete it once all code lives in the module directories.

Create application entry points for each service

Each service needs its own @SpringBootApplication class as an entry point. Because the entities and repositories now live in a different package (com.signalforge.common), Spring Boot's default component scanning will not find them automatically.

What are @EntityScan and @EnableJpaRepositories?

Spring Boot scans for @Entity classes and JPA repositories starting from the package where your main application class lives. When entities live in a separate module under a different base package, you need @EntityScan and @EnableJpaRepositories to point Spring at the right packages.

• Create the API service entry point at signalforge-api/src/main/java/com/signalforge/api/SignalForgeApiApplication.java with the following code:

package com.signalforge.api;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.domain.EntityScan;
import org.springframework.data.jpa.repository.config.EnableJpaRepositories;

// Scans for entities and repositories in the common module
@SpringBootApplication
@EntityScan(basePackages = "com.signalforge.common.entity")
@EnableJpaRepositories(basePackages = "com.signalforge.common.repository")
public class SignalForgeApiApplication {

    public static void main(String[] args) {
        SpringApplication.run(SignalForgeApiApplication.class, args);
    }
}

What does this class do?

• @EntityScan tells Spring Data JPA to find @Entity classes in com.signalforge.common.entity instead of only scanning the API module's own package.
• @EnableJpaRepositories tells Spring to find repository interfaces in the common module's com.signalforge.common.repository package.
• Without these annotations, Spring would throw Not a managed type errors at startup because it cannot find the entities.

• Create the API service configuration at signalforge-api/src/main/resources/application.yml with the following content:

spring:
  application:
    name: signalforge-api
  # Virtual Threads for high-concurrency request handling
  threads:
    virtual:
      enabled: true
  datasource:
    url: jdbc:postgresql://postgres:5432/signalforge
    username: signalforge
    password: signalforge
  jpa:
    hibernate:
      ddl-auto: validate
    open-in-view: false
  flyway:
    enabled: true
    locations: classpath:db/migration

server:
  port: 8080
  shutdown: graceful

management:
  endpoints:
    web:
      exposure:
        include: health, info, prometheus

What is different about this config?

This is a trimmed-down version of the original monolith config. It keeps the datasource, JPA, and Flyway settings because the API service handles migrations and event ingestion. Redis and Resilience4j config are not here because the API service does not perform deliveries.

• Create the Worker service entry point at signalforge-worker/src/main/java/com/signalforge/worker/SignalForgeWorkerApplication.java with the following code:

package com.signalforge.worker;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.domain.EntityScan;
import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
import org.springframework.scheduling.annotation.EnableScheduling;

// Enables the @Scheduled delivery worker loop
@SpringBootApplication
@EnableScheduling
@EntityScan(basePackages = "com.signalforge.common.entity")
@EnableJpaRepositories(basePackages = "com.signalforge.common.repository")
public class SignalForgeWorkerApplication {

    public static void main(String[] args) {
        SpringApplication.run(SignalForgeWorkerApplication.class, args);
    }
}

What does @EnableScheduling do here?

@EnableScheduling activates the @Scheduled annotation on DeliveryWorker. This annotation only appears on the Worker application class, not the API one. The API service does not claim or deliver events.

• Create the Worker service configuration at signalforge-worker/src/main/resources/application.yml with the following content:

spring:
  application:
    name: signalforge-worker
  # Virtual Threads for 10,000+ concurrent deliveries
  threads:
    virtual:
      enabled: true
  datasource:
    url: jdbc:postgresql://postgres:5432/signalforge
    username: signalforge
    password: signalforge
    hikari:
      maximum-pool-size: 30
  jpa:
    hibernate:
      ddl-auto: validate
    open-in-view: false
  flyway:
    # Migrations run from the API service only
    enabled: false
  data:
    redis:
      host: redis
      port: 6379

server:
  port: 8081
  shutdown: graceful

management:
  endpoints:
    web:
      exposure:
        include: health, prometheus

What is different about the worker config?

• flyway.enabled: false prevents the worker from running migrations. Only the API service manages the schema.
• hikari.maximum-pool-size: 30 caps database connections at 30. With Virtual Threads, thousands of virtual threads contend for these 30 connections, and HikariCP handles this correctly.
• Redis config is here because rate limiting and deduplication happen in the worker.
• The server port is 8081 so it does not conflict with the API service when running locally.

• Save both application classes and both application.yml files.
• Delete the old src/main/java/com/signalforge/SignalForgeApplication.java (the original monolith entry point) if it still exists.
• Compile the full project to verify all imports resolve by running:

mvn clean compile

You should see BUILD SUCCESS in the Reactor Summary for all three modules.

Seeing "package does not exist" or "cannot find symbol" errors?

These usually mean a moved class still has its old package declaration. Open the file and verify the package line matches the directory structure.

For example, Tenant.java in signalforge-common/src/main/java/com/signalforge/common/entity/ must declare package com.signalforge.common.entity; at the top.

Also update all import statements in files that reference moved classes. IntelliJ's **Optimize Imports** (Ctrl+Alt+O on Windows, Cmd+Option+O on macOS) fixes stale imports across the project.

help me fix compilation errors after restructuring my Maven multi-module project

Update Docker Compose and verify end-to-end delivery

The previous docker-compose.yml ran two copies of the same monolith. Now you need one container for the API service and a separate container for the Worker service.

• Create a Dockerfile for the API module at signalforge-api/Dockerfile with the following content:

# Stage 1: Build the entire multi-module project
FROM maven:3.9-eclipse-temurin-21 AS build
WORKDIR /app
COPY pom.xml .
COPY signalforge-common/ signalforge-common/
COPY signalforge-api/ signalforge-api/
COPY signalforge-worker/ signalforge-worker/
RUN mvn clean package -DskipTests -pl signalforge-api -am

# Stage 2: Run only the API JAR
FROM eclipse-temurin:21-jre-alpine
WORKDIR /app
COPY --from=build /app/signalforge-api/target/*.jar app.jar
EXPOSE 8080
ENTRYPOINT ["java", "-jar", "app.jar"]

What does this Dockerfile do?

• The build stage uses a full Maven image to compile the project. The -pl signalforge-api -am flag tells Maven to build the API module and its dependency (signalforge-common) only.
• The runtime stage uses a lightweight JRE Alpine image. Only the compiled JAR is copied over, keeping the image small and secure.

• Create a Dockerfile for the Worker module at signalforge-worker/Dockerfile with the following content:

# Stage 1: Build the entire multi-module project
FROM maven:3.9-eclipse-temurin-21 AS build
WORKDIR /app
COPY pom.xml .
COPY signalforge-common/ signalforge-common/
COPY signalforge-api/ signalforge-api/
COPY signalforge-worker/ signalforge-worker/
RUN mvn clean package -DskipTests -pl signalforge-worker -am

# Stage 2: Run only the Worker JAR
FROM eclipse-temurin:21-jre-alpine
WORKDIR /app
COPY --from=build /app/signalforge-worker/target/*.jar app.jar
EXPOSE 8081
ENTRYPOINT ["java", "-jar", "app.jar"]

• Replace the entire contents of docker-compose.yml at the project root with the following:

version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: signalforge
      POSTGRES_USER: signalforge
      POSTGRES_PASSWORD: signalforge
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ['CMD-SHELL', 'pg_isready -U signalforge']
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7
    ports:
      - '6379:6379'

  # API service: REST endpoints, event ingestion, Flyway migrations
  signalforge-api:
    build:
      context: .
      dockerfile: signalforge-api/Dockerfile
    ports:
      - '8080:8080'
    depends_on:
      postgres:
        condition: service_healthy
    environment:
      SPRING_DATASOURCE_URL: jdbc:postgresql://postgres:5432/signalforge

  # Worker service: delivery, retries, circuit breakers, metrics
  signalforge-worker:
    build:
      context: .
      dockerfile: signalforge-worker/Dockerfile
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_started
    environment:
      SPRING_DATASOURCE_URL: jdbc:postgresql://postgres:5432/signalforge
      SPRING_DATA_REDIS_HOST: redis

  # Nginx load balancer for the API service
  nginx:
    image: nginx:alpine
    ports:
      - '80:80'
    depends_on:
      - signalforge-api

  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - '9090:9090'

  grafana:
    image: grafana/grafana:latest
    ports:
      - '3000:3000'
    depends_on:
      - prometheus

volumes:
  postgres_data:

What changed from the previous Docker Compose?

• signalforge-1 and signalforge-2 (two monolith instances) are replaced by signalforge-api and signalforge-worker (two different services).
• Each service has its own Dockerfile, built from the project root context.
• PostgreSQL has a health check. The depends_on with condition: service_healthy ensures both services wait for PostgreSQL to be ready before starting.
• The worker does not expose a port to the host. Prometheus can still scrape it via the Docker network at signalforge-worker:8081.

• Update prometheus.yml to scrape the new service names. Replace the old signalforge-1:8080 and signalforge-2:8080 targets with signalforge-api:8080 and signalforge-worker:8081.
• Save all files.
• Tear down the old containers and rebuild by running:

docker compose down
docker compose up --build -d

The first build will take a few minutes because Maven downloads dependencies inside the Docker build stage.

• Verify both services are healthy by running:

docker compose ps

You should see signalforge-api and signalforge-worker both in the Up state.

• Confirm the API service Actuator is responding by running:

curl http://localhost:8080/actuator/health

You should see {"status":"UP"}.

Now run the end-to-end test to verify that events flow from the API service to the Worker service.

• Create a test tenant by running:

curl -s -X POST http://localhost:8080/api/v1/tenants \
  -H 'Content-Type: application/json' \
  -d '{"name": "test-tenant"}'

Note the id and apiKey from the response.

• Register a test endpoint for this tenant (replace the tenant ID with the one from the previous response):

curl -s -X POST http://localhost:8080/api/v1/tenants/[[TENANT_ID="your tenant UUID"]]/endpoints \
  -H 'Content-Type: application/json' \
  -d '{"url": "https://httpbin.org/post"}'

• Ingest a test event:

curl -s -X POST http://localhost:8080/api/v1/tenants/[[TENANT_ID="your tenant UUID"]]/events \
  -H 'Content-Type: application/json' \
  -d '{"event_type": "test.split", "payload": {"message": "hello from split services"}}'

• Wait 2-3 seconds for the Worker service to claim the delivery attempt, then check the delivery status:

curl -s http://localhost:8080/api/v1/tenants/[[TENANT_ID="your tenant UUID"]]/deliveries | python3 -m json.tool

You should see a delivery attempt with status DELIVERED and an HTTP status of 200. The API service ingested the event and the Worker service picked it up and delivered it.

Delivery still showing PENDING after 10+ seconds?

Check the worker logs by running docker compose logs signalforge-worker. If you see Not a managed type errors, double-check that @EntityScan and @EnableJpaRepositories point to com.signalforge.common.entity and com.signalforge.common.repository.

If the worker started but is not claiming work, verify @EnableScheduling is on SignalForgeWorkerApplication (not the API application class).

help me debug why my SignalForge Worker service is not claiming delivery attempts

Your monolith is now two independent services sharing a common library. Next up, you will add Kafka to decouple event ingestion from delivery fan-out, turning the Transactional Outbox into a relay that publishes events to a Kafka topic.

Kafka for Event Ingestion

Your API and worker services now run as separate containers, each focused on its own job. The API still writes delivery_attempts rows directly during event ingestion, though, so its transaction grows heavier with every endpoint a tenant registers.

In this step, you will add Redpanda (a Kafka-compatible broker) to your Docker stack and decouple ingestion from fan-out. The API service will write to an outbox table, a relay will publish to Kafka, and the worker will consume events and create delivery attempts independently.

In this step, get ready to:

• Add Redpanda and spring-kafka to the project.
• Build an outbox relay that publishes events to a Kafka topic.
• Build a Kafka consumer in the worker service that creates delivery attempts.

Add Redpanda and spring-kafka to the project

Redpanda is a Kafka-compatible streaming broker built as a single C++ binary. It includes a built-in Schema Registry and starts faster with less memory than a traditional Kafka cluster.

• Open docker-compose.yml and add the following Redpanda service below your existing services:

  # Kafka-compatible broker for event streaming
  redpanda:
    image: docker.redpanda.com/redpandadata/redpanda:v26.1.6
    container_name: redpanda
    command:
      - redpanda start
      - --smp 1
      - --memory 512M
      - --overprovisioned
      - --kafka-addr internal://0.0.0.0:9092,external://0.0.0.0:19092
      - --advertise-kafka-addr internal://redpanda:9092,external://localhost:19092
      - --schema-registry-addr internal://0.0.0.0:8081,external://0.0.0.0:18081
      - --mode dev-container
    ports:
      # Kafka API (external access from host)
      - "19092:19092"
      # Schema Registry (external access from host)
      - "18081:18081"
    volumes:
      - redpanda_data:/var/lib/redpanda/data
    healthcheck:
      test: ["CMD", "rpk", "cluster", "health"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 10s

What does this configuration do?

Redpanda exposes two listener addresses for the Kafka API. The internal address (redpanda:9092) is used by other Docker containers like your API and worker services. The external address (localhost:19092) lets you connect from your host machine.

The --mode dev-container flag disables fsync for faster startup. This is for development only.

• Also add redpanda_data to your volumes: section at the bottom of the file.
• Update your signalforge-api and signalforge-worker service definitions to add a depends_on entry for redpanda with condition: service_healthy.

Next, both modules need the Spring Kafka dependency so they can produce and consume messages.

• Open signalforge-api/pom.xml and add this dependency inside the <dependencies> block:

<!-- Kafka producer support for the outbox relay -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

• Open signalforge-worker/pom.xml and add the same dependency:

<!-- Kafka consumer support for event ingestion -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

Why no version number?

Spring Boot's parent BOM manages the spring-kafka version automatically. With Spring Boot 3.5.14 as your parent, it resolves the compatible spring-kafka 3.3.x version.

• Save both pom.xml files and reload Maven in IntelliJ (click the Maven refresh icon in the right sidebar, or press Ctrl+Shift+O).
• Restart Docker Compose to bring up Redpanda by running:

docker compose up -d

• Verify Redpanda is healthy by running:

docker compose exec redpanda rpk cluster health

You should see output showing the cluster is healthy with one broker and no issues.

Redpanda container not starting?

Make sure Docker Desktop has at least 4 GB of memory allocated. Redpanda requests 512 MB but shares resources with PostgreSQL, Redis, and your app containers.

Check container logs with docker compose logs redpanda to see if a port conflict is the cause. Port 19092 or 18081 might already be in use.

Help me troubleshoot my Redpanda Docker container.

Build the outbox relay in the API service

Right now, the API service creates delivery_attempt rows inside the same transaction that saves the event. You are going to replace that with an outbox table pattern.

The API writes an outbox row in the same transaction as the event. A separate relay component polls the outbox with SKIP LOCKED and publishes to Kafka. This guarantees zero event loss between your database write and Kafka publish.

Why not write directly to Kafka from the API?

If the API saves an event to PostgreSQL and then publishes to Kafka as two separate operations, either step could fail independently. The database write could succeed but the Kafka publish could fail, losing the event. This is the dual-write problem.

The Transactional Outbox pattern avoids this by writing both the event and the outbox row in a single database transaction. If the transaction succeeds, the relay is guaranteed to find the row and publish it. If it fails, nothing is written.

• Create the Flyway migration file V2__outbox_table.sql inside signalforge-common/src/main/resources/db/migration/ by right-clicking the migration folder and selecting New then File.
• Paste the following SQL into V2__outbox_table.sql:

-- Outbox table for reliable event publishing to Kafka
CREATE TABLE outbox_events (
  id          BIGSERIAL    PRIMARY KEY,
  event_id    UUID         NOT NULL REFERENCES events(id),
  tenant_id   UUID         NOT NULL,
  payload     JSONB        NOT NULL,
  status      VARCHAR(20)  NOT NULL DEFAULT 'PENDING',
  created_at  TIMESTAMPTZ  NOT NULL DEFAULT now()
);

-- Partial index so the outbox relay's SKIP LOCKED query hits an index scan
CREATE INDEX idx_outbox_pending ON outbox_events (id)
  WHERE status = 'PENDING';

What does this migration do?

• outbox_events stores a copy of each ingested event's payload, keyed to its event_id and tenant_id.
• The status column tracks whether the relay has published the row to Kafka (PENDING or PUBLISHED).
• The partial index on status = 'PENDING' ensures the relay's SKIP LOCKED query uses an index scan instead of scanning the full table.

• Save V2__outbox_table.sql.

Next, create the JPA entity that maps to this table.

• Create a new file OutboxEvent.java inside signalforge-common/src/main/java/com/signalforge/common/entity/.
• Paste the following entity code:

package com.signalforge.common.entity;

import jakarta.persistence.*;
import java.time.Instant;
import java.util.UUID;

// Maps to the outbox_events table created in V2
@Entity
@Table(name = "outbox_events")
public class OutboxEvent {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    // References the event that was ingested
    @Column(name = "event_id", nullable = false)
    private UUID eventId;

    // Tenant who owns this event
    @Column(name = "tenant_id", nullable = false)
    private UUID tenantId;

    // Full event payload as JSONB
    @Column(name = "payload", nullable = false, columnDefinition = "jsonb")
    private String payload;

    // PENDING until the relay publishes to Kafka, then PUBLISHED
    @Column(name = "status", nullable = false)
    private String status = "PENDING";

    @Column(name = "created_at", nullable = false)
    private Instant createdAt = Instant.now();

    public Long getId() { return id; }
    public void setId(Long id) { this.id = id; }

    public UUID getEventId() { return eventId; }
    public void setEventId(UUID eventId) { this.eventId = eventId; }

    public UUID getTenantId() { return tenantId; }
    public void setTenantId(UUID tenantId) { this.tenantId = tenantId; }

    public String getPayload() { return payload; }
    public void setPayload(String payload) { this.payload = payload; }

    public String getStatus() { return status; }
    public void setStatus(String status) { this.status = status; }

    public Instant getCreatedAt() { return createdAt; }
    public void setCreatedAt(Instant createdAt) { this.createdAt = createdAt; }
}

What does this entity do?

OutboxEvent mirrors the outbox_events table. The payload field holds a JSON string containing the event ID, tenant ID, event type, and the original event payload. This is the message body that gets published to Kafka.

• Save OutboxEvent.java.

Now you need to update EventService to write to the outbox instead of creating delivery attempts directly.

• Open signalforge-api/src/main/java/com/signalforge/api/service/EventService.java.
• In the ingestEvent() method, find the code that queries active endpoints and creates DeliveryAttempt rows. It looks like this:

// Create delivery attempts for all ACTIVE endpoints (Transactional Outbox)
List<Endpoint> endpoints = endpointRepository.findByTenantIdAndStatus(tenantId, "ACTIVE");
for (Endpoint endpoint : endpoints) {
    DeliveryAttempt attempt = new DeliveryAttempt();
    attempt.setEventId(event.getId());
    attempt.setEndpointId(endpoint.getId());
    attempt.setStatus("PENDING");
    attempt.setAttemptNumber(0);
    attempt.setNextRetryAt(Instant.now());
    deliveryAttemptRepository.save(attempt);
}

• Replace that entire block with the following outbox write:

// Write to outbox instead of creating delivery attempts directly
OutboxEvent outbox = new OutboxEvent();
outbox.setEventId(event.getId());
outbox.setTenantId(tenantId);
outbox.setPayload(buildOutboxPayload(event));
outboxEventRepository.save(outbox);

• Add a private helper method to EventService that builds the JSON payload:

// Serialize the event into a JSON string for the outbox
private String buildOutboxPayload(Event event) {
    try {
        return objectMapper.writeValueAsString(Map.of(
            "eventId", event.getId().toString(),
            "tenantId", event.getTenantId().toString(),
            "eventType", event.getEventType(),
            "payload", event.getPayload()
        ));
    } catch (Exception e) {
        throw new RuntimeException("Failed to serialize outbox payload", e);
    }
}

What changed in EventService?

The @Transactional method now saves the Event and an OutboxEvent in a single database transaction. It no longer queries endpoints or creates delivery attempts. That fan-out responsibility moves to the worker service.

• Make sure you have injected OutboxEventRepository and ObjectMapper into EventService via constructor injection.
• Create a simple OutboxEventRepository interface in signalforge-common extending JpaRepository<OutboxEvent, Long> if it does not already exist.
• Save EventService.java.

Now build the relay that polls the outbox and publishes to Kafka.

• Create a new package com.signalforge.api.outbox inside the signalforge-api module.
• Create OutboxRelay.java in that package and paste the following:

package com.signalforge.api.outbox;

import com.signalforge.common.entity.OutboxEvent;
import jakarta.persistence.EntityManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

@Component
public class OutboxRelay {

    private static final Logger log = LoggerFactory.getLogger(OutboxRelay.class);
    private static final String TOPIC = "signalforge.events.ingested";
    private static final int BATCH_SIZE = 20;

    private final EntityManager entityManager;
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OutboxRelay(EntityManager entityManager,
                       KafkaTemplate<String, String> kafkaTemplate) {
        this.entityManager = entityManager;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Scheduled(fixedDelay = 500)
    @Transactional
    public void relayOutboxEvents() {
        // Claim PENDING outbox rows using SKIP LOCKED to avoid contention
        @SuppressWarnings("unchecked")
        List<Long> claimedIds = entityManager.createNativeQuery("""
            WITH batch AS (
                SELECT id FROM outbox_events
                WHERE status = 'PENDING'
                ORDER BY id
                FOR UPDATE SKIP LOCKED
                LIMIT :batchSize
            )
            UPDATE outbox_events
            SET status = 'PUBLISHING'
            FROM batch
            WHERE outbox_events.id = batch.id
            RETURNING outbox_events.id
            """)
            .setParameter("batchSize", BATCH_SIZE)
            .getResultList();

        if (claimedIds.isEmpty()) {
            return;
        }

        // Load claimed rows and publish each to Kafka
        List<OutboxEvent> events = entityManager
            .createQuery("SELECT o FROM OutboxEvent o WHERE o.id IN :ids",
                OutboxEvent.class)
            .setParameter("ids", claimedIds)
            .getResultList();

        for (OutboxEvent outbox : events) {
            // Publish with tenant ID as the key for partition affinity
            kafkaTemplate.send(TOPIC,
                outbox.getTenantId().toString(),
                outbox.getPayload());

            // Mark as published so the relay doesn't pick it up again
            outbox.setStatus("PUBLISHED");

            log.info("Relayed event={} tenant={} to topic={}",
                outbox.getEventId(), outbox.getTenantId(), TOPIC);
        }
    }
}

What does the OutboxRelay do?

• Every 500 milliseconds, the @Scheduled method runs and claims up to 20 PENDING outbox rows using SELECT ... FOR UPDATE SKIP LOCKED. This is the same contention-free claiming pattern your delivery worker uses.
• For each claimed row, it publishes the payload to the signalforge.events.ingested Kafka topic using KafkaTemplate. The tenant ID is used as the message key so all events for one tenant land in the same partition.
• After publishing, it sets the row's status to PUBLISHED so the relay never picks it up again.

• Make sure @EnableScheduling is present on your SignalForgeApiApplication class. The API service needs scheduling enabled for the relay's @Scheduled method to run.

Finally, configure the Kafka producer settings for the API service.

• Open signalforge-api/src/main/resources/application.yml and add the following Kafka section under spring::

  # Kafka producer config for the outbox relay
  kafka:
    bootstrap-servers: redpanda:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

What does this configuration do?

bootstrap-servers: redpanda:9092 points the producer at Redpanda's internal Docker address. Spring Boot auto-configures a KafkaTemplate bean using these settings, which the OutboxRelay injects via constructor.

• Save application.yml and OutboxRelay.java.

Compilation errors in OutboxRelay?

If IntelliJ cannot resolve KafkaTemplate, make sure you reloaded Maven after adding the spring-kafka dependency. Click the Maven refresh icon in the right sidebar.

If OutboxEvent cannot be resolved, confirm it is in the signalforge-common module and that signalforge-api depends on signalforge-common in its pom.xml.

Help me fix compilation errors in my OutboxRelay.

✔️ Awesome, I've got everything!

Make sure you have saved all files: V2__outbox_table.sql, OutboxEvent.java, EventService.java, OutboxRelay.java, and application.yml.

ⓧ I'd like to double check the full code

Here is the complete OutboxRelay.java:

package com.signalforge.api.outbox;

import com.signalforge.common.entity.OutboxEvent;
import jakarta.persistence.EntityManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

@Component
public class OutboxRelay {

    private static final Logger log = LoggerFactory.getLogger(OutboxRelay.class);
    private static final String TOPIC = "signalforge.events.ingested";
    private static final int BATCH_SIZE = 20;

    private final EntityManager entityManager;
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OutboxRelay(EntityManager entityManager,
                       KafkaTemplate<String, String> kafkaTemplate) {
        this.entityManager = entityManager;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Scheduled(fixedDelay = 500)
    @Transactional
    public void relayOutboxEvents() {
        // Claim PENDING outbox rows using SKIP LOCKED to avoid contention
        @SuppressWarnings("unchecked")
        List<Long> claimedIds = entityManager.createNativeQuery("""
            WITH batch AS (
                SELECT id FROM outbox_events
                WHERE status = 'PENDING'
                ORDER BY id
                FOR UPDATE SKIP LOCKED
                LIMIT :batchSize
            )
            UPDATE outbox_events
            SET status = 'PUBLISHING'
            FROM batch
            WHERE outbox_events.id = batch.id
            RETURNING outbox_events.id
            """)
            .setParameter("batchSize", BATCH_SIZE)
            .getResultList();

        if (claimedIds.isEmpty()) {
            return;
        }

        // Load claimed rows and publish each to Kafka
        List<OutboxEvent> events = entityManager
            .createQuery("SELECT o FROM OutboxEvent o WHERE o.id IN :ids",
                OutboxEvent.class)
            .setParameter("ids", claimedIds)
            .getResultList();

        for (OutboxEvent outbox : events) {
            // Publish with tenant ID as the key for partition affinity
            kafkaTemplate.send(TOPIC,
                outbox.getTenantId().toString(),
                outbox.getPayload());

            // Mark as published so the relay doesn't pick it up again
            outbox.setStatus("PUBLISHED");

            log.info("Relayed event={} tenant={} to topic={}",
                outbox.getEventId(), outbox.getTenantId(), TOPIC);
        }
    }
}

And here is the complete V2__outbox_table.sql:

-- Outbox table for reliable event publishing to Kafka
CREATE TABLE outbox_events (
  id          BIGSERIAL    PRIMARY KEY,
  event_id    UUID         NOT NULL REFERENCES events(id),
  tenant_id   UUID         NOT NULL,
  payload     JSONB        NOT NULL,
  status      VARCHAR(20)  NOT NULL DEFAULT 'PENDING',
  created_at  TIMESTAMPTZ  NOT NULL DEFAULT now()
);

-- Partial index so the outbox relay's SKIP LOCKED query hits an index scan
CREATE INDEX idx_outbox_pending ON outbox_events (id)
  WHERE status = 'PENDING';

Build the Kafka consumer in the worker service

The worker service now needs to consume events from Kafka and create the delivery attempt rows that the API service no longer creates. This is the fan-out step that turns one ingested event into multiple delivery tasks.

• Open signalforge-worker/src/main/resources/application.yml and add the following Kafka consumer section under spring::

  # Kafka consumer config for event ingestion
  kafka:
    bootstrap-servers: redpanda:9092
    consumer:
      group-id: signalforge-worker
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

What do these consumer settings do?

• group-id: signalforge-worker assigns all worker instances to the same consumer group. Kafka distributes partitions across group members, so two worker instances split the work.
• auto-offset-reset: earliest tells the consumer to start reading from the beginning of the topic if no offset has been committed yet. This means no events are missed on first startup.

• Save application.yml.

Now create the consumer class that listens for ingested events and creates delivery attempts.

• Create a new package com.signalforge.worker.kafka inside the signalforge-worker module.
• Create EventIngestionConsumer.java in that package and paste the following:

package com.signalforge.worker.kafka;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.signalforge.common.entity.DeliveryAttempt;
import com.signalforge.common.entity.Endpoint;
import com.signalforge.common.repository.DeliveryAttemptRepository;
import com.signalforge.common.repository.EndpointRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.List;
import java.util.UUID;

@Component
public class EventIngestionConsumer {

    private static final Logger log =
        LoggerFactory.getLogger(EventIngestionConsumer.class);

    private final EndpointRepository endpointRepository;
    private final DeliveryAttemptRepository deliveryAttemptRepository;
    private final ObjectMapper objectMapper;

    public EventIngestionConsumer(
            EndpointRepository endpointRepository,
            DeliveryAttemptRepository deliveryAttemptRepository,
            ObjectMapper objectMapper) {
        this.endpointRepository = endpointRepository;
        this.deliveryAttemptRepository = deliveryAttemptRepository;
        this.objectMapper = objectMapper;
    }

    @KafkaListener(
        topics = "signalforge.events.ingested",
        groupId = "signalforge-worker"
    )
    @Transactional
    public void onEventIngested(String message) {
        try {
            // Parse the outbox payload published by the API relay
            JsonNode node = objectMapper.readTree(message);
            UUID eventId = UUID.fromString(node.get("eventId").asText());
            UUID tenantId = UUID.fromString(node.get("tenantId").asText());

            // Find all ACTIVE endpoints for this tenant
            List<Endpoint> endpoints = endpointRepository
                .findByTenantIdAndStatus(tenantId, "ACTIVE");

            // Create a delivery attempt for each active endpoint
            for (Endpoint endpoint : endpoints) {
                DeliveryAttempt attempt = new DeliveryAttempt();
                attempt.setEventId(eventId);
                attempt.setEndpointId(endpoint.getId());
                attempt.setStatus("PENDING");
                attempt.setAttemptNumber(0);
                attempt.setNextRetryAt(Instant.now());
                deliveryAttemptRepository.save(attempt);
            }

            log.info("Created {} delivery attempts for event={} tenant={}",
                endpoints.size(), eventId, tenantId);

        } catch (Exception e) {
            log.error("Failed to process ingested event", e);
            throw new RuntimeException(
                "Failed to process event message", e);
        }
    }
}

What does EventIngestionConsumer do?

• @KafkaListener subscribes to the signalforge.events.ingested topic. Spring Boot auto-configures the listener container factory using the consumer settings from application.yml.
• For each message, it parses the JSON payload to extract the eventId and tenantId, then queries for all ACTIVE endpoints belonging to that tenant.
• It creates a DeliveryAttempt row for each endpoint with status=PENDING. This is the same fan-out logic that used to live in the API service.
• If parsing or saving fails, the exception is rethrown so the Kafka consumer retries the message.

• Save EventIngestionConsumer.java.

Time to test the full pipeline end to end.

• Rebuild and restart both services by running:

docker compose up -d --build

• Wait for all containers to become healthy, then POST an event to your API service by running:

curl -X POST http://localhost:8080/api/v1/tenants/{your-tenant-id}/events \
  -H "Content-Type: application/json" \
  -d '{"eventType": "order.created", "payload": {"orderId": "12345"}}'

• Check the API service logs for a relay message by running:

docker compose logs signalforge-api | grep "Relayed event"

You should see a log line containing Relayed event= followed by the event UUID and the topic name.

• Check the worker service logs for the fan-out by running:

docker compose logs signalforge-worker | grep "Created.*delivery attempts"

You should see a log line showing the number of delivery attempts created for the event. If the tenant has two active endpoints, the log shows Created 2 delivery attempts.

No relay or consumer log messages?

If you see no relay messages, check that the API service has @EnableScheduling on its main application class. Without it, the @Scheduled method in OutboxRelay never runs.

If the relay publishes but the worker does not consume, verify that both services use redpanda:9092 as the bootstrap server (not localhost:19092, which is the external address for your host machine).

You can verify the Kafka topic exists by running:

docker compose exec redpanda rpk topic list

You should see signalforge.events.ingested in the list. If it is missing, the relay has not published any messages yet.

Help me debug my Kafka producer and consumer setup.

✔️ Awesome, I've got everything!

Make sure you have saved EventIngestionConsumer.java and the worker's application.yml.

ⓧ I'd like to double check the full code

Here is the complete EventIngestionConsumer.java:

package com.signalforge.worker.kafka;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.signalforge.common.entity.DeliveryAttempt;
import com.signalforge.common.entity.Endpoint;
import com.signalforge.common.repository.DeliveryAttemptRepository;
import com.signalforge.common.repository.EndpointRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.List;
import java.util.UUID;

@Component
public class EventIngestionConsumer {

    private static final Logger log =
        LoggerFactory.getLogger(EventIngestionConsumer.class);

    private final EndpointRepository endpointRepository;
    private final DeliveryAttemptRepository deliveryAttemptRepository;
    private final ObjectMapper objectMapper;

    public EventIngestionConsumer(
            EndpointRepository endpointRepository,
            DeliveryAttemptRepository deliveryAttemptRepository,
            ObjectMapper objectMapper) {
        this.endpointRepository = endpointRepository;
        this.deliveryAttemptRepository = deliveryAttemptRepository;
        this.objectMapper = objectMapper;
    }

    @KafkaListener(
        topics = "signalforge.events.ingested",
        groupId = "signalforge-worker"
    )
    @Transactional
    public void onEventIngested(String message) {
        try {
            // Parse the outbox payload published by the API relay
            JsonNode node = objectMapper.readTree(message);
            UUID eventId = UUID.fromString(node.get("eventId").asText());
            UUID tenantId = UUID.fromString(node.get("tenantId").asText());

            // Find all ACTIVE endpoints for this tenant
            List<Endpoint> endpoints = endpointRepository
                .findByTenantIdAndStatus(tenantId, "ACTIVE");

            // Create a delivery attempt for each active endpoint
            for (Endpoint endpoint : endpoints) {
                DeliveryAttempt attempt = new DeliveryAttempt();
                attempt.setEventId(eventId);
                attempt.setEndpointId(endpoint.getId());
                attempt.setStatus("PENDING");
                attempt.setAttemptNumber(0);
                attempt.setNextRetryAt(Instant.now());
                deliveryAttemptRepository.save(attempt);
            }

            log.info("Created {} delivery attempts for event={} tenant={}",
                endpoints.size(), eventId, tenantId);

        } catch (Exception e) {
            log.error("Failed to process ingested event", e);
            throw new RuntimeException(
                "Failed to process event message", e);
        }
    }
}

Event ingestion and delivery fan-out are now fully decoupled. The API service can accept thousands of events per second without worrying about how many endpoints each tenant has. Next up, you will build a delivery audit trail and dead-letter topic to give tenants full visibility into what happened to each event.

Delivery Audit Trail and Dead-Letter Topic

Your worker service now consumes events from Kafka and delivers them through the full resilience stack.

But right now, delivery outcomes vanish into the void. If a tenant asks "did my event get delivered?", you have no real-time record to show them. In this step, you'll build a delivery audit trail by publishing every delivery outcome to Kafka. You'll also route dead-lettered events to their own topic and expose the full delivery history through a REST API.

In this step, get ready to:

• Publish a delivery audit event to Kafka after every delivery attempt.
• Route dead-lettered events to a dedicated Kafka topic for operational alerting.
• Build a delivery history API that tenants can query to see exactly what happened to their events.

Create the delivery audit event DTO

Every delivery attempt produces an outcome. The worker needs a structured way to represent that outcome before publishing it to Kafka.

A Java record is perfect for this. It's an immutable data carrier that Kafka's JSON serializer can convert directly into a message payload.

• In the signalforge-common module, create a new file at src/main/java/com/signalforge/common/dto/DeliveryAuditEvent.java.
• Copy the following code into the file:

package com.signalforge.common.dto;

import java.time.Instant;
import java.util.UUID;

// Immutable record representing a delivery attempt outcome
public record DeliveryAuditEvent(
    UUID eventId,
    UUID endpointId,
    int attemptNumber,
    String status,
    Integer httpStatus,
    long durationMs,
    Instant timestamp
) {}

What does this record hold?

Each field maps to a piece of delivery telemetry that consumers of the audit trail will need.

• eventId and endpointId identify which event was delivered to which endpoint.
• status and httpStatus capture the delivery outcome and the HTTP response code from the target endpoint.
• durationMs records how long the HTTP call took, which feeds into latency metrics.
• Integer httpStatus is nullable because a connection timeout or DNS failure produces no HTTP status code at all.

• Save DeliveryAuditEvent.java.

Configure the worker's Kafka producer

The worker service currently only consumes from Kafka. To publish audit events, it needs producer configuration that tells Spring Kafka how to serialize your DeliveryAuditEvent records into JSON messages.

• Open signalforge-worker/src/main/resources/application.yml.
• Find the existing spring.kafka section.
• Add the following producer block underneath the existing consumer configuration:

    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

What does this configuration do?

The key-serializer converts the String message key (your event ID) into bytes. The JsonSerializer converts your DeliveryAuditEvent record into a JSON byte payload. Spring Boot auto-configures a KafkaTemplate bean using these serializers.

• Save application.yml.

Update the delivery worker to publish audit and dead-letter events

With the DTO and producer config in place, the delivery worker can now publish a structured audit event after every delivery attempt. It also needs to route events that have exhausted all retries to a dedicated dead-letter topic.

• Open signalforge-worker/src/main/java/com/signalforge/worker/DeliveryWorker.java.
• Add the following import statements at the top of the file:

import com.signalforge.common.dto.DeliveryAuditEvent;
import org.springframework.kafka.core.KafkaTemplate;
import java.time.Instant;

• Add a KafkaTemplate field alongside your existing dependencies, and include it in the constructor:

// Kafka producer for publishing audit and dead-letter events
private final KafkaTemplate<String, DeliveryAuditEvent> auditKafkaTemplate;

What's happening here?

Spring's auto-configured KafkaTemplate bean is injected through the constructor. The generic type <String, DeliveryAuditEvent> tells the compiler that message keys are strings (event IDs) and values are your audit records.

• Update the constructor to accept the new auditKafkaTemplate parameter and assign it to the field.

Now add the method that builds and publishes the audit event. This single method handles both the audit trail and the dead-letter routing.

• Add the following publishAuditEvent method to the DeliveryWorker class:

private void publishAuditEvent(DeliveryAttempt attempt, long durationMs) {
    // Build an immutable audit record from the delivery attempt outcome
    DeliveryAuditEvent auditEvent = new DeliveryAuditEvent(
        attempt.getEvent().getId(),
        attempt.getEndpoint().getId(),
        attempt.getAttemptNumber(),
        attempt.getStatus().name(),
        attempt.getHttpStatus(),
        durationMs,
        Instant.now()
    );

    // Publish to the audit topic for every delivery attempt
    auditKafkaTemplate.send("signalforge.delivery.audit",
        auditEvent.eventId().toString(), auditEvent);

    // Also publish to the dead-letter topic when all retries are exhausted
    if ("DEAD".equals(auditEvent.status())) {
        auditKafkaTemplate.send("signalforge.delivery.deadletter",
            auditEvent.eventId().toString(), auditEvent);
    }
}

What does this method do?

Every delivery attempt, whether it succeeds or fails, gets published to the signalforge.delivery.audit topic. This creates a real-time, immutable record of every HTTP callback your system makes.

The dead-letter check adds a second publish to signalforge.delivery.deadletter only when status is DEAD. A future consumer on this topic could alert the tenant or trigger manual review.

Using the event ID as the Kafka message key ensures that all audit events for the same event land in the same partition. This preserves ordering per event.

The final piece is wiring this method into the existing delivery logic.

• Find the section of your delivery loop where you make the HTTP POST call to the endpoint and record the duration.
• After you update the DeliveryAttempt status (whether to DELIVERED, FAILED, or DEAD) and save it to the database, add a call to publishAuditEvent:

// Publish the delivery outcome to the audit trail
publishAuditEvent(attempt, durationMs);

Place it after the database save

The audit publish should happen after you save the updated DeliveryAttempt to the database. This way, the database is your source of truth. If the Kafka publish fails, the delivery_attempts table still has the correct status.

• Save DeliveryWorker.java.
• Rebuild the worker module in IntelliJ by clicking Build in the top menu bar, then Build Project.
• Confirm the build succeeds with no compilation errors in the Build output panel at the bottom of IntelliJ.

Seeing a compilation error?

• If IntelliJ cannot resolve DeliveryAuditEvent, make sure you created the file in the signalforge-common module (not the worker module) and that the package declaration reads com.signalforge.common.dto.
• If KafkaTemplate cannot be resolved, confirm that spring-kafka is listed as a dependency in signalforge-worker/pom.xml. It was added in the previous step.
• Check that your constructor assigns the auditKafkaTemplate parameter to the field. A missing assignment causes a NullPointerException at runtime, not a compile error.

Still stuck? Ask for help.

Why two separate Kafka topics?

The audit topic (signalforge.delivery.audit) is a firehose. Every single delivery attempt lands here. Consumers of this topic build dashboards, analytics, and searchable delivery logs.

The dead-letter topic (signalforge.delivery.deadletter) is a targeted signal. Only permanently failed deliveries land here. Consumers of this topic send alerts, trigger manual review workflows, or notify tenants that their endpoint is unreachable.

Build the delivery history API

Tenants need a way to see exactly what happened to their events. The delivery_attempts table already holds a complete record of every delivery outcome. All you need is a REST endpoint that queries it.

This is the equivalent of Svix's delivery log portal, where customers can see attempt history, HTTP status codes, and timestamps for every event they sent.

• First, add a query method to your DeliveryAttemptRepository. Open the repository interface in the signalforge-common module and add this method signature:

// Find all delivery attempts for a specific event, ordered by creation time
List<DeliveryAttempt> findByEventIdOrderByCreatedAtAsc(UUID eventId);

What does this query do?

Spring Data JPA derives the SQL query from the method name. findByEventId becomes a WHERE event_id = ? clause. OrderByCreatedAtAsc sorts results chronologically so tenants see the delivery timeline in order.

• Save the repository file.

Now create the controller that exposes this data to tenants.

• In the signalforge-api module, create a new file at src/main/java/com/signalforge/api/controller/AuditController.java.
• Copy the following code into the file:

package com.signalforge.api.controller;

import com.signalforge.common.entity.DeliveryAttempt;
import com.signalforge.common.repository.DeliveryAttemptRepository;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.List;
import java.util.UUID;

@RestController
@RequestMapping("/api/v1/tenants/{tenantId}/events/{eventId}/deliveries")
public class AuditController {

    private final DeliveryAttemptRepository deliveryAttemptRepository;

    public AuditController(DeliveryAttemptRepository deliveryAttemptRepository) {
        this.deliveryAttemptRepository = deliveryAttemptRepository;
    }

    // Returns the full delivery history for a specific event
    @GetMapping
    public ResponseEntity<List<DeliveryAttempt>> getDeliveryHistory(
            @PathVariable UUID tenantId,
            @PathVariable UUID eventId) {
        List<DeliveryAttempt> history =
            deliveryAttemptRepository.findByEventIdOrderByCreatedAtAsc(eventId);
        return ResponseEntity.ok(history);
    }
}

What does this controller do?

• The @RequestMapping path nests under both tenant and event IDs, following the same REST resource hierarchy as your other controllers.
• The query reads directly from the delivery_attempts table. Both the API and worker services share the same PostgreSQL database, so the API can serve delivery history that the worker writes.
• The response includes every delivery attempt for the event: the status, HTTP response code, attempt number, and timestamps. This gives tenants full visibility into why a delivery succeeded or failed.

Extra for Experts: why not read from the Kafka audit topic instead?

The Kafka audit topic is optimized for streaming consumers, not random-access queries. Querying Kafka by event ID would require consuming the entire topic or maintaining a consumer-side index.

The delivery_attempts table is already indexed and supports the exact query pattern this API needs. The Kafka topics serve a different purpose: real-time event streaming for dashboards, alerting pipelines, and other downstream consumers.

• Save AuditController.java.
• Restart the API and worker services by running:

docker compose up -d --build

• Wait for both services to start, then test the audit API by running:

curl http://localhost:8080/api/v1/tenants/[[TENANT_ID="your tenant UUID"]]/events/[[EVENT_ID="your event UUID"]]/deliveries

• You should see a JSON array of delivery attempt objects. Each object contains the status, attemptNumber, httpStatus, and createdAt fields.

Getting an empty array or 404?

• An empty array means no delivery attempts exist for that event ID. Make sure you use UUIDs from a tenant and event you actually created. You can find valid IDs by querying the events table directly.
• If you get a 404, confirm the controller's @RequestMapping path exactly matches /api/v1/tenants/{tenantId}/events/{eventId}/deliveries.
• If you see a serialization error, check that DeliveryAttempt entity fields have proper getters. Jackson needs getter methods to serialize the entity to JSON.

Need help debugging the response?

To verify the Kafka topics are receiving messages, you can peek at the audit topic using Redpanda's built-in tooling.

• List the Kafka topics by running:

docker exec -it redpanda rpk topic list

• You should see signalforge.delivery.audit and signalforge.delivery.deadletter in the list alongside the existing signalforge.events.ingested topic.
• Consume a few messages from the audit topic to confirm events are flowing by running:

docker exec -it redpanda rpk topic consume signalforge.delivery.audit --num 5

• You should see JSON messages containing eventId, endpointId, status, and durationMs fields. If no events have been delivered yet, ingest a test event first through the API.

Not seeing the audit topics?

• Kafka auto-creates topics on first publish by default. If the topics don't exist yet, trigger a delivery by ingesting an event through the API. The worker delivers it and publishes the audit event, which creates the topic.
• If you see a serialization error in the worker logs, confirm the value-serializer in application.yml is set to org.springframework.kafka.support.serializer.JsonSerializer.

Still not working? Ask for help.

✔️ Awesome, I've got everything!

Double check that you've saved all three files: DeliveryAuditEvent.java, DeliveryWorker.java, and AuditController.java.

ⓧ I'd like to double check the full code

Here is the full code for each new or modified file in this step.

package com.signalforge.common.dto;

import java.time.Instant;
import java.util.UUID;

// Immutable record representing a delivery attempt outcome
public record DeliveryAuditEvent(
    UUID eventId,
    UUID endpointId,
    int attemptNumber,
    String status,
    Integer httpStatus,
    long durationMs,
    Instant timestamp
) {}

package com.signalforge.api.controller;

import com.signalforge.common.entity.DeliveryAttempt;
import com.signalforge.common.repository.DeliveryAttemptRepository;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.List;
import java.util.UUID;

@RestController
@RequestMapping("/api/v1/tenants/{tenantId}/events/{eventId}/deliveries")
public class AuditController {

    private final DeliveryAttemptRepository deliveryAttemptRepository;

    public AuditController(DeliveryAttemptRepository deliveryAttemptRepository) {
        this.deliveryAttemptRepository = deliveryAttemptRepository;
    }

    // Returns the full delivery history for a specific event
    @GetMapping
    public ResponseEntity<List<DeliveryAttempt>> getDeliveryHistory(
            @PathVariable UUID tenantId,
            @PathVariable UUID eventId) {
        List<DeliveryAttempt> history =
            deliveryAttemptRepository.findByEventIdOrderByCreatedAtAsc(eventId);
        return ResponseEntity.ok(history);
    }
}

For DeliveryWorker.java, confirm you have these additions:

• Imports for DeliveryAuditEvent, KafkaTemplate, and Instant.
• A KafkaTemplate<String, DeliveryAuditEvent> auditKafkaTemplate field assigned in the constructor.
• The publishAuditEvent method that publishes to signalforge.delivery.audit on every attempt and to signalforge.delivery.deadletter when status is DEAD.
• A publishAuditEvent(attempt, durationMs) call after every database save in the delivery loop.

// --- Key additions to DeliveryWorker.java ---

// New field
private final KafkaTemplate<String, DeliveryAuditEvent> auditKafkaTemplate;

// New method
private void publishAuditEvent(DeliveryAttempt attempt, long durationMs) {
    DeliveryAuditEvent auditEvent = new DeliveryAuditEvent(
        attempt.getEvent().getId(),
        attempt.getEndpoint().getId(),
        attempt.getAttemptNumber(),
        attempt.getStatus().name(),
        attempt.getHttpStatus(),
        durationMs,
        Instant.now()
    );

    // Publish to the audit trail topic after every delivery attempt
    auditKafkaTemplate.send("signalforge.delivery.audit",
        auditEvent.eventId().toString(), auditEvent);

    // Also publish to the dead-letter topic when all retries are exhausted
    if ("DEAD".equals(auditEvent.status())) {
        auditKafkaTemplate.send("signalforge.delivery.deadletter",
            auditEvent.eventId().toString(), auditEvent);
    }
}

// Call site (inside the delivery loop, after saving the attempt to the database)
publishAuditEvent(attempt, durationMs);

Your event delivery engine now has a complete audit trail. Every delivery attempt is published to Kafka in real time, dead-lettered events are routed to their own topic for alerting, and tenants can query the full delivery history through a REST API. Next up, you'll tie everything together with a full Docker Compose orchestration and project documentation.

Full Docker Compose Orchestration and Documentation

In the previous step, you built a delivery audit trail and dead-letter topic that give tenants full visibility into every delivery attempt.

Your system now has seven infrastructure services and two application services spread across separate containers. Without proper startup ordering, containers crash because they try to connect to services that are not ready yet. In this step, you will finalize your Docker Compose orchestration with health checks and dependency ordering, then document the entire system with a README.md and Architecture Decision Records.

In this step, get ready to:

• Finalize your Docker Compose file with health checks and dependency ordering so every service starts in the correct sequence.
• Write a comprehensive README with an architecture diagram, setup instructions, and a full API reference.
• Create Architecture Decision Records documenting the five key design choices you made throughout this project.

Finalize the Docker Compose file with health checks

Your existing docker-compose.yml starts all the services, but it has no way to guarantee startup order. Health checks let Docker verify that a container is actually ready to accept connections before starting containers that depend on it.

What are health checks in Docker Compose?

A healthcheck runs a command inside the container at a regular interval. If the command succeeds, Docker marks the container as healthy.

When you combine health checks with depends_on: condition: service_healthy, Docker waits for the dependency to become healthy before starting the dependent service. This prevents your Spring Boot apps from crashing because PostgreSQL or Kafka was not ready yet.

• Open docker-compose.yml in the root of your signalforge-engine project.
• Replace the entire contents of the file with the infrastructure services section below. This is the first half of the finalized compose file, covering PostgreSQL, Redis, and Redpanda:

services:

  # === Infrastructure Services ===

  postgres:
    image: postgres:16
    container_name: signalforge-postgres
    environment:
      POSTGRES_DB: signalforge
      POSTGRES_USER: signalforge
      POSTGRES_PASSWORD: signalforge
    ports:
      - "5432:5432"
    volumes:
      - postgres-data:/var/lib/postgresql/data
    # pg_isready confirms PostgreSQL accepts connections
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U signalforge -d signalforge"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7
    container_name: signalforge-redis
    ports:
      - "6379:6379"
    # PING returns PONG when Redis is ready
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 5s
      retries: 5

  redpanda:
    image: redpandadata/redpanda:latest
    container_name: signalforge-redpanda
    command:
      - redpanda start
      - --smp 1
      - --memory 512M
      - --overprovisioned
      - --kafka-addr internal://0.0.0.0:9092,external://0.0.0.0:19092
      - --advertise-kafka-addr internal://redpanda:9092,external://localhost:19092
    ports:
      - "19092:19092"
      - "9644:9644"
    # rpk cluster health returns healthy when the broker is ready
    healthcheck:
      test: ["CMD-SHELL", "rpk cluster health | grep -E 'Healthy:.+true' || exit 1"]
      interval: 15s
      timeout: 3s
      retries: 10

What do these health checks do?

• pg_isready is a built-in PostgreSQL utility that checks whether the server is accepting connections on the given database.
• redis-cli ping sends a PING command to Redis and expects PONG back.
• rpk cluster health queries Redpanda's cluster status. The grep filter ensures the broker reports itself as truly healthy before dependents start.

• Add the monitoring services, application services, and volumes section directly below the Redpanda service. Copy the following into your docker-compose.yml after the Redpanda block:

  # === Monitoring Services ===

  prometheus:
    image: prom/prometheus:latest
    container_name: signalforge-prometheus
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"
    depends_on:
      signalforge-api:
        condition: service_healthy
      signalforge-worker:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:9090/-/healthy"]
      interval: 10s
      timeout: 5s
      retries: 5

  grafana:
    image: grafana/grafana:latest
    container_name: signalforge-grafana
    ports:
      - "3000:3000"
    environment:
      GF_SECURITY_ADMIN_PASSWORD: admin
    depends_on:
      prometheus:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:3000/api/health"]
      interval: 10s
      timeout: 5s
      retries: 5

  # === Application Services ===

  signalforge-api:
    build:
      context: .
      dockerfile: signalforge-api/Dockerfile
    container_name: signalforge-api
    ports:
      - "8080:8080"
    environment:
      SPRING_DATASOURCE_URL: jdbc:postgresql://postgres:5432/signalforge
      SPRING_DATASOURCE_USERNAME: signalforge
      SPRING_DATASOURCE_PASSWORD: signalforge
      SPRING_DATA_REDIS_HOST: redis
      SPRING_KAFKA_BOOTSTRAP_SERVERS: redpanda:9092
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy
      redpanda:
        condition: service_healthy
    # start_period gives Spring Boot time to initialize
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:8080/actuator/health"]
      interval: 10s
      timeout: 5s
      retries: 10
      start_period: 30s

  signalforge-worker:
    build:
      context: .
      dockerfile: signalforge-worker/Dockerfile
    container_name: signalforge-worker
    environment:
      SPRING_DATASOURCE_URL: jdbc:postgresql://postgres:5432/signalforge
      SPRING_DATASOURCE_USERNAME: signalforge
      SPRING_DATASOURCE_PASSWORD: signalforge
      SPRING_DATA_REDIS_HOST: redis
      SPRING_KAFKA_BOOTSTRAP_SERVERS: redpanda:9092
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy
      redpanda:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:8081/actuator/health"]
      interval: 10s
      timeout: 5s
      retries: 10
      start_period: 30s

  nginx:
    image: nginx:alpine
    container_name: signalforge-nginx
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      signalforge-api:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:80/health"]
      interval: 10s
      timeout: 5s
      retries: 5

# === Persistent Volumes ===

volumes:
  postgres-data:

What does this configuration do?

• depends_on: condition: service_healthy tells Docker to wait until the dependency's health check passes before starting the dependent container.
• start_period: 30s on the Spring Boot services gives the JVM time to initialize before Docker starts counting health check failures.
• Prometheus depends on both app services because it needs their /actuator/prometheus endpoints to be available before scraping begins.
• Grafana depends on Prometheus because its dashboards query Prometheus as a data source.

Prometheus needs its own configuration file to know which services to scrape.

• Create a file called prometheus.yml in the root of your signalforge-engine project (next to docker-compose.yml) with the following content:

# Scrape both SignalForge services every 5 seconds
global:
  scrape_interval: 5s

scrape_configs:
  # API service metrics
  - job_name: 'signalforge-api'
    metrics_path: '/actuator/prometheus'
    static_configs:
      - targets: ['signalforge-api:8080']

  # Worker service metrics
  - job_name: 'signalforge-worker'
    metrics_path: '/actuator/prometheus'
    static_configs:
      - targets: ['signalforge-worker:8081']

What does this config do?

This tells Prometheus to scrape Micrometer metrics from both the API and worker services every 5 seconds. Prometheus uses the Docker Compose container names (signalforge-api, signalforge-worker) as hostnames, which Docker's internal DNS resolves automatically.

• Save both docker-compose.yml and prometheus.yml.
• Bring all services up from scratch by running this command in your terminal:

docker compose down -v && docker compose up --build -d

• Wait about 60 seconds for all services to start, then check their status by running:

docker compose ps

• Every service should show a status of healthy in the output. You should see all nine containers (postgres, redis, redpanda, prometheus, grafana, signalforge-api, signalforge-worker, and nginx) listed as running and healthy.

Some containers showing unhealthy or restarting?

• Check which container is failing by running docker compose logs <service-name> to see its logs.
• If signalforge-api or signalforge-worker fail, increase the start_period to 60s. Spring Boot may need more time to start on slower machines.
• If Redpanda fails its health check, check that ports 19092 and 9644 are not already in use by another process.

Why are my Docker Compose containers failing health checks?

✔️ Awesome, I've got everything!

Make sure you have saved both docker-compose.yml and prometheus.yml.

ⓧ I'd like to double check the full code

services:

  # === Infrastructure Services ===

  postgres:
    image: postgres:16
    container_name: signalforge-postgres
    environment:
      POSTGRES_DB: signalforge
      POSTGRES_USER: signalforge
      POSTGRES_PASSWORD: signalforge
    ports:
      - "5432:5432"
    volumes:
      - postgres-data:/var/lib/postgresql/data
    # pg_isready confirms PostgreSQL accepts connections
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U signalforge -d signalforge"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7
    container_name: signalforge-redis
    ports:
      - "6379:6379"
    # PING returns PONG when Redis is ready
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 5s
      retries: 5

  redpanda:
    image: redpandadata/redpanda:latest
    container_name: signalforge-redpanda
    command:
      - redpanda start
      - --smp 1
      - --memory 512M
      - --overprovisioned
      - --kafka-addr internal://0.0.0.0:9092,external://0.0.0.0:19092
      - --advertise-kafka-addr internal://redpanda:9092,external://localhost:19092
    ports:
      - "19092:19092"
      - "9644:9644"
    # rpk cluster health returns healthy when the broker is ready
    healthcheck:
      test: ["CMD-SHELL", "rpk cluster health | grep -E 'Healthy:.+true' || exit 1"]
      interval: 15s
      timeout: 3s
      retries: 10

  # === Monitoring Services ===

  prometheus:
    image: prom/prometheus:latest
    container_name: signalforge-prometheus
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"
    depends_on:
      signalforge-api:
        condition: service_healthy
      signalforge-worker:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:9090/-/healthy"]
      interval: 10s
      timeout: 5s
      retries: 5

  grafana:
    image: grafana/grafana:latest
    container_name: signalforge-grafana
    ports:
      - "3000:3000"
    environment:
      GF_SECURITY_ADMIN_PASSWORD: admin
    depends_on:
      prometheus:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:3000/api/health"]
      interval: 10s
      timeout: 5s
      retries: 5

  # === Application Services ===

  signalforge-api:
    build:
      context: .
      dockerfile: signalforge-api/Dockerfile
    container_name: signalforge-api
    ports:
      - "8080:8080"
    environment:
      SPRING_DATASOURCE_URL: jdbc:postgresql://postgres:5432/signalforge
      SPRING_DATASOURCE_USERNAME: signalforge
      SPRING_DATASOURCE_PASSWORD: signalforge
      SPRING_DATA_REDIS_HOST: redis
      SPRING_KAFKA_BOOTSTRAP_SERVERS: redpanda:9092
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy
      redpanda:
        condition: service_healthy
    # start_period gives Spring Boot time to initialize
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:8080/actuator/health"]
      interval: 10s
      timeout: 5s
      retries: 10
      start_period: 30s

  signalforge-worker:
    build:
      context: .
      dockerfile: signalforge-worker/Dockerfile
    container_name: signalforge-worker
    environment:
      SPRING_DATASOURCE_URL: jdbc:postgresql://postgres:5432/signalforge
      SPRING_DATASOURCE_USERNAME: signalforge
      SPRING_DATASOURCE_PASSWORD: signalforge
      SPRING_DATA_REDIS_HOST: redis
      SPRING_KAFKA_BOOTSTRAP_SERVERS: redpanda:9092
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy
      redpanda:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:8081/actuator/health"]
      interval: 10s
      timeout: 5s
      retries: 10
      start_period: 30s

  nginx:
    image: nginx:alpine
    container_name: signalforge-nginx
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      signalforge-api:
        condition: service_healthy
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:80/health"]
      interval: 10s
      timeout: 5s
      retries: 5

# === Persistent Volumes ===

volumes:
  postgres-data:

# Scrape both SignalForge services every 5 seconds
global:
  scrape_interval: 5s

scrape_configs:
  # API service metrics
  - job_name: 'signalforge-api'
    metrics_path: '/actuator/prometheus'
    static_configs:
      - targets: ['signalforge-api:8080']

  # Worker service metrics
  - job_name: 'signalforge-worker'
    metrics_path: '/actuator/prometheus'
    static_configs:
      - targets: ['signalforge-worker:8081']

Write the project README

A well-documented project is what separates a portfolio piece from a throwaway experiment. Your README is the first thing anyone sees when they open the repository.

• Create a file called README.md in the root of your signalforge-engine project.
• Copy the following content into it. This includes a Mermaid architecture diagram, setup instructions, a full API reference, and pointers to your ADRs:

# SignalForge - Multi-Tenant Event Delivery Engine

A production-grade event delivery engine that accepts events via REST API
and reliably delivers them as HTTP callbacks to customer-registered endpoints.
Guarantees zero event loss through a Transactional Outbox, retries with
exponential backoff, per-endpoint circuit breakers, and a full observability stack.

## Architecture

```mermaid
flowchart LR
    subgraph API Service
        A[REST API] --> B[EventService]
        B --> C[Outbox Table]
        C --> D[Outbox Relay]
    end

    D -->|Kafka: signalforge.events.ingested| E

    subgraph Worker Service
        E[EventIngestionConsumer] --> F[delivery_attempts]
        F --> G[DeliveryWorker]
        G -->|SKIP LOCKED| F
        G --> H[HTTP POST to endpoint]
        G -->|Kafka: signalforge.delivery.audit| I[Audit Trail]
        G -->|Kafka: signalforge.delivery.deadletter| J[Dead Letter]
    end

    subgraph Infrastructure
        K[(PostgreSQL 16)]
        L[(Redis 7)]
        M[Redpanda / Kafka]
        N[Prometheus]
        O[Grafana]
    end
```

## Quick Start

### Prerequisites

- Java 21 (Eclipse Temurin)
- Maven 3.9+
- Docker Desktop with WSL 2 backend

### Run Everything

```bash
# Build the project
mvn clean package -DskipTests

# Start all services
docker compose up --build -d

# Verify all containers are healthy
docker compose ps
```

The API is available at `http://localhost:80` (via Nginx)
or directly at `http://localhost:8080`.

Grafana dashboard: `http://localhost:3000` (admin / admin)

## API Reference

### Tenants

| Method | Endpoint | Description |
|--------|----------|-------------|
| POST | `/api/v1/tenants` | Register a new tenant |
| GET | `/api/v1/tenants/{id}` | Get tenant details |

### Endpoints

| Method | Endpoint | Description |
|--------|----------|-------------|
| POST | `/api/v1/tenants/{tenantId}/endpoints` | Register a callback URL |
| GET | `/api/v1/tenants/{tenantId}/endpoints` | List all endpoints |
| PATCH | `/api/v1/tenants/{tenantId}/endpoints/{endpointId}` | Enable or disable |
| GET | `/api/v1/tenants/{tenantId}/endpoints/{endpointId}/health` | Circuit breaker state |

### Events

| Method | Endpoint | Description |
|--------|----------|-------------|
| POST | `/api/v1/tenants/{tenantId}/events` | Ingest a new event |

Supports idempotency via the `Idempotency-Key` header.
Duplicate submissions return 200 instead of 201.

### Delivery

| Method | Endpoint | Description |
|--------|----------|-------------|
| GET | `/api/v1/tenants/{tenantId}/deliveries` | Delivery attempts (filterable by status, endpoint) |
| GET | `/api/v1/tenants/{tenantId}/events/{eventId}/deliveries` | Full delivery history for one event |

### Health

| Method | Endpoint | Description |
|--------|----------|-------------|
| GET | `/actuator/health` | Service health |
| GET | `/actuator/prometheus` | Prometheus metrics |

## Design Decisions

All major architecture decisions are documented as ADRs in `docs/adr/`:

- [ADR-001: PostgreSQL as Queue](docs/adr/ADR-001-postgresql-as-queue.md)
- [ADR-002: Transactional Outbox](docs/adr/ADR-002-transactional-outbox.md)
- [ADR-003: Virtual Threads](docs/adr/ADR-003-virtual-threads.md)
- [ADR-004: Per-Endpoint Circuit Breakers](docs/adr/ADR-004-per-endpoint-circuit-breakers.md)
- [ADR-005: Kafka for V3](docs/adr/ADR-005-kafka-v3.md)

## Key Concepts

- **Transactional Outbox**: Events and delivery attempts are written in the
  same database transaction. If the transaction fails, nothing is written.
  If it succeeds, delivery is guaranteed.
- **SKIP LOCKED**: The delivery worker claims batches of work using
  `SELECT ... FOR UPDATE SKIP LOCKED`, which atomically locks rows and
  skips those already claimed by other workers. Zero double-delivery.
- **Exponential Backoff with Jitter**: Failed deliveries are retried up to
  11 times over approximately 5 days, with randomized delays to prevent
  thundering herd effects.
- **Per-Endpoint Circuit Breakers**: A single degraded endpoint cannot
  consume all delivery capacity. When an endpoint's failure rate exceeds
  50%, its circuit breaker opens and deliveries are re-scheduled.
- **Virtual Threads (Java 21)**: Each HTTP delivery runs on a virtual thread,
  enabling 10,000+ concurrent deliveries without thread pool exhaustion.

## Load Testing

```bash
# Create a tenant
curl -X POST http://localhost:80/api/v1/tenants \
  -H "Content-Type: application/json" \
  -d '{"name": "load-test-tenant"}'

# Register an endpoint (use a tool like webhook.site for testing)
curl -X POST http://localhost:80/api/v1/tenants/{TENANT_ID}/endpoints \
  -H "Content-Type: application/json" \
  -d '{"url": "https://webhook.site/your-unique-url"}'

# Ingest 100 events in a loop
for i in $(seq 1 100); do
  curl -X POST http://localhost:80/api/v1/tenants/{TENANT_ID}/events \
    -H "Content-Type: application/json" \
    -d "{\"event_type\": \"test.load\", \"payload\": {\"index\": $i}}"
done
```

Monitor delivery progress on the Grafana dashboard at `http://localhost:3000`.

## Tech Stack

- Java 21 (Temurin) with Virtual Threads
- Spring Boot 3.5.14
- PostgreSQL 16 with Flyway migrations
- Redis 7 for rate limiting and deduplication
- Redpanda (Kafka-compatible) for event streaming
- Resilience4j for circuit breakers
- Micrometer + Prometheus + Grafana for observability
- Nginx as a reverse proxy
- Docker Compose for orchestration

What does this README cover?

• The Mermaid diagram renders as an interactive flowchart on GitHub, showing the data flow from API ingestion through Kafka to the delivery worker.
• The Quick Start section lets anyone clone and run the full system with three terminal commands.
• The API Reference documents every endpoint across all controllers you built in earlier steps.
• The Load Testing section gives a copy-paste script for sending events through the full pipeline.

• Save README.md.
• Open the file in IntelliJ's Markdown preview (click the split editor icon in the top right of the editor tab) to confirm the Mermaid diagram and tables render correctly.

Mermaid diagram not rendering in preview?

IntelliJ's built-in Markdown preview does not always render Mermaid diagrams. The diagram will render correctly on GitHub when you push the repository. You can also paste the Mermaid block into mermaid.live to preview it immediately.

Create Architecture Decision Records

Architecture Decision Records (ADRs) document the reasoning behind key design choices. They are invaluable during interviews because they show you can articulate trade-offs, not just write code.

• Create the ADR directory by running this command from the root of your signalforge-engine project:

mkdir -p docs/adr

• Confirm the directory exists by running:

ls docs/adr

You should see an empty directory listed. You will now create five ADR files inside it.

• Create docs/adr/ADR-001-postgresql-as-queue.md with the following content:

# ADR-001: PostgreSQL as Queue vs Dedicated Message Broker (V1)

## Status
Accepted

## Context
The delivery worker needs a queue of pending work. Common approaches include
a dedicated message broker (RabbitMQ, Redis Streams) or using the relational
database that already stores the business data.

In V1, the system is a monolith. All business data (tenants, endpoints, events,
delivery_attempts) lives in PostgreSQL. Introducing a separate broker adds
operational complexity and creates a dual-write problem: the event must be
written to both PostgreSQL and the broker atomically.

## Decision
Use PostgreSQL as the work queue for V1 via `SELECT ... FOR UPDATE SKIP LOCKED`.
The delivery_attempts table serves as both the persistent record and the queue.

## Consequences

### Positive
- **Zero additional infrastructure** in V1. Only PostgreSQL is required.
- **Transactional consistency**: delivery_attempts are created in the same
  transaction as the Event row. If the transaction fails, nothing is written.
- **SKIP LOCKED** provides contention-free, atomic work claiming. Multiple
  worker instances claim different rows without double-delivery.
- **Partial index** on `(next_retry_at, id) WHERE status IN ('PENDING', 'FAILED')`
  ensures the worker query uses an efficient index scan.

### Negative
- PostgreSQL is not optimized for high-throughput queue workloads. At extreme
  scale (100K+ events/second), a dedicated broker is more appropriate.
- Frequent UPDATE operations on delivery_attempts create dead tuples. Autovacuum
  must be tuned for this table.

### Mitigation
In V3, Kafka is introduced for event ingestion, decoupling upstream producers
from the delivery pipeline. PostgreSQL remains the queue for the delivery
worker where transactional guarantees matter most.

• Create docs/adr/ADR-002-transactional-outbox.md with the following content:

# ADR-002: Transactional Outbox vs Dual-Write

## Status
Accepted

## Context
When an event is ingested, the system must write the Event row to PostgreSQL
and ensure delivery_attempts (V1) or a Kafka message (V3) are also created.
A naive approach writes to both destinations separately (dual-write), but
if the application crashes between the two writes, data is lost or duplicated.

## Decision
Use the Transactional Outbox pattern. In V1, the Event row and all
delivery_attempt rows are written in a single `@Transactional` method.
In V3, the Event row and an outbox_events row are written in the same
transaction. A polling relay reads unpublished outbox rows with SKIP LOCKED
and publishes them to Kafka.

## Consequences

### Positive
- **Zero event loss**: if the transaction commits, delivery is guaranteed.
  If it rolls back, nothing is written.
- **No distributed transaction coordinator** (2PC) required.
- The outbox relay is idempotent. Kafka consumers track processed event IDs
  to handle the at-least-once delivery guarantee.

### Negative
- The polling relay introduces a small delay (up to 500ms) between event
  ingestion and Kafka publishing.
- The outbox_events table accumulates rows that must be cleaned up after
  successful publishing.

### Alternatives Considered
- **Dual-write**: rejected due to the crash-between-writes failure mode.
- **Change Data Capture (Debezium)**: viable but adds significant operational
  complexity (Kafka Connect, connector configuration, schema evolution).
  The polling relay is simpler for this project's scale.

• Create docs/adr/ADR-003-virtual-threads.md with the following content:

# ADR-003: Virtual Threads vs Reactive (WebFlux)

## Status
Accepted

## Context
Each delivery attempt makes an outbound HTTP POST that blocks for 100ms to
30 seconds. With a traditional 200-thread pool, maximum concurrency is 200
simultaneous deliveries. The system needs 10,000+ concurrent deliveries.

Two approaches exist:
1. **Reactive (Spring WebFlux)**: non-blocking I/O with `Mono`/`Flux`. Requires
   rewriting all code to a reactive style and using reactive database drivers.
2. **Virtual Threads (Java 21)**: lightweight threads managed by the JVM.
   Blocking code works unchanged. Enabled with a single Spring Boot property.

## Decision
Use Java 21 Virtual Threads with `spring.threads.virtual.enabled=true`.

## Consequences

### Positive
- **Blocking code works as-is**: `RestClient`, JDBC, `Thread.sleep()` all work
  without modification. The JVM unmounts the virtual thread from the carrier
  OS thread while it blocks.
- **Simpler mental model**: standard imperative Java. No reactive operator chains,
  no `subscribeOn`/`publishOn` confusion.
- **HikariCP integration**: `Semaphore.acquire()` in HikariCP unmounts the
  virtual thread, so thousands of virtual threads can contend for a small
  connection pool (30 connections) without deadlock.
- **Initial stack size ~1KB** vs ~1MB for platform threads. The JVM can host
  millions of virtual threads.

### Negative
- `synchronized` blocks pin the virtual thread to a carrier thread. All
  `synchronized` usage must be replaced with `ReentrantLock`.
- Debugging is different: thread dumps show virtual thread IDs, and
  traditional thread-pool metrics become less meaningful.

### Detection
Use `-Djdk.tracePinnedThreads=full` to detect pinning at runtime.

• Create docs/adr/ADR-004-per-endpoint-circuit-breakers.md with the following content:

# ADR-004: Per-Endpoint Circuit Breakers vs Global Circuit Breaker

## Status
Accepted

## Context
When a customer endpoint becomes degraded (returning 500s or timing out),
the delivery worker wastes capacity retrying deliveries that will fail.
A circuit breaker stops calling a failing endpoint and re-schedules
deliveries until the endpoint recovers.

Two granularity levels exist:
1. **Global circuit breaker**: one breaker for all endpoints. If any endpoint
   is degraded, all deliveries are paused.
2. **Per-endpoint circuit breaker**: each endpoint gets its own breaker.
   A degraded endpoint is isolated without affecting healthy endpoints.

## Decision
Use per-endpoint circuit breakers via Resilience4j. Each endpoint ID maps to
a named CircuitBreaker instance in a registry.

## Consequences

### Positive
- **Fault isolation**: one degraded tenant endpoint does not block deliveries
  to other tenants or other endpoints of the same tenant.
- **Endpoint-level health visibility**: the `/health` API endpoint exposes
  CLOSED, OPEN, or HALF_OPEN state per endpoint.
- **Configurable per endpoint** in the future (different thresholds for
  different SLA tiers).

### Negative
- Memory overhead: each circuit breaker maintains a sliding window of
  call results. With thousands of endpoints, this adds up.
- Complexity: the worker must check circuit breaker state before each delivery
  and re-schedule if OPEN.

### Configuration
- Failure rate threshold: 50%
- Slow call threshold: 80%
- Slow call duration: 10 seconds
- Wait duration in open state: 60 seconds
- Permitted calls in half-open state: 5

• Create docs/adr/ADR-005-kafka-v3.md with the following content:

# ADR-005: Kafka for V3 (Event Ingestion Decoupling and Audit Trail)

## Status
Accepted

## Context
In V1 and V2, the API service writes delivery_attempts directly to PostgreSQL
in the same transaction as the Event. This couples ingestion throughput to
delivery fan-out. If a tenant has 1,000 active endpoints, a single event
ingestion inserts 1,001 rows in one transaction.

V3 splits the system into separate API and Worker services. The API service
needs a way to signal the Worker that new events are available without tight
coupling.

## Decision
Introduce Kafka (via Redpanda) for three purposes:
1. **Event ingestion decoupling**: the API service writes to an outbox table
   and a relay publishes to `signalforge.events.ingested`. The Worker consumes
   and creates delivery_attempts.
2. **Delivery audit trail**: after each delivery attempt, the Worker publishes
   to `signalforge.delivery.audit` for an immutable audit log.
3. **Dead-letter topic**: when a delivery reaches DEAD status, the Worker
   publishes to `signalforge.delivery.deadletter` for alerting and review.

## Consequences

### Positive
- **Independent scaling**: API and Worker services scale independently.
  More workers can be added to handle higher delivery volumes.
- **Decoupled ingestion**: the API transaction is small (Event + outbox row).
  Fan-out to delivery_attempts happens asynchronously in the Worker.
- **Immutable audit trail**: Kafka topics retain delivery history independent
  of the mutable delivery_attempts table.

### Negative
- Additional infrastructure: Redpanda must be deployed and monitored.
- Eventual consistency: there is a small delay between event ingestion and
  delivery_attempt creation.
- Consumer idempotency is required to handle at-least-once delivery.

### Why Not Kafka From Day One?
V1 is a monolith where PostgreSQL handles both storage and queuing. Adding
Kafka from the start would increase complexity without a clear benefit at
monolith scale. Kafka is justified in V3 because the system is now distributed.

Why write ADRs?

ADRs follow a standard format: Context (the problem), Decision (what you chose), and Consequences (the trade-offs). Each one captures not just what you decided, but why you rejected the alternatives.

In a system design interview, being able to articulate "we chose X over Y because of Z" is the difference between a mid-level and senior-level answer. These five ADRs cover the most common distributed systems trade-offs interviewers ask about.

• Save all five ADR files.
• Confirm all files exist by running:

ls docs/adr/

• You should see all five files listed:

ADR-001-postgresql-as-queue.md
ADR-002-transactional-outbox.md
ADR-003-virtual-threads.md
ADR-004-per-endpoint-circuit-breakers.md
ADR-005-kafka-v3.md

Missing ADR files?

• Make sure you created the files inside docs/adr/ and not in the project root or another directory.
• Check that each file name matches exactly, including the .md extension.

Help me find where my ADR files ended up in my project.

Your SignalForge engine is now fully documented and runnable with a single docker compose up command. The README, architecture diagram, and ADRs transform this from a personal project into a portfolio piece that demonstrates production thinking. If you are ready for an extra challenge, the Secret Mission adds HMAC request signing and a delivery replay API.

Secret mission

Add Request Signing and Delivery Replay

Your delivery engine guarantees events reach their destination. But how do receiving endpoints know those events are actually from SignalForge? And what happens when dead-lettered events need a second chance? Extend your system with HMAC-SHA256 request signing and a self-service delivery replay API.

Before you go

Clean Up Your Resources

This project runs entirely on your local machine via Docker Compose. No cloud resources are used and no ongoing costs are incurred.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• PostgreSQL 16 Docker container (database: signalforge).
• Redis Docker container.
• Redpanda (Kafka-compatible broker) Docker container.
• Schema Registry Docker container.
• Prometheus Docker container.
• Grafana Docker container.
• SignalForge API service container.
• SignalForge Worker service container.
• Nginx or Traefik load balancer container.
• Local Maven project files (signalforge-engine directory on your Desktop).
• Docker volumes storing PostgreSQL data, Redis data, and Redpanda data.

Keep everything running

No action needed. Choose this if you are still actively building or want to keep experimenting with the system.

• Your Docker Compose services will continue running in the background.
• All data in PostgreSQL, Redis, and Redpanda is preserved in Docker volumes.
• Prometheus and Grafana dashboards remain accessible at their configured ports.

Pause - I'll come back to this later

Shut down all running containers to free up memory and CPU, while keeping your data intact in Docker volumes.

• Stop all containers by running this command from your project directory:

docker compose stop

• This stops all containers without removing them or their data.
• When you are ready to resume, run docker compose start to bring everything back up with all your data intact.

Delete - I don't want to use this again

Remove all project containers, networks, and volumes to start completely fresh.

• Remove all containers and their associated volumes by running this command from your project directory:

docker compose down -v

• The -v flag removes Docker volumes, which means all PostgreSQL data, Redis data, and Redpanda data will be permanently deleted.
• To also remove the Docker images used by this project, run:

docker compose down -v --rmi all

• To delete the project source code, remove the signalforge-engine folder from your Desktop.

Mission accomplished

That's a Wrap!

Nice work. You just built a production-grade, distributed event delivery engine from scratch. This is the same infrastructure that companies like Svix charge $490/month for, and you designed every layer of it yourself.

You've learned how to:

• Built a multi-tenant event delivery pipeline with a Transactional Outbox for zero event loss, a PostgreSQL SKIP LOCKED delivery worker for contention-free work claiming, and an exponential backoff retry system with jitter and dead-letter handling.
• Powered 10,000+ concurrent HTTP deliveries with Java 21 Virtual Threads, per-endpoint circuit breakers, and token-bucket rate limiters.
• Split the system into distributed API and Worker services communicating through Kafka, with a full Prometheus and Grafana observability dashboard tracking p99 latency, success rates, and queue depth.
• Added HMAC-SHA256 request signing so endpoints can verify event authenticity, and built a delivery replay API for operational self-service.

Ready to quiz yourself?