Outbox & Durable Workflows

npm i --save @nestjs/outbox @nestjs/microservices

import { PGlite } from '@electric-sql/pglite';
import { sql } from 'drizzle-orm';
import { drizzle } from 'drizzle-orm/pglite';
import { jsonb, pgTable, text } from 'drizzle-orm/pg-core';

export const DATABASE = Symbol('DATABASE');

export const orders = pgTable('orders', {
  id: text('id').primaryKey(),
  userId: text('user_id').notNull(),
  total: jsonb('total').$type<number>().notNull(),
  status: text('status').notNull(),
});

export const schema = { orders };

export function createDatabase() {
  // PGlite is PostgreSQL compiled to WebAssembly, running inside the Node.js process.
  const client = new PGlite();
  return { client, db: drizzle(client, { schema }) };
}

export type Database = ReturnType<typeof createDatabase>['db'];
export type Transaction = Parameters<Parameters<Database['transaction']>[0]>[0];

/** Your own migration: only the application's table. The outbox store creates its own schema. */
export async function migrateOrders(db: Database) {
  await db.execute(sql`create table if not exists orders (
    id text primary key,
    user_id text not null,
    total jsonb not null,
    status text not null
  )`);
}

import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { ClientProxyTransport, OutboxModule, OutboxStorage } from '@nestjs/outbox';
import { fromDrizzle, PostgresOutboxStore } from '@nestjs/outbox/postgres';
import { createDatabase, DATABASE, migrateOrders, type Database } from './database.js';
import { MailerService, OrderEmailsHandler } from './handlers.js';
import { OrdersService } from './orders.service.js';

export const ANALYTICS_SERVICE = 'ANALYTICS_SERVICE';

@Module({
  imports: [
    OutboxModule.forRootAsync({
      imports: [
        ClientsModule.registerAsync([
          {
            name: ANALYTICS_SERVICE,
            // Read at startup, not at import time.
            useFactory: () => ({
              transport: Transport.TCP,
              options: { 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: 5, backoff: { delay: '1s', maxDelay: '1m' } },
      }),
    }),
  ],
  providers: [
    {
      provide: DATABASE,
      useFactory: async () => {
        const { db } = createDatabase();
        await migrateOrders(db);
        return db;
      },
    },
    {
      // Messages, dead letters and inbox records in your database, in the nest_outbox schema.
      provide: PostgresOutboxStore,
      inject: [DATABASE, OutboxStorage],
      useFactory: (db: Database, outboxStorage: OutboxStorage) =>
        new PostgresOutboxStore({ executor: fromDrizzle(db) }, outboxStorage),
    },
    MailerService,
    OrderEmailsHandler,
    OrdersService,
  ],
  exports: [OrdersService],
})
export class OutboxDemoModule {}

import { randomUUID } from 'node:crypto';
import { BadRequestException, Inject, Injectable } from '@nestjs/common';
import { Outbox } from '@nestjs/outbox';
import { DATABASE, orders, type Database, type Transaction } from './database.js';

export interface Order {
  id: string;
  userId: string;
  total: number;
  status: 'placed' | 'cancelled';
}

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

  async placeOrder(userId: string, total: number, failAfterAdd = false): Promise<Order> {
    if (total <= 0) {
      throw new BadRequestException('The total must be positive');
    }
    const order: Order = { id: randomUUID(), userId, total, status: 'placed' };

    await this.db.transaction(async (tx) => {
      await tx.insert(orders).values(order);

      // The messages are rows written through the same tx: they 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 },
      ]);

      if (failAfterAdd) {
        throw new Error('Connection terminated unexpectedly');
      }
    });

    this.outbox.notify(); // wake the relay now instead of waiting for the next poll
    return order;
  }
}

import { Injectable } from '@nestjs/common';
import { NonRetryableMessageError, OnOutboxMessage } from '@nestjs/outbox';
import type { Order } from './orders.service.js';

/** Stand-in for an email provider's SDK. */
@Injectable()
export class MailerService {
  readonly sent: string[] = [];
  failTimes = 0;

  async sendOrderConfirmation(order: Order): Promise<void> {
    if (this.failTimes > 0) {
      this.failTimes--;
      throw new Error('mail provider unavailable');
    }
    this.sent.push(order.id);
  }
}

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

  // The consumer name identifies this handler's inbox: keep it stable.
  @OnOutboxMessage('order.placed', { consumer: 'order-confirmation-email' })
  async sendConfirmation(order: Order) {
    if (order.total > 1_000_000) {
      // Retrying will not make it valid: dead-letter at once, reason "rejected".
      throw new NonRetryableMessageError(`Order ${order.id} exceeds the mail provider limit`);
    }
    await this.mailer.sendOrderConfirmation(order);
  }
}

import { Controller } from '@nestjs/common';
import { EventPattern, Payload } from '@nestjs/microservices';

@Controller()
export class AnalyticsController {
  @EventPattern('analytics.order.placed')
  record(@Payload() envelope: { id: string; key: string; payload: unknown }) {
    // Deduplicate by envelope.id in the analytics service's own database.
  }
}

import { Controller, Delete, Get, HttpCode, HttpStatus, Param, Post, Query, UseGuards } from '@nestjs/common';
import { OutboxDeadLetters } from '@nestjs/outbox';
import { AdminGuard } from './admin.guard.js'; // your real authorization: dead letters hold payloads

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

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

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

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

import { Controller, Module, type INestMicroservice } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { EventPattern, Payload, Transport } from '@nestjs/microservices';
import { OutboxDeadLetters, OutboxRelay } from '@nestjs/outbox';
import { Test, type TestingModule } from '@nestjs/testing';
import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest';
import { MailerService } from '../src/outbox-workflows/handlers.js';
import { OrdersService } from '../src/outbox-workflows/orders.service.js';
import { OutboxDemoModule } from '../src/outbox-workflows/outbox-demo.module.js';

const received: { topic: string; key: string }[] = [];

// A real Nest microservice over TCP, standing in for the analytics service.
@Controller()
class AnalyticsController {
  @EventPattern('analytics.order.placed')
  record(@Payload() envelope: { topic: string; key: string }) {
    received.push(envelope);
  }
}
@Module({ controllers: [AnalyticsController] })
class AnalyticsModule {}

describe('outbox', () => {
  let moduleRef: TestingModule;
  let analytics: INestMicroservice;
  let orders: OrdersService;
  let relay: OutboxRelay;
  let mailer: MailerService;

  beforeAll(async () => {
    process.env.OUTBOX_RELAY = 'off'; // no poll loop: the test drives the relay
    process.env.ANALYTICS_PORT = '47011';
    analytics = await NestFactory.createMicroservice(AnalyticsModule, {
      transport: Transport.TCP,
      options: { host: '127.0.0.1', port: 47011 },
      logger: false,
    });
    await analytics.listen();
    moduleRef = await Test.createTestingModule({ imports: [OutboxDemoModule] }).compile();
    await moduleRef.init();
    orders = moduleRef.get(OrdersService);
    relay = moduleRef.get(OutboxRelay);
    mailer = moduleRef.get(MailerService);
  });

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

  it('publishes only after the commit, to the handler and to the microservice', async () => {
    const order = await orders.placeOrder('user-1', 2499);
    expect(mailer.sent).toEqual([]); // nothing is published inside the transaction
    expect(await relay.stats()).toMatchObject({ pending: 2 });

    expect(await relay.runOnce()).toMatchObject({ claimed: 2, published: 2 });

    expect(mailer.sent).toEqual([order.id]);
    await vi.waitFor(() => expect(received).toHaveLength(1));
    expect(received[0]).toMatchObject({ topic: 'analytics.order.placed', key: order.id });
    expect(await relay.stats()).toMatchObject({ pending: 0, deadLetters: 0 });
  });

  it('writes nothing when the transaction rolls back', async () => {
    await expect(orders.placeOrder('user-1', 100, true)).rejects.toThrow('Connection terminated');
    expect(await relay.stats()).toMatchObject({ pending: 0 });
    expect(await relay.runOnce()).toMatchObject({ claimed: 0 });
  });

  it('retries a failed handler with backoff', async () => {
    mailer.failTimes = 1;
    const now = Date.now();
    const clock = vi.spyOn(Date, 'now'); // the relay and the store read the time from Date.now()
    await orders.placeOrder('user-2', 500);

    expect(await relay.runOnce()).toMatchObject({ retried: 1 });
    clock.mockReturnValue(now + 10_000); // move past the backoff
    expect(await relay.runOnce()).toMatchObject({ published: 1, retried: 0 });
    clock.mockRestore();

    expect(await relay.stats()).toMatchObject({ pending: 0, deadLetters: 0 });
  });

  it('dead-letters a non-retryable failure and requeues it', async () => {
    await orders.placeOrder('user-3', 2_000_000);
    expect(await relay.runOnce()).toMatchObject({ deadLettered: 1 });

    const deadLetters = moduleRef.get(OutboxDeadLetters);
    const [letter] = await deadLetters.list({ topic: 'order.placed' });
    expect(letter.history).toHaveLength(1);
    expect(await deadLetters.requeue(letter.id)).toBeTruthy();
    expect((await relay.stats()).pending).toBeGreaterThan(0);
  });
});

npm i --save @nestjs/workflows

import { Module } from '@nestjs/common';
import { WorkflowsModule, WorkflowStorage } from '@nestjs/workflows';
import { fromDrizzle, PostgresWorkflowStore } from '@nestjs/workflows/postgres';
import { createDatabase, DATABASE, migrateOrders, type Database } from '../database.js';
import { OrderFulfilmentWorkflow } from './order-fulfilment.workflow.js';
import { InventoryService, MailService, PaymentsService } from './services.js';

@Module({
  imports: [
    WorkflowsModule.forRoot({
      worker: {
        concurrency: 10, // instances this process executes at once
        leaseDuration: '30s', // how long a crashed process keeps its instances
        shutdownTimeout: '10s', // how long a deploy waits for running steps
      },
    }),
  ],
  providers: [
    {
      provide: DATABASE,
      useFactory: async () => {
        const { db } = createDatabase();
        await migrateOrders(db);
        return db;
      },
    },
    {
      // Instances, journals and signals in your database, in the nest_workflows schema.
      provide: PostgresWorkflowStore,
      inject: [DATABASE, WorkflowStorage],
      useFactory: (db: Database, workflowStorage: WorkflowStorage) =>
        new PostgresWorkflowStore({ executor: fromDrizzle(db) }, workflowStorage),
    },
    PaymentsService,
    InventoryService,
    MailService,
    OrderFulfilmentWorkflow,
  ],
  exports: [DATABASE, PaymentsService, InventoryService, MailService],
})
export class WorkflowsDemoModule {}

import { Injectable } from '@nestjs/common';
import { NonRetryableStepError } from '@nestjs/workflows';
import type { Order } from './order.js';

/** In-memory stand-ins for a payment provider, a warehouse and a mail provider. They deduplicate by idempotency key. */
@Injectable()
export class PaymentsService {
  readonly charges = new Map<string, number>();
  readonly refunds = new Set<string>();

  async charge(order: Order, idempotencyKey: string) {
    if (order.total > 100_000) {
      throw new NonRetryableStepError(`The payment provider declined order ${order.id}.`);
    }
    this.charges.set(idempotencyKey, order.total); // a repeated key returns the first charge
    return { chargeId: `ch_${idempotencyKey}`, amount: order.total };
  }

  async refund(chargeId: string, idempotencyKey: string) {
    this.refunds.add(idempotencyKey);
  }
}

@Injectable()
export class InventoryService {
  stock = new Map<string, number>([['kibble', 10]]);
  readonly reservations = new Map<string, Order['items']>();

  async reserve(items: Order['items'], idempotencyKey: string) {
    if (this.reservations.has(idempotencyKey)) {
      return { reservationId: idempotencyKey };
    }
    for (const item of items) {
      if ((this.stock.get(item.productId) ?? 0) < item.quantity) {
        throw new NonRetryableStepError(`Not enough stock for "${item.productId}"`);
      }
    }
    for (const item of items) {
      this.stock.set(item.productId, this.stock.get(item.productId)! - item.quantity);
    }
    this.reservations.set(idempotencyKey, items);
    return { reservationId: idempotencyKey };
  }

  async release(reservationId: string) {
    for (const item of this.reservations.get(reservationId) ?? []) {
      this.stock.set(item.productId, this.stock.get(item.productId)! + item.quantity);
    }
    this.reservations.delete(reservationId);
  }
}

@Injectable()
export class MailService {
  readonly reviewRequests: string[] = [];

  async sendReviewRequest(order: Order, idempotencyKey: string) {
    this.reviewRequests.push(idempotencyKey);
  }
}

import { WorkflowSignal } from '@nestjs/workflows';
import type { CarrierEvent } from './order.js';

/** One definition, shared by the sender and the workflow: neither can misspell the name or the payload. */
export const shipmentDelivered = new WorkflowSignal<CarrierEvent>('shipment.delivered');
export const shipmentReturned = new WorkflowSignal<CarrierEvent>('shipment.returned');

import { Workflow, type WorkflowContext, type WorkflowRunner } from '@nestjs/workflows';
import { shipmentDelivered, shipmentReturned } from './carrier-events.js';
import type { FulfilmentResult, Order } from './order.js';
import { InventoryService, MailService, PaymentsService } from './services.js';

/** One fulfilment per order: the order id is the workflow instance id. */
export const fulfilmentId = (orderId: string) => `order-${orderId}`;

@Workflow('order-fulfilment', { timeout: '30d' })
export class OrderFulfilmentWorkflow implements WorkflowRunner<Order, FulfilmentResult> {
  constructor(
    private readonly payments: PaymentsService,
    private readonly inventory: InventoryService,
    private readonly mail: MailService,
  ) {}

  async run(ctx: WorkflowContext, order: Order): Promise<FulfilmentResult> {
    const charge = await ctx.step(
      'charge-payment',
      ({ idempotencyKey }) => this.payments.charge(order, idempotencyKey),
      {
        retry: { attempts: 5, backoff: { delay: '2s' } },
        // Runs, in reverse order, if the workflow later fails or is cancelled.
        compensate: (charge, { idempotencyKey }) => this.payments.refund(charge.chargeId, idempotencyKey),
      },
    );
    ctx.setStatus('paid');

    await ctx.step(
      'reserve-stock',
      ({ idempotencyKey }) => this.inventory.reserve(order.items, idempotencyKey),
      { compensate: ({ reservationId }) => this.inventory.release(reservationId) },
    );
    ctx.setStatus('awaiting-delivery');

    // Parks the instance in the database: no memory, timer or worker is held for up to 3 days.
    const outcome = await ctx.waitForAny('await-delivery', {
      delivered: ctx.signalWait(shipmentDelivered, { key: order.id }),
      returned: ctx.signalWait(shipmentReturned, { key: order.id }),
      timedOut: ctx.timer('3d'),
    });
    if (outcome.key === 'returned') {
      ctx.fail(`The carrier returned order ${order.id} to the warehouse.`);
    }
    if (outcome.key === 'timedOut') {
      ctx.fail(`Order ${order.id} was not delivered within 3 days.`);
    }
    const delivery = outcome.value; // narrowed to a CarrierEvent

    // Point of no return: nothing from here on refunds the charge or releases the stock.
    ctx.commit('delivered');
    ctx.setStatus('delivered');

    await ctx.sleep('before-review-request', '7d');
    await ctx.step('send-review-request', ({ idempotencyKey }) => this.mail.sendReviewRequest(order, idempotencyKey));

    return { chargeId: charge.chargeId, trackingNumber: delivery.trackingNumber };
  }
}

import { Inject, Injectable } from '@nestjs/common';
import { WorkflowClient } from '@nestjs/workflows';
import { DATABASE, orders, type Database } from '../database.js';
import { fulfilmentId, OrderFulfilmentWorkflow } from './order-fulfilment.workflow.js';
import type { Order } from './order.js';

@Injectable()
export class OrdersTxService {
  constructor(
    @Inject(DATABASE) private readonly db: Database,
    private readonly workflowClient: WorkflowClient,
  ) {}

  async place(order: Order, fail = false) {
    return this.db.transaction(async (tx) => {
      await tx.insert(orders).values({ id: order.id, userId: order.userId, total: order.total, status: 'placed' });
      // Same transaction: the fulfilment exists if and only if the order does.
      const fulfilment = await this.workflowClient.start(OrderFulfilmentWorkflow, order, {
        id: fulfilmentId(order.id),
        transaction: tx,
        priority: 2, // lower runs first
      });
      if (fail) {
        throw new Error('insert or update violates foreign key constraint');
      }
      return { ...order, fulfilment };
    });
  }
}

// In a webhook handler: the order's new status and the signal commit together.
await this.db.transaction(async (tx) => {
  await tx.update(orders).set({ status: 'delivered' }).where(eq(orders.id, event.reference));
  await this.workflowClient.signal(shipmentDelivered, event, { key: event.reference, transaction: tx });
});

import { Workflow, type WorkflowContext, type WorkflowRunner } from '@nestjs/workflows';
import { shipmentDelivered, shipmentReturned } from './carrier-events.js';
import type { CarrierEvent, Order } from './order.js';

/** Waits for the carrier: a child workflow with its own steps, retries and status. */
@Workflow('shipment')
export class ShipmentWorkflow implements WorkflowRunner<Order, CarrierEvent> {
  async run(ctx: WorkflowContext, order: Order): Promise<CarrierEvent> {
    await ctx.step('book-pickup', async () => ({ pickupId: `pickup-${order.id}` }));

    const outcome = await ctx.waitForAny('await-delivery', {
      delivered: ctx.signalWait(shipmentDelivered, { key: order.id }),
      returned: ctx.signalWait(shipmentReturned, { key: order.id }),
      timedOut: ctx.timer('3d'),
    });
    if (outcome.key !== 'delivered') {
      ctx.fail(`Order ${order.id} was not delivered (${outcome.key}).`);
    }
    return outcome.value;
  }
}

@Workflow('order-with-shipment')
export class OrderWithShipmentWorkflow implements WorkflowRunner<Order, string> {
  async run(ctx: WorkflowContext, order: Order): Promise<string> {
    await ctx.step('charge', async () => ({ ok: true }), { compensate: async () => undefined });
    // Starts the child and waits for it. A failed child throws ChildWorkflowFailedError.
    const delivery = await ctx.executeChild(ShipmentWorkflow, order);
    return delivery.trackingNumber;
  }
}

@Workflow('invoice-batch', {
  schedules: [
    {
      id: 'monthly-invoices',
      cron: '0 2 1 * *', // 02:00 on the 1st of every month, in the zone below
      tz: 'America/New_York',
      missed: 'once', // a deploy or an outage at 02:00 delays the batch instead of skipping a month
      input: ({ at }) => ({ month: lastMonth(at) }),
    },
  ],
})
export class InvoiceBatchWorkflow implements WorkflowRunner<{ month: string }, { rendered: number }> {
  async run(ctx: WorkflowContext, { month }: { month: string }) {
    return ctx.step(
      'render-invoices',
      async ({ progress, heartbeat, signal }: WorkflowStepContext<{ cursor: number | null; rendered: number }>) => {
        let { cursor, rendered } = progress ?? { cursor: 0, rendered: 0 };
        while (cursor !== null) {
          const page = await this.invoices.renderPage(month, cursor, signal);
          cursor = page.nextCursor;
          rendered += page.rendered;
          await heartbeat({ cursor, rendered }); // the checkpoint a retry receives as `progress`
        }
        return { rendered };
      },
      { heartbeatTimeout: '2m' },
    );
  }
}

import { Workflow, type WorkflowContext, type WorkflowRunner } from '@nestjs/workflows';
import type { FulfilmentResult, Order } from './order.js';
import { InventoryService, MailService, PaymentsService } from './services.js';

// Changing what a replay would see (here, a step inserted before reserve-stock) ships as a new version.
@Workflow('order-fulfilment', { version: 2 })
export class OrderFulfilmentWorkflowV2 implements WorkflowRunner<Order, FulfilmentResult> {
  constructor(
    private readonly payments: PaymentsService,
    private readonly inventory: InventoryService,
    private readonly mail: MailService,
  ) {}

  async run(ctx: WorkflowContext, order: Order): Promise<FulfilmentResult> {
    const charge = await ctx.step('charge-payment', ({ idempotencyKey }) => this.payments.charge(order, idempotencyKey));
    await ctx.step('send-order-confirmation', async () => ({ sent: true }));
    await ctx.step('reserve-stock', ({ idempotencyKey }) => this.inventory.reserve(order.items, idempotencyKey));
    return { chargeId: charge.chargeId, trackingNumber: 'n/a' };
  }
}

import { Inject, Module } from '@nestjs/common';
import { CommandBus, CommandHandler, CqrsModule, EventBus, type ICommandHandler } from '@nestjs/cqrs';
import { WorkflowsModule, WorkflowStorage, Workflow, type WorkflowContext, type WorkflowRunner } from '@nestjs/workflows';
import { fromDrizzle, PostgresWorkflowStore } from '@nestjs/workflows/postgres';
import { SignalOn, StartOn, WorkflowsCqrsModule } from '@nestjs/workflows/cqrs';
import { createDatabase, DATABASE, migrateOrders, orders, type Database } from '../database.js';
import { shipmentDelivered } from './carrier-events.js';
import type { CarrierEvent, Order } from './order.js';

export class OrderPlacedEvent {
  constructor(readonly order: Order) {}
}
export class OrderDeliveredEvent {
  constructor(readonly delivery: CarrierEvent) {}
}
export class PlaceOrderCommand {
  constructor(readonly order: Order, readonly fail = false) {}
}

@Workflow('cqrs-fulfilment')
@StartOn(OrderPlacedEvent, {
  id: (event) => `order-${event.order.id}`, // required: makes a re-published event find its instance
  input: (event) => event.order,
})
@SignalOn(OrderDeliveredEvent, {
  signal: shipmentDelivered,
  key: (event) => event.delivery.reference,
  id: (event) => event.delivery.trackingNumber, // a re-published event stores no second signal
  payload: (event) => event.delivery,
})
export class CqrsFulfilmentWorkflow implements WorkflowRunner<Order, string> {
  async run(ctx: WorkflowContext, order: Order): Promise<string> {
    const delivery = await ctx.waitForSignal('await-delivery', shipmentDelivered, { key: order.id });
    return delivery!.trackingNumber;
  }
}

@CommandHandler(PlaceOrderCommand)
export class PlaceOrderHandler implements ICommandHandler<PlaceOrderCommand> {
  constructor(
    @Inject(DATABASE) private readonly db: Database,
    private readonly eventBus: EventBus,
  ) {}

  async execute({ order, fail }: PlaceOrderCommand): Promise<Order> {
    return this.db.transaction(async (tx) => {
      await tx.insert(orders).values({ id: order.id, userId: order.userId, total: order.total, status: 'placed' });
      // The transaction rides along as the dispatcher context: the workflow this event
      // starts commits with the order, or rolls back with it.
      await this.eventBus.publish(new OrderPlacedEvent(order), { transaction: tx });
      if (fail) {
        throw new Error('constraint violated');
      }
      return order;
    });
  }
}

@Module({
  imports: [CqrsModule.forRoot(), WorkflowsModule.forRoot(), WorkflowsCqrsModule],
  providers: [
    {
      provide: DATABASE,
      useFactory: async () => {
        const { db } = createDatabase();
        await migrateOrders(db);
        return db;
      },
    },
    {
      provide: PostgresWorkflowStore,
      inject: [DATABASE, WorkflowStorage],
      useFactory: (db: Database, workflowStorage: WorkflowStorage) =>
        new PostgresWorkflowStore({ executor: fromDrizzle(db) }, workflowStorage),
    },
    PlaceOrderHandler,
    CqrsFulfilmentWorkflow,
  ],
})
export class CqrsDemoModule {}

import { Test, type TestingModule } from '@nestjs/testing';
import { ManualWorkflowClock, WORKFLOWS_MODULE_OPTIONS, WorkflowClient, WorkflowWorker } from '@nestjs/workflows';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { shipmentDelivered } from '../src/outbox-workflows/wf/carrier-events.js';
import { fulfilmentId } from '../src/outbox-workflows/wf/order-fulfilment.workflow.js';
import type { Order } from '../src/outbox-workflows/wf/order.js';
import { InventoryService, MailService, PaymentsService } from '../src/outbox-workflows/wf/services.js';
import { WorkflowsDemoModule } from '../src/outbox-workflows/wf/workflows-demo.module.js';

const order = (id: string, quantity = 1): Order => ({
  id,
  userId: 'u1',
  items: [{ productId: 'kibble', quantity, price: 2499 }],
  total: 2499 * quantity,
});

describe('order fulfilment workflow', () => {
  let clock: ManualWorkflowClock;
  let moduleRef: TestingModule;
  let client: WorkflowClient;
  let worker: WorkflowWorker;

  beforeEach(async () => {
    clock = new ManualWorkflowClock(); // time only moves when the test says so
    moduleRef = await Test.createTestingModule({ imports: [WorkflowsDemoModule] })
      .overrideProvider(WORKFLOWS_MODULE_OPTIONS)
      .useValue({ clock, worker: { enabled: false, shutdownTimeout: 50 } }) // no polling: drain() runs due instances
      .compile();
    await moduleRef.init();
    client = moduleRef.get(WorkflowClient);
    worker = moduleRef.get(WorkflowWorker);
  });

  afterEach(() => moduleRef.close());

  it('completes across a signal and a 7-day sleep', async () => {
    const o = order('a1');
    await client.start('order-fulfilment', o, { id: fulfilmentId(o.id) });
    await worker.drain(); // charge, reserve, then park

    let instance = await client.getStatus(fulfilmentId(o.id), { journal: true });
    expect(instance).toMatchObject({ status: 'suspended', customStatus: 'awaiting-delivery' });
    expect(instance!.waits.map((wait) => wait.signal)).toEqual(['shipment.delivered', 'shipment.returned']);

    await client.signal(shipmentDelivered, { type: 'shipment.delivered', reference: o.id, trackingNumber: 'TRK1' }, { key: o.id });
    await worker.drain(); // takes the signal, commits, parks on the 7-day sleep
    expect(await client.getStatus(fulfilmentId(o.id))).toMatchObject({ status: 'suspended', customStatus: 'delivered' });

    clock.advance('7d');
    await worker.drain();
    instance = await client.getStatus(fulfilmentId(o.id), { journal: true });
    expect(instance).toMatchObject({ status: 'completed', output: { trackingNumber: 'TRK1' } });
    expect(moduleRef.get(MailService).reviewRequests).toHaveLength(1);
  });

  it('refunds and releases the stock when the carrier never delivers', async () => {
    const o = order('a2');
    await client.start('order-fulfilment', o, { id: fulfilmentId(o.id) });
    await worker.drain();

    clock.advance('3d');
    await worker.drain(); // the wait times out and the compensations run

    const instance = await client.getStatus(fulfilmentId(o.id), { journal: true });
    expect(instance).toMatchObject({ status: 'failed', error: { message: `Order ${o.id} was not delivered within 3 days.` } });
    expect(instance!.journal.filter((entry) => entry.kind === 'compensation').map((entry) => entry.name)).toEqual([
      '$compensate:reserve-stock', // reverse order
      '$compensate:charge-payment',
    ]);
    expect(moduleRef.get(PaymentsService).refunds.size).toBe(1);
    expect(moduleRef.get(InventoryService).stock.get('kibble')).toBe(10);
  });

  it('cancels with compensation', async () => {
    const o = order('a3');
    await client.start('order-fulfilment', o, { id: fulfilmentId(o.id) });
    await worker.drain();

    expect(await client.cancel(fulfilmentId(o.id), 'Customer changed their mind')).toMatchObject({ accepted: true });
    await worker.drain();

    expect(await client.getStatus(fulfilmentId(o.id))).toMatchObject({ status: 'cancelled' });
  });

  it('starts idempotently, and fails fast on non-retryable errors', async () => {
    const o = order('a4');
    expect(await client.start('order-fulfilment', o, { id: fulfilmentId(o.id) })).toMatchObject({ created: true });
    expect(await client.start('order-fulfilment', o, { id: fulfilmentId(o.id) })).toMatchObject({ created: false });
    await expect(client.start('order-fulfilment', order('a4', 2), { id: fulfilmentId(o.id) })).rejects.toThrow();

    await client.start('order-fulfilment', { ...order('a5'), total: 200_000 }, { id: fulfilmentId('a5') }); // declined card
    await client.start('order-fulfilment', order('a6', 11), { id: fulfilmentId('a6') }); // out of stock
    await worker.drain();

    expect((await client.getStatus(fulfilmentId('a5')))!.error?.message).toContain('declined');
    expect((await client.getStatus(fulfilmentId('a6')))!.error?.message).toContain('Not enough stock');
  });
});