feat(iios): tenant-scoped /v1/dlq REST + chaos endpoint + smoke (P9 slice 5)
GET /v1/dlq lists the caller tenant's dead-letters; POST /v1/dlq/:id/replay fences by assertOwns (KG-02) + audits each replay. Dev-only POST /v1/dev/chaos/poison + ChaosConsumer (fails first delivery, succeeds on replay) drive smoke-dlq.mjs: poison → QUARANTINED → cross-tenant denied → owner replay → RESOLVED. Adds dev.chaos.v1 contract event. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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');
|
||||
}
|
||||
|
||||
@@ -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<string>();
|
||||
|
||||
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<void> {
|
||||
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<boolean> {
|
||||
const res = await this.prisma.iiosProcessedEvent.createMany({
|
||||
data: [{ consumerName: this.consumer, eventId }],
|
||||
skipDuplicates: true,
|
||||
});
|
||||
return res.count > 0;
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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 {}
|
||||
|
||||
Reference in New Issue
Block a user