import type { OutboxAttempt, OutboxDeadLetterReason } from '@nestjs/outbox';
import { bigint, index, integer, jsonb, pgTable, primaryKey, text, timestamp } from 'drizzle-orm/pg-core';
import type { OrderItem } from '../orders/order.js';

export const products = pgTable('products', {
  id: text('id').primaryKey(),
  name: text('name').notNull(),
  /** In cents. */
  price: integer('price').notNull(),
  inStock: integer('in_stock').notNull(),
  reserved: integer('reserved').notNull().default(0),
});

export const orders = pgTable('orders', {
  id: text('id').primaryKey(),
  userId: text('user_id').notNull(),
  items: jsonb('items').$type<OrderItem[]>().notNull(),
  /** In cents. */
  total: integer('total').notNull(),
  status: text('status').$type<'placed' | 'cancelled'>().notNull(),
});

// The outbox's tables, read and written by DrizzleOutboxStore.

export const outboxMessages = pgTable(
  'outbox_messages',
  {
    /** Order within a key: numbered at insert, in commit order (DrizzleOutboxStore.add() locks the key). */
    seq: bigint('seq', { mode: 'number' }).primaryKey().generatedByDefaultAsIdentity(),
    id: text('id').notNull().unique(),
    topic: text('topic').notNull(),
    // A null payload is stored as NULL.
    payload: jsonb('payload').$type<unknown>(),
    headers: jsonb('headers').$type<Record<string, string>>().notNull(),
    key: text('key'),
    createdAt: timestamp('created_at', { withTimezone: true }).notNull(),
    availableAt: timestamp('available_at', { withTimezone: true }).notNull(),
    attempts: integer('attempts').notNull().default(0),
    lastError: text('last_error'),
    history: jsonb('history').$type<OutboxAttempt[]>().notNull().default([]),
    leaseOwner: text('lease_owner'),
    leaseUntil: timestamp('lease_until', { withTimezone: true }),
  },
  // A key's older messages, for the claim's per-key check.
  (table) => [index('outbox_messages_key_seq').on(table.key, table.seq)],
);

export const outboxDeadLetters = pgTable(
  'outbox_dead_letters',
  {
    id: text('id').primaryKey(),
    /** The message's place in its key, so a requeue puts it back there. */
    seq: bigint('seq', { mode: 'number' }).notNull(),
    topic: text('topic').notNull(),
    payload: jsonb('payload').$type<unknown>(),
    headers: jsonb('headers').$type<Record<string, string>>().notNull(),
    key: text('key'),
    createdAt: timestamp('created_at', { withTimezone: true }).notNull(),
    attempts: integer('attempts').notNull(),
    lastError: text('last_error'),
    history: jsonb('history').$type<OutboxAttempt[]>().notNull(),
    reason: text('reason').$type<OutboxDeadLetterReason>().notNull(),
    failedAt: timestamp('failed_at', { withTimezone: true }).notNull(),
  },
  (table) => [index('outbox_dead_letters_topic_failed_at').on(table.topic, table.failedAt)],
);

export const outboxInbox = pgTable(
  'outbox_inbox',
  {
    consumer: text('consumer').notNull(),
    messageId: text('message_id').notNull(),
    processedAt: timestamp('processed_at', { withTimezone: true }).notNull(),
  },
  (table) => [
    // One record per consumer and message: the unique key two deliveries meet at.
    primaryKey({ columns: [table.consumer, table.messageId] }),
    index('outbox_inbox_processed_at').on(table.processedAt),
  ],
);

import { DrizzleModule } from '@nestjs/drizzle';
import { drizzle } from 'drizzle-orm/node-postgres';
import * as schema from './database/schema.js';
// ...
DrizzleModule.forRootAsync({
  // A pg pool on DATABASE_URL, closed in onApplicationShutdown(), after the relay drained.
  useFactory: () => ({ drizzle, connection: process.env.DATABASE_URL!, schema }),
}),

import type { NodePgDatabase } from 'drizzle-orm/node-postgres';
import type * as schema from './schema.js';

export type Database = NodePgDatabase<typeof schema>;
/** The `tx` that `db.transaction()` passes its callback. */
export type Transaction = Parameters<Parameters<Database['transaction']>[0]>[0];

CREATE TABLE "outbox_messages" (
	"seq" bigint PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY (sequence name "outbox_messages_seq_seq" INCREMENT BY 1 MINVALUE 1 MAXVALUE 9223372036854775807 START WITH 1 CACHE 1),
	"id" text NOT NULL,
	"topic" text NOT NULL,
	"payload" jsonb,
	"headers" jsonb NOT NULL,
	"key" text,
	"created_at" timestamp with time zone NOT NULL,
	"available_at" timestamp with time zone NOT NULL,
	"attempts" integer DEFAULT 0 NOT NULL,
	"last_error" text,
	"history" jsonb DEFAULT '[]'::jsonb NOT NULL,
	"lease_owner" text,
	"lease_until" timestamp with time zone,
	CONSTRAINT "outbox_messages_id_unique" UNIQUE("id")
);
...

import { Injectable } from '@nestjs/common';
import { InjectDrizzle } from '@nestjs/drizzle';
import {
  OutboxStorage,
  OutboxTransactionRequiredError,
  type OutboxClaimRequest,
  type OutboxDeadLetter,
  type OutboxDeadLetterFilter,
  type OutboxDeadLetterQuery,
  type OutboxDeadLetterUpdate,
  type OutboxInboxStore,
  type OutboxMessage,
  type OutboxRescheduleUpdate,
  type OutboxStore,
  type OutboxStoreStats,
} from '@nestjs/outbox';
import { and, desc, eq, getTableColumns, gt, inArray, isNull, lt, lte, notExists, or, sql, type SQL } from 'drizzle-orm';
import { alias } from 'drizzle-orm/pg-core';
import { createHash } from 'node:crypto';
import type { Database, Transaction } from './drizzle.js';
import { outboxDeadLetters, outboxInbox, outboxMessages } from './schema.js';

/** Advisory lock classes (the two-number form): any two numbers no other code of yours locks on. */
const CLAIM_LOCK = 20_260_901;
const KEY_LOCK = 20_260_902;

@Injectable()
export class DrizzleOutboxStore implements OutboxStore<Transaction>, OutboxInboxStore<Transaction> {
  constructor(
    @InjectDrizzle() private readonly db: Database,
    storage: OutboxStorage,
  ) {
    storage.registerSource({ messages: this, inbox: this });
  }

  async add(tx: Transaction, messages: readonly OutboxMessage[]): Promise<void> {
    assertTransaction(tx);
    if (messages.length === 0) return;
    // Commit order: a transaction adding a message with the same key waits here until this
    // one commits or rolls back, so rows are numbered in the order they become visible.
    for (const lock of keyLocks(messages)) {
      await tx.execute(sql`SELECT pg_advisory_xact_lock(${KEY_LOCK}::int, ${lock}::int)`);
    }
    await tx.insert(outboxMessages).values(
      messages.map((message) => ({
        id: message.id,
        topic: message.topic,
        payload: message.payload,
        headers: message.headers,
        key: message.key,
        createdAt: new Date(message.createdAt),
        availableAt: new Date(message.availableAt),
      })),
    );
  }

  claim({ owner, now, leaseMs, limit }: OutboxClaimRequest): Promise<OutboxMessage[]> {
    return this.db.transaction(
      async (tx) => {
        // Claims take turns, so each one sees the leases of the one before it.
        await tx.execute(sql`SELECT pg_advisory_xact_lock(${CLAIM_LOCK}::int, 0)`);
        // Due rows in seq order, stopping at `limit` claimable ones: a primary-key index scan,
        // whatever the backlog. SKIP LOCKED passes over a row a relay is writing right now.
        const due = await tx
          .select({ seq: outboxMessages.seq })
          .from(outboxMessages)
          .where(this.claimable(now))
          .orderBy(outboxMessages.seq)
          .limit(limit)
          .for('update', { skipLocked: true });
        if (due.length === 0) return [];
        const seqs = due.map((row) => row.seq);
        await tx
          .update(outboxMessages)
          .set({ leaseOwner: owner, leaseUntil: new Date(now + leaseMs) })
          .where(inArray(outboxMessages.seq, seqs));

        // A last look before committing: each claimed row's predecessor in its key must be
        // ours too. A row whose older sibling was skipped as locked, or came back meanwhile (a
        // dead letter requeued right then), goes back with the rest of its key.
        const previous = alias(outboxMessages, 'previous');
        const rows = await tx
          .select({
            ...getTableColumns(outboxMessages),
            behindOther: sql<boolean | null>`(${tx
              .select({ other: sql`${previous.leaseOwner} IS DISTINCT FROM ${owner}` })
              .from(previous)
              .where(and(eq(previous.key, outboxMessages.key), lt(previous.seq, outboxMessages.seq)))
              .orderBy(desc(previous.seq))
              .limit(1)})`,
          })
          .from(outboxMessages)
          .where(inArray(outboxMessages.seq, seqs))
          .orderBy(outboxMessages.seq);
        const blocked = new Set<string>();
        const giveBack: number[] = [];
        const batch: OutboxMessage[] = [];
        for (const { behindOther, ...row } of rows) {
          if (row.key !== null && (behindOther === true || blocked.has(row.key))) {
            blocked.add(row.key);
            giveBack.push(row.seq);
          } else {
            batch.push(toMessage(row));
          }
        }
        if (giveBack.length > 0) {
          await tx
            .update(outboxMessages)
            .set({ leaseOwner: null, leaseUntil: null })
            .where(inArray(outboxMessages.seq, giveBack));
        }
        return batch;
      },
      // Each statement sees what committed before it started (PostgreSQL's default).
      { isolationLevel: 'read committed' },
    );
  }

  async markPublished(id: string, owner: string): Promise<boolean> {
    const deleted = await this.db
      .delete(outboxMessages)
      .where(leasedBy(id, owner))
      .returning({ id: outboxMessages.id });
    return deleted.length === 1;
  }

  async reschedule(id: string, owner: string, update: OutboxRescheduleUpdate): Promise<boolean> {
    const updated = await this.db
      .update(outboxMessages)
      .set({
        attempts: update.attempts,
        availableAt: new Date(update.availableAt),
        lastError: update.error.error,
        history: sql`${outboxMessages.history} || ${JSON.stringify([update.error])}::jsonb`,
        leaseOwner: null,
        leaseUntil: null,
      })
      .where(leasedBy(id, owner))
      .returning({ id: outboxMessages.id });
    return updated.length === 1;
  }

  deadLetter(id: string, owner: string, update: OutboxDeadLetterUpdate): Promise<boolean> {
    return this.db.transaction(async (tx) => {
      const [row] = await tx.delete(outboxMessages).where(leasedBy(id, owner)).returning();
      if (!row) return false;
      const deadLetter = {
        id: row.id,
        seq: row.seq,
        topic: row.topic,
        payload: row.payload,
        headers: row.headers,
        key: row.key,
        createdAt: row.createdAt,
        attempts: update.attempts,
        lastError: update.error.error,
        history: [...row.history, update.error],
        reason: update.reason,
        failedAt: new Date(update.failedAt),
      };
      // Replaces an earlier dead letter with this id (a producer that reused a custom id).
      await tx
        .insert(outboxDeadLetters)
        .values(deadLetter)
        .onConflictDoUpdate({ target: outboxDeadLetters.id, set: deadLetter });
      return true;
    });
  }

  async release(ids: readonly string[], owner: string): Promise<number> {
    const released = await this.db
      .update(outboxMessages)
      .set({ leaseOwner: null, leaseUntil: null })
      .where(and(eq(outboxMessages.leaseOwner, owner), inArray(outboxMessages.id, [...ids])))
      .returning({ id: outboxMessages.id });
    return released.length;
  }

  async stats(now: number): Promise<OutboxStoreStats> {
    const at = new Date(now);
    const [pending, ready, leased, deadLetters, [oldest]] = await Promise.all([
      this.db.$count(outboxMessages),
      this.db.$count(outboxMessages, this.claimable(now)),
      this.db.$count(outboxMessages, gt(outboxMessages.leaseUntil, at)),
      this.db.$count(outboxDeadLetters),
      // Waiting since it became due; once an attempt failed, since it was added.
      this.db
        .select({
          dueAt: sql<Date | null>`min(CASE
            WHEN ${gt(outboxMessages.attempts, 0)} THEN ${outboxMessages.createdAt}
            WHEN ${lte(outboxMessages.availableAt, at)} THEN ${outboxMessages.availableAt} END)`.mapWith(
            outboxMessages.createdAt,
          ),
        })
        .from(outboxMessages),
    ]);
    return { pending, ready, leased, deadLetters, oldestDueAt: oldest?.dueAt?.getTime() ?? null };
  }

  async listDeadLetters({ topic, key, limit = 50, offset = 0 }: OutboxDeadLetterQuery): Promise<OutboxDeadLetter[]> {
    const rows = await this.db
      .select()
      .from(outboxDeadLetters)
      .where(
        and(
          topic === undefined ? undefined : eq(outboxDeadLetters.topic, topic),
          key === undefined ? undefined : eq(outboxDeadLetters.key, key),
        ),
      )
      .orderBy(desc(outboxDeadLetters.failedAt), desc(outboxDeadLetters.id))
      .limit(limit)
      .offset(offset);
    return rows.map(toDeadLetter);
  }

  async getDeadLetter(id: string): Promise<OutboxDeadLetter | undefined> {
    const [row] = await this.db.select().from(outboxDeadLetters).where(eq(outboxDeadLetters.id, id));
    return row && toDeadLetter(row);
  }

  requeueDeadLetters(filter: OutboxDeadLetterFilter, now: number): Promise<number> {
    const where = deadLetterFilter(filter);
    return this.db.transaction(async (tx) => {
      const rows = await tx.delete(outboxDeadLetters).where(where).returning();
      if (rows.length === 0) return 0;
      // The original seq: back ahead of the messages added after it, with the same key.
      await tx.insert(outboxMessages).values(
        rows.map((row) => ({
          seq: row.seq,
          id: row.id,
          topic: row.topic,
          payload: row.payload,
          headers: row.headers,
          key: row.key,
          createdAt: row.createdAt,
          availableAt: new Date(now),
          lastError: row.lastError,
          history: row.history,
        })),
      );
      return rows.length;
    });
  }

  async purgeDeadLetters(filter: OutboxDeadLetterFilter): Promise<number> {
    const purged = await this.db
      .delete(outboxDeadLetters)
      .where(deadLetterFilter(filter))
      .returning({ id: outboxDeadLetters.id });
    return purged.length;
  }

  async recordInbox(tx: Transaction | undefined, consumer: string, messageId: string, now: number): Promise<boolean> {
    if (tx !== undefined) assertTransaction(tx);
    // One statement on the unique key: a concurrent delivery waits for this transaction.
    const inserted = await (tx ?? this.db)
      .insert(outboxInbox)
      .values({ consumer, messageId, processedAt: new Date(now) })
      .onConflictDoNothing()
      .returning({ consumer: outboxInbox.consumer });
    return inserted.length === 1;
  }

  async hasInbox(consumer: string, messageId: string): Promise<boolean> {
    const found = await this.db.$count(
      outboxInbox,
      and(eq(outboxInbox.consumer, consumer), eq(outboxInbox.messageId, messageId)),
    );
    return found > 0;
  }

  async pruneInbox(before: number): Promise<number> {
    const pruned = await this.db
      .delete(outboxInbox)
      .where(lt(outboxInbox.processedAt, new Date(before)))
      .returning({ consumer: outboxInbox.consumer });
    return pruned.length;
  }

  /**
   * Due, unleased, and no older message with the same key is delayed or leased. The
   * NOT EXISTS probes the (key, seq) index from the key's oldest row.
   */
  private claimable(now: number): SQL {
    const at = new Date(now);
    const older = alias(outboxMessages, 'older');
    return and(
      lte(outboxMessages.availableAt, at),
      or(isNull(outboxMessages.leaseUntil), lte(outboxMessages.leaseUntil, at)),
      or(
        isNull(outboxMessages.key),
        notExists(
          this.db
            .select({ one: sql`1` })
            .from(older)
            .where(
              and(
                eq(older.key, outboxMessages.key),
                lt(older.seq, outboxMessages.seq),
                or(gt(older.availableAt, at), gt(older.leaseUntil, at)),
              ),
            ),
        ),
      ),
    )!;
  }
}

/** Drizzle's `tx` has rollback(); the database itself doesn't, and would write outside the transaction. */
function assertTransaction(tx: Transaction) {
  if (typeof (tx as Partial<Transaction> | undefined)?.rollback !== 'function') {
    throw new OutboxTransactionRequiredError('Pass the tx that db.transaction() gives you, not the database.');
  }
}

/** One lock per key, sorted, so two transactions locking the same keys can't deadlock. */
function keyLocks(messages: readonly OutboxMessage[]): number[] {
  const keys = new Set(messages.flatMap((message) => (message.key === null ? [] : [message.key])));
  const locks = [...keys].map((key) => createHash('sha256').update(key).digest().readInt32BE(0));
  return [...new Set(locks)].sort((a, b) => a - b);
}

/** The message, if `owner` still holds its lease: every relay write is fenced by it. */
function leasedBy(id: string, owner: string): SQL {
  return and(eq(outboxMessages.id, id), eq(outboxMessages.leaseOwner, owner))!;
}

/** The filter's fields combined with AND; an empty filter is refused unless it says `all`. */
function deadLetterFilter({ ids, topic, key, failedBefore, all }: OutboxDeadLetterFilter): SQL | undefined {
  const conditions: SQL[] = [];
  if (ids) conditions.push(inArray(outboxDeadLetters.id, ids));
  if (topic !== undefined) conditions.push(eq(outboxDeadLetters.topic, topic));
  if (key !== undefined) conditions.push(eq(outboxDeadLetters.key, key));
  if (failedBefore !== undefined) conditions.push(lt(outboxDeadLetters.failedAt, new Date(+failedBefore)));
  if (conditions.length === 0 && !all) {
    throw new Error('Refusing an empty dead-letter filter; pass { all: true }');
  }
  return and(...conditions);
}

function toMessage(row: typeof outboxMessages.$inferSelect): OutboxMessage {
  return {
    id: row.id,
    topic: row.topic,
    payload: row.payload,
    headers: row.headers,
    key: row.key,
    createdAt: row.createdAt.getTime(),
    availableAt: row.availableAt.getTime(),
    attempts: row.attempts,
    lastError: row.lastError,
  };
}

function toDeadLetter(row: typeof outboxDeadLetters.$inferSelect): OutboxDeadLetter {
  return {
    id: row.id,
    topic: row.topic,
    payload: row.payload,
    headers: row.headers,
    key: row.key,
    createdAt: row.createdAt.getTime(),
    attempts: row.attempts,
    lastError: row.lastError,
    reason: row.reason,
    failedAt: row.failedAt.getTime(),
    history: row.history,
  };
}

import { outboxInboxStoreContract, outboxStoreContract, type OutboxStoreHarness } from '@nestjs/outbox/testing';
// ...
/** A store on emptied tables, built the way Nest builds it: with the database and a registry. */
async function freshStore(db: Database): Promise<OutboxStoreHarness<Transaction> & { store: DrizzleOutboxStore }> {
  await db.execute(sql`TRUNCATE outbox_messages, outbox_dead_letters, outbox_inbox RESTART IDENTITY`);
  const store = new DrizzleOutboxStore(db, new OutboxStorage());
  return { store, transaction: (work) => db.transaction(work), notATransaction: db };
}

describe('DrizzleOutboxStore on PGlite: the store contract', () => {
  const client = new PGlite();
  const db = drizzlePglite(client, { schema }) as unknown as Database;
  beforeAll(() => migratePglite(db as never, { migrationsFolder }));
  afterAll(() => client.close());

  // The concurrency cases run too; with one connection, PGlite runs them one transaction at a time.
  for (const c of outboxStoreContract(() => freshStore(db), { concurrent: true })) it(c.name, c.run);
  describe('the inbox contract', () => {
    for (const c of outboxInboxStoreContract(() => freshStore(db), { concurrent: true })) it(c.name, c.run);
  });
});

import { Module } from '@nestjs/common';
import { DrizzleModule } from '@nestjs/drizzle';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { ClientProxyTransport, OutboxModule } from '@nestjs/outbox';
import { drizzle } from 'drizzle-orm/node-postgres';
import { DrizzleOutboxStore } from './database/drizzle-outbox.store.js';
import * as schema from './database/schema.js';
import { InventoryModule } from './inventory/inventory.module.js';
import { NotificationsModule } from './notifications/notifications.module.js';
import { OrdersModule } from './orders/orders.module.js';
import { OutboxAdminModule } from './outbox-admin/outbox-admin.module.js';

export const ANALYTICS_SERVICE = 'ANALYTICS_SERVICE';

@Module({
  imports: [
    DrizzleModule.forRootAsync({
      // A pg pool on DATABASE_URL, closed in onApplicationShutdown(), after the relay drained.
      useFactory: () => ({ drizzle, connection: process.env.DATABASE_URL!, schema }),
    }),
    OutboxModule.forRootAsync({
      imports: [
        ClientsModule.registerAsync([
          {
            name: ANALYTICS_SERVICE,
            useFactory: () => ({
              transport: Transport.TCP,
              options: {
                host: process.env.ANALYTICS_HOST ?? '127.0.0.1',
                port: Number(process.env.ANALYTICS_PORT ?? 4001),
              },
            }),
          },
        ]),
      ],
      transports: { analytics: ClientProxyTransport(ANALYTICS_SERVICE) },
      useFactory: () => ({
        // Each message goes to exactly one transport; `local` runs @OnOutboxMessage() handlers.
        route: (message) => (message.topic.startsWith('analytics.') ? 'analytics' : 'local'),
        relay: {
          enabled: process.env.OUTBOX_RELAY !== 'off',
          pollInterval: '1s',
          lease: '30s',
          publishTimeout: '10s',
        },
        retry: {
          attempts: 10,
          backoff: { delay: '1s', maxDelay: '1m' },
        },
      }),
    }),
    OrdersModule,
    NotificationsModule,
    InventoryModule,
    OutboxAdminModule,
  ],
  // Registers itself as the outbox's store.
  providers: [DrizzleOutboxStore],
})
export class AppModule {}

export interface OrderItem {
  productId: string;
  quantity: number;
  /** Unit price in cents, copied from the catalog when the order is placed. */
  price: number;
}

export interface Order {
  id: string;
  userId: string;
  items: OrderItem[];
  /** In cents. */
  total: number;
  status: 'placed' | 'cancelled';
}

export class PlaceOrderDto {
  userId: string;
  items: { productId: string; quantity: number }[];
}

import { BadRequestException, Injectable } from '@nestjs/common';
import { InjectDrizzle } from '@nestjs/drizzle';
import { Outbox } from '@nestjs/outbox';
import { inArray } from 'drizzle-orm';
import { randomUUID } from 'node:crypto';
import type { Database, Transaction } from '../database/drizzle.js';
import { orders, products } from '../database/schema.js';
import type { Order, PlaceOrderDto } from './order.js';

@Injectable()
export class OrdersService {
  constructor(
    @InjectDrizzle() private readonly db: Database,
    private readonly outbox: Outbox<Transaction>,
  ) {}

  async placeOrder({ userId, items }: PlaceOrderDto): Promise<Order> {
    if (!items?.length) throw new BadRequestException('An order needs at least one item');

    const order = await this.db.transaction(async (tx) => {
      const ids = items.map((item) => item.productId);
      const prices = await tx.select().from(products).where(inArray(products.id, ids));
      const lines = items.map(({ productId, quantity }) => {
        const product = prices.find((row) => row.id === productId);
        if (!product) throw new BadRequestException(`Unknown product "${productId}"`);
        if (!Number.isInteger(quantity) || quantity < 1) {
          throw new BadRequestException(`Invalid quantity for "${productId}"`);
        }
        return { productId, quantity, price: product.price };
      });
      const order: Order = {
        id: randomUUID(),
        userId,
        items: lines,
        total: lines.reduce((sum, line) => sum + line.price * line.quantity, 0),
        status: 'placed',
      };
      await tx.insert(orders).values(order);

      // Drizzle's tx: the messages commit or roll back with the order.
      await this.outbox.add(tx, [
        { topic: 'order.placed', payload: order },
        { topic: 'analytics.order.placed', key: order.id, payload: order },
      ]);
      return order;
    });

    this.outbox.notify(); // publish now instead of at the next poll
    return order;
  }
}

import { Body, Controller, HttpCode, HttpStatus, Param, Post } from '@nestjs/common';
import { PlaceOrderDto } from './order.js';
import { OrdersService } from './orders.service.js';

@Controller('orders')
export class OrdersController {
  constructor(private readonly ordersService: OrdersService) {}

  @Post()
  place(@Body() dto: PlaceOrderDto) {
    return this.ordersService.placeOrder(dto);
  }

  @Post(':id/cancel')
  @HttpCode(HttpStatus.OK)
  cancel(@Param('id') id: string) {
    return this.ordersService.cancelOrder(id);
  }
}

import { Injectable, Logger } from '@nestjs/common';
import type { Order } from '../orders/order.js';

/** Stand-in for your email provider's SDK. */
@Injectable()
export class MailerService {
  private readonly logger = new Logger(MailerService.name);

  async sendOrderConfirmation(order: Order): Promise<void> {
    this.logger.log(`Order confirmation for ${order.id} sent to ${order.userId}`);
  }
}

import { Injectable } from '@nestjs/common';
import { OnOutboxMessage } from '@nestjs/outbox';
import type { Order } from '../orders/order.js';
import { MailerService } from './mailer.service.js';

@Injectable()
export class OrderEmailsHandler {
  constructor(private readonly mailerService: MailerService) {}

  // The consumer's inbox skips messages it has already handled.
  @OnOutboxMessage('order.placed', { consumer: 'order-confirmation-email' })
  async sendConfirmation(order: Order) {
    await this.mailerService.sendOrderConfirmation(order);
  }
}

import { Injectable, Logger } from '@nestjs/common';
import { InjectDrizzle } from '@nestjs/drizzle';
import { NonRetryableMessageError, OnOutboxMessage, type OutboxHandlerContext } from '@nestjs/outbox';
import { and, eq, gte, sql } from 'drizzle-orm';
import type { Database, Transaction } from '../database/drizzle.js';
import { products } from '../database/schema.js';
import type { Order } from '../orders/order.js';

@Injectable()
export class StockReservationHandler {
  private readonly logger = new Logger(StockReservationHandler.name);

  constructor(@InjectDrizzle() private readonly db: Database) {}

  @OnOutboxMessage('order.placed', { consumer: 'stock-reservation' })
  async reserve(order: Order, ctx: OutboxHandlerContext<Transaction>) {
    await this.db.transaction(async (tx) => {
      // Records the message id through tx, then runs the callback only if this consumer
      // hasn't processed it yet. Await it: the record commits with the reservation.
      await ctx.processInTransaction(tx, async () => {
        for (const { productId, quantity } of order.items) {
          const reserved = await tx
            .update(products)
            .set({ inStock: sql`${products.inStock} - ${quantity}`, reserved: sql`${products.reserved} + ${quantity}` })
            .where(and(eq(products.id, productId), gte(products.inStock, quantity)))
            .returning({ id: products.id });
          if (reserved.length === 0) {
            // Retrying won't create stock: dead-letter it now and let a human decide.
            throw new NonRetryableMessageError(`Not enough stock for "${productId}" (order ${order.id})`);
          }
        }
        this.logger.log(`Reserved stock for order ${order.id}`);
      });
    });
  }
}

async cancelOrder(id: string): Promise<Order> {
  const order = await this.db.transaction(async (tx) => {
    const [row] = await tx.select().from(orders).where(eq(orders.id, id)).for('update');
    if (!row) throw new NotFoundException(`Order ${id} not found`);
    if (row.status !== 'placed') throw new ConflictException(`Order ${id} is ${row.status}`);

    await tx.update(orders).set({ status: 'cancelled' }).where(eq(orders.id, id));
    const order: Order = { ...row, status: 'cancelled' };
    await this.outbox.add(tx, { topic: 'analytics.order.cancelled', key: order.id, payload: order });
    return order;
  });

  this.outbox.notify();
  return order;
}

import { bigint, index, integer, pgTable, primaryKey, text, timestamp } from 'drizzle-orm/pg-core';

export const orderEvents = pgTable(
  'order_events',
  {
    /** Arrival order. */
    seq: bigint('seq', { mode: 'number' }).primaryKey().generatedAlwaysAsIdentity(),
    orderId: text('order_id').notNull(),
    event: text('event').$type<'placed' | 'cancelled'>().notNull(),
    /** The change to revenue, in cents: the order's total when placed, minus it when cancelled. */
    amount: integer('amount').notNull(),
    recordedAt: timestamp('recorded_at', { withTimezone: true }).notNull().defaultNow(),
  },
  (table) => [index('order_events_order_id').on(table.orderId)],
);

// The inbox, read and written by DrizzleInboxStore: the message ids this service has processed.
export const outboxInbox = pgTable(
  'outbox_inbox',
  {
    consumer: text('consumer').notNull(),
    messageId: text('message_id').notNull(),
    processedAt: timestamp('processed_at', { withTimezone: true }).notNull(),
  },
  (table) => [
    // One record per consumer and message: the unique key two deliveries meet at.
    primaryKey({ columns: [table.consumer, table.messageId] }),
    index('outbox_inbox_processed_at').on(table.processedAt),
  ],
);

import { Injectable } from '@nestjs/common';
import { InjectDrizzle } from '@nestjs/drizzle';
import { OutboxStorage, OutboxTransactionRequiredError, type OutboxInboxStore } from '@nestjs/outbox';
import { and, eq, lt } from 'drizzle-orm';
import type { Database, Transaction } from './drizzle.js';
import { outboxInbox } from './schema.js';

/** The inbox half of the order API's DrizzleOutboxStore: all a consumer-only service needs. */
@Injectable()
export class DrizzleInboxStore implements OutboxInboxStore<Transaction> {
  constructor(
    @InjectDrizzle() private readonly db: Database,
    storage: OutboxStorage,
  ) {
    // Consumer only: the inbox is the one contract this service registers.
    storage.registerSource({ inbox: this });
  }

  async recordInbox(tx: Transaction | undefined, consumer: string, messageId: string, now: number): Promise<boolean> {
    if (tx !== undefined && typeof (tx as Partial<Transaction>).rollback !== 'function') {
      throw new OutboxTransactionRequiredError('Pass the tx that db.transaction() gives you, not the database.');
    }
    // One statement on the unique key: a concurrent delivery waits for this transaction.
    const inserted = await (tx ?? this.db)
      .insert(outboxInbox)
      .values({ consumer, messageId, processedAt: new Date(now) })
      .onConflictDoNothing()
      .returning({ consumer: outboxInbox.consumer });
    return inserted.length === 1;
  }

  async hasInbox(consumer: string, messageId: string): Promise<boolean> {
    const found = await this.db.$count(
      outboxInbox,
      and(eq(outboxInbox.consumer, consumer), eq(outboxInbox.messageId, messageId)),
    );
    return found > 0;
  }

  async pruneInbox(before: number): Promise<number> {
    const pruned = await this.db
      .delete(outboxInbox)
      .where(lt(outboxInbox.processedAt, new Date(before)))
      .returning({ consumer: outboxInbox.consumer });
    return pruned.length;
  }
}

import { Module } from '@nestjs/common';
import { DrizzleModule } from '@nestjs/drizzle';
import { OutboxModule } from '@nestjs/outbox';
import { drizzle } from 'drizzle-orm/node-postgres';
import { AnalyticsController } from './analytics.controller.js';
import { DrizzleInboxStore } from './database/drizzle-inbox.store.js';
import * as schema from './database/schema.js';
import { OrderStatsService } from './order-stats.service.js';

@Module({
  imports: [
    DrizzleModule.forRootAsync({
      // This service's own database, not the order API's.
      useFactory: () => ({ drizzle, connection: process.env.DATABASE_URL!, schema }),
    }),
    // Consumer only: the store holds this service's inbox, and no relay runs.
    OutboxModule.forRoot({ relay: { enabled: false } }),
  ],
  controllers: [AnalyticsController],
  providers: [DrizzleInboxStore, OrderStatsService],
})
export class AnalyticsModule {}

import { Injectable, Logger } from '@nestjs/common';
import { InjectDrizzle } from '@nestjs/drizzle';
import { OutboxInbox } from '@nestjs/outbox';
import { asc, eq, sql } from 'drizzle-orm';
import type { Database, Transaction } from './database/drizzle.js';
import { orderEvents } from './database/schema.js';

/** The fields of the order API's `Order` that this service reads. */
export interface OrderSnapshot {
  id: string;
  total: number;
}

export type OrderEvent = 'placed' | 'cancelled';

@Injectable()
export class OrderStatsService {
  private readonly logger = new Logger(OrderStatsService.name);

  constructor(
    @InjectDrizzle() private readonly db: Database,
    private readonly outboxInbox: OutboxInbox,
  ) {}

  /**
   * Records an order event once per message id. The inbox record and the event commit in
   * one transaction, so a redelivered message changes nothing. `false` for a duplicate.
   */
  async record(messageId: string, event: OrderEvent, order: OrderSnapshot): Promise<boolean> {
    const outcome = await this.db.transaction(async (tx) =>
      // Records the message id through tx, then runs the callback only if it is new here.
      this.outboxInbox.processInTransaction(tx, 'analytics', messageId, async () => {
        const amount = event === 'placed' ? order.total : -order.total;
        await tx.insert(orderEvents).values({ orderId: order.id, event, amount });
        return this.revenue(tx);
      }),
    );
    if (outcome.duplicate) return false;
    this.logger.log(`Order ${order.id} ${event}, revenue is now ${outcome.result}`);
    return true;
  }

  /** Total revenue in cents. */
  async revenue(db: Database | Transaction = this.db): Promise<number> {
    const [row] = await db
      .select({ total: sql<number>`coalesce(sum(${orderEvents.amount}), 0)`.mapWith(Number) })
      .from(orderEvents);
    return row!.total;
  }

  /** An order's events, in the order they were recorded. */
  async timeline(orderId: string): Promise<OrderEvent[]> {
    const rows = await this.db
      .select({ event: orderEvents.event })
      .from(orderEvents)
      .where(eq(orderEvents.orderId, orderId))
      .orderBy(asc(orderEvents.seq));
    return rows.map((row) => row.event);
  }
}

import { Controller, Logger } from '@nestjs/common';
import { EventPattern, Payload } from '@nestjs/microservices';
import type { OutboxEnvelope } from '@nestjs/outbox';
import { OrderStatsService, type OrderEvent, type OrderSnapshot } from './order-stats.service.js';

@Controller()
export class AnalyticsController {
  private readonly logger = new Logger(AnalyticsController.name);
  /** The last event handled for each key: one order's events run one at a time. */
  private readonly queues = new Map<string, Promise<void>>();

  constructor(private readonly orderStatsService: OrderStatsService) {}

  @EventPattern('analytics.order.placed')
  onPlaced(@Payload() envelope: OutboxEnvelope<OrderSnapshot>) {
    return this.apply('placed', envelope);
  }

  @EventPattern('analytics.order.cancelled')
  onCancelled(@Payload() envelope: OutboxEnvelope<OrderSnapshot>) {
    return this.apply('cancelled', envelope);
  }

  private apply(event: OrderEvent, envelope: OutboxEnvelope<OrderSnapshot>) {
    // Events arrive in the order the relay published them, but handlers are asynchronous:
    // without a queue, a cancellation could commit before the placement it follows.
    return this.inOrder(envelope.key ?? envelope.id, async () => {
      // The envelope id is the outbox message id: stable across redeliveries.
      const recorded = await this.orderStatsService.record(envelope.id, event, envelope.payload);
      if (!recorded) this.logger.warn(`Skipped duplicate ${envelope.topic} ${envelope.id}`);
    });
  }

  private inOrder(key: string, work: () => Promise<void>): Promise<void> {
    const next = (this.queues.get(key) ?? Promise.resolve()).then(work, work);
    this.queues.set(key, next);
    const forget = () => {
      if (this.queues.get(key) === next) this.queues.delete(key);
    };
    next.then(forget, forget);
    return next;
  }
}

import { NestFactory } from '@nestjs/core';
import { Transport, type MicroserviceOptions } from '@nestjs/microservices';
import { AnalyticsModule } from './analytics.module.js';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(AnalyticsModule, {
    transport: Transport.TCP,
    options: { host: '127.0.0.1', port: Number(process.env.PORT ?? 4001) },
  });
  app.enableShutdownHooks();
  await app.listen();
}
await bootstrap();

import {
  Controller,
  Delete,
  Get,
  HttpCode,
  HttpStatus,
  Param,
  Post,
  Query,
  UseGuards,
} from '@nestjs/common';
import { OutboxDeadLetters } from '@nestjs/outbox';
import { AdminGuard } from './admin.guard.js';
import { OutboxMetrics } from './outbox-metrics.service.js';

@Controller('admin/outbox')
@UseGuards(AdminGuard)
export class OutboxAdminController {
  constructor(
    private readonly outboxDeadLetters: OutboxDeadLetters,
    private readonly outboxMetrics: OutboxMetrics,
  ) {}

  @Get('dead-letters')
  list(@Query('topic') topic?: string) {
    return this.outboxDeadLetters.list({ topic });
  }

  @Post('dead-letters/:id/requeue')
  @HttpCode(HttpStatus.OK)
  async requeue(@Param('id') id: string) {
    return { requeued: await this.outboxDeadLetters.requeue(id) };
  }

  @Delete('dead-letters/:id')
  async purge(@Param('id') id: string) {
    return { purged: await this.outboxDeadLetters.purge(id) };
  }

  @Get('stats')
  stats() {
    return this.outboxMetrics.snapshot();
  }
}

import { Injectable, type CanActivate, type ExecutionContext } from '@nestjs/common';
import { createHash, timingSafeEqual } from 'node:crypto';

type HttpRequest = { headers: Record<string, string | string[] | undefined> };

/** Placeholder: protect these routes with your application's real authorization. */
@Injectable()
export class AdminGuard implements CanActivate {
  canActivate(context: ExecutionContext): boolean {
    const request = context.switchToHttp().getRequest<HttpRequest>();
    const token = process.env.ADMIN_TOKEN;
    const given = request.headers['x-admin-token'];
    return !!token && typeof given === 'string' && sameSecret(given, token);
  }
}

/** Compares in constant time, so response times don't give the token away. */
function sameSecret(given: string, expected: string): boolean {
  const digest = (value: string) => createHash('sha256').update(value).digest();
  return timingSafeEqual(digest(given), digest(expected));
}

import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module.js';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  // On SIGTERM the relay stops claiming, finishes in-flight publishes and releases its leases.
  app.enableShutdownHooks();
  await app.listen(process.env.PORT ?? 3000);
}
await bootstrap();

import { Injectable, type OnModuleInit } from '@nestjs/common';
import { OutboxEvents, OutboxRelay } from '@nestjs/outbox';

@Injectable()
export class OutboxMetrics implements OnModuleInit {
  readonly counters = { published: 0, retried: 0, deadLettered: 0, leaseLost: 0 };
  /** Time from add() to a successful publish, for the latest message. */
  lastPublishLagMs = 0;

  constructor(
    private readonly outboxEvents: OutboxEvents,
    private readonly outboxRelay: OutboxRelay,
  ) {}

  onModuleInit() {
    this.outboxEvents.events$.subscribe((event) => {
      switch (event.type) {
        case 'published':
          this.counters.published++;
          this.lastPublishLagMs = Date.now() - event.message.createdAt;
          break;
        case 'retry-scheduled':
          this.counters.retried++;
          break;
        case 'dead-lettered':
          this.counters.deadLettered++;
          break;
        case 'lease-lost':
          this.counters.leaseLost++;
          break;
      }
    });
  }

  async snapshot() {
    // pending, ready, leased, deadLetters, lagMs (how long the longest-waiting message has waited)...
    const stats = await this.outboxRelay.stats();
    return { ...stats, ...this.counters, lastPublishLagMs: this.lastPublishLagMs };
  }
}

import { PGlite } from '@electric-sql/pglite';
import { getDrizzleToken } from '@nestjs/drizzle';
import { Outbox, OutboxRelay } from '@nestjs/outbox';
import { Test, type TestingModule } from '@nestjs/testing';
import { drizzle } from 'drizzle-orm/pglite';
import { migrate } from 'drizzle-orm/pglite/migrator';
import { fileURLToPath } from 'node:url';
import { of } from 'rxjs';
import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest';
import { ANALYTICS_SERVICE, AppModule } from '../src/app.module.js';
import * as schema from '../src/database/schema.js';
import { MailerService } from '../src/notifications/mailer.service.js';
import { OrdersService } from '../src/orders/orders.service.js';

describe('OrdersService (outbox)', () => {
  const client = new PGlite(); // PostgreSQL, in-process
  const db = drizzle(client, { schema });
  const mailer = { sendOrderConfirmation: vi.fn() };
  const analytics = { emit: vi.fn(() => of(undefined)) };
  let moduleRef: TestingModule;
  let orders: OrdersService;
  let relay: OutboxRelay;

  beforeAll(async () => {
    // The real migrations, the outbox's tables included.
    await migrate(db, { migrationsFolder: fileURLToPath(new URL('../drizzle', import.meta.url)) });
    process.env.OUTBOX_RELAY = 'off'; // no poll loop: the test drives the relay
    moduleRef = await Test.createTestingModule({ imports: [AppModule] })
      .overrideProvider(getDrizzleToken())
      .useValue(db) // DrizzleOutboxStore injects it too
      .overrideProvider(MailerService)
      .useValue(mailer)
      .overrideProvider(ANALYTICS_SERVICE)
      .useValue(analytics)
      .compile();
    await moduleRef.init();
    orders = moduleRef.get(OrdersService);
    relay = moduleRef.get(OutboxRelay);
  });

  afterAll(async () => {
    await moduleRef.close();
    await client.close();
  });

  it('publishes order events only after the order commits', async () => {
    const order = await orders.placeOrder({ userId: 'user-42', items: [{ productId: 'salmon-kibble-2kg', quantity: 1 }] });
    expect(mailer.sendOrderConfirmation).not.toHaveBeenCalled();

    await relay.runOnce(); // claim and publish one batch

    expect(mailer.sendOrderConfirmation).toHaveBeenCalledWith(order);
    expect(analytics.emit).toHaveBeenCalledWith(
      'analytics.order.placed',
      expect.objectContaining({ key: order.id, payload: order }),
    );
    expect(await relay.stats()).toMatchObject({ pending: 0, deadLetters: 0 });
  });

  it('publishes nothing when the transaction rolls back', async () => {
    const outbox = moduleRef.get(Outbox);
    const add = outbox.add.bind(outbox);
    vi.spyOn(outbox, 'add').mockImplementationOnce(async (tx, messages) => {
      await add(tx, messages); // the messages are written...
      throw new Error('Connection terminated unexpectedly'); // ...then the transaction fails
    });

    await expect(
      orders.placeOrder({ userId: 'user-42', items: [{ productId: 'salmon-kibble-2kg', quantity: 1 }] }),
    ).rejects.toThrow('Connection terminated unexpectedly');

    expect(await relay.stats()).toMatchObject({ pending: 0 });
    expect(await relay.runOnce()).toMatchObject({ claimed: 0 });
  });
});