feat(iios): wire projectors + relay into DLQ (P9 chaos-safe)

Projectors register a replay handler and route bus errors to dlq.onConsumerFailure (dead-letter + clear the claim-then-fail claim) instead of swallowing them. Relay dead-letters an outbox event after IIOS_OUTBOX_MAX_ATTEMPTS instead of retrying forever. Chaos test proves poison → DLQ → replay resolves while sibling events keep flowing (KG-13/KG-06). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-03 01:43:57 +05:30
parent 208fb0b4f0
commit d3d557766e
14 changed files with 128 additions and 18 deletions
@@ -0,0 +1,76 @@
+import { randomUUID } from 'node:crypto';
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
+import { PrismaClient } from '@prisma/client';
+import { resetDb } from '../test-utils/reset-db';
+import { makeFakePorts, makeFakeInference } from '@insignia/iios-testkit';
+import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts';
+import { ActorResolver } from '../identity/actor.resolver';
+import { IngestService } from '../interactions/ingest.service';
+import { AiJobService } from '../ai/ai.service';
+import { AiBudgetGuard } from '../ai/ai-budget.guard';
+import { AiProjector } from '../ai/ai.projector';
+import { OutboxBus } from './outbox.bus';
+import { DlqService } from './dlq.service';
+import type { PrismaService } from '../prisma/prisma.service';
+
+const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public';
+const prisma = new PrismaClient({ datasources: { db: { url } } });
+const asService = prisma as unknown as PrismaService;
+const actors = new ActorResolver(asService);
+const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors);
+
+async function ingestInbound(text: string): Promise<{ interactionId: string; scopeId: string; event: CloudEvent }> {
+  const ingest = new IngestService(asService, makeFakePorts(), actors);
+  const res = await ingest.ingest(
+    {
+      scope: { orgId: 'org_demo', appId: 'portal-demo' },
+      channel: { type: 'WEBHOOK', externalChannelId: 'webhook' },
+      source: { handleKind: 'EXTERNAL_VISITOR', externalId: 'ext-1' },
+      kind: 'MESSAGE',
+      thread: { externalThreadId: `wh-${randomUUID().slice(0, 6)}` },
+      parts: [{ kind: 'TEXT', bodyText: text }],
+      occurredAt: new Date().toISOString(),
+      providerEventId: `pe-${randomUUID()}`,
+    },
+    `ing-${randomUUID()}`,
+  );
+  const interaction = await prisma.iiosInteraction.findUniqueOrThrow({ where: { id: res.interactionId } });
+  const ev = (await prisma.iiosOutboxEvent.findFirstOrThrow({
+    where: { eventType: IIOS_EVENTS.interactionNormalized, aggregateId: res.interactionId },
+  })).cloudEvent as unknown as CloudEvent;
+  return { interactionId: res.interactionId, scopeId: interaction.scopeId, event: ev };
+}
+
+beforeAll(async () => { await prisma.$connect(); });
+afterAll(async () => { await prisma.$disconnect(); });
+beforeEach(async () => { await resetDb(prisma); });
+
+describe('chaos: poison consumer → DLQ → replay (P9)', () => {
+  it('a consumer failure is dead-lettered (not swallowed), the lane keeps flowing, and replay resolves it', async () => {
+    const dlq = new DlqService(asService);
+    const projector = new AiProjector(asService, new OutboxBus(), dlq, ai());
+    dlq.registerHandler('ai-projector', (e) => projector.onNormalized(e)); // healthy handler for replay
+
+    const poison = await ingestInbound('There is a party at 9 PM tonight!');
+    // Simulate the consumer crashing on the poison (exactly what the projector's .catch does).
+    await prisma.iiosProcessedEvent.create({ data: { consumerName: 'ai-projector', eventId: poison.event.id } }); // claimed-then-failed
+    await dlq.onConsumerFailure('ai-projector', poison.event, new Error('chaos: AI down'));
+
+    // Dead-lettered (tenant-tagged), claim cleared, and NOT processed.
+    const item = await prisma.iiosDlqItem.findFirstOrThrow({ where: { consumerName: 'ai-projector' } });
+    expect(item.scopeId).toBe(poison.scopeId);
+    expect(item.status).toBe('QUARANTINED');
+    expect(await prisma.iiosProcessedEvent.count({ where: { consumerName: 'ai-projector', eventId: poison.event.id } })).toBe(0);
+    expect(await prisma.iiosAiJob.count({ where: { interactionId: poison.interactionId } })).toBe(0);
+
+    // Lane not starved: a SIBLING event still processes through the healthy consumer.
+    const sibling = await ingestInbound('Team standup notes are posted.');
+    await projector.onNormalized(sibling.event);
+    expect(await prisma.iiosAiJob.count({ where: { interactionId: sibling.interactionId } })).toBe(1);
+
+    // Replay the poison → the registered handler re-runs it → RESOLVED + now processed.
+    const r = await dlq.replay(item.id);
+    expect(r.status).toBe('RESOLVED');
+    expect(await prisma.iiosAiJob.count({ where: { interactionId: poison.interactionId } })).toBe(1);
+  });
+});