diff --git a/packages/iios-contracts/src/events.ts b/packages/iios-contracts/src/events.ts index 4c78d27..713d509 100644 --- a/packages/iios-contracts/src/events.ts +++ b/packages/iios-contracts/src/events.ts @@ -52,6 +52,7 @@ export const IIOS_EVENTS = { meetingSummarized: 'com.insignia.iios.calendar.meeting.summarized.v1', notificationQueued: 'com.insignia.iios.notification.queued.v1', deliveryAttempted: 'com.insignia.iios.delivery.attempted.v1', + devChaos: 'com.insignia.iios.dev.chaos.v1', } as const; export type IiosEventType = (typeof IIOS_EVENTS)[keyof typeof IIOS_EVENTS]; diff --git a/packages/iios-service/scripts/smoke-dlq.mjs b/packages/iios-service/scripts/smoke-dlq.mjs new file mode 100644 index 0000000..b68863b --- /dev/null +++ b/packages/iios-service/scripts/smoke-dlq.mjs @@ -0,0 +1,71 @@ +// P9 slice-5 DLQ smoke: a poison chaos event is dead-lettered (not lost), shows up +// tenant-scoped in /v1/dlq, another tenant can neither see nor replay it, and the +// owner's replay resolves it (KG-13 / KG-06). Requires the service running with +// IIOS_DEV_TOKENS=1 (the relay ticks by default, publishing the poison to the consumer). +import 'dotenv/config'; + +const SERVICE = process.env.SMOKE_URL ?? 'http://localhost:3200'; +const APP_ID = 'portal-demo'; +const assert = (c, m) => { if (!c) { console.error('✗', m); process.exit(1); } console.log('✓', m); }; +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); + +async function devToken(userId, orgId) { + const r = await fetch(`${SERVICE}/v1/dev/token`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ appId: APP_ID, userId, name: userId, orgId }), + }); + if (!r.ok) throw new Error(`devToken ${r.status} (service with IIOS_DEV_TOKENS=1?)`); + return (await r.json()).token; +} +async function call(token, path, method = 'GET', body) { + return fetch(`${SERVICE}${path}`, { + method, + headers: { 'content-type': 'application/json', authorization: `Bearer ${token}` }, + body: body ? JSON.stringify(body) : undefined, + }); +} +async function json(token, path, method = 'GET', body) { + const r = await call(token, path, method, body); + if (!r.ok) throw new Error(`${method} ${path} ${r.status}: ${await r.text()}`); + return r.json(); +} + +const A = await devToken('a1', 'org_A'); +const B = await devToken('b1', 'org_B'); + +// A injects a poison chaos event. +const { eventId } = await json(A, '/v1/dev/chaos/poison', 'POST'); +assert(eventId.startsWith('evt_chaos_'), `A injected a poison chaos event (${eventId})`); + +// Poll until the relay publishes it and the chaos consumer dead-letters it. +let item; +for (let i = 0; i < 40; i++) { + const list = await json(A, '/v1/dlq'); + item = list.find((d) => d.sourceId === eventId); + if (item) break; + await sleep(250); +} +assert(item, 'poison surfaced in A\'s DLQ (relay published → consumer failed → dead-lettered)'); +assert(item.status === 'QUARANTINED', `DLQ item is QUARANTINED (got ${item.status})`); +assert(item.consumerName === 'chaos-consumer', `dead-lettered by the chaos-consumer lane (got ${item.consumerName})`); + +// Tenant B cannot see A's DLQ item. +const bList = await json(B, '/v1/dlq'); +assert(!bList.some((d) => d.sourceId === eventId), "tenant B's DLQ excludes A's item (isolation)"); + +// Tenant B cannot replay A's DLQ item. +const bReplay = await call(B, `/v1/dlq/${item.id}/replay`, 'POST'); +assert(bReplay.status === 403, `tenant B replaying A's item → 403 (got ${bReplay.status}) — KG-02`); + +// A replays → resolves. +const replayed = await json(A, `/v1/dlq/${item.id}/replay`, 'POST'); +assert(replayed.status === 'RESOLVED', `A's replay resolves the poison (got ${replayed.status})`); + +// It now shows RESOLVED in A's DLQ. +const afterList = await json(A, '/v1/dlq'); +const after = afterList.find((d) => d.sourceId === eventId); +assert(after && after.status === 'RESOLVED', 'DLQ item is RESOLVED after replay'); + +console.log('\nP9 DLQ smoke: PASS'); +process.exit(0); diff --git a/packages/iios-service/src/dev/dev.controller.ts b/packages/iios-service/src/dev/dev.controller.ts index 764a310..a1ebd25 100644 --- a/packages/iios-service/src/dev/dev.controller.ts +++ b/packages/iios-service/src/dev/dev.controller.ts @@ -1,9 +1,13 @@ import { randomUUID } from 'node:crypto'; -import { BadRequestException, Body, Controller, ForbiddenException, Param, Post } from '@nestjs/common'; +import { BadRequestException, Body, Controller, ForbiddenException, Headers, Param, Post } from '@nestjs/common'; import jwt from 'jsonwebtoken'; +import { IIOS_EVENTS } from '@insignia/iios-contracts'; import { hmacSign } from '@insignia/iios-adapter-sdk'; import { InboundService } from '../adapters/inbound.service'; import { AdapterRegistry } from '../adapters/adapter.registry'; +import { PrismaService } from '../prisma/prisma.service'; +import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver'; +import { SessionVerifier } from '../platform/session.verifier'; /** * Dev-only helpers (gated by IIOS_DEV_TOKENS=1). Never enable in production. @@ -13,6 +17,9 @@ export class DevController { constructor( private readonly inbound: InboundService, private readonly registry: AdapterRegistry, + private readonly prisma: PrismaService, + private readonly actors: ActorResolver, + private readonly session: SessionVerifier, ) {} @Post('token') @@ -55,6 +62,46 @@ export class DevController { }); } + /** + * Inject a poison chaos event into the outbox. The relay publishes it, the + * ChaosConsumer throws on first delivery → it is dead-lettered under the caller's + * scope, then `POST /v1/dlq/:id/replay` resolves it. Drives the DLQ smoke. + */ + @Post('chaos/poison') + async poison(@Headers('authorization') auth?: string): Promise<{ eventId: string; scopeId: string }> { + this.assertDev(); + const principal = this.principal(auth); + const scope = await this.actors.resolveScope(principal); // find-or-create so GET /v1/dlq matches + const id = `evt_chaos_${randomUUID()}`; + const cloudEvent = { + specversion: '1.0', + id, + type: IIOS_EVENTS.devChaos, + source: `iios/dev/${principal.appId}`, + subject: `chaos/${id}`, + time: new Date().toISOString(), + datacontenttype: 'application/json', + insignia: { scopeSnapshotId: scope.id, idempotencyKey: id, dataClass: 'internal' }, + data: { poison: true }, + }; + await this.prisma.iiosOutboxEvent.create({ + data: { + aggregateType: 'dev-chaos', + aggregateId: id, + eventType: IIOS_EVENTS.devChaos, + cloudEvent: cloudEvent as unknown as object, + partitionKey: `${principal.orgId}:${principal.appId}:chaos`, + }, + }); + return { eventId: id, scopeId: scope.id }; + } + + private principal(authorization?: string): MessagePrincipal { + const token = (authorization ?? '').replace(/^Bearer\s+/i, ''); + if (!token) throw new BadRequestException('Authorization bearer token is required'); + return this.session.verify(token); + } + private assertDev(): void { if (process.env.IIOS_DEV_TOKENS !== '1') throw new ForbiddenException('dev endpoints disabled'); } diff --git a/packages/iios-service/src/outbox/chaos.consumer.ts b/packages/iios-service/src/outbox/chaos.consumer.ts new file mode 100644 index 0000000..fb634e7 --- /dev/null +++ b/packages/iios-service/src/outbox/chaos.consumer.ts @@ -0,0 +1,45 @@ +import { Injectable, OnModuleInit } from '@nestjs/common'; +import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts'; +import { PrismaService } from '../prisma/prisma.service'; +import { OutboxBus } from './outbox.bus'; +import { DlqService } from './dlq.service'; + +/** + * Dev-only chaos consumer (P9). Drives the DLQ smoke deterministically: it THROWS on + * the first delivery of a chaos event (so the event is dead-lettered) and SUCCEEDS on + * replay (so the DLQ item resolves). Registered like any projector — same claim() + + * onConsumerFailure wiring — so it exercises the real dead-letter/replay path. + */ +@Injectable() +export class ChaosConsumer implements OnModuleInit { + private readonly consumer = 'chaos-consumer'; + private readonly seen = new Set(); + + constructor( + private readonly prisma: PrismaService, + private readonly bus: OutboxBus, + private readonly dlq: DlqService, + ) {} + + onModuleInit(): void { + this.dlq.registerHandler(this.consumer, (e) => this.onChaos(e)); + this.bus.on(IIOS_EVENTS.devChaos, (p) => void this.onChaos(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err))); + } + + async onChaos(event: CloudEvent): Promise { + if (!(await this.claim(event.id))) return; // idempotent + if (!this.seen.has(event.id)) { + this.seen.add(event.id); + throw new Error('chaos: first delivery always fails'); // → dead-lettered, claim cleared + } + // replay: succeeds (the claim persists, marking it processed exactly-once) + } + + private async claim(eventId: string): Promise { + const res = await this.prisma.iiosProcessedEvent.createMany({ + data: [{ consumerName: this.consumer, eventId }], + skipDuplicates: true, + }); + return res.count > 0; + } +} diff --git a/packages/iios-service/src/outbox/dlq.controller.ts b/packages/iios-service/src/outbox/dlq.controller.ts new file mode 100644 index 0000000..de48ae2 --- /dev/null +++ b/packages/iios-service/src/outbox/dlq.controller.ts @@ -0,0 +1,46 @@ +import { BadRequestException, Controller, ForbiddenException, Get, Headers, NotFoundException, Param, Post, Query } from '@nestjs/common'; +import { PrismaService } from '../prisma/prisma.service'; +import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver'; +import { SessionVerifier } from '../platform/session.verifier'; +import { recordAudit } from '../observability/audit'; +import { DlqService } from './dlq.service'; + +/** + * DLQ ops surface (P9, KG-13). Strictly tenant-scoped: a caller only ever sees and + * replays dead-letters owned by their own scope (assertOwns fences replay by id). + */ +@Controller('v1/dlq') +export class DlqController { + constructor( + private readonly prisma: PrismaService, + private readonly dlq: DlqService, + private readonly actors: ActorResolver, + private readonly session: SessionVerifier, + ) {} + + @Get() + async list(@Query('status') status?: string, @Headers('authorization') auth?: string) { + const principal = this.principal(auth); + const scope = await this.actors.findScope(principal); + if (!scope) return []; // a caller with no scope owns nothing + return this.dlq.listDlq(scope.id, status); + } + + @Post(':id/replay') + async replay(@Param('id') id: string, @Headers('authorization') auth?: string) { + const principal = this.principal(auth); + const item = await this.prisma.iiosDlqItem.findUnique({ where: { id } }); + if (!item) throw new NotFoundException('dlq item not found'); + if (!item.scopeId) throw new ForbiddenException('dlq item is not tenant-scoped'); + const scope = await this.actors.assertOwns(principal, item.scopeId); // KG-02 fence + const result = await this.dlq.replay(id); + await recordAudit(this.prisma, { action: 'dlq.replay', resourceType: 'dlq_item', resourceId: id, scopeId: scope.id }); + return result; + } + + private principal(authorization?: string): MessagePrincipal { + const token = (authorization ?? '').replace(/^Bearer\s+/i, ''); + if (!token) throw new BadRequestException('Authorization bearer token is required'); + return this.session.verify(token); + } +} diff --git a/packages/iios-service/src/outbox/outbox.module.ts b/packages/iios-service/src/outbox/outbox.module.ts index e68547e..5158a4f 100644 --- a/packages/iios-service/src/outbox/outbox.module.ts +++ b/packages/iios-service/src/outbox/outbox.module.ts @@ -2,9 +2,12 @@ import { Module } from '@nestjs/common'; import { OutboxBus } from './outbox.bus'; import { OutboxRelay } from './outbox.relay'; import { DlqService } from './dlq.service'; +import { DlqController } from './dlq.controller'; +import { ChaosConsumer } from './chaos.consumer'; @Module({ - providers: [OutboxBus, OutboxRelay, DlqService], + controllers: [DlqController], + providers: [OutboxBus, OutboxRelay, DlqService, ChaosConsumer], exports: [OutboxBus, OutboxRelay, DlqService], }) export class OutboxModule {}