feat(events): add leased outbox delivery and scoped operational APIs

src/checkout/checkout.store.ts

@@ -9,6 +9,7 @@ import { checkoutSnapshot } from './checkout-snapshot';
 import { holdOrderStock } from './stock-allocation';
 import { orderView } from './order-view';
 import { snapshotPrice } from '../pricing/snapshot-price';
+import { enqueueEvent } from '../events/enqueue-event';
 
 @Injectable()
 export class CheckoutStore {
@@ -68,6 +69,7 @@ export class CheckoutStore {
       });
       await holdOrderStock(tx, actor, order.id, expiresAt, lines);
       const pricing = await snapshotPrice(tx, order);
+      await enqueueEvent(tx, actor.organizationId, order.id, 'order.created');
       await tx.cartLine.deleteMany({ where: { cartId } });
       await tx.cart.update({
         where: { id: cartId },

src/checkout/order.store.ts

@@ -6,6 +6,7 @@ import { recordAudit } from '../identity/audit';
 import { AppError } from '../common/errors/app-error';
 import { lockStock } from '../inventory/stock-lock';
 import { orderView, orderStatus } from './order-view';
+import { enqueueEvent } from '../events/enqueue-event';
 
 @Injectable()
 export class OrderStore {
@@ -85,6 +86,7 @@ export class OrderStore {
           data: { status: 'CANCELLED' },
           include: { lines: true, pricing: true },
         });
+        await enqueueEvent(tx, actor.organizationId, id, 'order.cancelled');
         await recordAudit(
           tx,
           actor.organizationId,

src/events/delivery.port.ts

@@ -0,0 +1,11 @@
+export interface DeliveryMessage {
+  id: string;
+  organizationId: string;
+  orderId: string;
+  kind: string;
+}
+export abstract class DeliveryPort {
+  abstract assertConfigured(): void;
+  // Delivery is at-least-once. The adapter must deduplicate using message.id.
+  abstract deliver(message: DeliveryMessage): Promise<void>;
+}

src/events/delivery.service.ts

@@ -0,0 +1,42 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { DeliveryPort } from './delivery.port';
+import { DeliveryStore } from './delivery.store';
+@Injectable()
+export class DeliveryService {
+  private readonly logger = new Logger('CommerceDelivery');
+  constructor(
+    private readonly delivery: DeliveryPort,
+    private readonly store: DeliveryStore,
+  ) {}
+  async dispatchOne(): Promise<boolean> {
+    this.delivery.assertConfigured();
+    const row = await this.store.claim();
+    if (!row) return false;
+    try {
+      await this.delivery.deliver({
+        id: row.event.id,
+        organizationId: row.event.organizationId,
+        orderId: row.event.orderId,
+        kind: row.event.kind,
+      });
+      const result = await this.store.acknowledge(row.eventId, row.leaseToken!);
+      if (!result.count)
+        this.logger.warn(
+          JSON.stringify({
+            event: 'DELIVERY_LEASE_LOST',
+            eventId: row.eventId,
+          }),
+        );
+    } catch {
+      await this.store.fail(row.eventId, row.leaseToken!, row.attempts);
+      this.logger.error(
+        JSON.stringify({
+          event: 'DELIVERY_FAILED',
+          eventId: row.eventId,
+          attempt: row.attempts,
+        }),
+      );
+    }
+    return true;
+  }
+}

src/events/delivery.store.ts

@@ -0,0 +1,57 @@
+import { Injectable } from '@nestjs/common';
+import { randomUUID } from 'node:crypto';
+import { DatabaseService } from '../database/database.service';
+
+@Injectable()
+export class DeliveryStore {
+  constructor(private readonly db: DatabaseService) {}
+  claim() {
+    return this.db.$transaction(async (tx) => {
+      const now = new Date();
+      const rows = await tx.$queryRaw<{ event_id: string }[]>`
+        SELECT event_id FROM event_deliveries
+        WHERE delivered_at IS NULL AND attempts < 5 AND available_at <= ${now}
+          AND (lease_expires_at IS NULL OR lease_expires_at <= ${now})
+        ORDER BY available_at, event_id FOR UPDATE SKIP LOCKED LIMIT 1`;
+      if (!rows[0]) return null;
+      return tx.eventDelivery.update({
+        where: { eventId: rows[0].event_id },
+        data: {
+          attempts: { increment: 1 },
+          leaseToken: randomUUID(),
+          leaseExpiresAt: new Date(now.getTime() + 60000),
+        },
+        include: { event: true },
+      });
+    });
+  }
+  acknowledge(eventId: string, leaseToken: string) {
+    return this.db.eventDelivery.updateMany({
+      where: {
+        eventId,
+        leaseToken,
+        deliveredAt: null,
+        leaseExpiresAt: { gt: new Date() },
+      },
+      data: {
+        deliveredAt: new Date(),
+        leaseToken: null,
+        leaseExpiresAt: null,
+        lastErrorCode: null,
+      },
+    });
+  }
+  fail(eventId: string, leaseToken: string, attempts: number) {
+    return this.db.eventDelivery.updateMany({
+      where: { eventId, leaseToken, deliveredAt: null },
+      data: {
+        leaseToken: null,
+        leaseExpiresAt: null,
+        lastErrorCode: 'DELIVERY_FAILED',
+        availableAt: new Date(
+          Date.now() + Math.min(3600, 30 * 2 ** (attempts - 1)) * 1000,
+        ),
+      },
+    });
+  }
+}

src/events/disabled-delivery.ts

@@ -0,0 +1,12 @@
+import { Injectable } from '@nestjs/common';
+import { AppError } from '../common/errors/app-error';
+import { DeliveryPort } from './delivery.port';
+@Injectable()
+export class DisabledDelivery extends DeliveryPort {
+  assertConfigured(): void {
+    throw new AppError('DELIVERY_UNAVAILABLE');
+  }
+  async deliver(): Promise<void> {
+    this.assertConfigured();
+  }
+}

src/events/enqueue-event.ts

@@ -0,0 +1,13 @@
+import type { Prisma } from '../generated/prisma/client';
+export async function enqueueEvent(
+  tx: Prisma.TransactionClient,
+  organizationId: string,
+  orderId: string,
+  kind: 'order.created' | 'order.cancelled',
+) {
+  return tx.commerceEvent.upsert({
+    where: { orderId_kind: { orderId, kind } },
+    create: { organizationId, orderId, kind, delivery: { create: {} } },
+    update: {},
+  });
+}

src/operations/operations.controller.ts

@@ -0,0 +1,41 @@
+import {
+  Controller,
+  Get,
+  Param,
+  ParseUUIDPipe,
+  Post,
+  Query,
+} from '@nestjs/common';
+import {
+  CurrentPrincipal,
+  RequirePermission,
+} from '../identity/access.decorator';
+import type { Principal } from '../identity/identity.types';
+import { pageSchema, type PageInput } from '../identity/identity.schemas';
+import { SchemaPipe } from '../common/validation.pipe';
+import { OperationsStore } from './operations.store';
+@Controller('admin/operations')
+export class OperationsController {
+  constructor(private readonly operations: OperationsStore) {}
+  @Get('summary')
+  @RequirePermission('operations.read')
+  summary(@CurrentPrincipal() actor: Principal) {
+    return this.operations.summary(actor);
+  }
+  @Get('events')
+  @RequirePermission('operations.read')
+  events(
+    @CurrentPrincipal() actor: Principal,
+    @Query(new SchemaPipe(pageSchema)) page: PageInput,
+  ) {
+    return this.operations.events(actor, page);
+  }
+  @Post('events/:id/retry')
+  @RequirePermission('notifications.retry')
+  retry(
+    @CurrentPrincipal() actor: Principal,
+    @Param('id', ParseUUIDPipe) id: string,
+  ) {
+    return this.operations.retry(actor, id);
+  }
+}

src/operations/operations.module.ts

@@ -3,9 +3,21 @@ import { DatabaseModule } from '../database/database.module';
 import { IdentityModule } from '../identity/identity.module';
 import { PricingController } from '../pricing/pricing.controller';
 import { PricingStore } from '../pricing/pricing.store';
+import { OperationsController } from './operations.controller';
+import { OperationsStore } from './operations.store';
+import { DeliveryPort } from '../events/delivery.port';
+import { DisabledDelivery } from '../events/disabled-delivery';
+import { DeliveryStore } from '../events/delivery.store';
+import { DeliveryService } from '../events/delivery.service';
 @Module({
   imports: [DatabaseModule, IdentityModule],
-  controllers: [PricingController],
-  providers: [PricingStore],
+  controllers: [PricingController, OperationsController],
+  providers: [
+    PricingStore,
+    OperationsStore,
+    DeliveryStore,
+    DeliveryService,
+    { provide: DeliveryPort, useClass: DisabledDelivery },
+  ],
 })
 export class OperationsModule {}

src/operations/operations.store.ts

@@ -0,0 +1,119 @@
+import { Injectable } from '@nestjs/common';
+import { DatabaseService } from '../database/database.service';
+import { AccessStore } from '../identity/access.store';
+import type { Principal } from '../identity/identity.types';
+import type { PageInput } from '../identity/identity.schemas';
+import { AppError } from '../common/errors/app-error';
+import { recordAudit } from '../identity/audit';
+
+@Injectable()
+export class OperationsStore {
+  constructor(
+    private readonly db: DatabaseService,
+    private readonly access: AccessStore,
+  ) {}
+  async summary(actor: Principal) {
+    const organizationId = actor.organizationId;
+    const now = new Date();
+    return this.db.$transaction(
+      async (tx) => ({
+        pendingOrders: await tx.order.count({
+          where: {
+            organizationId,
+            status: 'PENDING_PAYMENT',
+            expiresAt: { gt: now },
+          },
+        }),
+        expiredOrders: await tx.order.count({
+          where: {
+            organizationId,
+            status: 'PENDING_PAYMENT',
+            expiresAt: { lte: now },
+          },
+        }),
+        cancelledOrders: await tx.order.count({
+          where: { organizationId, status: 'CANCELLED' },
+        }),
+        unpricedActiveOrders: await tx.order.count({
+          where: {
+            organizationId,
+            status: 'PENDING_PAYMENT',
+            expiresAt: { gt: now },
+            pricing: null,
+          },
+        }),
+        pendingDeliveries: await tx.eventDelivery.count({
+          where: {
+            event: { organizationId },
+            deliveredAt: null,
+            attempts: { lt: 5 },
+          },
+        }),
+        failedDeliveries: await tx.eventDelivery.count({
+          where: {
+            event: { organizationId },
+            deliveredAt: null,
+            attempts: { gte: 5 },
+          },
+        }),
+      }),
+      { isolationLevel: 'RepeatableRead' },
+    );
+  }
+  events(actor: Principal, page: PageInput) {
+    return this.db.commerceEvent.findMany({
+      where: { organizationId: actor.organizationId },
+      take: page.limit,
+      skip: page.offset,
+      orderBy: [{ createdAt: 'desc' }, { id: 'desc' }],
+      select: {
+        id: true,
+        orderId: true,
+        kind: true,
+        createdAt: true,
+        delivery: {
+          select: {
+            attempts: true,
+            availableAt: true,
+            deliveredAt: true,
+            lastErrorCode: true,
+          },
+        },
+      },
+    });
+  }
+  retry(actor: Principal, id: string) {
+    return this.access.mutate(actor, 'notifications.retry', async (tx) => {
+      const event = await tx.commerceEvent.findFirst({
+        where: { id, organizationId: actor.organizationId },
+      });
+      if (!event) throw new AppError('EVENT_NOT_FOUND');
+      const updated = await tx.eventDelivery.updateMany({
+        where: {
+          eventId: id,
+          deliveredAt: null,
+          OR: [
+            { leaseExpiresAt: null },
+            { leaseExpiresAt: { lte: new Date() } },
+          ],
+        },
+        data: {
+          attempts: 0,
+          availableAt: new Date(),
+          leaseToken: null,
+          leaseExpiresAt: null,
+          lastErrorCode: null,
+        },
+      });
+      if (!updated.count) throw new AppError('EVENT_NOT_RETRYABLE');
+      await recordAudit(
+        tx,
+        actor.organizationId,
+        actor.userId,
+        'notification.retry.requested',
+        id,
+      );
+      return { queued: true };
+    });
+  }
+}

test/delivery-service.spec.ts

@@ -0,0 +1,73 @@
+import { Logger } from '@nestjs/common';
+import { DeliveryService } from '../src/events/delivery.service';
+import type { DeliveryStore } from '../src/events/delivery.store';
+describe('delivery lease and logging policy', () => {
+  afterEach(() => jest.restoreAllMocks());
+  it('warns when delivery succeeds after lease ownership is lost', async () => {
+    const warning = jest
+      .spyOn(Logger.prototype, 'warn')
+      .mockImplementation(() => undefined);
+    const store = {
+      claim: jest.fn().mockResolvedValue({
+        eventId: 'event',
+        leaseToken: 'lease',
+        attempts: 1,
+        event: {
+          id: 'event',
+          organizationId: 'org',
+          orderId: 'order',
+          kind: 'order.created',
+        },
+      }),
+      acknowledge: jest.fn().mockResolvedValue({ count: 0 }),
+      fail: jest.fn(),
+    };
+    const service = new DeliveryService(
+      { assertConfigured() {}, async deliver() {} },
+      store as unknown as DeliveryStore,
+    );
+    expect(await service.dispatchOne()).toBe(true);
+    expect(warning).toHaveBeenCalledWith(
+      JSON.stringify({ event: 'DELIVERY_LEASE_LOST', eventId: 'event' }),
+    );
+    expect(store.fail).not.toHaveBeenCalled();
+  });
+  it('never logs adapter exception text or notification content', async () => {
+    const logging = jest
+      .spyOn(Logger.prototype, 'error')
+      .mockImplementation(() => undefined);
+    const store = {
+      claim: jest.fn().mockResolvedValue({
+        eventId: 'event',
+        leaseToken: 'lease',
+        attempts: 2,
+        event: {
+          id: 'event',
+          organizationId: 'org',
+          orderId: 'order',
+          kind: 'order.created',
+        },
+      }),
+      acknowledge: jest.fn(),
+      fail: jest.fn().mockResolvedValue({ count: 1 }),
+    };
+    const service = new DeliveryService(
+      {
+        assertConfigured() {},
+        async deliver() {
+          throw new Error('private-address-and-secret');
+        },
+      },
+      store as unknown as DeliveryStore,
+    );
+    await service.dispatchOne();
+    expect(logging).toHaveBeenCalledWith(
+      JSON.stringify({
+        event: 'DELIVERY_FAILED',
+        eventId: 'event',
+        attempt: 2,
+      }),
+    );
+    expect(JSON.stringify(logging.mock.calls)).not.toContain('private-address');
+  });
+});

test/events-concurrency.spec.ts

@@ -0,0 +1,28 @@
+import { identityApp, type IdentityApp } from './helpers/identity-app';
+import { checkoutFixture } from './helpers/checkout';
+import { DeliveryStore } from '../src/events/delivery.store';
+const native = process.env.TEST_DATABASE_URL ? describe : describe.skip;
+native('native PostgreSQL event claims', () => {
+  let ctx: IdentityApp;
+  beforeAll(async () => {
+    ctx = await identityApp();
+  }, 60000);
+  afterAll(async () => {
+    await ctx.close();
+  });
+  it('assigns distinct events to simultaneous workers', async () => {
+    for (let index = 0; index < 2; index++) {
+      const f = await checkoutFixture(ctx);
+      await ctx
+        .api()
+        .post('/api/v1/checkout')
+        .auth(f.actor.token, { type: 'bearer' })
+        .send(f.input)
+        .expect(201);
+    }
+    const store = ctx.app.get(DeliveryStore);
+    const claims = await Promise.all([store.claim(), store.claim()]);
+    expect(claims.every(Boolean)).toBe(true);
+    expect(new Set(claims.map((claim) => claim!.eventId)).size).toBe(2);
+  });
+});

test/events.spec.ts

@@ -0,0 +1,200 @@
+import { identityApp, type IdentityApp } from './helpers/identity-app';
+import { checkoutFixture } from './helpers/checkout';
+import { secondActor } from './helpers/commerce';
+import { DeliveryService } from '../src/events/delivery.service';
+import { DeliveryStore } from '../src/events/delivery.store';
+import { DisabledDelivery } from '../src/events/disabled-delivery';
+import * as audit from '../src/identity/audit';
+
+describe('transactional commerce event delivery', () => {
+  let ctx: IdentityApp;
+  beforeAll(async () => {
+    ctx = await identityApp();
+  }, 60000);
+  afterAll(async () => {
+    await ctx.close();
+  });
+  beforeEach(async () => {
+    await ctx.clearLimits();
+    await ctx.db.eventDelivery.updateMany({
+      data: { deliveredAt: new Date(), leaseToken: null, leaseExpiresAt: null },
+    });
+  });
+  afterEach(() => jest.restoreAllMocks());
+  async function event() {
+    const f = await checkoutFixture(ctx);
+    const order = await ctx
+      .api()
+      .post('/api/v1/checkout')
+      .auth(f.actor.token, { type: 'bearer' })
+      .send(f.input)
+      .expect(201);
+    return {
+      ...f,
+      order: order.body,
+      event: await ctx.db.commerceEvent.findFirstOrThrow({
+        where: { orderId: order.body.id },
+      }),
+    };
+  }
+  it('emits one event per committed transition and none after rollback', async () => {
+    const f = await event();
+    await ctx
+      .api()
+      .post('/api/v1/checkout')
+      .auth(f.actor.token, { type: 'bearer' })
+      .send(f.input)
+      .expect(201);
+    await ctx
+      .api()
+      .post('/api/v1/orders/' + f.order.id + '/cancel')
+      .auth(f.actor.token, { type: 'bearer' })
+      .expect(201);
+    await ctx
+      .api()
+      .post('/api/v1/orders/' + f.order.id + '/cancel')
+      .auth(f.actor.token, { type: 'bearer' })
+      .expect(201);
+    expect(
+      await ctx.db.commerceEvent.count({ where: { orderId: f.order.id } }),
+    ).toBe(2);
+    const next = await checkoutFixture(ctx);
+    jest
+      .spyOn(audit, 'recordAudit')
+      .mockRejectedValueOnce(new Error('Synthetic final write failure'));
+    await ctx
+      .api()
+      .post('/api/v1/checkout')
+      .auth(next.actor.token, { type: 'bearer' })
+      .send(next.input)
+      .expect(500);
+    expect(
+      await ctx.db.commerceEvent.count({
+        where: { order: { userId: next.actor.userId } },
+      }),
+    ).toBe(0);
+    await expect(
+      ctx.executeSql(`DELETE FROM commerce_events WHERE id = '${f.event.id}'`),
+    ).rejects.toThrow();
+  });
+  it('keeps delivery disabled by default and uses a test adapter with stable event IDs', async () => {
+    const f = await event();
+    await expect(
+      ctx.app.get(DeliveryService).dispatchOne(),
+    ).rejects.toMatchObject({ code: 'DELIVERY_UNAVAILABLE' });
+    await expect(new DisabledDelivery().deliver()).rejects.toThrow();
+    const adapter = {
+      assertConfigured: jest.fn(),
+      deliver: jest.fn().mockResolvedValue(undefined),
+    };
+    const service = new DeliveryService(adapter, ctx.app.get(DeliveryStore));
+    expect(await service.dispatchOne()).toBe(true);
+    expect(adapter.deliver).toHaveBeenCalledWith({
+      id: f.event.id,
+      orderId: f.order.id,
+      organizationId: f.actor.organizationId,
+      kind: 'order.created',
+    });
+    expect(await service.dispatchOne()).toBe(false);
+    expect(
+      (
+        await ctx.db.eventDelivery.findUniqueOrThrow({
+          where: { eventId: f.event.id },
+        })
+      ).deliveredAt,
+    ).not.toBeNull();
+  });
+  it('backs off failures, bounds attempts and permits authorized manual retries', async () => {
+    const f = await event();
+    const store = ctx.app.get(DeliveryStore);
+    const service = new DeliveryService(
+      {
+        assertConfigured() {},
+        deliver: jest
+          .fn()
+          .mockRejectedValue(new Error('Private transport error')),
+      },
+      store,
+    );
+    await service.dispatchOne();
+    const failed = await ctx.db.eventDelivery.findUniqueOrThrow({
+      where: { eventId: f.event.id },
+    });
+    expect(failed).toMatchObject({
+      attempts: 1,
+      lastErrorCode: 'DELIVERY_FAILED',
+      leaseToken: null,
+    });
+    expect(failed.availableAt.getTime()).toBeGreaterThan(Date.now());
+    expect(await store.claim()).toBeNull();
+    await ctx.db.eventDelivery.update({
+      where: { eventId: f.event.id },
+      data: { attempts: 5, availableAt: new Date(0) },
+    });
+    expect(await store.claim()).toBeNull();
+    const summary = await ctx
+      .api()
+      .get('/api/v1/admin/operations/summary')
+      .auth(ctx.token, { type: 'bearer' })
+      .expect(200);
+    expect(summary.body.failedDeliveries).toBe(1);
+    await ctx
+      .api()
+      .post('/api/v1/admin/operations/events/' + f.event.id + '/retry')
+      .auth(f.actor.token, { type: 'bearer' })
+      .expect(403);
+    await ctx
+      .api()
+      .post('/api/v1/admin/operations/events/' + f.event.id + '/retry')
+      .auth(ctx.token, { type: 'bearer' })
+      .expect(201);
+    expect((await store.claim())?.attempts).toBe(1);
+  });
+  it('rejects stale lease acknowledgements and protects cross-organization event access', async () => {
+    const f = await event();
+    const store = ctx.app.get(DeliveryStore);
+    const first = (await store.claim())!;
+    await ctx
+      .api()
+      .post('/api/v1/admin/operations/events/' + f.event.id + '/retry')
+      .auth(ctx.token, { type: 'bearer' })
+      .expect(409);
+    await ctx.db.eventDelivery.update({
+      where: { eventId: f.event.id },
+      data: { leaseExpiresAt: new Date(0) },
+    });
+    const next = (await store.claim())!;
+    expect(next.leaseToken).not.toBe(first.leaseToken);
+    expect((await store.acknowledge(f.event.id, first.leaseToken!)).count).toBe(
+      0,
+    );
+    expect((await store.fail(f.event.id, first.leaseToken!, 1)).count).toBe(0);
+    await store.acknowledge(f.event.id, next.leaseToken!);
+    await ctx
+      .api()
+      .post('/api/v1/admin/operations/events/' + f.event.id + '/retry')
+      .auth(ctx.token, { type: 'bearer' })
+      .expect(409);
+    const other = await secondActor(ctx, false, [
+      'operations.read',
+      'notifications.retry',
+    ]);
+    const list = await ctx
+      .api()
+      .get('/api/v1/admin/operations/events')
+      .auth(other.token, { type: 'bearer' })
+      .expect(200);
+    expect(list.body).toEqual([]);
+    await ctx
+      .api()
+      .post('/api/v1/admin/operations/events/' + f.event.id + '/retry')
+      .auth(other.token, { type: 'bearer' })
+      .expect(404);
+    const own = await ctx
+      .api()
+      .get('/api/v1/admin/operations/events?limit=1')
+      .auth(ctx.token, { type: 'bearer' })
+      .expect(200);
+    expect(own.body[0].delivery.leaseToken).toBeUndefined();
+  });
+});