diff --git a/src/checkout/checkout.store.ts b/src/checkout/checkout.store.ts index 4890caf..00d1cf7 100644 --- a/src/checkout/checkout.store.ts +++ b/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 }, diff --git a/src/checkout/order.store.ts b/src/checkout/order.store.ts index 48e7cec..6e57c98 100644 --- a/src/checkout/order.store.ts +++ b/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, diff --git a/src/events/delivery.port.ts b/src/events/delivery.port.ts new file mode 100644 index 0000000..ef3b829 --- /dev/null +++ b/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; +} diff --git a/src/events/delivery.service.ts b/src/events/delivery.service.ts new file mode 100644 index 0000000..f8757e4 --- /dev/null +++ b/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 { + 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; + } +} diff --git a/src/events/delivery.store.ts b/src/events/delivery.store.ts new file mode 100644 index 0000000..0acd8d6 --- /dev/null +++ b/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, + ), + }, + }); + } +} diff --git a/src/events/disabled-delivery.ts b/src/events/disabled-delivery.ts new file mode 100644 index 0000000..db80310 --- /dev/null +++ b/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 { + this.assertConfigured(); + } +} diff --git a/src/events/enqueue-event.ts b/src/events/enqueue-event.ts new file mode 100644 index 0000000..d999146 --- /dev/null +++ b/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: {}, + }); +} diff --git a/src/operations/operations.controller.ts b/src/operations/operations.controller.ts new file mode 100644 index 0000000..d242e65 --- /dev/null +++ b/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); + } +} diff --git a/src/operations/operations.module.ts b/src/operations/operations.module.ts index 9638100..6c1031b 100644 --- a/src/operations/operations.module.ts +++ b/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 {} diff --git a/src/operations/operations.store.ts b/src/operations/operations.store.ts new file mode 100644 index 0000000..3abc7a3 --- /dev/null +++ b/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 }; + }); + } +} diff --git a/test/delivery-service.spec.ts b/test/delivery-service.spec.ts new file mode 100644 index 0000000..dc567aa --- /dev/null +++ b/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'); + }); +}); diff --git a/test/events-concurrency.spec.ts b/test/events-concurrency.spec.ts new file mode 100644 index 0000000..79a2a64 --- /dev/null +++ b/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); + }); +}); diff --git a/test/events.spec.ts b/test/events.spec.ts new file mode 100644 index 0000000..ee2decf --- /dev/null +++ b/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(); + }); +});