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 { ProjectionCursorService } from '../projection/projection-cursor.service'; 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(), new ProjectionCursorService(asService)); 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); }); });