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) <noreply@anthropic.com>

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 {}

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);
+  }
+}

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 {}

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<number> {
+    const eventIds = await this.prisma.$transaction(async (tx) => {
+      const picked = await tx.$queryRaw<Array<{ eventId: string }>>`
+        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<boolean> {
+    const res = await this.prisma.iiosProcessedEvent.createMany({
+      data: [{ consumerName: this.consumer, eventId }],
+      skipDuplicates: true,
+    });
+    return res.count > 0;
+  }
+}

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<void> {
+  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);
+  });
+});

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);
+  }
+}

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 {}

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 })),
+      })),
+    };
+  }
+}

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<void> {
+  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');
+  });
+});