Recipes

Reliable Integration Events

Transactional outbox pattern for at-least-once integration event delivery — OutboxStore, OutboxIntegrationEventTransport, OutboxRelay, fire-and-forget tradeoff, the fire-and-forget compiler warning and its suppressDiagnostics opt-out.

Fire-and-forget integration events work fine until the process crashes between committing an aggregate write and dispatching the event. When that happens, the write lands but the downstream handler never runs — a ghost record that never triggers its consequences. The transactional outbox closes that gap: the event persists in the same transaction as the aggregate, so either both commit or neither does.

The problem with fire-and-forget

await this.repository.save(order);
await this.orderPlaced.publish({ orderId: order.id }); // lost on crash here

The producer is a typed IntegrationEvent class injected by identity (this.orderPlaced), and publish(data) routes through the bound transport. If the process terminates after save returns but before the transport delivers the event, the order exists in the database but the event is gone. The downstream context never learns the order was placed, and no error is visible anywhere.

The outbox pattern

Instead of emitting directly to the queue, the event is written to an outbox table in the same database transaction as the aggregate write. After the commit, a relay picks up pending rows and dispatches them. A crash after the commit leaves the outbox row durable — the relay replays it on the next boot:

await this.repository.runInTransaction(async () => {
  await this.repository.save(order);
  await this.orderPlaced.publish({ orderId: order.id }); // buffered, not yet delivered
}); // onBeforeCommit: outbox row written atomically; onCommit: relay scheduled

The producer call is unchanged from the in-process default — the same await this.orderPlaced.publish(...). The difference is entirely in which IntegrationEventTransport the event publishes through. The relay delivers after the commit, so the downstream handler receives the event if and only if the aggregate write committed.

The producer event

The producer is a typed IntegrationEvent class — one class per cross-context event, carrying a plain-interface payload. The orders controller injects it by class identity and calls publish(data); the compiler bakes the prefixed channel from the "order.placed" type argument and wires the publisher over the bound transport:

src/orders/events.ts

import { IntegrationEvent } from "@heximon/integration";

export interface OrderPlacedPayload {
  readonly orderId: string;
  readonly customer: string;
  readonly totalCents: number;
}

export class OrderPlaced extends IntegrationEvent<"order.placed", OrderPlacedPayload> {}

List it under integration: { events } in the producing module (alongside the outbox bindings below). The matching consumer is an unchanged IntegrationEventHandler<"order.placed", OrderPlacedPayload> in the downstream context (name-first type argument), listed under integration: { eventHandlers }.

The outbox wiring

The transactional outbox tier ships in @heximon/integration as the OutboxStore / OutboxRelay / OutboxIntegrationEventTransport classes (plus the relay's in-process delivery terminal). Because the compiler reads constructor signatures from app source, you bind them by authoring thin nominal-satisfier subclasses in your own module:

`OutboxStore` — persistence

OutboxStore is the abstract DI token. Implement it (or extend DrizzleOutboxStore from @heximon/drizzle/core) with a concrete subclass that exposes its own constructor:

src/outbox/outbox-store.ts

import { OutboxStore } from "@heximon/integration";
import { DrizzleOutboxStore } from "@heximon/drizzle/core";
import { Database } from "../database/database";
import { outbox } from "./schema";

export class AppOutboxStore extends DrizzleOutboxStore implements OutboxStore {
  public constructor(database: Database) {
    super(database, outbox);
  }
}

Add the outbox columns to your Drizzle schema by spreading outboxColumns():

src/database/schema.ts

import { outboxColumns } from "@heximon/drizzle/sqlite-core"; // or pg-core / mysql-core
import { sqliteTable } from "drizzle-orm/sqlite-core";

export const outbox = sqliteTable("outbox", {
  ...outboxColumns(),
});

outboxColumns() includes correlationId and causationId as nullable columns (new rows carry the ambient tracing ids; pre-existing rows read back as null). For an existing outbox table, add a migration:

ALTER TABLE outbox ADD COLUMN correlation_id TEXT;
ALTER TABLE outbox ADD COLUMN causation_id TEXT;

`OutboxRelay` and its in-process terminal

OutboxRelay drains pending rows after each commit and replays rows left pending on boot, delivering each claimed row through an IntegrationEventTransport terminal. That terminal is the relay's own in-process delivery target — a concrete MemoryIntegrationEventTransport subclass bound separately from the producer's transport (the producer publishes through the outbox transport; the relay fans each drained row out through the in-process channel table):

src/outbox/outbox-relay.ts

import { Context } from "@heximon/runtime";
import { OutboxRelay } from "@heximon/integration";
import { QueueChannelDispatcher } from "@heximon/queue";
import { MemoryIntegrationEventTransport } from "@heximon/queue/memory";
import { AppOutboxStore } from "./outbox-store";

export class AppRelayTransport extends MemoryIntegrationEventTransport {
  public constructor(dispatcher: QueueChannelDispatcher) {
    super(dispatcher);
  }
}

export class AppOutboxRelay extends OutboxRelay {
  public constructor(store: AppOutboxStore, target: AppRelayTransport, context: Context) {
    super(store, target, context);
  }
}

On Cloudflare Workers, bind CloudflareQueueIntegrationEventTransport (@heximon/queue/cloudflare) as the relay's terminal instead — the durable counterpart of MemoryIntegrationEventTransport. Rather than running the handlers inline on the producing isolate, it sends each drained row onto the app's own Cloudflare Queue, and that same worker's queue(batch, env, ctx) export drains it later, decoupled from the producing request and covered by the platform's own retry / backoff / DLQ:

src/database/database.module.ts

import { Platform } from "@heximon/runtime";
import { CloudflareQueueIntegrationEventTransport } from "@heximon/queue/cloudflare";

{
  provide: CloudflareQueueIntegrationEventTransport,
  useFactory: (): CloudflareQueueIntegrationEventTransport =>
    new CloudflareQueueIntegrationEventTransport(Platform.binding("OUTBOX_QUEUE")),
},

Declare the queue once in heximon.config.ts:

heximon.config.ts

cloudflare: {
  queues: [{ queue: "ticketing-outbox", binding: "OUTBOX_QUEUE" }],
},

The consume side needs no new code — the worker's generic queue() export already routes a drained message by its channel into the same IntegrationEventHandlers, and vp dev provisions the consumer in Miniflare too, so the round trip runs locally against real workerd.

On Vercel Functions, bind VercelQueueIntegrationEventTransport (@heximon/queue/vercel) as the terminal instead — the same durable pattern for a freezing platform. It sends each drained row onto the app's own Vercel Queue — always onto the one fixed heximon-integration-events topic (the dotted integration channel is not a valid Vercel topic name; it rides the envelope, which the consumer dispatches by) — and the separate generated functions/queues/consumer.func drains it later, decoupled from the frozen-between-invocations function that produced it.

The build registers the topic's consumer trigger automatically whenever the app binds its own transport. The send is an optional constructor argument (omit it to lazily import the @vercel/queue SDK on first publish; pass one — the in-process platform simulator — in tests):

src/database/database.module.ts

import { VercelQueueIntegrationEventTransport } from "@heximon/queue/vercel";

{
  provide: VercelQueueIntegrationEventTransport,
  useFactory: (): VercelQueueIntegrationEventTransport => new VercelQueueIntegrationEventTransport(),
},

`OutboxIntegrationEventTransport` — the transactional producer transport

OutboxIntegrationEventTransport is an IntegrationEventTransport satisfier that wraps the buffer → flush → schedule lifecycle. Binding it replaces the in-process memory default, so every typed IntegrationEvent.publish becomes reliable:

src/outbox/outbox.ts

import { Context, TransactionContext } from "@heximon/runtime";
import { OutboxIntegrationEventTransport } from "@heximon/integration";
import { IntegrationEventTransport } from "@heximon/queue";
import { Database } from "../database/database";
import { AppOutboxRelay } from "./outbox-relay";
import { AppOutboxStore } from "./outbox-store";

export class AppOutboxTransport
  extends OutboxIntegrationEventTransport
  implements IntegrationEventTransport
{
  public constructor(
    database: Database,
    store: AppOutboxStore,
    relay: AppOutboxRelay,
    context: Context,
    transactionContext: TransactionContext,
  ) {
    super(database, store, relay, context, transactionContext);
  }
}

Declare all four in your module's providers — bare, no useFactory, since every constructor above is authored in app source and the compiler reads its parameter types directly. That includes AppOutboxRelay: OutboxRelay's OnBootstrap hook (the crash-recovery boot replay) is inherited, not redeclared, and still fires automatically at boot.

Because both AppOutboxTransport (the producer transport) and AppRelayTransport (the relay's terminal) satisfy the abstract IntegrationEventTransport, disambiguate the producer's binding explicitly with { provide: IntegrationEventTransport, useExisting: AppOutboxTransport } — the typed event publishes through the outbox transport. The relay's own constructor sidesteps the same ambiguity from the other side: it asks for AppRelayTransport by its concrete class, not the abstract token the alias above resolves, so the two terminals never collide:

src/outbox/outbox.module.ts

import { Module, TransactionContext } from "@heximon/runtime";
import { IntegrationEventTransport } from "@heximon/queue";
import { DatabaseModule } from "../database/database.module";
import { AppOutboxRelay, AppRelayTransport } from "./outbox-relay";
import { AppOutboxStore } from "./outbox-store";
import { AppOutboxTransport } from "./outbox";

export class OutboxModule extends Module({
  imports: [DatabaseModule],
  providers: [
    AppOutboxStore,
    AppRelayTransport,
    AppOutboxRelay,
    AppOutboxTransport,
    // Both AppOutboxTransport and AppRelayTransport satisfy IntegrationEventTransport, so disambiguate:
    // the typed event publishes through the outbox transport; the relay injects its concrete terminal.
    { provide: IntegrationEventTransport, useExisting: AppOutboxTransport },
    TransactionContext, // bare: its constructor is already a plain DI token (just the app Context)
  ],
  exports: [AppOutboxStore, AppRelayTransport, AppOutboxRelay, AppOutboxTransport, IntegrationEventTransport],
}) {}

The compiler warning — `integration.fire-and-forget`

When the compiler discovers IntegrationEventHandler classes but no durable transport is bound, it emits a warning-severity diagnostic:

warning [integration/fire-and-forget]  OrderPlacedHandler: 1 IntegrationEventHandler discovered, but
fire-and-forget delivery is not explicitly opted into. Integration events dispatched without a
transactional outbox are fire-and-forget: a process crash between the aggregate write and the dispatch
loses the event silently. Bind the transactional outbox pattern (OutboxIntegrationEventTransport +
OutboxRelay) for at-least-once delivery — see the "reliable integration events" recipe in the docs.

The build always completes — the warning never becomes an error. If fire-and-forget is deliberately acceptable for a given app (tests, development, a throwaway prototype), suppress it by listing the code in suppressDiagnostics — the silence-direction mirror of strictDiagnostics:

vite.config.ts

import { DiagnosticCode } from "@heximon/build";

heximon({ suppressDiagnostics: [DiagnosticCode.FireAndForgetIntegrationEvents] });

Correlation and causation across the outbox boundary

The correlationId and causationId are captured from the ambient context at emit time and persisted on the outbox row. The relay reads them from the stored row — not from the (empty) context at drain time — and seeds them into the consumer's context via QueueChannelDispatcher.dispatch. This means:

A boot-time replay (rows left pending across a process restart) threads the same tracing ids into the handler as the original dispatch would have.
The handler's own DomainEvents.emit calls inherit the correct correlationId (the originating request) and causationId (the integration event's stable row id).

No extra wiring is needed — the three-class setup above handles it automatically.

The columns are nullable in the outbox table. Rows written before this feature existed read back as null; the relay converts null to undefined before passing to the dispatcher, so a legacy row causes no ids to be seeded (the same as if the event had been emitted outside an active context).

At-least-once delivery

The outbox guarantees at-least-once delivery — not exactly-once. A crash after the relay dispatches the event but before it deletes the outbox row will cause re-delivery. Every handler that writes state should guard against double-processing:

const existing = await this.repository.getByOrderId(payload.orderId);
if (existing !== undefined) {
  return; // already processed — skip the duplicate
}

The QueueIdempotencyInterceptor from @heximon/queue/idempotency provides a framework-level guard on top of the handler's own check — it dedups queue and integration-event dispatch on the envelope's stable id, so a relay redelivery is a no-op the second time. (EventIdempotencyInterceptor from @heximon/events/idempotency is the sibling for the serial in-process EventHandler tier, keyed on DomainEvent.eventId — a different dispatch path from the relay.)