From 072c541e09aeb6a22d60bdb6b25852ba7d3fae3a Mon Sep 17 00:00:00 2001 From: maaz519 Date: Tue, 30 Jun 2026 21:29:00 +0530 Subject: [PATCH] feat(service): outbox relay + thread read endpoint + replay (P1.5, P1 complete) OutboxRelay: FOR UPDATE SKIP LOCKED claim, in-process bus publish, PENDING-> PUBLISHED with 30s backoff + idempotent-consumer ledger. GET /v1/threads/:id/ messages (policy-scoped). Replay + read specs green; vitest fileParallelism off (shared DB). End-to-end HTTP smoke verified: idempotent ingest, read-back, 400 on missing key. 19 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/iios-service/src/app.module.ts | 4 +- .../iios-service/src/outbox/outbox.bus.ts | 25 +++++++ .../iios-service/src/outbox/outbox.module.ts | 9 +++ .../iios-service/src/outbox/outbox.relay.ts | 69 +++++++++++++++++++ .../iios-service/src/outbox/replay.spec.ts | 60 ++++++++++++++++ .../src/threads/threads.controller.ts | 12 ++++ .../src/threads/threads.module.ts | 10 +++ .../src/threads/threads.service.ts | 46 +++++++++++++ .../iios-service/src/threads/threads.spec.ts | 42 +++++++++++ vitest.config.ts | 3 + 10 files changed, 279 insertions(+), 1 deletion(-) create mode 100644 packages/iios-service/src/outbox/outbox.bus.ts create mode 100644 packages/iios-service/src/outbox/outbox.module.ts create mode 100644 packages/iios-service/src/outbox/outbox.relay.ts create mode 100644 packages/iios-service/src/outbox/replay.spec.ts create mode 100644 packages/iios-service/src/threads/threads.controller.ts create mode 100644 packages/iios-service/src/threads/threads.module.ts create mode 100644 packages/iios-service/src/threads/threads.service.ts create mode 100644 packages/iios-service/src/threads/threads.spec.ts diff --git a/packages/iios-service/src/app.module.ts b/packages/iios-service/src/app.module.ts index d9ff377..5e4af9b 100644 --- a/packages/iios-service/src/app.module.ts +++ b/packages/iios-service/src/app.module.ts @@ -2,10 +2,12 @@ import { Module } from '@nestjs/common'; import { PrismaModule } from './prisma/prisma.module'; import { PlatformModule } from './platform/platform.module'; import { InteractionsModule } from './interactions/interactions.module'; +import { OutboxModule } from './outbox/outbox.module'; +import { ThreadsModule } from './threads/threads.module'; import { HealthController } from './health.controller'; @Module({ - imports: [PrismaModule, PlatformModule, InteractionsModule], + imports: [PrismaModule, PlatformModule, InteractionsModule, OutboxModule, ThreadsModule], controllers: [HealthController], }) export class AppModule {} diff --git a/packages/iios-service/src/outbox/outbox.bus.ts b/packages/iios-service/src/outbox/outbox.bus.ts new file mode 100644 index 0000000..172277f --- /dev/null +++ b/packages/iios-service/src/outbox/outbox.bus.ts @@ -0,0 +1,25 @@ +import { Injectable } from '@nestjs/common'; +import { EventEmitter } from 'node:events'; + +/** + * In-process event bus (P1 delivery surface). A real broker (Kafka/Redpanda) + * replaces this in P9; the relay contract stays the same. Delivery surfaces are + * never the source of truth — the outbox table is. + */ +@Injectable() +export class OutboxBus { + private readonly emitter = new EventEmitter(); + + publish(eventType: string, payload: unknown): void { + this.emitter.emit(eventType, payload); + this.emitter.emit('*', eventType, payload); + } + + on(eventType: string, handler: (payload: unknown) => void): void { + this.emitter.on(eventType, handler); + } + + onAny(handler: (eventType: string, payload: unknown) => void): void { + this.emitter.on('*', handler); + } +} diff --git a/packages/iios-service/src/outbox/outbox.module.ts b/packages/iios-service/src/outbox/outbox.module.ts new file mode 100644 index 0000000..abdcf2b --- /dev/null +++ b/packages/iios-service/src/outbox/outbox.module.ts @@ -0,0 +1,9 @@ +import { Module } from '@nestjs/common'; +import { OutboxBus } from './outbox.bus'; +import { OutboxRelay } from './outbox.relay'; + +@Module({ + providers: [OutboxBus, OutboxRelay], + exports: [OutboxBus, OutboxRelay], +}) +export class OutboxModule {} diff --git a/packages/iios-service/src/outbox/outbox.relay.ts b/packages/iios-service/src/outbox/outbox.relay.ts new file mode 100644 index 0000000..38f72c1 --- /dev/null +++ b/packages/iios-service/src/outbox/outbox.relay.ts @@ -0,0 +1,69 @@ +import { Injectable } from '@nestjs/common'; +import { PrismaService } from '../prisma/prisma.service'; +import { OutboxBus } from './outbox.bus'; + +/** + * Transactional-outbox relay (Bottom-Up §12). At-least-once by design; the + * idempotent-consumer ledger (IiosProcessedEvent) makes effects exactly-once. + * + * relayOnce(): claim PENDING rows with FOR UPDATE SKIP LOCKED, publish each to + * the bus once, mark PUBLISHED. On failure, reset to PENDING with a 30s backoff. + * A scheduler (P9) calls this on an interval; tests call it directly. + */ +@Injectable() +export class OutboxRelay { + private readonly consumer = 'outbox-relay'; + + constructor( + private readonly prisma: PrismaService, + private readonly bus: OutboxBus, + ) {} + + async relayOnce(batchSize = 100): Promise { + const eventIds = await this.prisma.$transaction(async (tx) => { + const picked = await tx.$queryRaw>` + SELECT "eventId" FROM "IiosOutboxEvent" + WHERE status = 'PENDING' AND "nextAttemptAt" <= now() + ORDER BY "createdAt" ASC + LIMIT ${batchSize} + FOR UPDATE SKIP LOCKED`; + const ids = picked.map((r) => r.eventId); + if (ids.length > 0) { + await tx.iiosOutboxEvent.updateMany({ + where: { eventId: { in: ids } }, + data: { status: 'SENDING', attempts: { increment: 1 } }, + }); + } + return ids; + }); + + let published = 0; + for (const eventId of eventIds) { + const evt = await this.prisma.iiosOutboxEvent.findUniqueOrThrow({ where: { eventId } }); + try { + const fresh = await this.consumeOnce(eventId); + if (fresh) this.bus.publish(evt.eventType, evt.cloudEvent); + await this.prisma.iiosOutboxEvent.update({ + where: { eventId }, + data: { status: 'PUBLISHED', publishedAt: new Date() }, + }); + published++; + } catch { + await this.prisma.iiosOutboxEvent.update({ + where: { eventId }, + data: { status: 'PENDING', nextAttemptAt: new Date(Date.now() + 30_000) }, + }); + } + } + return published; + } + + /** Records the event as processed exactly once. Returns false if already seen. */ + private async consumeOnce(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/replay.spec.ts b/packages/iios-service/src/outbox/replay.spec.ts new file mode 100644 index 0000000..9da35f0 --- /dev/null +++ b/packages/iios-service/src/outbox/replay.spec.ts @@ -0,0 +1,60 @@ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import { PrismaClient } from '@prisma/client'; +import { makeFakePorts, portalMessageBasic, replayTwice, isIdempotent } from '@insignia/iios-testkit'; +import { IngestService } from '../interactions/ingest.service'; +import { OutboxRelay } from './outbox.relay'; +import { OutboxBus } from './outbox.bus'; +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; + +async function clean(): Promise { + await prisma.iiosOutboxEvent.deleteMany(); + await prisma.iiosProcessedEvent.deleteMany(); + await prisma.iiosMessagePart.deleteMany(); + await prisma.iiosInteraction.deleteMany(); + await prisma.iiosThreadParticipant.deleteMany(); + await prisma.iiosThread.deleteMany(); + await prisma.iiosActorRef.deleteMany(); + await prisma.iiosSourceHandle.deleteMany(); + await prisma.iiosChannel.deleteMany(); + await prisma.iiosScope.deleteMany(); +} + +beforeAll(async () => { await prisma.$connect(); }); +afterAll(async () => { await prisma.$disconnect(); }); +beforeEach(async () => { await clean(); }); + +describe('outbox relay + replay', () => { + it('replaying a fixture twice produces no duplicate state', async () => { + const ingest = new IngestService(asService, makeFakePorts()); + const result = await replayTwice( + portalMessageBasic, + (f) => ingest.ingest(f, f.providerEventId!), + () => prisma.iiosInteraction.count(), + ); + expect(isIdempotent(result)).toBe(true); + expect(result.countAfterSecond).toBe(1); + }); + + it('relay publishes exactly one event per interaction, then nothing new', async () => { + const ingest = new IngestService(asService, makeFakePorts()); + const bus = new OutboxBus(); + const received: string[] = []; + bus.onAny((eventType) => received.push(eventType)); + const relay = new OutboxRelay(asService, bus); + + await ingest.ingest(portalMessageBasic, portalMessageBasic.providerEventId!); + + const published = await relay.relayOnce(); + expect(published).toBe(1); + expect(received).toEqual(['com.insignia.iios.interaction.normalized.v1']); + expect(await prisma.iiosOutboxEvent.count({ where: { status: 'PUBLISHED' } })).toBe(1); + expect(await prisma.iiosProcessedEvent.count()).toBe(1); + + // A second relay pass finds nothing PENDING. + expect(await relay.relayOnce()).toBe(0); + }); +}); diff --git a/packages/iios-service/src/threads/threads.controller.ts b/packages/iios-service/src/threads/threads.controller.ts new file mode 100644 index 0000000..8fc1f64 --- /dev/null +++ b/packages/iios-service/src/threads/threads.controller.ts @@ -0,0 +1,12 @@ +import { Controller, Get, Param } from '@nestjs/common'; +import { ThreadsService } from './threads.service'; + +@Controller('v1/threads') +export class ThreadsController { + constructor(private readonly threads: ThreadsService) {} + + @Get(':id/messages') + async messages(@Param('id') id: string) { + return this.threads.getMessages(id); + } +} diff --git a/packages/iios-service/src/threads/threads.module.ts b/packages/iios-service/src/threads/threads.module.ts new file mode 100644 index 0000000..03e14d1 --- /dev/null +++ b/packages/iios-service/src/threads/threads.module.ts @@ -0,0 +1,10 @@ +import { Module } from '@nestjs/common'; +import { ThreadsController } from './threads.controller'; +import { ThreadsService } from './threads.service'; + +@Module({ + controllers: [ThreadsController], + providers: [ThreadsService], + exports: [ThreadsService], +}) +export class ThreadsModule {} diff --git a/packages/iios-service/src/threads/threads.service.ts b/packages/iios-service/src/threads/threads.service.ts new file mode 100644 index 0000000..2f7da82 --- /dev/null +++ b/packages/iios-service/src/threads/threads.service.ts @@ -0,0 +1,46 @@ +import { Inject, Injectable, NotFoundException } from '@nestjs/common'; +import type { IiosPlatformPorts } from '@insignia/iios-contracts'; +import { PrismaService } from '../prisma/prisma.service'; +import { PLATFORM_PORTS } from '../platform/platform-ports'; +import { decideOrThrow } from '../platform/fail-closed'; + +export interface ThreadMessage { + interactionId: string; + actorId: string | null; + kind: string; + occurredAt: Date; + parts: Array<{ kind: string; bodyText: string | null; contentRef: string | null }>; +} + +@Injectable() +export class ThreadsService { + constructor( + private readonly prisma: PrismaService, + @Inject(PLATFORM_PORTS) private readonly ports: IiosPlatformPorts, + ) {} + + async getMessages(threadId: string): Promise<{ threadId: string; messages: ThreadMessage[] }> { + const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } }); + if (!thread) throw new NotFoundException('thread not found'); + + // Reads are policy-scoped too (fail-closed). + await decideOrThrow(this.ports, { action: 'iios.thread.read', threadId, scopeId: thread.scopeId }); + + const interactions = await this.prisma.iiosInteraction.findMany({ + where: { threadId }, + orderBy: { occurredAt: 'asc' }, + include: { parts: { orderBy: { partIndex: 'asc' } } }, + }); + + return { + threadId, + messages: interactions.map((i) => ({ + interactionId: i.id, + actorId: i.actorId, + kind: i.kind, + occurredAt: i.occurredAt, + parts: i.parts.map((p) => ({ kind: p.kind, bodyText: p.bodyText, contentRef: p.contentRef })), + })), + }; + } +} diff --git a/packages/iios-service/src/threads/threads.spec.ts b/packages/iios-service/src/threads/threads.spec.ts new file mode 100644 index 0000000..42c9257 --- /dev/null +++ b/packages/iios-service/src/threads/threads.spec.ts @@ -0,0 +1,42 @@ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import { PrismaClient } from '@prisma/client'; +import { makeFakePorts, portalMessageBasic } from '@insignia/iios-testkit'; +import { IngestService } from '../interactions/ingest.service'; +import { ThreadsService } from './threads.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; + +async function clean(): Promise { + await prisma.iiosOutboxEvent.deleteMany(); + await prisma.iiosProcessedEvent.deleteMany(); + await prisma.iiosMessagePart.deleteMany(); + await prisma.iiosInteraction.deleteMany(); + await prisma.iiosThreadParticipant.deleteMany(); + await prisma.iiosThread.deleteMany(); + await prisma.iiosActorRef.deleteMany(); + await prisma.iiosSourceHandle.deleteMany(); + await prisma.iiosChannel.deleteMany(); + await prisma.iiosScope.deleteMany(); +} + +beforeAll(async () => { await prisma.$connect(); }); +afterAll(async () => { await prisma.$disconnect(); }); +beforeEach(async () => { await clean(); }); + +describe('GET thread messages', () => { + it('returns the stored message part for the thread', async () => { + const ports = makeFakePorts(); + const ingest = new IngestService(asService, ports); + const res = await ingest.ingest(portalMessageBasic, 'idem-thread-1'); + + const threads = new ThreadsService(asService, ports); + const out = await threads.getMessages(res.threadId); + + expect(out.messages).toHaveLength(1); + expect(out.messages[0]?.parts[0]?.bodyText).toBe('my payment failed'); + expect(out.messages[0]?.kind).toBe('MESSAGE'); + }); +}); diff --git a/vitest.config.ts b/vitest.config.ts index c6fdefe..f780b06 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -14,5 +14,8 @@ export default defineConfig({ }, test: { include: ['packages/**/src/**/*.{test,spec}.ts', 'test/**/*.{test,spec}.ts'], + // DB-backed specs share one Postgres and truncate tables between tests, so + // test files must not run in parallel against it. + fileParallelism: false, }, });