Webhook sống sót qua phía khách hàng

export async function handleEvent(event: EventEnvelope, logger: Logger) {
  logger.log("info", "event.handled", { event_id: event.id, type: event.type });
}

// 3.4: the claim and the effect commit together or not at all.
await claimEvent(db, CONSUMER, event.id, async () => {
  await handleEvent(event, logger);
});

//   claim BEFORE posting  → a crash in the gap loses the webhook silently.
//   post BEFORE reporting → a crash in the gap re-posts. The customer receives
//                           it twice and CAN tell, because the envelope carries
//                           the event id they deduplicate on.

import {
  createCipheriv,
  createDecipheriv,
  randomBytes,
} from "node:crypto";
 
// Webhook signing secrets at rest (chapter 3.5).
//
// DECISION (chapter 3.5): envelope encryption, not the salted hash chapter 3.2
// used for API keys. NFR-SEC-02 permits either — "salted hashes or under
// envelope encryption" — and this is the branch that applies, for a reason worth
// stating because the two credentials look so alike:
//
//   an API key is VERIFIED. A caller presents it, we hash what arrived and
//   compare digests. The original is never needed again, so keeping it would be
//   a liability with no upside.
//
//   a signing secret is USED. Every delivery computes an HMAC with it. A hash
//   cannot be used, only compared — so hashing it here would not be "more
//   secure", it would make the feature impossible.
//
// The cost is real and is not hidden: this is the first customer credential the
// platform can turn back into plaintext. That raises obligations a hash did not.
// The key lives in configuration and never in the database (a key stored beside
// the ciphertext it protects is a filing convention, not encryption), the
// plaintext exists only in memory and only for the duration of a signature, and
// it appears in no log line at any level (NFR-SEC-06).
 
const ALGORITHM = "aes-256-gcm";
const KEY_BYTES = 32;
const IV_BYTES = 12; // 96 bits, the size GCM is specified for
const TAG_BYTES = 16;
const SECRET_BYTES = 32; // 256 bits, the budget 3.2 gave an API key secret
 
/** The stored value is `iv | tag | ciphertext`, base64. One column, no schema
 * for the reader to decode, and no second place for the IV to drift out of sync
 * with the payload it belongs to. */
const IV_END = IV_BYTES;
const TAG_END = IV_BYTES + TAG_BYTES;
 
export const WEBHOOK_SECRET_KEY_ENV = "RELAY_WEBHOOK_SECRET_KEY";
 
/** Resolved per call rather than at import time: a module that throws on import
 * takes the whole service down at startup over a feature most requests never
 * touch, and it makes the failure unreachable in a unit test. */
function encryptionKey(): Buffer {
  const configured = process.env[WEBHOOK_SECRET_KEY_ENV];
  if (!configured) {
    // Development only, and deliberately loud about it. A silent default here
    // would mean production quietly encrypting with a key from a tutorial.
    if (process.env.NODE_ENV === "production") {
      throw new Error(
        `${WEBHOOK_SECRET_KEY_ENV} is required in production: webhook signing secrets cannot be stored without it`,
      );
    }
    return Buffer.alloc(KEY_BYTES, "relay-development-key-not-for-production");
  }
  const key = Buffer.from(configured, "base64");
  if (key.length !== KEY_BYTES) {
    throw new Error(
      `${WEBHOOK_SECRET_KEY_ENV} must decode to ${KEY_BYTES} bytes, got ${key.length}`,
    );
  }
  return key;
}
 
/** A new signing secret, shown to the customer once and never again. */
export function mintSigningSecret(): string {
  return randomBytes(SECRET_BYTES).toString("base64url");
}
 
export function encryptSecret(secret: string): string {
  // A fresh IV per call, which is what makes two encryptions of the same secret
  // differ. A deterministic ciphertext would leak equality: an operator reading
  // the table could see which endpoints share a secret without decrypting one.
  const iv = randomBytes(IV_BYTES);
  const cipher = createCipheriv(ALGORITHM, encryptionKey(), iv);
  const ciphertext = Buffer.concat([
    cipher.update(secret, "utf8"),
    cipher.final(),
  ]);
  return Buffer.concat([iv, cipher.getAuthTag(), ciphertext]).toString("base64");
}
 
export function decryptSecret(stored: string): string {
  const raw = Buffer.from(stored, "base64");
  if (raw.length <= TAG_END) {
    throw new Error("webhook secret ciphertext is too short to be valid");
  }
  const decipher = createDecipheriv(
    ALGORITHM,
    encryptionKey(),
    raw.subarray(0, IV_END),
  );
  // GCM's tag is why a tampered value THROWS rather than returning plausible
  // garbage. Signing with silently-wrong bytes would break every signature for
  // that endpoint while looking like it worked.
  decipher.setAuthTag(raw.subarray(IV_END, TAG_END));
  return Buffer.concat([
    decipher.update(raw.subarray(TAG_END)),
    decipher.final(),
  ]).toString("utf8");
}
 
/** The rotation window: 24 hours (contracts/webhooks.md §Rotation).
 *
 * Long enough that a customer can roll a configuration change across their fleet
 * without a deploy window becoming an outage; short enough that a secret they
 * rotated *because it leaked* stops working the same day. Fixed here because it
 * is a promise a recipient writes code against, not a tuning parameter. */
export const ROTATION_WINDOW_MS = 24 * 60 * 60 * 1000;
 
/** Which secrets a delivery must be signed with right now.
 *
 * Two during the window, one after it — and the "after it" half is the part that
 * matters. A previous secret that never expires is not a rotation, it is a
 * second permanent credential, and a customer who rotated because of a leak
 * would still be accepting the leaked one. */
export function activeSigningSecrets(
  endpoint: {
    secretCiphertext: string;
    secretPreviousCiphertext: string | null;
    secretRotatedAt: Date | null;
  },
  now: Date = new Date(),
): string[] {
  const current = decryptSecret(endpoint.secretCiphertext);
  if (!endpoint.secretPreviousCiphertext || !endpoint.secretRotatedAt) {
    return [current];
  }
  const elapsed = now.getTime() - endpoint.secretRotatedAt.getTime();
  if (elapsed >= ROTATION_WINDOW_MS) return [current];
  return [current, decryptSecret(endpoint.secretPreviousCiphertext)];
}

-- Chapter 3.5 — webhook endpoints, the delivery schedule, and dead letters.
--
-- REVIEW DISPOSITION: drizzle-kit generated this from schema.ts and it was read
-- line by line before being applied (the ADR-16 workflow). Nothing was
-- rewritten. Five things were checked rather than assumed:
--
--   * all three tables carry environment_id NOT NULL with a foreign key to
--     environments. Unlike 3.3's outbox and 3.4's consumed_events, these are
--     tenant data, and the rule distinguishing the two cases is stated in
--     schema.ts rather than left for the next chapter to infer;
--   * webhook_deliveries has UNIQUE (event_id, endpoint_id). That constraint IS
--     the idempotence of expansion — one event produces at most one delivery per
--     endpoint however many times the event is redelivered, enforced by the
--     database rather than by care;
--   * webhook_deliveries_due_idx is PARTIAL, on next_attempt_at WHERE state =
--     'pending'. It covers the relay's only query and nothing else, so delivered
--     rows cost nothing to keep — the same shape as 3.3's outbox index;
--   * the state CHECK admits exactly pending / delivered / dead. There is no
--     fourth state to get stuck in;
--   * webhook_endpoints.deleted_at exists because DELETION IS SOFT. The foreign
--     keys from deliveries and dead letters are ON DELETE NO ACTION on purpose:
--     a hard delete would have to cascade, and cascading would erase a
--     customer's dead letters, which FR-WHK-04 says to retain for seven days.
--     Chapter 3.2 reached the same conclusion for api_keys.revoked_at.
 
CREATE TABLE "webhook_dead_letters" (
	"id" uuid PRIMARY KEY NOT NULL,
	"environment_id" uuid NOT NULL,
	"endpoint_id" uuid NOT NULL,
	"event_id" uuid NOT NULL,
	"payload" jsonb NOT NULL,
	"last_status" integer,
	"last_error" text,
	"attempts" integer NOT NULL,
	"dead_lettered_at" timestamp with time zone DEFAULT now() NOT NULL
);
--> statement-breakpoint
CREATE TABLE "webhook_deliveries" (
	"id" uuid PRIMARY KEY NOT NULL,
	"environment_id" uuid NOT NULL,
	"endpoint_id" uuid NOT NULL,
	"event_id" uuid NOT NULL,
	"payload" jsonb NOT NULL,
	"attempt" integer DEFAULT 1 NOT NULL,
	"next_attempt_at" timestamp with time zone DEFAULT now() NOT NULL,
	"dispatched_at" timestamp with time zone,
	"state" text DEFAULT 'pending' NOT NULL,
	"created_at" timestamp with time zone DEFAULT now() NOT NULL,
	CONSTRAINT "webhook_deliveries_event_endpoint_unique" UNIQUE("event_id","endpoint_id"),
	CONSTRAINT "webhook_deliveries_state_check" CHECK ("webhook_deliveries"."state" IN ('pending','delivered','dead'))
);
--> statement-breakpoint
CREATE TABLE "webhook_endpoints" (
	"id" uuid PRIMARY KEY NOT NULL,
	"environment_id" uuid NOT NULL,
	"url" text NOT NULL,
	"event_types" jsonb NOT NULL,
	"secret_ciphertext" text NOT NULL,
	"secret_previous_ciphertext" text,
	"secret_rotated_at" timestamp with time zone,
	"enabled" boolean DEFAULT true NOT NULL,
	"created_at" timestamp with time zone DEFAULT now() NOT NULL,
	"deleted_at" timestamp with time zone
);
--> statement-breakpoint
ALTER TABLE "webhook_dead_letters" ADD CONSTRAINT "webhook_dead_letters_environment_id_environments_id_fk" FOREIGN KEY ("environment_id") REFERENCES "public"."environments"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "webhook_dead_letters" ADD CONSTRAINT "webhook_dead_letters_endpoint_id_webhook_endpoints_id_fk" FOREIGN KEY ("endpoint_id") REFERENCES "public"."webhook_endpoints"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "webhook_deliveries" ADD CONSTRAINT "webhook_deliveries_environment_id_environments_id_fk" FOREIGN KEY ("environment_id") REFERENCES "public"."environments"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "webhook_deliveries" ADD CONSTRAINT "webhook_deliveries_endpoint_id_webhook_endpoints_id_fk" FOREIGN KEY ("endpoint_id") REFERENCES "public"."webhook_endpoints"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "webhook_endpoints" ADD CONSTRAINT "webhook_endpoints_environment_id_environments_id_fk" FOREIGN KEY ("environment_id") REFERENCES "public"."environments"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
CREATE INDEX "webhook_dead_letters_environment_idx" ON "webhook_dead_letters" USING btree ("environment_id");--> statement-breakpoint
CREATE INDEX "webhook_deliveries_due_idx" ON "webhook_deliveries" USING btree ("next_attempt_at") WHERE "webhook_deliveries"."state" = 'pending';--> statement-breakpoint
CREATE INDEX "webhook_endpoints_environment_idx" ON "webhook_endpoints" USING btree ("environment_id");

import {
  Injectable,
  NotFoundException,
  UnprocessableEntityException,
} from "@nestjs/common";
 
import { Repository, type WebhookEndpointRow } from "../db/repository";
import { encryptSecret, mintSigningSecret } from "./secret";
 
// The management surface's rules (chapter 3.5). Two of them are worth stating
// here rather than leaving to the controller, because both are the kind of thing
// that looks like validation and is actually a security boundary.
 
/** FR-WHK-01, and the error must say the number. Chapter 3.2's lesson about
 * error messages that name the mistake applies to a limit as much as to a
 * credential: "too many endpoints" leaves the reader counting. */
export const MAX_ENDPOINTS_PER_ENVIRONMENT = 5;
 
/** DECISION (chapter 3.5, research R9): no requirement mandates this check, and
 * without it the dispatcher is a request-forgery primitive pointed at whatever
 * it can reach — a tenant supplies the URL, and the platform fetches it.
 *
 * Deliberately NOT built: DNS re-resolution at delivery time to defeat
 * rebinding, an egress proxy, an allowlist. Each is a real hardening step and
 * each needs infrastructure this platform does not have. Naming them is more
 * honest than implying the simple check is complete; NFR-SEC-07's OWASP scan is
 * where this comes back. */
const BLOCKED_HOSTS = /^(localhost|0\.0\.0\.0|\[?::1\]?)$/i;
const BLOCKED_RANGES = [
  /^127\./,
  /^10\./,
  /^192\.168\./,
  /^172\.(1[6-9]|2\d|3[01])\./,
  /^169\.254\./, // link-local, and the cloud metadata endpoint lives here
  /^fc00:/i,
  /^fe80:/i,
];
 
export interface CreateEndpointInput {
  url: string;
  event_types: string[];
}
 
/** What a customer receives once and never again. */
export interface EndpointWithSecret extends WebhookEndpointRow {
  secret: string;
}
 
@Injectable()
export class WebhooksService {
  constructor(private readonly repo: Repository) {}
 
  async create(input: CreateEndpointInput): Promise<EndpointWithSecret> {
    this.assertDeliverableUrl(input.url);
    this.assertEventTypes(input.event_types);
 
    const existing = await this.repo.countEndpoints();
    if (existing >= MAX_ENDPOINTS_PER_ENVIRONMENT) {
      throw new UnprocessableEntityException(
        `an environment may have at most ${MAX_ENDPOINTS_PER_ENVIRONMENT} webhook endpoints; this one already has ${existing}`,
      );
    }
 
    // Minted here, encrypted here, returned here — and never readable again.
    // The plaintext exists in this method's scope and nowhere else.
    const secret = mintSigningSecret();
    const row = await this.repo.createEndpoint({
      url: input.url,
      eventTypes: input.event_types,
      secretCiphertext: encryptSecret(secret),
    });
    return { ...row, secret };
  }
 
  list(): Promise<WebhookEndpointRow[]> {
    return this.repo.listEndpoints();
  }
 
  async get(id: string): Promise<WebhookEndpointRow> {
    const row = await this.repo.getEndpoint(id);
    // 404 rather than 403, for an endpoint that exists in another environment as
    // much as for one that never existed: existence itself must not leak
    // (FR-TEN-05).
    if (!row) throw new NotFoundException("no such webhook endpoint");
    return row;
  }
 
  async rotateSecret(id: string): Promise<EndpointWithSecret> {
    const secret = mintSigningSecret();
    const row = await this.repo.rotateEndpointSecret(id, encryptSecret(secret));
    if (!row) throw new NotFoundException("no such webhook endpoint");
    return { ...row, secret };
  }
 
  async setEnabled(id: string, enabled: boolean): Promise<WebhookEndpointRow> {
    const row = await this.repo.setEndpointEnabled(id, enabled);
    if (!row) throw new NotFoundException("no such webhook endpoint");
    return row;
  }
 
  async remove(id: string): Promise<void> {
    const deleted = await this.repo.deleteEndpoint(id);
    if (!deleted) throw new NotFoundException("no such webhook endpoint");
  }
 
  private assertDeliverableUrl(raw: string): void {
    let parsed: URL;
    try {
      parsed = new URL(raw);
    } catch {
      throw new UnprocessableEntityException("url must be a valid absolute URL");
    }
    if (parsed.protocol !== "https:") {
      throw new UnprocessableEntityException(
        "url must use https — a signature over a plaintext channel protects the body, not the reader",
      );
    }
    const host = parsed.hostname;
    if (BLOCKED_HOSTS.test(host) || BLOCKED_RANGES.some((r) => r.test(host))) {
      throw new UnprocessableEntityException(
        "url must not point at a loopback, link-local or private address",
      );
    }
  }
 
  private assertEventTypes(types: string[]): void {
    if (!Array.isArray(types) || types.length === 0) {
      throw new UnprocessableEntityException(
        "event_types must list at least one event type",
      );
    }
  }
}

import {
  Body,
  Controller,
  Delete,
  Get,
  HttpCode,
  Param,
  Post,
  UseGuards,
} from "@nestjs/common";
 
import { Accepts, CredentialGuard } from "../auth/credential.guard";
import { WebhooksService } from "./webhooks.service";
import type { CreateEndpointInput } from "./webhooks.service";
 
// The webhook management surface (chapter 3.5, FR-WHK-01 and FR-WHK-08).
//
// `Accepts("application")` and nothing else: configuring where a tenant's events
// are delivered is the tenant's software acting for itself, not an end user
// acting for themselves. An end-user token reaching this route would mean any
// logged-in person in a customer's product could redirect that customer's
// events, which is a very short sentence describing a very large incident.
@Controller("v1/webhooks")
@UseGuards(CredentialGuard)
@Accepts("application")
export class WebhooksController {
  constructor(private readonly webhooks: WebhooksService) {}
 
  /** 201 with the secret — the only time it is ever returned. */
  @Post()
  create(@Body() body: CreateEndpointInput) {
    return this.webhooks.create(body);
  }
 
  @Get()
  list() {
    return this.webhooks.list();
  }
 
  @Get(":id")
  get(@Param("id") id: string) {
    return this.webhooks.get(id);
  }
 
  /** 200, not Nest's default 201: rotation replaces a secret on an endpoint that
   * already exists. A 201 would tell a client something was created and hand it
   * no location for it.
   *
   * Opens the 24-hour rotation window and returns the new secret once. The
   * outgoing secret keeps signing until the window closes, so a recipient
   * accepting either is correct throughout (contracts/webhooks.md §Rotation). */
  @Post(":id/rotate-secret")
  @HttpCode(200)
  rotateSecret(@Param("id") id: string) {
    return this.webhooks.rotateSecret(id);
  }
 
  @Post(":id/enable")
  @HttpCode(200)
  enable(@Param("id") id: string) {
    return this.webhooks.setEnabled(id, true);
  }
 
  @Post(":id/disable")
  @HttpCode(200)
  disable(@Param("id") id: string) {
    return this.webhooks.setEnabled(id, false);
  }
 
  /** SOFT delete. The endpoint stops receiving deliveries and disappears from
   * every read; the row survives because its dead letters must (FR-WHK-04). */
  @Delete(":id")
  @HttpCode(204)
  async remove(@Param("id") id: string): Promise<void> {
    await this.webhooks.remove(id);
  }
}

import {
  Inject,
  Injectable,
  Module,
  Scope,
  type OnModuleDestroy,
} from "@nestjs/common";
import { REQUEST } from "@nestjs/core";
 
import { createLogger } from "@relay/service-kit";
 
import { AuthModule } from "../auth/auth.module";
import { createDb, createPool, type Db } from "../db/client";
import { Repository } from "../db/repository";
import type { RequestWithTenant } from "../messages/request-with-tenant";
import { createJetStreamPublisher } from "../outbox/jetstream.publisher";
import {
  createDeliveryRelay,
  ensureDeliveriesStream,
  type DeliveryRelay,
} from "./delivery-relay";
import { WebhooksController } from "./webhooks.controller";
import { WebhooksService } from "./webhooks.service";
 
export const DELIVERY_RELAY = "DELIVERY_RELAY";
 
/** The api's SECOND relay (chapter 3.5, research R13), started with the service
 * exactly as 3.3's is. An event spine that only runs when someone remembers is
 * not a spine, and the same is true of a retry schedule.
 *
 * `RELAY_DELIVERY_RELAY=off` is the sibling of `RELAY_OUTBOX_RELAY=off`, and it
 * exists for the same reason: suites that assert on delivery rows would flap if
 * a background loop marked them dispatched mid-assertion. Chapter 3.3's finding
 * 4 — a background daemon and a test lane do not share a table quietly — applies
 * again, and this time it is anticipated rather than discovered. */
export function deliveryRelayEnabled(): boolean {
  return (process.env.RELAY_DELIVERY_RELAY ?? "on").toLowerCase() !== "off";
}
 
@Injectable()
export class DeliveryRelayService implements OnModuleDestroy {
  constructor(@Inject(DELIVERY_RELAY) private readonly relay: DeliveryRelay) {}
 
  start(): void {
    if (deliveryRelayEnabled()) this.relay.start();
  }
 
  async onModuleDestroy(): Promise<void> {
    await this.relay.stop();
  }
}
 
// The same wiring chapter 2.2 established and 3.2 re-pointed: the repository is
// the plain class, constructed per request with the environment the middleware
// resolved from a verified credential. A request cannot name a tenant it has not
// proved it may act for, and this module adds no new way to try.
@Module({
  imports: [AuthModule],
  controllers: [WebhooksController],
  providers: [
    {
      provide: "DB",
      useFactory: (): Db => createDb(createPool()),
      scope: Scope.DEFAULT,
    },
    {
      provide: Repository,
      scope: Scope.REQUEST,
      inject: ["DB", REQUEST],
      useFactory: (db: Db, req: RequestWithTenant) =>
        new Repository(db, req.principal?.environmentId ?? ""),
    },
    WebhooksService,
    {
      provide: DELIVERY_RELAY,
      useFactory: (): DeliveryRelay =>
        createDeliveryRelay({
          db: createDb(createPool()) as Db,
          // Its OWN stream, not the events one. Chapter 3.3's publisher
          // ensured EVENTS because that was the only stream; this one must
          // bring DELIVERIES into existence or every publish is a 503.
          publisher: createJetStreamPublisher({ ensure: ensureDeliveriesStream }),
          logger: createLogger("deliveries"),
        }),
    },
    DeliveryRelayService,
  ],
  exports: [DELIVERY_RELAY, DeliveryRelayService],
})
export class WebhooksModule {}

import { createHmac } from "node:crypto";
 
// The signature a customer verifies (chapter 3.5, FR-WHK-08).
//
// This is the platform's first contract addressed to code it did not write and
// cannot read. A REST consumer can retry against a sandbox and read an error
// message; a webhook consumer finds out at 3 a.m. that their check has been
// wrong for a week. So the construction below is deliberately dull: no length
// prefixes, no canonical JSON, no nested encoding. Everything a recipient needs
// is the request and the shared secret, and the recipe fits in five lines of
// their language.
//
// Kept in its own file, and behind a function, for SAD risk R7's reason: "isolate
// HMAC/crypto behind an interface" so a profiling-driven swap stays contained.
 
/** Travels with the value, not the header name, so a second algorithm is an
 * additional signature rather than a breaking change to the header set. */
export const SIGNATURE_SCHEME = "v1";
 
export const TIMESTAMP_HEADER = "relay-webhook-timestamp";
export const SIGNATURE_HEADER = "relay-webhook-signature";
 
/** `<scheme>:<timestamp>:<raw body>`.
 *
 * The timestamp is INSIDE the signed string rather than merely alongside it —
 * otherwise it is decoration, and a captured request replays forever. The body
 * is the raw bytes as they will be transmitted: signing a parsed-and-
 * re-serialised body is the single most common way a first integration fails,
 * and it fails in the direction that looks like the platform's bug. */
function canonicalString(timestamp: string, rawBody: string): string {
  return `${SIGNATURE_SCHEME}:${timestamp}:${rawBody}`;
}
 
export function signDelivery({
  rawBody,
  timestamp,
  secret,
}: {
  rawBody: string;
  timestamp: string;
  secret: string;
}): string {
  return createHmac("sha256", secret)
    .update(canonicalString(timestamp, rawBody))
    .digest("hex");
}
 
/** One signature per valid secret. During a rotation window an endpoint has two,
 * and a recipient holding either must be able to verify — that is what makes a
 * 24-hour window survivable without a synchronised deploy on the customer's
 * side (contracts/webhooks.md §Rotation). */
export function signatureHeaders({
  rawBody,
  timestamp,
  secrets,
}: {
  rawBody: string;
  timestamp: string;
  secrets: string[];
}): Record<string, string> {
  const signatures = secrets.map(
    (secret) =>
      `${SIGNATURE_SCHEME}=${signDelivery({ rawBody, timestamp, secret })}`,
  );
  return {
    [TIMESTAMP_HEADER]: timestamp,
    [SIGNATURE_HEADER]: signatures.join(","),
  };
}

it("verifies against a verifier written from the documentation, not from the signer", () => {

Q1  delayed redelivery survives a restart?
    NAKKED  delay_ms=90000  at 1786500140980   (process then exits)
    ARRIVED at 1786500230983 in a fresh process — 3 ms late.  YES.
 
Q2  do delayed messages hold the ack-pending budget?
    max_ack_pending=3; nak'd 3 with a 300 s delay; fetched afterwards: 0
    num_ack_pending=3  num_pending=2            YES — and that is the problem.

@@ -1,10 +1,11 @@
 import { sql } from "drizzle-orm";
 import {
   bigserial,
   bigint,
+  boolean,
   check,
   index,
   integer,
   jsonb,
   pgTable,
   primaryKey,
@@ -356,6 +357,159 @@ export const consumedEvents = pgTable(
     handledAt: timestamp("handled_at", { withTimezone: true })
       .notNull()
       .defaultNow(),
   },
   (t) => [primaryKey({ columns: [t.consumer, t.eventId] })],
 );
+
+// ---------------------------------------------------------------------------
+// Webhooks (chapter 3.5). Three tables, and all three carry `environment_id`.
+//
+// DECISION (chapter 3.5): 3.3's `outbox` and 3.4's `consumed_events` each
+// omitted the tenant column and each recorded it as a deliberate exception. Two
+// exceptions in consecutive chapters is a pattern, and a pattern without a
+// stated rule is how a third chapter gets it wrong by resemblance. THE RULE: a
+// table below the tenant boundary may omit `environment_id` only when it holds
+// the platform's own bookkeeping AND no tenant-visible content. An endpoint is
+// customer configuration; a dead letter holds a payload that was being sent to a
+// customer. Both fail the test on both halves, so both are scoped and both join
+// chapter 3.7's cross-tenant gauntlet as targets.
+// ---------------------------------------------------------------------------
+
+// DECISION (chapter 3.5): no source document defines this table. FR-WHK-01 and
+// FR-WHK-08 require the behaviour — up to five endpoints per environment, each
+// with an independently rotatable signing secret — and leave the shape open.
+//
+// The secret is stored ENCRYPTED, not hashed, and the difference is the point.
+// An API key (3.2) is VERIFIED: a caller presents it, we hash what arrived and
+// compare. A signing secret is USED: we must compute an HMAC with it, which
+// needs the secret itself. A hash cannot be used, only compared. NFR-SEC-02
+// permits "salted hashes OR envelope encryption" and this is the branch that
+// applies — two credentials, one requirement, two mechanisms, because the verbs
+// differ (research R3).
+export const webhookEndpoints = pgTable(
+  "webhook_endpoints",
+  {
+    id: uuid("id").primaryKey(),
+    environmentId: uuid("environment_id")
+      .notNull()
+      .references(() => environments.id),
+    url: text("url").notNull(),
+    // The subscription set. An endpoint receives only these (FR-WHK-02).
+    eventTypes: jsonb("event_types").notNull(),
+    secretCiphertext: text("secret_ciphertext").notNull(),
+    // Non-null only during a rotation window: both secrets sign, so a recipient
+    // accepting either is correct throughout (contracts/webhooks.md §Rotation).
+    secretPreviousCiphertext: text("secret_previous_ciphertext"),
+    secretRotatedAt: timestamp("secret_rotated_at", { withTimezone: true }),
+    // An owner can pause an endpoint. What disables one AUTOMATICALLY after
+    // continuous failure is FR-WHK-07's, in the follow-on chapter — the column
+    // exists now so that chapter adds a rule rather than a migration.
+    enabled: boolean("enabled").notNull().default(true),
+    createdAt: timestamp("created_at", { withTimezone: true })
+      .notNull()
+      .defaultNow(),
+    // DELETION IS SOFT, and this column is why. Chapter 3.2 made the same
+    // choice for `api_keys.revokedAt` and gave the reason: "a deleted row loses
+    // the record of what once had access". Here the stakes are higher — a hard
+    // delete would have to cascade, and cascading would erase the customer's
+    // dead letters, which FR-WHK-04 says to retain for seven days. So a deleted
+    // endpoint stops receiving deliveries and stops appearing in listings, and
+    // the record of what it was survives its deletion.
+    deletedAt: timestamp("deleted_at", { withTimezone: true }),
+  },
+  (t) => [index("webhook_endpoints_environment_idx").on(t.environmentId)],
+);
+
+// The retry schedule — and it is chapter 3.3's outbox with one more column.
+//
+// DECISION (chapter 3.5, research R1, MEASURED): the obvious implementation is
+// to let the broker hold the delay between attempts. It was measured against a
+// real broker and disqualified: a delayed redelivery survives a restart to
+// within 3 ms, but a message waiting out its delay HOLDS AN ACKNOWLEDGEMENT
+// SLOT the whole time. With `max_ack_pending=3`, three messages nak'd for five
+// minutes made two available messages unfetchable. Scaled up, a handful of dead
+// customer endpoints starve deliveries to healthy ones — exactly what FR-WHK-05
+// forbids, and invisible until an incident.
+//
+// So a delivery that is not due yet is a ROW, not a message the broker holds.
+// The whole schedule is `next_attempt_at`; the api's relay publishes a delivery
+// only once it is already due. Nothing waits in the broker.
+export const webhookDeliveries = pgTable(
+  "webhook_deliveries",
+  {
+    id: uuid("id").primaryKey(),
+    environmentId: uuid("environment_id")
+      .notNull()
+      .references(() => environments.id),
+    endpointId: uuid("endpoint_id")
+      .notNull()
+      .references(() => webhookEndpoints.id),
+    // Chapter 3.3's envelope id — the customer's deduplication key, stable
+    // across every attempt and across a dead-letter replay (spec FR-018).
+    eventId: uuid("event_id").notNull(),
+    payload: jsonb("payload").notNull(),
+    // 1..7 — the INDEX INTO THE TIER TABLE, not a free-running counter.
+    // `attempt = 5` means "the 5-minute tier", which is what makes recomputing
+    // `next_attempt_at` total rather than incremental. Seven, not six: FR-WHK-03
+    // names six retry delays and the initial delivery makes seven requests —
+    // see webhooks/schedule.ts's DECISION for why the delay list won.
+    attempt: integer("attempt").notNull().default(1),
+    nextAttemptAt: timestamp("next_attempt_at", { withTimezone: true })
+      .notNull()
+      .defaultNow(),
+    // Set when the relay publishes it; cleared when the next attempt is
+    // scheduled. The relay's claim, in the shape `outbox.published_at` has.
+    dispatchedAt: timestamp("dispatched_at", { withTimezone: true }),
+    state: text("state").notNull().default("pending"),
+    createdAt: timestamp("created_at", { withTimezone: true })
+      .notNull()
+      .defaultNow(),
+  },
+  (t) => [
+    // One event produces at most one delivery per endpoint. This is what makes
+    // expansion idempotent at the database rather than by care (research R2).
+    unique("webhook_deliveries_event_endpoint_unique").on(t.eventId, t.endpointId),
+    check(
+      "webhook_deliveries_state_check",
+      sql`${t.state} IN ('pending','delivered','dead')`,
+    ),
+    // The relay's only query: pending rows that are due, oldest first. Partial,
+    // for the reason 3.3's outbox index is partial — it covers only what the
+    // relay reads, so delivered rows cost nothing to keep.
+    index("webhook_deliveries_due_idx")
+      .on(t.nextAttemptAt)
+      .where(sql`${t.state} = 'pending'`),
+  ],
+);
+
+// DECISION (chapter 3.5): FR-WHK-04 requires exhausted events to be retained
+// for seven days, inspectable and replayable, and leaves the shape open.
+//
+// This is the first store in the platform whose PURPOSE is retaining data that
+// failed to leave. Retention is therefore a liability rather than a feature: a
+// dead-letter table with no expiry is a place tenant data accumulates until an
+// audit finds it. Pruning is named and deferred; the chapter says what happens
+// on day eight and means it.
+export const webhookDeadLetters = pgTable(
+  "webhook_dead_letters",
+  {
+    id: uuid("id").primaryKey(),
+    environmentId: uuid("environment_id")
+      .notNull()
+      .references(() => environments.id),
+    endpointId: uuid("endpoint_id")
+      .notNull()
+      .references(() => webhookEndpoints.id),
+    // Reused on replay, so a customer who deduplicates correctly is unharmed by
+    // an operator replaying something they already received.
+    eventId: uuid("event_id").notNull(),
+    payload: jsonb("payload").notNull(),
+    lastStatus: integer("last_status"),
+    lastError: text("last_error"),
+    attempts: integer("attempts").notNull(),
+    deadLetteredAt: timestamp("dead_lettered_at", { withTimezone: true })
+      .notNull()
+      .defaultNow(),
+  },
+  (t) => [index("webhook_dead_letters_environment_idx").on(t.environmentId)],
+);

// The retry schedule (chapter 3.5, FR-WHK-03).
//
// Seven attempts — one immediate, then FR-WHK-03's six retries — and the delays
// are data rather than arithmetic. An exponential
// formula would be shorter and would not let each step carry its reason — and
// the reasons are the part a reader needs, because "exponential backoff" is a
// shape, not a decision.
//
// ---------------------------------------------------------------------------
// DECISION (chapter 3.5, taken by the author on 2026-08-10): FR-WHK-03 is
// internally inconsistent, and this is the reading taken.
//
// It says: "Failed deliveries shall be retried with exponential backoff at
// approximately 1 s, 5 s, 30 s, 5 min, 30 min, 2 h — six attempts in total."
//
// Six delays are listed, and an initial delivery followed by six retries is
// SEVEN requests — which the same sentence's "six attempts in total" forbids.
// The two halves cannot both be satisfied.
//
// THE DELAY LIST WINS. "Six attempts" is read as the six RETRIES, which is how
// the sentence's first clause uses the word ("shall be retried ... at"), and it
// keeps every number the requirement actually names — including the 2 h tier
// that gives a customer a working day's outage before their events are
// dead-lettered. Dropping that tier to preserve a count would have quietly
// shortened the platform's most customer-visible promise from two hours to
// thirty-six minutes.
//
// So: attempt 1 is immediate, attempts 2-7 follow at 1 s, 5 s, 30 s, 5 min,
// 30 min and 2 h, and the schedule spans about two hours thirty-six minutes.
//
// The SRS wording should be corrected to say "six retries" rather than "six
// attempts in total"; that is an amendment, not a chapter's decision to take
// alone, and it is recorded here until it happens.
//
// Worth noticing that this switch cost one table entry and one constant. The
// column, the relay and every invariant were untouched — which is a property of
// R1's re-plan, where the schedule stopped being broker state and became data.
// ---------------------------------------------------------------------------
 
const SECOND = 1000;
const MINUTE = 60 * SECOND;
const HOUR = 60 * MINUTE;
 
/** Delay before each attempt, indexed by `attempt - 1`.
 *
 * `attempt` is the INDEX INTO THIS TABLE, not a free-running counter. That is
 * what makes recomputing `next_attempt_at` total rather than incremental: given
 * a delivery row, the next due time is a lookup, and a bug in one branch cannot
 * leave a delivery drifting on a schedule nobody can reconstruct. */
export const RETRY_TIERS_MS: readonly number[] = [
  // Attempt 1 — immediate. A webhook that waits before its FIRST try would make
  // every customer's integration feel broken for a reason none of them could
  // see.
  0,
  // 1 s — the deploy-restart tier. Most first failures are a customer's process
  // being replaced, and a second is long enough for the new one to be listening.
  1 * SECOND,
  // 5 s — the transient tier: a connection reset, a cold lambda, a brief 502.
  5 * SECOND,
  // 30 s — long enough that a customer's own retry-and-recover has had a turn.
  30 * SECOND,
  // 5 min — the deploy tier: long enough for a rollback or a restart to have
  // finished, rather than catching a customer mid-deploy twice.
  5 * MINUTE,
  // 30 min — past here the endpoint is not blinking, it is down, and a human is
  // already involved on the customer's side.
  30 * MINUTE,
  // 2 h — the last one, and the reason the delay list won over the count. It
  // gives a customer most of a working day's outage to be noticed and fixed
  // before their events stop being retried at all. Anything still failing here
  // needs the dead-letter store and its seven days.
  2 * HOUR,
];
 
/** Seven requests: the initial delivery plus FR-WHK-03's six retries.
 * Exceeding it dead-letters. */
export const MAX_ATTEMPTS = RETRY_TIERS_MS.length;
 
/** When attempt N falls due, measured from the moment its predecessor failed.
 *
 * Returns null when there is no attempt N — the caller's signal to dead-letter
 * rather than to schedule. Making "no next tier" a value rather than an
 * exception keeps the outcome path a single expression. */
export function nextAttemptAt(attempt: number, from: Date = new Date()): Date | null {
  if (attempt < 1 || attempt > MAX_ATTEMPTS) return null;
  const delay = RETRY_TIERS_MS[attempt - 1];
  if (delay === undefined) return null;
  return new Date(from.getTime() + delay);
}

@@ -1,9 +1,9 @@
 import { randomUUID } from "node:crypto";
 
-import { and, asc, desc, eq, gt, lt, sql, type SQL } from "drizzle-orm";
+import { and, asc, desc, eq, gt, isNull, lt, sql, type SQL } from "drizzle-orm";
 
 import type { Db } from "./client";
 import {
   apiKeys,
   applications,
   channels,
@@ -13,14 +13,19 @@ import {
   members,
   memberships,
   messages,
   organisations,
   outbox,
   users,
+  webhookDeadLetters,
+  webhookDeliveries,
+  webhookEndpoints,
 } from "./schema";
 import { messageCreatedEvent } from "../outbox/event";
+import { nextAttemptAt } from "../webhooks/schedule";
+import { activeSigningSecrets } from "../webhooks/secret";
 import {
   mintApiKey,
   parseApiKeyCredential,
   prefixMatchesKind,
   secretMatches,
   type EnvironmentKind,
@@ -292,12 +297,393 @@ export async function drainOutbox(
     return published.length;
   });
 }
 
 /** How far behind the relay is. The single number worth alarming on later, and
  * the one the chapter shows going up while the broker is down. */
+/** The name this consumer claims events under. One name, because the ledger is
+ * keyed per consumer and the dispatcher is one consumer however many processes
+ * run it (chapter 3.4's data model). */
+export const DISPATCHER_CONSUMER = "dispatcher";
+
+/** Turn one event into one delivery per matching endpoint — **in one
+ * transaction** (chapter 3.5, research R2).
+ *
+ * Admin surface, like `drainOutbox`: one dispatcher serves every environment, so
+ * this cannot go through the scoped Repository. It is still safe, because the
+ * environment comes from the EVENT rather than from a caller's parameter.
+ *
+ * The claim is chapter 3.4's, unchanged, and it is doing more work here than it
+ * did there. The broker will redeliver — that is what at-least-once means — and
+ * an event expanded twice would double every webhook it produced. Because the
+ * claim and the N inserts share a transaction, "expansion runs exactly once"
+ * stops being something the code must be careful about and becomes a property of
+ * the database.
+ *
+ * An event no endpoint subscribes to is still CLAIMED, with zero rows created.
+ * Leaving it unclaimed would make every redelivery re-ask the same question
+ * forever. */
+export async function expandEventToDeliveries(
+  db: Db,
+  event: {
+    eventId: string;
+    environmentId: string;
+    type: string;
+    payload: unknown;
+  },
+): Promise<{ created: number; duplicate: boolean }> {
+  let created = 0;
+  const result = await claimEvent(
+    db,
+    DISPATCHER_CONSUMER,
+    event.eventId,
+    async () => {
+      const endpoints = await db
+        .select({
+          id: webhookEndpoints.id,
+          eventTypes: webhookEndpoints.eventTypes,
+        })
+        .from(webhookEndpoints)
+        .where(
+          and(
+            eq(webhookEndpoints.environmentId, event.environmentId),
+            eq(webhookEndpoints.enabled, true),
+            isNull(webhookEndpoints.deletedAt),
+          ),
+        );
+
+      // Subscription filtering happens HERE rather than at delivery time: a
+      // delivery row that exists and is never sent is a retry schedule with a
+      // permanent no-op in it, and an operator reading the table would have no
+      // way to tell it from work that is stuck.
+      const matching = endpoints.filter((e) =>
+        (e.eventTypes as string[]).includes(event.type),
+      );
+      if (matching.length === 0) return;
+
+      await db.insert(webhookDeliveries).values(
+        matching.map((endpoint) => ({
+          id: randomUUID(),
+          environmentId: event.environmentId,
+          endpointId: endpoint.id,
+          eventId: event.eventId,
+          payload: event.payload,
+        })),
+      );
+      created = matching.length;
+    },
+  );
+  return { created, duplicate: result === "duplicate" };
+}
+
+/** What the api did with an attempt's outcome. */
+export type DeliveryOutcome = "delivered" | "rescheduled" | "dead_lettered";
+
+/** Record one attempt's result, and decide what happens next — **in one
+ * transaction** (chapter 3.5).
+ *
+ * The three terminal paths are here together on purpose. Splitting "record the
+ * failure" from "schedule the next attempt" would allow a delivery marked failed
+ * with no next attempt scheduled: a webhook that stops without anyone being
+ * told, which is the failure mode a retry system exists to prevent.
+ *
+ * IDEMPOTENT on `(delivery_id, attempt)`. The dispatcher posts, then reports,
+ * then acknowledges; a crash between the POST and the acknowledgement means the
+ * delivery is redelivered and reported again. Recognising the repeat and
+ * returning the same answer is what makes that redelivery harmless — the POST
+ * itself may duplicate, and the customer absorbs it on the event id, but the
+ * SCHEDULE must not advance twice for one attempt or the tiers would collapse.
+ */
+export async function recordAttemptOutcome(
+  db: Db,
+  input: {
+    deliveryId: string;
+    attempt: number;
+    status?: number;
+    error?: string;
+  },
+): Promise<{ outcome: DeliveryOutcome; nextAttemptAt: Date | null }> {
+  return db.transaction(async (tx) => {
+    const [delivery] = await tx
+      .select({
+        id: webhookDeliveries.id,
+        environmentId: webhookDeliveries.environmentId,
+        endpointId: webhookDeliveries.endpointId,
+        eventId: webhookDeliveries.eventId,
+        payload: webhookDeliveries.payload,
+        attempt: webhookDeliveries.attempt,
+        state: webhookDeliveries.state,
+        nextAttemptAt: webhookDeliveries.nextAttemptAt,
+        dispatchedAt: webhookDeliveries.dispatchedAt,
+      })
+      .from(webhookDeliveries)
+      .where(eq(webhookDeliveries.id, input.deliveryId))
+      .for("update");
+
+    if (!delivery) throw new DeliveryNotFoundError(input.deliveryId);
+
+    // The idempotence check. A report for an attempt this delivery has already
+    // moved past is a repeat: answer with what was decided the first time and
+    // change nothing.
+    if (delivery.state !== "pending" || delivery.attempt !== input.attempt) {
+      return {
+        outcome:
+          delivery.state === "delivered"
+            ? ("delivered" as const)
+            : delivery.state === "dead"
+              ? ("dead_lettered" as const)
+              : ("rescheduled" as const),
+        nextAttemptAt:
+          delivery.state === "pending" ? delivery.nextAttemptAt : null,
+      };
+    }
+
+    const succeeded =
+      input.status !== undefined && input.status >= 200 && input.status < 300;
+
+    if (succeeded) {
+      await tx
+        .update(webhookDeliveries)
+        .set({ state: "delivered", dispatchedAt: null })
+        .where(eq(webhookDeliveries.id, delivery.id));
+      return { outcome: "delivered" as const, nextAttemptAt: null };
+    }
+
+    const next = nextAttemptAt(delivery.attempt + 1);
+    if (next) {
+      await tx
+        .update(webhookDeliveries)
+        .set({
+          attempt: delivery.attempt + 1,
+          nextAttemptAt: next,
+          // Cleared so the relay can pick it up again when it falls due.
+          dispatchedAt: null,
+        })
+        .where(eq(webhookDeliveries.id, delivery.id));
+      return { outcome: "rescheduled" as const, nextAttemptAt: next };
+    }
+
+    // Attempts exhausted. The dead letter and the state change commit together —
+    // a delivery marked dead with no dead letter behind it would be a failure
+    // with no record, which is exactly what FR-WHK-04's seven days are for.
+    await tx.insert(webhookDeadLetters).values({
+      id: randomUUID(),
+      environmentId: delivery.environmentId,
+      endpointId: delivery.endpointId,
+      eventId: delivery.eventId,
+      payload: delivery.payload,
+      lastStatus: input.status ?? null,
+      lastError: input.error ?? null,
+      attempts: delivery.attempt,
+    });
+    await tx
+      .update(webhookDeliveries)
+      .set({ state: "dead", dispatchedAt: null })
+      .where(eq(webhookDeliveries.id, delivery.id));
+    return { outcome: "dead_lettered" as const, nextAttemptAt: null };
+  });
+}
+
+/** Everything the dispatcher needs to sign and post one delivery.
+ *
+ * Admin surface: one dispatcher serves every environment. The environment is
+ * read from the DELIVERY rather than supplied by the caller, so this cannot be
+ * pointed at a tenant by anyone who does not already hold a delivery id.
+ *
+ * Returns DECRYPTED secrets — one, or two inside a 24-hour rotation window. This
+ * is the only place in the platform that hands a customer credential back in
+ * plaintext, and the obligations that come with it are stated in
+ * contracts/dispatcher.md rather than assumed. */
+export async function deliveryMaterial(
+  db: Db,
+  deliveryId: string,
+): Promise<{
+  delivery_id: string;
+  endpoint_id: string;
+  environment_id: string;
+  event_id: string;
+  url: string;
+  attempt: number;
+  secrets: string[];
+  payload: unknown;
+} | null> {
+  const [row] = await db
+    .select({
+      id: webhookDeliveries.id,
+      environmentId: webhookDeliveries.environmentId,
+      endpointId: webhookDeliveries.endpointId,
+      eventId: webhookDeliveries.eventId,
+      attempt: webhookDeliveries.attempt,
+      payload: webhookDeliveries.payload,
+      url: webhookEndpoints.url,
+      enabled: webhookEndpoints.enabled,
+      deletedAt: webhookEndpoints.deletedAt,
+      secretCiphertext: webhookEndpoints.secretCiphertext,
+      secretPreviousCiphertext: webhookEndpoints.secretPreviousCiphertext,
+      secretRotatedAt: webhookEndpoints.secretRotatedAt,
+    })
+    .from(webhookDeliveries)
+    .innerJoin(
+      webhookEndpoints,
+      eq(webhookDeliveries.endpointId, webhookEndpoints.id),
+    )
+    .where(eq(webhookDeliveries.id, deliveryId));
+
+  if (!row) return null;
+  // An endpoint paused or removed after the delivery was scheduled gets nothing.
+  // The spec's edge case: events already in the retry schedule for a removed
+  // endpoint must not be delivered.
+  if (!row.enabled || row.deletedAt) return null;
+
+  return {
+    delivery_id: row.id,
+    endpoint_id: row.endpointId,
+    environment_id: row.environmentId,
+    event_id: row.eventId,
+    url: row.url,
+    attempt: row.attempt,
+    secrets: activeSigningSecrets({
+      secretCiphertext: row.secretCiphertext,
+      secretPreviousCiphertext: row.secretPreviousCiphertext,
+      secretRotatedAt: row.secretRotatedAt,
+    }),
+    payload: row.payload,
+  };
+}
+
+/** A delivery that is due, as the relay claims it. */
+export interface DueDeliveryRow {
+  id: string;
+  environment_id: string;
+  endpoint_id: string;
+  event_id: string;
+  attempt: number;
+}
+
+/** Claim the deliveries that are due and hand each to `publish` — chapter 3.3's
+ * `drainOutbox` with ONE MORE PREDICATE (research R13).
+ *
+ * That is the whole point, and it is worth not obscuring: the reader built this
+ * loop two chapters ago. `SELECT … FOR UPDATE SKIP LOCKED`, publish, mark. The
+ * only difference is `AND next_attempt_at <= now()`, and that difference is the
+ * entire retry schedule.
+ *
+ * `SKIP LOCKED` matters here for the reason it mattered there: the api runs more
+ * than once, so two relays draining one table is the ordinary deployment rather
+ * than an edge case.
+ *
+ * NOTHING WAITS IN THE BROKER. A delivery enters the stream only once it is
+ * already due — which is the property research R1 measured the alternative
+ * against and found it wanting: a broker-held delay holds an acknowledgement
+ * slot the whole time it waits, so dead endpoints starve healthy ones. */
+export async function drainDueDeliveries(
+  db: Db,
+  limit: number,
+  publish: (row: DueDeliveryRow) => Promise<void>,
+): Promise<number> {
+  return db.transaction(async (tx) => {
+    const claimed = (await tx.execute(
+      sql`SELECT id, environment_id, endpoint_id, event_id, attempt
+            FROM webhook_deliveries
+           WHERE state = 'pending'
+             AND dispatched_at IS NULL
+             AND next_attempt_at <= now()
+           ORDER BY next_attempt_at, id
+           LIMIT ${limit}
+             FOR UPDATE SKIP LOCKED`,
+    )) as unknown as { rows: DueDeliveryRow[] };
+
+    const dispatched: string[] = [];
+    try {
+      for (const row of claimed.rows) {
+        await publish(row);
+        dispatched.push(row.id);
+      }
+    } finally {
+      // In the `finally` for 3.3's reason: whatever went wrong with row N+1,
+      // rows 1..N really did reach the broker and must not be published twice by
+      // this instance's next pass.
+      if (dispatched.length > 0) {
+        await tx.execute(
+          sql`UPDATE webhook_deliveries SET dispatched_at = now()
+               WHERE id = ANY(${sql.raw(
+                 `ARRAY[${dispatched.map((id) => `'${id}'`).join(",")}]::uuid[]`,
+               )})`,
+        );
+      }
+    }
+    return dispatched.length;
+  });
+}
+
+/** How many deliveries are waiting to become due. The number an operator watches
+ * when a customer says "we stopped receiving webhooks". */
+export async function pendingDeliveryDepth(db: Db): Promise<number> {
+  const result = (await db.execute(
+    sql`SELECT count(*)::int AS pending
+          FROM webhook_deliveries
+         WHERE state = 'pending'`,
+  )) as unknown as { rows: { pending: number }[] };
+  return result.rows[0]?.pending ?? 0;
+}
+
+/** Put a dead-lettered delivery back on the schedule.
+ *
+ * Resets the EXISTING delivery row rather than inserting a new one, and that is
+ * forced by the shape of the data rather than chosen: `UNIQUE (event_id,
+ * endpoint_id)` is what makes expansion idempotent, so a second row for the same
+ * pair cannot exist. Reusing the row therefore preserves the original
+ * `event_id` — the identifier the customer deduplicates on — by construction
+ * instead of by remembering to copy it.
+ *
+ * **Current configuration, automatically.** The URL and the signing secrets are
+ * read from the endpoint at SEND time by `deliveryMaterial`, never stored on the
+ * delivery. So a replay of something that failed against a broken URL goes to
+ * whatever the endpoint says today, which is the whole reason anyone asks for a
+ * replay. Nothing here has to arrange that.
+ *
+ * **The dead-letter record is left alone.** FR-WHK-04 retains it for seven days,
+ * and a replay is a new attempt rather than an erasure of the fact that the
+ * attempts once ran out. Deleting it would remove the only evidence a customer's
+ * endpoint was broken at the moment somebody retried it.
+ *
+ * Returns false when there is no such dead letter — the controller's 404. */
+export async function replayDeadLetter(
+  db: Db,
+  deadLetterId: string,
+): Promise<boolean> {
+  return db.transaction(async (tx) => {
+    const [dead] = await tx
+      .select({
+        eventId: webhookDeadLetters.eventId,
+        endpointId: webhookDeadLetters.endpointId,
+      })
+      .from(webhookDeadLetters)
+      .where(eq(webhookDeadLetters.id, deadLetterId));
+    if (!dead) return false;
+
+    await tx
+      .update(webhookDeliveries)
+      .set({
+        state: "pending",
+        // Tier 1: a replay is a fresh chance, not a continuation of a schedule
+        // that has already run out.
+        attempt: 1,
+        nextAttemptAt: new Date(),
+        dispatchedAt: null,
+      })
+      .where(
+        and(
+          eq(webhookDeliveries.eventId, dead.eventId),
+          eq(webhookDeliveries.endpointId, dead.endpointId),
+        ),
+      );
+    return true;
+  });
+}
+
 export async function outboxDepth(db: Db): Promise<number> {
   const result = (await db.execute(
     sql`SELECT count(*)::int AS pending FROM outbox WHERE published_at IS NULL`,
   )) as unknown as { rows: { pending: number }[] };
   return result.rows[0]?.pending ?? 0;
 }
@@ -608,21 +994,257 @@ export class ChannelNotFoundError extends Error {
  * millisecond precision) — the driver hands back a Date or a string
  * depending on the column and the query shape. */
 function toIso(value: Date | string): string {
   return value instanceof Date ? value.toISOString() : String(value);
 }
 
+/** A webhook endpoint as the management surface returns it. The ciphertext is
+ * absent by construction rather than by filtering — a read path that can return
+ * it is one refactor away from a response that does. */
+export interface WebhookEndpointRow {
+  id: string;
+  url: string;
+  event_types: string[];
+  enabled: boolean;
+  secret_rotated_at: string | null;
+  created_at: string;
+}
+
+/** Raised when an outcome names a delivery that is not there. A caller error,
+ * not a platform one — the controller turns it into a 404. */
+export class DeliveryNotFoundError extends Error {
+  constructor(id: string) {
+    super(`no such delivery: ${id}`);
+    this.name = "DeliveryNotFoundError";
+  }
+}
+
+export interface WebhookDeliveryRow {
+  id: string;
+  endpoint_id: string;
+  event_id: string;
+  attempt: number;
+  state: string;
+  next_attempt_at: string;
+  /** Non-null once the relay has published it. Exposed because the drain is
+   * GLOBAL — one dispatcher serves every environment — so a test that asserts on
+   * what ITS call to the drain returned is asserting on which suite got there
+   * first. Chapter 3.3's finding 3, in its third chapter. */
+  dispatched_at: string | null;
+}
+
+export interface WebhookDeadLetterRow {
+  id: string;
+  endpoint_id: string;
+  event_id: string;
+  last_status: number | null;
+  last_error: string | null;
+  attempts: number;
+  dead_lettered_at: string;
+}
+
 export class Repository {
   // Constructor parameter properties — the shorthand chapter 1.4 released
   // for this service when ADR-15 spent erasableSyntaxOnly on decorator
   // metadata. The guarantee still holds in the gateway and every package.
   constructor(
     private readonly db: Db,
     private readonly environmentId: string,
   ) {}
 
+  // ---------------------------------------------------------------------
+  // Webhook endpoints (chapter 3.5). Scoped like everything else on this class:
+  // the environment comes from the constructor and never from a caller, so a
+  // handler cannot ask for another tenant's endpoints even by accident
+  // (constitution I).
+  //
+  // Every read here excludes soft-deleted rows. That is the whole contract of
+  // `deleted_at`: the row survives so its dead letters can, and it is invisible
+  // to everything else.
+  // ---------------------------------------------------------------------
+
+  private get liveEndpoints() {
+    return and(
+      eq(webhookEndpoints.environmentId, this.environmentId),
+      isNull(webhookEndpoints.deletedAt),
+    );
+  }
+
+  async countEndpoints(): Promise<number> {
+    const rows = await this.db
+      .select({ id: webhookEndpoints.id })
+      .from(webhookEndpoints)
+      .where(this.liveEndpoints);
+    return rows.length;
+  }
+
+  async createEndpoint(input: {
+    url: string;
+    eventTypes: string[];
+    secretCiphertext: string;
+  }): Promise<WebhookEndpointRow> {
+    const id = randomUUID();
+    await this.db.insert(webhookEndpoints).values({
+      id,
+      environmentId: this.environmentId,
+      url: input.url,
+      eventTypes: input.eventTypes,
+      secretCiphertext: input.secretCiphertext,
+    });
+    const row = await this.getEndpoint(id);
+    if (!row) throw new Error("endpoint vanished immediately after insert");
+    return row;
+  }
+
+  async listEndpoints(): Promise<WebhookEndpointRow[]> {
+    return this.selectEndpoints(this.liveEndpoints);
+  }
+
+  async getEndpoint(id: string): Promise<WebhookEndpointRow | null> {
+    const rows = await this.selectEndpoints(
+      and(eq(webhookEndpoints.id, id), this.liveEndpoints),
+    );
+    return rows[0] ?? null;
+  }
+
+  /** Opens a 24-hour rotation window: the outgoing secret keeps signing until it
+   * closes, so a recipient accepting either is correct throughout
+   * (contracts/webhooks.md §Rotation). */
+  async rotateEndpointSecret(
+    id: string,
+    secretCiphertext: string,
+  ): Promise<WebhookEndpointRow | null> {
+    const current = await this.getEndpoint(id);
+    if (!current) return null;
+    const [previous] = await this.db
+      .select({ secretCiphertext: webhookEndpoints.secretCiphertext })
+      .from(webhookEndpoints)
+      .where(and(eq(webhookEndpoints.id, id), this.liveEndpoints));
+    await this.db
+      .update(webhookEndpoints)
+      .set({
+        secretCiphertext,
+        secretPreviousCiphertext: previous?.secretCiphertext ?? null,
+        secretRotatedAt: new Date(),
+      })
+      .where(and(eq(webhookEndpoints.id, id), this.liveEndpoints));
+    return this.getEndpoint(id);
+  }
+
+  async setEndpointEnabled(
+    id: string,
+    enabled: boolean,
+  ): Promise<WebhookEndpointRow | null> {
+    await this.db
+      .update(webhookEndpoints)
+      .set({ enabled })
+      .where(and(eq(webhookEndpoints.id, id), this.liveEndpoints));
+    return this.getEndpoint(id);
+  }
+
+  /** SOFT. A hard delete would have to cascade, and cascading would erase the
+   * customer's dead letters — which FR-WHK-04 says to retain for seven days.
+   * Returns false when there was nothing live to delete, which the controller
+   * turns into the same 404 a foreign tenant gets. */
+  async deleteEndpoint(id: string): Promise<boolean> {
+    const existing = await this.getEndpoint(id);
+    if (!existing) return false;
+    await this.db
+      .update(webhookEndpoints)
+      .set({ deletedAt: new Date() })
+      .where(and(eq(webhookEndpoints.id, id), this.liveEndpoints));
+    return true;
+  }
+
+  /** Every delivery this event produced, for this environment. Scoped like
+   * everything else on this class — the drain is global, but a tenant's view of
+   * its own deliveries is not. */
+  async listDeliveriesForEvent(eventId: string): Promise<WebhookDeliveryRow[]> {
+    const rows = await this.db
+      .select({
+        id: webhookDeliveries.id,
+        endpointId: webhookDeliveries.endpointId,
+        eventId: webhookDeliveries.eventId,
+        attempt: webhookDeliveries.attempt,
+        state: webhookDeliveries.state,
+        nextAttemptAt: webhookDeliveries.nextAttemptAt,
+        dispatchedAt: webhookDeliveries.dispatchedAt,
+      })
+      .from(webhookDeliveries)
+      .where(
+        and(
+          eq(webhookDeliveries.environmentId, this.environmentId),
+          eq(webhookDeliveries.eventId, eventId),
+        ),
+      )
+      .orderBy(asc(webhookDeliveries.id));
+    return rows.map((r) => ({
+      id: r.id,
+      endpoint_id: r.endpointId,
+      event_id: r.eventId,
+      attempt: r.attempt,
+      state: r.state,
+      next_attempt_at: r.nextAttemptAt.toISOString(),
+      dispatched_at: r.dispatchedAt?.toISOString() ?? null,
+    }));
+  }
+
+  /** A tenant's dead letters, newest first. Scoped: a dead letter holds a
+   * payload that was being sent to this customer, which is why the table carries
+   * `environment_id` where 3.3's outbox and 3.4's ledger did not. */
+  async listDeadLetters(): Promise<WebhookDeadLetterRow[]> {
+    const rows = await this.db
+      .select({
+        id: webhookDeadLetters.id,
+        endpointId: webhookDeadLetters.endpointId,
+        eventId: webhookDeadLetters.eventId,
+        lastStatus: webhookDeadLetters.lastStatus,
+        lastError: webhookDeadLetters.lastError,
+        attempts: webhookDeadLetters.attempts,
+        deadLetteredAt: webhookDeadLetters.deadLetteredAt,
+      })
+      .from(webhookDeadLetters)
+      .where(eq(webhookDeadLetters.environmentId, this.environmentId))
+      .orderBy(desc(webhookDeadLetters.deadLetteredAt));
+    return rows.map((r) => ({
+      id: r.id,
+      endpoint_id: r.endpointId,
+      event_id: r.eventId,
+      last_status: r.lastStatus,
+      last_error: r.lastError,
+      attempts: r.attempts,
+      dead_lettered_at: r.deadLetteredAt.toISOString(),
+    }));
+  }
+
+  private async selectEndpoints(
+    where: ReturnType<typeof and>,
+  ): Promise<WebhookEndpointRow[]> {
+    const rows = await this.db
+      .select({
+        id: webhookEndpoints.id,
+        url: webhookEndpoints.url,
+        eventTypes: webhookEndpoints.eventTypes,
+        enabled: webhookEndpoints.enabled,
+        secretRotatedAt: webhookEndpoints.secretRotatedAt,
+        createdAt: webhookEndpoints.createdAt,
+      })
+      .from(webhookEndpoints)
+      .where(where);
+    // Never the ciphertext. A read path that can return it is one refactor away
+    // from a response that does.
+    return rows.map((r) => ({
+      id: r.id,
+      url: r.url,
+      event_types: r.eventTypes as string[],
+      enabled: r.enabled,
+      secret_rotated_at: r.secretRotatedAt?.toISOString() ?? null,
+      created_at: r.createdAt.toISOString(),
+    }));
+  }
+
   async createUser(externalId: string, displayName?: string): Promise<UserRow> {
     const id = randomUUID();
     await this.db.insert(users).values({
       id,
       environmentId: this.environmentId,
       externalId,

import { createRequire } from "node:module";
import { createServer, type Server } from "node:http";
import { dirname, join } from "node:path";
import { randomUUID } from "node:crypto";
import { fileURLToPath } from "node:url";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
 
import { boot, type System } from "./harness.js";
 
// Invariant 14, first half (chapter 3.5): with the dispatcher absent, end users
// are served exactly as before.
//
// This is the journey that shows what the service split BOUGHT. Inside the api,
// "webhooks do not delay end users" would be a claim about event loops and
// connection pools — something you argue for. Across a process boundary it is a
// claim about processes, and you settle it by not starting one.
//
// The dispatcher is never started in this file. It does not exist for the
// duration, and the platform is expected not to notice.
//
// THE OTHER HALF — that a dispatcher starting later drains the backlog — lives
// in the dispatcher's own suite, which can start and stop one. Asserting it here
// would mean this journey managing a service the harness does not own, and a
// journey that boots half a platform to prove something a unit of it can prove
// is a slower test with a weaker claim.
 
const require_ = createRequire(import.meta.url);
const REPO = join(dirname(fileURLToPath(import.meta.url)), "..", "..", "..");
 
/** A customer's server. Nothing should reach it while the dispatcher is absent,
 * and that silence is the assertion. */
function customerEndpoint() {
  const received: string[] = [];
  const server: Server = createServer((req, res) => {
    let body = "";
    req.on("data", (c) => (body += String(c)));
    req.on("end", () => {
      received.push(body);
      res.writeHead(200).end("ok");
    });
  });
  return {
    received,
    listen: () =>
      new Promise<string>((resolve) =>
        server.listen(0, () => {
          const addr = server.address();
          resolve(
            `http://127.0.0.1:${typeof addr === "object" && addr ? addr.port : 0}`,
          );
        }),
      ),
    close: () => server.close(),
  };
}
 
function apiInternals() {
  const dist = join(REPO, "services", "api", "dist");
  const client = require_(join(dist, "db", "client.js")) as {
    createDb: (p: unknown) => unknown;
    createPool: () => unknown;
  };
  const seeder = require_(join(dist, "db", "repository.js")) as {
    Repository: new (
      db: unknown,
      env: string,
    ) => {
      createEndpoint: (i: {
        url: string;
        eventTypes: string[];
        secretCiphertext: string;
      }) => Promise<{ id: string }>;
    };
  };
  const secrets = require_(join(dist, "webhooks", "secret.js")) as {
    encryptSecret: (s: string) => string;
    mintSigningSecret: () => string;
  };
  return { db: client.createDb(client.createPool()), seeder, secrets };
}
 
describe("journey: webhooks never stand between the platform and its users", () => {
  let system: System;
  let endpoint: ReturnType<typeof customerEndpoint>;
  let endpointUrl: string;
 
  beforeAll(async () => {
    endpoint = customerEndpoint();
    endpointUrl = await endpoint.listen();
    // No dispatcher. The api and one gateway come up; nothing consumes
    // deliveries, and nothing is supposed to need to.
    system = await boot({ gateways: 1 });
  }, 120_000);
 
  afterAll(async () => {
    await system?.stop();
    endpoint?.close();
  });
 
  it("invariant 14: an end user is served while the dispatcher does not exist", async () => {
    // `dispatcher` here is the FLEET DISPATCHER — a person, from chapter 2.8's
    // cast — not this chapter's service. Renamed locally, because a file about
    // the webhook dispatcher that also has a variable called `dispatcher`
    // meaning something else is a trap for whoever reads it next.
    const {
      environmentId,
      channel,
      dispatcher: fleetDispatcher,
      tuan,
    } = await system.seedConversation();
 
    // The clients are constructed, not connected — the journey connects them to
    // the gateways it wants them on.
    await fleetDispatcher.connect(system.gateways[0]!);
    await tuan.connect(system.gateways[0]!);
 
    // A configured endpoint, so there IS webhook work for the absent service to
    // be failing to do. Seeded through the repository on purpose: this journey
    // is about the delivery path, and the management surface has its own suite.
    const { db, seeder, secrets } = apiInternals();
    const repo = new seeder.Repository(db, environmentId);
    await repo.createEndpoint({
      url: `${endpointUrl}/hook`,
      eventTypes: ["message.created"],
      secretCiphertext: secrets.encryptSecret(secrets.mintSigningSecret()),
    });
 
    const text = `north ramp ${randomUUID().slice(0, 6)}`;
    fleetDispatcher.send(channel, text);
 
    // The whole assertion: message delivery does not consult the webhook path,
    // does not wait for it, and does not care that the service which would run
    // it was never started.
    await tuan.waitFor(
      (frame) =>
        frame.type === "message.created" &&
        (frame.payload as { text?: string }).text === text,
      `tuan hears "${text}" with no dispatcher running`,
      15_000,
    );
 
    expect(tuan.timeline(channel).some((m) => m.text === text)).toBe(true);
 
    // And the customer's endpoint heard nothing, because nothing is dispatching.
    // If this ever fails, some other process is delivering webhooks and every
    // other assertion in this file is measuring the wrong thing.
    expect(endpoint.received).toHaveLength(0);
  }, 60_000);
});

@@ -18,13 +18,46 @@ export interface ApplicationPrincipal {
 export interface UserPrincipal {
   kind: "user";
   environmentId: string;
   userExternalId: string;
 }
 
-export type Principal = ApplicationPrincipal | UserPrincipal;
+/** The platform acting for itself (chapter 3.5).
+ *
+ * The dispatcher is the first caller of the internal seam that is neither a
+ * tenant's software nor an end user: it consumes every environment's events and
+ * asks the api to write on behalf of all of them.
+ *
+ * THE SHORTCUT THAT WOULD HAVE WORKED AND BEEN WRONG: mint the dispatcher an API
+ * key. It would authenticate on the first try — and an `application` principal
+ * is scoped to exactly ONE environment by construction (3.1, 3.2), so a
+ * dispatcher holding one either cannot serve other tenants or has been quietly
+ * granted cross-tenant reach through the credential type whose entire meaning is
+ * that it has none. Principle I is a correctness property, and that is the shape
+ * its erosion would take.
+ *
+ * So this kind carries NO `environmentId`. That is not an omission — it is what
+ * stops it being usable anywhere a tenant is expected: every environment-scoped
+ * provider reads `environmentId`, and a principal without one cannot silently
+ * become a tenant's. It is accepted only where a route opts in, and never on a
+ * public route. */
+export interface PlatformPrincipal {
+  kind: "platform";
+  /** Which internal service presented it, for logs. Never the credential. */
+  service: string;
+  /** Present and always undefined, so the environment-scoped providers that read
+   * `principal?.environmentId` keep compiling AND keep getting nothing. A
+   * platform principal reaching a tenant-scoped repository yields the empty
+   * scope, which the guard has already refused before any handler runs. */
+  environmentId?: undefined;
+}
+
+export type Principal =
+  | ApplicationPrincipal
+  | UserPrincipal
+  | PlatformPrincipal;
 export type PrincipalKind = Principal["kind"];
 
 /** The request as everything downstream of the middleware sees it. The
  * principal is optional at the type level for one honest reason: a request that
  * presented nothing has none, and pre-credential routes (signup) are reached
  * exactly that way. */
@@ -34,13 +67,15 @@ export interface RequestWithPrincipal {
 }
 
 /** How a credential class is named to a human. Used by the wrong-credential
  * error, which must say what was presented and what was expected — and must
  * never quote the credential (NFR-SEC-06). */
 export function describePrincipalKind(kind: PrincipalKind): string {
-  return kind === "application" ? "an API key" : "an end-user token";
+  if (kind === "application") return "an API key";
+  if (kind === "platform") return "an internal platform credential";
+  return "an end-user token";
 }
 
 /** The `Bearer <credential>` half of RFC 6750, and nothing else. Query strings
  * are refused by omission: URLs reach logs and referrer headers, and NFR-SEC-06
  * forbids a credential in either. (The WebSocket upgrade is the one exception,
  * and it lives in the gateway where a browser gives no other choice.) */

@@ -330,7 +330,34 @@ describe("credentials", () => {
     const channel = await repo.createChannel("signup-key", "public");
     expect(
       (await post({ text: "bootstrapped" }, first.apiKey!.secret, channel.id))
         .status,
     ).toBe(201);
   });
+
+  // --- chapter 3.5: the third principal ----------------------------------
+
+  describe("the internal platform credential", () => {
+    // SET, not read. This began as `process.env.RELAY_INTERNAL_CREDENTIAL` with
+    // an early return when it was absent — and CI never set it, so the one
+    // assertion standing between a platform credential and a public route
+    // silently did nothing on every build. A security test that skips itself is
+    // worse than no test: it reports green about a question it never asked.
+    const PLATFORM = "rk_svc_credentials_itest_0123456789abcdef01234";
+    process.env["RELAY_INTERNAL_CREDENTIAL"] = PLATFORM;
+
+    it("is refused on a public route, whatever else it can do", async () => {
+
+      // The whole point of the kind. It reaches every environment, so a public
+      // route accepting it would be a cross-tenant hole with a valid credential
+      // in front of it. 403 `wrong_credential_type` — the route's default is
+      // application-or-user, and platform is neither.
+      const res = await post({ text: "should never land" }, PLATFORM);
+
+      expect(res.status).toBe(403);
+      const body = (await res.json()) as { code?: string; message?: string };
+      expect(body.code).toBe("wrong_credential_type");
+      // And it must not quote the credential back (NFR-SEC-06).
+      expect(JSON.stringify(body)).not.toContain(PLATFORM);
+    });
+  });
 });

@@ -52,12 +52,20 @@ export class DevTokenController {
   @UseGuards(CredentialGuard)
   async mint(
     @Body(new ZodValidationPipe(devTokenRequestSchema)) body: DevTokenRequest,
     @Req() req: RequestWithPrincipal,
   ): Promise<{ token: string; expires_at: string }> {
     const principal = req.principal!;
+    // The guard above already refused anything but an API key, so this is
+    // unreachable — but chapter 3.5 added a third principal kind that carries no
+    // environment at all, and an assumption the compiler cannot see is one a
+    // later refactor can quietly break. Narrowing here costs a line and makes
+    // `@Accepts("application")` a fact rather than a promise.
+    if (principal.kind !== "application") {
+      throw new BadRequestException("an API key is required");
+    }
     const environment = await environmentSigningSecret(
       this.db,
       principal.environmentId,
     );
     if (!environment) throw new BadRequestException("unknown environment");

@@ -136,12 +136,124 @@ export const internalMembershipsResponseSchema = z.strictObject({
 export const internalSessionResponseSchema = z.strictObject({
   environment_id: z.string().min(1),
   user: z.string().min(1),
   channel_ids: z.array(z.string().min(1)),
 });
 
+/** The deliveries stream (chapter 3.5), and its subject grammar.
+ *
+ * Here rather than in either service, for the reason chapter 3.4 moved the event
+ * grammar here: a consumer that assembles its own subject filter receives
+ * nothing the day the grammar changes — no error, no warning, just an empty
+ * stream position. Two sides, one definition.
+ *
+ * The stream carries only work that is ALREADY DUE. A delivery waiting out a
+ * retry tier is a row in Postgres, not a message the broker is holding — which
+ * is what keeps a dead endpoint from occupying an acknowledgement slot for two
+ * hours (research R1, measured). */
+export const DELIVERIES_STREAM = "DELIVERIES";
+export const DELIVERY_SUBJECT_PREFIX = "deliveries";
+export const ALL_DELIVERIES_SUBJECT = `${DELIVERY_SUBJECT_PREFIX}.>`;
+
+/** One subject per environment, so a future per-tenant dispatcher shard is a
+ * filter change rather than a redesign. */
+export function deliverySubjectFor(environmentId: string): string {
+  if (!environmentId) throw new Error("an environment id is required");
+  return `${DELIVERY_SUBJECT_PREFIX}.${environmentId}`;
+}
+
+// ---------------------------------------------------------------------------
+// The dispatch contract (chapter 3.5, constitution IV).
+//
+// The dispatcher owns no database. "Only the API service writes to PostgreSQL…
+// Other services obtain writes and backfill reads via the API service's internal
+// endpoints." These are those endpoints, and they live here for the reason
+// chapter 2.5 put the gateway's here: both sides validate against ONE definition,
+// so the day a field is renamed the other side fails loudly instead of reading
+// `undefined` three layers away.
+// ---------------------------------------------------------------------------
+
+/** dispatcher → api: turn one event into one delivery per matching endpoint.
+ *
+ * A CLAIMED write: the api reuses chapter 3.4's deduplication ledger, so an event
+ * expanded twice would double every webhook it produced and cannot. The claim and
+ * all N delivery rows commit in one transaction (research R2). */
+export const internalExpandRequestSchema = z.strictObject({
+  event_id: z.string().uuid(),
+  environment_id: z.string().min(1),
+  type: z.string().min(1),
+  /** The envelope as it will be delivered, byte-identical to what 3.3 published.
+   * The dispatcher does not author payloads; it moves them. */
+  payload: z.unknown(),
+});
+
+export const internalExpandResponseSchema = z.strictObject({
+  /** How many deliveries this event produced. Zero is normal and not an error:
+   * no endpoint in that environment subscribes to this type. */
+  created: z.number().int().nonnegative(),
+  /** True when the ledger recognised the event as already expanded. The
+   * dispatcher acknowledges either way — that is what makes a redelivery safe. */
+  duplicate: z.boolean(),
+});
+
+/** dispatcher → api: everything needed to sign and post one delivery.
+ *
+ * The response carries DECRYPTED signing secrets, which is a real widening of
+ * where a customer credential exists. Internal network only, never logged, held
+ * for the signature and no longer (contracts/dispatcher.md). */
+export const internalDeliveryMaterialSchema = z.strictObject({
+  delivery_id: z.string().uuid(),
+  endpoint_id: z.string().uuid(),
+  environment_id: z.string().min(1),
+  event_id: z.string().uuid(),
+  url: z.string().url(),
+  attempt: z.number().int().positive(),
+  /** One or two: two during a 24-hour rotation window, so a recipient holding
+   * either can verify (contracts/webhooks.md §Rotation). */
+  secrets: z.array(z.string().min(1)).min(1).max(2),
+  payload: z.unknown(),
+});
+
+/** dispatcher → api: what happened when we posted.
+ *
+ * NOT a claim — the POST already happened on somebody else's machine and cannot
+ * be undone. Idempotent on `(delivery_id, attempt)`, so a redelivery arriving
+ * after a successful report is recognised and simply acknowledged rather than
+ * posted again (research R5). */
+export const internalDeliveryOutcomeRequestSchema = z.strictObject({
+  delivery_id: z.string().uuid(),
+  attempt: z.number().int().positive(),
+  /** Absent when the attempt never got a response — a timeout or a refused
+   * connection. The platform can only believe a status code it received. */
+  status: z.number().int().optional(),
+  error: z.string().max(2000).optional(),
+  latency_ms: z.number().int().nonnegative(),
+});
+
+export const internalDeliveryOutcomeResponseSchema = z.strictObject({
+  /** What the api did with it: delivered, scheduled for another tier, or moved
+   * to the dead-letter store because the attempts are exhausted. */
+  outcome: z.enum(["delivered", "rescheduled", "dead_lettered"]),
+  /** Present when rescheduled — when the next attempt falls due. */
+  next_attempt_at: z.iso.datetime().optional(),
+});
+
+export type InternalExpandRequest = z.infer<typeof internalExpandRequestSchema>;
+export type InternalExpandResponse = z.infer<
+  typeof internalExpandResponseSchema
+>;
+export type InternalDeliveryMaterial = z.infer<
+  typeof internalDeliveryMaterialSchema
+>;
+export type InternalDeliveryOutcomeRequest = z.infer<
+  typeof internalDeliveryOutcomeRequestSchema
+>;
+export type InternalDeliveryOutcomeResponse = z.infer<
+  typeof internalDeliveryOutcomeResponseSchema
+>;
+
 export type InternalSendRequest = z.infer<typeof internalSendRequestSchema>;
 export type InternalSessionResponse = z.infer<
   typeof internalSessionResponseSchema
 >;
 export type InternalSendResponse = z.infer<typeof internalSendResponseSchema>;
 export type InternalMembershipsResponse = z.infer<

import {
  Body,
  Controller,
  Inject,
  NotFoundException,
  Post,
  HttpCode,
  UseGuards,
} from "@nestjs/common";
 
import {
  internalDeliveryOutcomeRequestSchema,
  internalExpandRequestSchema,
  type InternalDeliveryOutcomeRequest,
  type InternalDeliveryOutcomeResponse,
  type InternalExpandRequest,
  type InternalExpandResponse,
} from "@relay/protocol";
 
import { Accepts, CredentialGuard } from "../auth/credential.guard";
import type { Db } from "../db/client";
import {
  DeliveryNotFoundError,
  deliveryMaterial,
  expandEventToDeliveries,
  recordAttemptOutcome,
  replayDeadLetter,
} from "../db/repository";
import { ZodValidationPipe } from "../messages/zod-validation.pipe";
 
// The dispatcher's only road to state (chapter 3.5, constitution IV).
//
// "Only the API service writes to PostgreSQL… Other services obtain writes and
// backfill reads via the API service's internal endpoints." The dispatcher owns
// no database, so everything it needs is here — and that constraint is not a
// workaround, it is the reason chapter 3.4's claim-and-effect-in-one-transaction
// pattern stops applying and the chapter has something to say.
//
// `@Accepts("platform")` and nothing else. These routes reach EVERY environment,
// which is exactly why no tenant credential may use them: an API key is scoped to
// one environment by construction, and a route that accepted one here would
// either be useless to the dispatcher or would have to ignore the scope — and
// ignoring a tenant scope is the shape a cross-tenant hole takes.
@Controller("internal/dispatch")
@UseGuards(CredentialGuard)
@Accepts("platform")
export class DispatchController {
  constructor(@Inject("DB") private readonly db: Db) {}
 
  /** One event becomes one delivery per matching endpoint — claimed, so it
   * happens exactly once however often the broker redelivers (research R2). */
  @Post("expand")
  @HttpCode(200)
  async expand(
    @Body(new ZodValidationPipe(internalExpandRequestSchema))
    body: InternalExpandRequest,
  ): Promise<InternalExpandResponse> {
    const result = await expandEventToDeliveries(this.db, {
      eventId: body.event_id,
      environmentId: body.environment_id,
      type: body.type,
      payload: body.payload,
    });
    return { created: result.created, duplicate: result.duplicate };
  }
 
  /** Everything needed to sign and post one delivery, including the DECRYPTED
   * signing secrets — one, or two during a rotation window.
   *
   * This is the one response in the platform that carries a customer credential
   * in plaintext. Internal network only, never logged at any level, and held by
   * the dispatcher for the duration of a signature and no longer
   * (contracts/dispatcher.md §The secret crosses a process boundary). */
  @Post("material")
  @HttpCode(200)
  async material(@Body() body: { delivery_id?: string }) {
    if (!body?.delivery_id) throw new NotFoundException("no such delivery");
    const material = await deliveryMaterial(this.db, body.delivery_id);
    if (!material) throw new NotFoundException("no such delivery");
    return material;
  }
 
  /** What happened when we posted. Idempotent on `(delivery_id, attempt)`: the
   * dispatcher posts, reports, then acknowledges, so a crash in the last gap
   * means this arrives twice. The POST may duplicate — the customer absorbs that
   * on the event id — but the SCHEDULE must not advance twice for one attempt. */
  @Post("outcome")
  @HttpCode(200)
  async outcome(
    @Body(new ZodValidationPipe(internalDeliveryOutcomeRequestSchema))
    body: InternalDeliveryOutcomeRequest,
  ): Promise<InternalDeliveryOutcomeResponse> {
    try {
      // Spread rather than assign: `exactOptionalPropertyTypes` is on, so an
      // explicit `undefined` is not the same as an absent key — and the
      // difference is real here, because "no status" means the attempt never got
      // a response at all.
      const result = await recordAttemptOutcome(this.db, {
        deliveryId: body.delivery_id,
        attempt: body.attempt,
        ...(body.status !== undefined ? { status: body.status } : {}),
        ...(body.error !== undefined ? { error: body.error } : {}),
      });
      return {
        outcome: result.outcome,
        ...(result.nextAttemptAt
          ? { next_attempt_at: result.nextAttemptAt.toISOString() }
          : {}),
      };
    } catch (error) {
      if (error instanceof DeliveryNotFoundError) {
        throw new NotFoundException("no such delivery");
      }
      throw error;
    }
  }
 
  /** Put a dead-lettered delivery back on the schedule, with the endpoint's
   * CURRENT url and secret — which is automatic, because delivery material is
   * read at send time rather than frozen into the delivery.
   *
   * Internal for now. FR-WHK-04 also asks for dead letters to be inspectable and
   * replayable FROM THE DASHBOARD, and the dashboard is chapter 5.2's; this is
   * the mechanism that screen will call, built where it can be tested today. */
  @Post("replay")
  @HttpCode(200)
  async replay(@Body() body: { dead_letter_id?: string }) {
    if (!body?.dead_letter_id) throw new NotFoundException("no such dead letter");
    const replayed = await replayDeadLetter(this.db, body.dead_letter_id);
    if (!replayed) throw new NotFoundException("no such dead letter");
    return { replayed: true };
  }
}

@@ -1,16 +1,38 @@
-import { Module } from "@nestjs/common";
+import { Module, Scope } from "@nestjs/common";
 
 import { MessagesModule } from "../messages/messages.module";
 import { AuthModule } from "../auth/auth.module";
+import { createDb, createPool, type Db } from "../db/client";
 import { BackfillController } from "./backfill.controller";
 import { InternalController } from "./internal.controller";
+import { DispatchController } from "./dispatch.controller";
 import { SessionController } from "./session.controller";
 
 // The internal routes reuse MessagesModule's providers wholesale — the
 // request-scoped Repository, the guard, the service. One write path, two
 // doors (ADR-04/05).
+//
+// Chapter 3.5 adds the dispatch controller, which needs an UNSCOPED connection
+// rather than the request-scoped Repository: one dispatcher serves every
+// environment, so its operations take the tenant from the row they touch rather
+// than from a principal. `MessagesModule` provides "DB" but does not export it,
+// so this module declares its own — the same DEFAULT-scoped factory every other
+// module here uses, and a smaller change than widening 2.2's exports for a
+// reason 2.2 has nothing to do with.
 @Module({
   imports: [MessagesModule, AuthModule],
-  controllers: [InternalController, BackfillController, SessionController],
+  controllers: [
+    InternalController,
+    BackfillController,
+    SessionController,
+    DispatchController,
+  ],
+  providers: [
+    {
+      provide: "DB",
+      useFactory: (): Db => createDb(createPool()),
+      scope: Scope.DEFAULT,
+    },
+  ],
 })
 export class InternalModule {}

import {
  ALL_DELIVERIES_SUBJECT,
  DELIVERIES_STREAM,
  deliverySubjectFor,
} from "@relay/protocol";
import type { Logger } from "@relay/service-kit";
import {
  DiscardPolicy,
  RetentionPolicy,
  StorageType,
  type NatsConnection,
} from "nats";
 
import type { Db } from "../db/client";
import { drainDueDeliveries, type DueDeliveryRow } from "../db/repository";
import type { Publisher } from "../outbox/publisher";
 
// The second relay (chapter 3.5, research R13).
//
// This is the moment chapter 3.3's outbox stops being a thing that moves EVENTS
// and becomes a shape: `SELECT … FOR UPDATE SKIP LOCKED`, publish, mark, for any
// work the platform owes itself and must not lose. The second instance is what
// makes it a pattern rather than a trick, and the reader has already built it.
//
// The one difference from 3.3's relay is a predicate — `next_attempt_at <=
// now()` — and that predicate IS the retry schedule. Nothing waits in the
// broker; a delivery enters the stream only once it is already due (research R1,
// measured).
 
/** One delivery, as it reaches the dispatcher, is deliberately thin: the id and
 * the attempt, and nothing else. The dispatcher fetches the URL, the payload and
 * the signing secrets over the internal seam when it is ready to send, so a
 * customer credential never sits in a broker (contracts/dispatcher.md).
 *
 * The subject grammar lives in `@relay/protocol` — 3.4's lesson, applied again:
 * two sides, one definition, or a consumer silently receives nothing. */
 
/** The api creates this stream because the api publishes to it. A publisher
 * whose stream does not exist gets a 503 back from the broker, which is a
 * confusing way to discover that JetStream does not create streams on demand.
 *
 * Its age bound is sized for "how long may the DISPATCHER be down", not "how
 * long is the longest retry tier" — because nothing waits here. Conflating those
 * two numbers is how the first design went wrong (research R1). */
const DELIVERIES_MAX_AGE_NS = 7 * 24 * 60 * 60 * 1_000_000_000;
const DELIVERIES_MAX_BYTES = 1024 * 1024 * 1024;
 
export async function ensureDeliveriesStream(nc: NatsConnection): Promise<void> {
  const jsm = await nc.jetstreamManager();
  const mutable = {
    subjects: [ALL_DELIVERIES_SUBJECT],
    max_age: DELIVERIES_MAX_AGE_NS,
    max_bytes: DELIVERIES_MAX_BYTES,
    discard: DiscardPolicy.Old,
  };
  const existing = await jsm.streams.info(DELIVERIES_STREAM).catch(() => null);
  if (existing === null) {
    await jsm.streams.add({
      name: DELIVERIES_STREAM,
      retention: RetentionPolicy.Limits,
      storage: StorageType.File,
      ...mutable,
    });
    return;
  }
  // Retention and storage are immutable on an existing stream — chapter 3.4
  // measured that (its research R1), and the lesson transfers unchanged.
  await jsm.streams.update(DELIVERIES_STREAM, { ...existing.config, ...mutable });
}
 
/** Small, for 3.3's reason: a batch is held inside one transaction, and a long
 * one holds row locks while it publishes. */
const BATCH_SIZE = 50;
 
/** The poll interval when there is nothing due. Deliveries become due on a clock
 * rather than on an insert, so unlike 3.3's relay there is no "wake me on write"
 * shortcut to be tempted by — a timer is the correctness path here, not a
 * fallback behind one. */
const IDLE_INTERVAL_MS = 250;
 
export interface DeliveryRelay {
  start(): void;
  stop(): Promise<void>;
  /** One pass, for tests and for the walk script — the same code path `start`
   * runs, so nothing is proven about a loop only tests exercise. */
  drainOnce(): Promise<number>;
}
 
export function createDeliveryRelay({
  db,
  publisher,
  logger,
  batchSize = BATCH_SIZE,
  intervalMs = IDLE_INTERVAL_MS,
}: {
  db: Db;
  publisher: Publisher;
  logger: Logger;
  batchSize?: number;
  intervalMs?: number;
}): DeliveryRelay {
  let running = false;
  let loop: Promise<void> = Promise.resolve();
 
  async function drainOnce(): Promise<number> {
    return drainDueDeliveries(db, batchSize, async (row: DueDeliveryRow) => {
      // The same three-field port chapter 3.3 defined, unchanged. That the
      // second relay needed no new seam is the evidence that 3.3's abstraction
      // was drawn in the right place.
      await publisher.publish({
        subject: deliverySubjectFor(row.environment_id),
        // The deduplication key is the delivery id AND THE ATTEMPT, and the
        // second half was missing until a walk against a failing endpoint went
        // looking for attempt 2 and found nothing had been sent.
        //
        // `row.id` alone is stable across all seven attempts, so every retry
        // republished under the key the first attempt had already used and
        // JetStream collapsed it inside the duplicate window. The publish
        // reported success, no message reached the dispatcher, and the row kept
        // the `dispatched_at` its claim had set — which only an outcome report
        // clears, and no outcome was ever coming. Every failing delivery stopped
        // dead after one attempt and the retry schedule below it never ran.
        //
        // Per attempt, a republish after a crash is still recognisably the same
        // work, which is what the deduplication is for. A NEW attempt is
        // genuinely new work and has to be allowed to say so.
        id: `${row.id}:${row.attempt}`,
        payload: {
          delivery_id: row.id,
          endpoint_id: row.endpoint_id,
          event_id: row.event_id,
          attempt: row.attempt,
        },
      });
    });
  }
 
  async function run(): Promise<void> {
    while (running) {
      try {
        const dispatched = await drainOnce();
        if (dispatched > 0) {
          // Counts, never payloads — and never a signing secret, which is why
          // this relay publishes ids rather than material (NFR-SEC-06).
          logger.log("info", "deliveries.dispatched", { count: dispatched });
          continue; // straight back for more; a backlog should not wait out the idle interval
        }
      } catch (error) {
        // A broker that is down is an expected state, not a crash. Rows stay
        // pending and due, so the backlog drains when it returns — the same
        // buffering 3.3's relay promises for events.
        logger.log("error", "deliveries.drain_failed", { error: String(error) });
      }
      await new Promise((resolve) => setTimeout(resolve, intervalMs));
    }
  }
 
  return {
    start() {
      if (running) return;
      running = true;
      loop = run();
    },
    async stop() {
      running = false;
      await loop;
    },
    drainOnce,
  };
}

@@ -93,24 +93,33 @@ export async function ensureStream(nc: NatsConnection): Promise<void> {
   }
   await jsm.streams.update(STREAM, { ...existing.config, ...mutable });
 }
 
 export function createJetStreamPublisher({
   url = process.env.RELAY_NATS_URL ?? DEFAULT_NATS_URL,
-}: { url?: string } = {}): Publisher {
+  ensure = ensureStream,
+}: {
+  url?: string;
+  /** Which stream this publisher is responsible for bringing into existence.
+   * Defaults to the EVENTS stream this chapter created. Chapter 3.5 passes its
+   * own: a publisher that ensures the wrong stream publishes into nothing and
+   * gets a 503 back, which is a confusing way to learn that streams are not
+   * created on demand. */
+  ensure?: (nc: NatsConnection) => Promise<void>;
+} = {}): Publisher {
   let connection: NatsConnection | null = null;
   let js: JetStreamClient | null = null;
 
   /** Connection is LAZY and re-attempted. The api must start and accept writes
    * with the broker unreachable — a service that refuses to boot without its
    * event spine has made the spine a dependency of the write path, which is the
    * opposite of what an outbox is for (research R9). */
   async function client(): Promise<JetStreamClient> {
     if (js && connection && !connection.isClosed()) return js;
     const nc = await connect({ servers: url });
-    await ensureStream(nc);
+    await ensure(nc);
     connection = nc;
     js = nc.jetstream();
     return js;
   }
 
   return {

import {
  AckPolicy,
  connect,
  DeliverPolicy,
  type JsMsg,
  type NatsConnection,
} from "nats";
import {
  ALL_DELIVERIES_SUBJECT,
  ALL_EVENTS_SUBJECT,
  DELIVERIES_STREAM,
} from "@relay/protocol";
import { createLogger, type Logger } from "@relay/service-kit";
 
import { createApiClient, type ApiClient } from "./api-client.js";
import { ATTEMPT_TIMEOUT_MS, deliverOnce, type DeliveryJob } from "./deliver.js";
import { expandOnce } from "./expand.js";
 
// The dispatcher (chapter 3.5) — the first service in this platform that exists
// because of constitution IV rather than in spite of it.
//
// TWO CONSUMERS, and they do different jobs on purpose:
//
//   events.>      -> expand one event into one delivery row per matching
//                    endpoint. A claimed write, so a redelivered event cannot
//                    double a customer's webhooks.
//   deliveries.>  -> post one delivery that the api's relay has decided is DUE.
//                    Nothing waits here: a delivery not yet due is a row, not a
//                    message the broker is holding (research R1, measured).
//
// Frameworkless and ESM, mirroring the gateway. ADR-15 binds NestJS to the API
// service only, and a second Nest application would be adopting a framework by
// momentum rather than by decision.
 
const EVENTS_STREAM = "EVENTS";
 
/** One durable per job. A durable name is a POSITION in a stream shared by every
 * instance using it, so two dispatchers divide the work rather than each
 * receiving everything (chapter 3.4's research R8).
 *
 * Overridable, and chapter 3.4's suites explain why: a test that shares the
 * production durable inherits every message every previous run left behind, and
 * a batch of twenty-five is quickly all backlog. A per-run durable with
 * `DeliverPolicy.New` is the same trick 2.1 used for environments and 2.6 for
 * subjects — the only namespace a stream position has is its name. */
export const EXPAND_DURABLE = "dispatcher-expand";
export const DELIVER_DURABLE = "dispatcher-deliver";
 
const BATCH = 25;
/** How long the broker waits for an acknowledgement before deciding the
 * dispatcher died and handing the message to someone else. Overridable ONLY so
 * tests can observe redelivery: at thirty seconds, an assertion that an
 * unacknowledged message comes back has to wait thirty seconds, so in practice
 * it is written with a short window, never observes anything, and passes no
 * matter what the code does. That is how the `term` decision below went
 * unmeasured — the test looked right and proved nothing. */
export const ACK_WAIT_MS = 30_000;
 
/** The DELIVERY attempt budget lives in the api's tier table, not here. This
 * bound is only about how many times the BROKER redelivers a job the dispatcher
 * failed to process at all — a crash, or an api that was unreachable. Confusing
 * the two would let a broker redelivery consume a customer's retry. */
const MAX_DELIVER = 10;
const MAX_ACK_PENDING = 100;
 
const DEFAULT_NATS_URL = "nats://localhost:4222";
const DEFAULT_API_URL = "http://127.0.0.1:4000";
 
export interface Dispatcher {
  ready(): Promise<void>;
  start(): void;
  stop(): Promise<void>;
  /** One pass of each consumer, for tests and the walk script — the same code
   * path `start` runs, so nothing is proven about a loop only tests exercise. */
  pollOnce(): Promise<{ expanded: number; delivered: number }>;
}
 
export function createDispatcher({
  natsUrl = process.env["RELAY_NATS_URL"] ?? DEFAULT_NATS_URL,
  apiUrl = process.env["RELAY_API_URL"] ?? DEFAULT_API_URL,
  credential = process.env["RELAY_INTERNAL_CREDENTIAL"] ?? "",
  logger = createLogger("dispatcher"),
  api = createApiClient(apiUrl, credential),
  attemptTimeoutMs = ATTEMPT_TIMEOUT_MS,
  durables = { expand: EXPAND_DURABLE, deliver: DELIVER_DURABLE },
  deliverPolicy = DeliverPolicy.All,
  ackWaitMs = { expand: ACK_WAIT_MS, deliver: ACK_WAIT_MS },
}: {
  natsUrl?: string;
  apiUrl?: string;
  credential?: string;
  logger?: Logger;
  api?: ApiClient;
  attemptTimeoutMs?: number;
  durables?: { expand: string; deliver: string };
  deliverPolicy?: DeliverPolicy;
  /** PER CONSUMER, and the split is load-bearing. Shortening this for the whole
   * dispatcher to make one expand assertion observable also shortened it for
   * the DELIVER consumer, whose attempts run to a timeout — so under coverage
   * instrumentation a batch outlived its acknowledgement window, the broker
   * decided the dispatcher had died, and redelivered work that was in fact in
   * progress. Two customers received the same webhook twice and two unrelated
   * tests failed. One knob for two consumers with different time budgets is the
   * bug; this is the fix. */
  ackWaitMs?: { expand: number; deliver: number };
} = {}): Dispatcher {
  let connection: NatsConnection | null = null;
  let running = false;
  let loop: Promise<void> = Promise.resolve();
 
  /** Lazy, like every other broker client in this workspace: an unreachable
   * broker must not stop the service from starting, and an unreachable API
   * service must not either — the work is durable in both directions. */
  async function connection_(): Promise<NatsConnection> {
    if (connection && !connection.isClosed()) return connection;
    const nc = await connect({ servers: natsUrl });
    const jsm = await nc.jetstreamManager();
 
    // The DELIVERIES stream is created by the API SERVICE, which publishes to
    // it. The dispatcher only consumes, so it does not define the stream — two
    // definitions of one stream is a drift waiting for the day they disagree.
    // Consumer creation below simply fails until the api has been up once, and
    // the poll loop retries; nothing is lost, because nothing is due yet either.
 
    for (const [stream, durable, ackWait] of [
      [EVENTS_STREAM, durables.expand, ackWaitMs.expand],
      [DELIVERIES_STREAM, durables.deliver, ackWaitMs.deliver],
    ] as const) {
      await jsm.consumers
        .add(stream, {
          durable_name: durable,
          ack_policy: AckPolicy.Explicit,
          deliver_policy: deliverPolicy,
          ack_wait: ackWait * 1_000_000,
          max_deliver: MAX_DELIVER,
          max_ack_pending: MAX_ACK_PENDING,
          ...(stream === EVENTS_STREAM
            ? { filter_subject: ALL_EVENTS_SUBJECT }
            : { filter_subject: ALL_DELIVERIES_SUBJECT }),
        })
        // Created if absent, left alone if present: two dispatchers starting
        // together must share a position rather than fight over it.
        .catch(() => undefined);
    }
    connection = nc;
    return nc;
  }
 
  async function drain(
    stream: string,
    durable: string,
    handle: (msg: JsMsg) => Promise<"ack" | "term" | "retry">,
  ): Promise<number> {
    const nc = await connection_();
    const consumer = await nc.jetstream().consumers.get(stream, durable);
    const messages = await consumer.fetch({ max_messages: BATCH, expires: 1_000 });
    let handled = 0;
    for await (const msg of messages) {
      let decision: "ack" | "term" | "retry" = "retry";
      try {
        decision = await handle(msg);
      } catch (error) {
        // Not acknowledging IS how this asks for the work back. The api being
        // unreachable lands here and must not consume anything.
        logger.log("error", "dispatcher.handle_failed", {
          stream,
          error: String(error),
        });
      }
      if (decision === "ack") {
        msg.ack();
        handled++;
      } else if (decision === "term") {
        msg.term();
        handled++;
      }
    }
    return handled;
  }
 
  /** Deliveries, grouped by endpoint and run in parallel.
   *
   * THIS IS THE ISOLATION PROPERTY, and it is structural rather than a tuning
   * choice. FR-WHK-05 says one customer's endpoint must not delay deliveries to
   * another; processing a batch one message at a time makes that false the
   * moment a single endpoint hangs, because every delivery behind it waits out
   * a timeout it had nothing to do with.
   *
   * ONE in-flight attempt per endpoint, endpoints concurrent. Serialising WITHIN
   * an endpoint matters too: two attempts at once against the same customer
   * would let a retry overtake the attempt it is retrying, and a customer whose
   * server is already struggling is the last one to hand extra concurrency
   * (research R7). */
  async function drainByEndpoint(): Promise<number> {
    const nc = await connection_();
    const consumer = await nc.jetstream().consumers.get(
      DELIVERIES_STREAM,
      durables.deliver,
    );
    const messages = await consumer.fetch({ max_messages: BATCH, expires: 1_000 });
 
    const byEndpoint = new Map<string, { msg: JsMsg; job: DeliveryJob }[]>();
    for await (const msg of messages) {
      let job: DeliveryJob;
      try {
        job = msg.json<DeliveryJob>();
      } catch {
        logger.log("error", "deliver.undecodable", { seq: msg.seq });
        msg.term();
        continue;
      }
      const queue = byEndpoint.get(job.endpoint_id) ?? [];
      queue.push({ msg, job });
      byEndpoint.set(job.endpoint_id, queue);
    }
 
    let handled = 0;
    await Promise.all(
      [...byEndpoint.values()].map(async (queue) => {
        for (const { msg, job } of queue) {
          try {
            await deliverOnce(api, logger, job, attemptTimeoutMs);
          } catch (error) {
            // The API SERVICE is unreachable — the one failure this must not
            // absorb. Leaving the message unacknowledged is how the work comes
            // back; a customer's 500 never lands here, because that is an
            // outcome rather than an error.
            logger.log("error", "dispatcher.handle_failed", {
              stream: DELIVERIES_STREAM,
              error: String(error),
            });
            continue;
          }
          // Acknowledged whatever the customer said. A 500 is not the
          // dispatcher's failure to handle the job — it is the job's OUTCOME,
          // already recorded by the api, which has scheduled the next tier or
          // dead-lettered it. Redelivering here would retry outside the
          // schedule the customer was promised.
          msg.ack();
          handled++;
        }
      }),
    );
    return handled;
  }
 
  async function pollOnce(): Promise<{ expanded: number; delivered: number }> {
    const expanded = await drain(EVENTS_STREAM, durables.expand, async (msg) => {
      // `msg.json()` THROWS on bytes that are not JSON, and a throw here would
      // land in drain()'s catch and be treated as "ask for the work back" — so
      // bytes that can never parse would be redelivered until the broker's
      // attempt budget ran out. Chapter 3.4 settled this: the same bytes fail the
      // same way every time, so an unparseable payload is terminated on the FIRST
      // attempt. Decoding defensively is what keeps that promise here.
      let raw: unknown;
      try {
        raw = msg.json();
      } catch {
        logger.log("error", "expand.undecodable", { seq: msg.seq });
        return "term";
      }
      const outcome = await expandOnce(api, logger, raw);
      // A duplicate is acknowledged because it genuinely has been handled — by a
      // previous delivery of this same event.
      return outcome === "unparseable" ? "term" : "ack";
    });
 
    const delivered = await drainByEndpoint();
 
    return { expanded, delivered };
  }
 
  async function run(): Promise<void> {
    while (running) {
      try {
        const { expanded, delivered } = await pollOnce();
        if (expanded > 0 || delivered > 0) continue;
      } catch (error) {
        logger.log("error", "dispatcher.poll_failed", { error: String(error) });
      }
      await new Promise((resolve) => setTimeout(resolve, 200));
    }
  }
 
  return {
    /** Force the consumers into existence without processing anything. A suite
     * using `DeliverPolicy.New` must create its position BEFORE it publishes, or
     * the message it is about to send lands before the consumer exists and is
     * never seen. */
    async ready(): Promise<void> {
      await connection_();
    },
    start() {
      if (running) return;
      running = true;
      loop = run();
    },
    async stop() {
      running = false;
      await loop;
      if (connection && !connection.isClosed()) await connection.drain();
      connection = null;
    },
    pollOnce,
  };
}
 
// Entry point. `import.meta` rather than `require.main`: this package is ESM,
// unlike the api (ADR-15's dialect split).
if (import.meta.url === `file://${process.argv[1]}`) {
  const dispatcher = createDispatcher();
  dispatcher.start();
  for (const signal of ["SIGINT", "SIGTERM"] as const) {
    process.on(signal, () => {
      void dispatcher.stop().then(() => process.exit(0));
    });
  }
}