feat(service): P3.2 InboxProjector + message.read event + scheduled relay

Event-driven projector creates/collapses NEEDS_REPLY items from message.sent
(traceId propagated; inbox.item.created emitted) and resolves them to DONE on
message.read; idempotent via processed-event ledger. markRead now emits a
message.read outbox event (atomic). OutboxRelay runs on a 500ms timer in the
app (off in tests). Shared resetDb (TRUNCATE CASCADE) fixes cross-spec FK
isolation. 32 tests green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/iios-contracts/src/events.ts

@@ -35,6 +35,7 @@ export const IIOS_EVENTS = {
   rawReceived: 'com.insignia.iios.raw.received.v1',
   interactionNormalized: 'com.insignia.iios.interaction.normalized.v1',
   messageSent: 'com.insignia.iios.message.sent.v1',
+  messageRead: 'com.insignia.iios.message.read.v1',
   messagePersisted: 'com.insignia.iios.message.persisted.v1',
   inboxItemCreated: 'com.insignia.iios.inbox.item.created.v1',
   ticketCreated: 'com.insignia.iios.support.ticket.created.v1',

packages/iios-service/src/app.module.ts

@@ -6,11 +6,21 @@ import { InteractionsModule } from './interactions/interactions.module';
 import { OutboxModule } from './outbox/outbox.module';
 import { ThreadsModule } from './threads/threads.module';
 import { MessageModule } from './messaging/message.module';
+import { InboxModule } from './inbox/inbox.module';
 import { HealthController } from './health.controller';
 import { DevController } from './dev/dev.controller';
 
 @Module({
-  imports: [PrismaModule, PlatformModule, IdentityModule, InteractionsModule, OutboxModule, ThreadsModule, MessageModule],
+  imports: [
+    PrismaModule,
+    PlatformModule,
+    IdentityModule,
+    InteractionsModule,
+    OutboxModule,
+    ThreadsModule,
+    MessageModule,
+    InboxModule,
+  ],
   controllers: [HealthController, DevController],
 })
 export class AppModule {}

packages/iios-service/src/inbox/inbox.module.ts

@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+import { OutboxModule } from '../outbox/outbox.module';
+import { InboxProjector } from './inbox.projector';
+
+@Module({
+  imports: [OutboxModule],
+  providers: [InboxProjector],
+  exports: [InboxProjector],
+})
+export class InboxModule {}

packages/iios-service/src/inbox/inbox.projector.ts

@@ -0,0 +1,136 @@
+import { Injectable, OnModuleInit } from '@nestjs/common';
+import { Prisma } from '@prisma/client';
+import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts';
+import { PrismaService } from '../prisma/prisma.service';
+import { OutboxBus } from '../outbox/outbox.bus';
+
+interface MessageSentData {
+  interactionId: string;
+  threadId: string;
+  senderActorId: string;
+}
+interface MessageReadData {
+  threadId: string;
+  actorId: string;
+  interactionId: string;
+}
+
+/**
+ * Projects message events into Inbox work-items (P3). Inbox reacts to events —
+ * it never imports the message layer. Deterministic + idempotent (collapse one
+ * OPEN NEEDS_REPLY per owner+thread; dedupe each event via the processed-event
+ * ledger). Reading a thread resolves the owner's item to DONE.
+ */
+@Injectable()
+export class InboxProjector implements OnModuleInit {
+  private readonly consumer = 'inbox-projector';
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly bus: OutboxBus,
+  ) {}
+
+  onModuleInit(): void {
+    this.bus.on(IIOS_EVENTS.messageSent, (p) => void this.onMessageSent(p as CloudEvent).catch(() => {}));
+    this.bus.on(IIOS_EVENTS.messageRead, (p) => void this.onMessageRead(p as CloudEvent).catch(() => {}));
+  }
+
+  async onMessageSent(event: CloudEvent): Promise<void> {
+    if (!(await this.claim(event.id))) return;
+    const data = event.data as MessageSentData;
+    const traceId = event.insignia?.correlationId;
+    const thread = await this.prisma.iiosThread.findUnique({ where: { id: data.threadId } });
+    if (!thread) return;
+
+    const participants = await this.prisma.iiosThreadParticipant.findMany({ where: { threadId: data.threadId } });
+    for (const p of participants) {
+      if (p.actorId === data.senderActorId) continue;
+      await this.upsertNeedsReply(thread.scopeId, p.actorId, data.threadId, data.interactionId, traceId, thread.subject);
+    }
+  }
+
+  async onMessageRead(event: CloudEvent): Promise<void> {
+    if (!(await this.claim(event.id))) return;
+    const data = event.data as MessageReadData;
+    const item = await this.prisma.iiosInboxItem.findFirst({
+      where: {
+        threadId: data.threadId,
+        ownerActorId: data.actorId,
+        kind: 'NEEDS_REPLY',
+        state: { in: ['OPEN', 'SNOOZED'] },
+      },
+    });
+    if (!item) return;
+    await this.prisma.$transaction([
+      this.prisma.iiosInboxItem.update({ where: { id: item.id }, data: { state: 'DONE' } }),
+      this.prisma.iiosInboxItemStateHistory.create({
+        data: { inboxItemId: item.id, fromState: item.state, toState: 'DONE', reasonCode: 'message_read' },
+      }),
+    ]);
+  }
+
+  private async upsertNeedsReply(
+    scopeId: string,
+    ownerActorId: string,
+    threadId: string,
+    sourceInteractionId: string,
+    traceId: string | undefined,
+    subject: string | null,
+  ): Promise<void> {
+    const existing = await this.prisma.iiosInboxItem.findFirst({
+      where: { ownerActorId, threadId, kind: 'NEEDS_REPLY', state: { in: ['OPEN', 'SNOOZED'] } },
+    });
+    if (existing) {
+      await this.prisma.iiosInboxItem.update({
+        where: { id: existing.id },
+        data: { sourceInteractionId, traceId: traceId ?? existing.traceId, state: 'OPEN' },
+      });
+      return;
+    }
+
+    const item = await this.prisma.iiosInboxItem.create({
+      data: {
+        scopeId,
+        ownerActorId,
+        kind: 'NEEDS_REPLY',
+        title: subject ?? 'New messages',
+        threadId,
+        sourceInteractionId,
+        traceId,
+      },
+    });
+    await this.prisma.iiosInboxItemStateHistory.create({
+      data: { inboxItemId: item.id, toState: 'OPEN', reasonCode: 'message_sent' },
+    });
+
+    const created: CloudEvent = {
+      specversion: '1.0',
+      id: `evt_${item.id}`,
+      type: IIOS_EVENTS.inboxItemCreated,
+      source: `iios/inbox/${scopeId}`,
+      subject: `inbox_item/${item.id}`,
+      time: new Date().toISOString(),
+      datacontenttype: 'application/json',
+      insignia: { scopeSnapshotId: scopeId, correlationId: traceId, idempotencyKey: `inbox:${item.id}`, dataClass: 'internal' },
+      data: { inboxItemId: item.id, ownerActorId, threadId, kind: 'NEEDS_REPLY' },
+    };
+    await this.prisma.iiosOutboxEvent.create({
+      data: {
+        aggregateType: 'inbox_item',
+        aggregateId: item.id,
+        eventType: IIOS_EVENTS.inboxItemCreated,
+        cloudEvent: created as unknown as Prisma.InputJsonValue,
+        partitionKey: `${scopeId}:${threadId}`,
+      },
+    });
+  }
+
+  /** Idempotent claim: false if this event id was already processed. */
+  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;
+  }
+}

packages/iios-service/src/inbox/inbox.spec.ts

@@ -0,0 +1,105 @@
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
+import { PrismaClient } from '@prisma/client';
+import { resetDb } from '../test-utils/reset-db';
+import { makeFakePorts } from '@insignia/iios-testkit';
+import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts';
+import { MessageService, type MessagePrincipal } from '../messaging/message.service';
+import { ActorResolver } from '../identity/actor.resolver';
+import { OutboxBus } from '../outbox/outbox.bus';
+import { InboxProjector } from './inbox.projector';
+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 ms = () => new MessageService(asService, makeFakePorts(), actors);
+const projector = () => new InboxProjector(asService, new OutboxBus());
+
+const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Alice' };
+const bob: MessagePrincipal = { userId: 'bob', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Bob' };
+
+async function clean(): Promise<void> {
+  await resetDb(prisma);
+}
+
+async function actorIdFor(userId: string): Promise<string> {
+  const handle = await prisma.iiosSourceHandle.findFirstOrThrow({ where: { externalId: userId } });
+  const actor = await prisma.iiosActorRef.findFirstOrThrow({ where: { sourceHandleId: handle.id } });
+  return actor.id;
+}
+async function itemsFor(userId: string) {
+  return prisma.iiosInboxItem.findMany({ where: { ownerActorId: await actorIdFor(userId) } });
+}
+async function eventsOf(type: string) {
+  const rows = await prisma.iiosOutboxEvent.findMany({ where: { eventType: type }, orderBy: { createdAt: 'asc' } });
+  return rows.map((r) => r.cloudEvent as unknown as CloudEvent);
+}
+
+beforeAll(async () => { await prisma.$connect(); });
+afterAll(async () => { await prisma.$disconnect(); });
+beforeEach(async () => { await clean(); });
+
+describe('InboxProjector (P3 work surface)', () => {
+  it('message.sent → NEEDS_REPLY for the recipient (not the sender), with the message traceId', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice);
+    await m.openThread(threadId, bob);
+    const sent = await m.send(threadId, alice, { content: 'hi bob' }, 'k1');
+
+    const [event] = await eventsOf(IIOS_EVENTS.messageSent);
+    await projector().onMessageSent(event);
+
+    const bobItems = await itemsFor('bob');
+    expect(bobItems).toHaveLength(1);
+    expect(bobItems[0]?.kind).toBe('NEEDS_REPLY');
+    expect(bobItems[0]?.state).toBe('OPEN');
+    expect(bobItems[0]?.traceId).toBe(sent.traceId);
+    expect(await itemsFor('alice')).toHaveLength(0);
+  });
+
+  it('collapses per thread: two messages → still 1 open item', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice);
+    await m.openThread(threadId, bob);
+    await m.send(threadId, alice, { content: '1' }, 'k1');
+    await m.send(threadId, alice, { content: '2' }, 'k2');
+
+    const proj = projector();
+    for (const e of await eventsOf(IIOS_EVENTS.messageSent)) await proj.onMessageSent(e);
+
+    expect(await itemsFor('bob')).toHaveLength(1);
+  });
+
+  it('is idempotent: replaying the same event id creates no duplicate', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice);
+    await m.openThread(threadId, bob);
+    await m.send(threadId, alice, { content: 'hi' }, 'k1');
+
+    const [event] = await eventsOf(IIOS_EVENTS.messageSent);
+    const proj = projector();
+    await proj.onMessageSent(event);
+    await proj.onMessageSent(event); // replay
+
+    expect(await itemsFor('bob')).toHaveLength(1);
+  });
+
+  it('message.read → the recipient item goes DONE', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice);
+    await m.openThread(threadId, bob);
+    const sent = await m.send(threadId, alice, { content: 'hi' }, 'k1');
+
+    const proj = projector();
+    await proj.onMessageSent((await eventsOf(IIOS_EVENTS.messageSent))[0]!);
+    expect((await itemsFor('bob'))[0]?.state).toBe('OPEN');
+
+    await m.markRead(threadId, bob, sent.id);
+    await proj.onMessageRead((await eventsOf(IIOS_EVENTS.messageRead))[0]!);
+
+    expect((await itemsFor('bob'))[0]?.state).toBe('DONE');
+    const history = await prisma.iiosInboxItemStateHistory.findMany({ where: { toState: 'DONE' } });
+    expect(history.length).toBeGreaterThanOrEqual(1);
+  });
+});

packages/iios-service/src/interactions/ingest.spec.ts

@@ -1,5 +1,6 @@
 import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
 import { PrismaClient } from '@prisma/client';
+import { resetDb } from '../test-utils/reset-db';
 import { makeFakePorts, portalMessageBasic } from '@insignia/iios-testkit';
 import { PolicyDeniedError } from '@insignia/iios-contracts';
 import { IngestService } from './ingest.service';
@@ -9,16 +10,7 @@ const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/i
 const prisma = new PrismaClient({ datasources: { db: { url } } });
 
 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();
+  await resetDb(prisma);
 }
 
 const svc = (ports = makeFakePorts()) => new IngestService(prisma as unknown as PrismaService, ports);

packages/iios-service/src/messaging/message.service.ts

@@ -165,16 +165,39 @@ export class MessageService {
     if (!thread) throw new NotFoundException('thread not found');
     const actor = await this.actors.resolveActor(thread.scopeId, principal);
 
-    await this.prisma.iiosMessageReceipt.upsert({
+    const readEvent: CloudEvent = {
+      specversion: '1.0',
+      id: `evt_read_${interactionId}_${actor.id}`,
+      type: IIOS_EVENTS.messageRead,
+      source: `iios/message/${thread.scopeId}`,
+      subject: `interaction/${interactionId}`,
+      time: new Date().toISOString(),
+      datacontenttype: 'application/json',
+      insignia: { scopeSnapshotId: thread.scopeId, idempotencyKey: `read:${interactionId}:${actor.id}`, dataClass: 'internal' },
+      data: { threadId, actorId: actor.id, interactionId },
+    };
+
+    await this.prisma.$transaction([
+      this.prisma.iiosMessageReceipt.upsert({
         where: { interactionId_actorId_receiptKind: { interactionId, actorId: actor.id, receiptKind: 'READ' } },
         create: { interactionId, actorId: actor.id, receiptKind: 'READ' },
         update: { occurredAt: new Date() },
-    });
-    await this.prisma.iiosUnreadCounter.upsert({
+      }),
+      this.prisma.iiosUnreadCounter.upsert({
         where: { threadId_actorId: { threadId, actorId: actor.id } },
         create: { threadId, actorId: actor.id, unreadCount: 0, lastReadInteractionId: interactionId },
         update: { unreadCount: 0, lastReadInteractionId: interactionId },
-    });
+      }),
+      this.prisma.iiosOutboxEvent.create({
+        data: {
+          aggregateType: 'interaction',
+          aggregateId: interactionId,
+          eventType: IIOS_EVENTS.messageRead,
+          cloudEvent: readEvent as unknown as Prisma.InputJsonValue,
+          partitionKey: `${thread.scopeId}:${threadId}`,
+        },
+      }),
+    ]);
     return { interactionId, actorId: actor.id };
   }
 

packages/iios-service/src/messaging/message.spec.ts

@@ -1,5 +1,6 @@
 import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
 import { PrismaClient } from '@prisma/client';
+import { resetDb } from '../test-utils/reset-db';
 import { makeFakePorts } from '@insignia/iios-testkit';
 import { MessageService, type MessagePrincipal } from './message.service';
 import { ActorResolver } from '../identity/actor.resolver';
@@ -14,18 +15,7 @@ const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'po
 const bob: MessagePrincipal = { userId: 'bob', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Bob' };
 
 async function clean(): Promise<void> {
-  await prisma.iiosUnreadCounter.deleteMany();
-  await prisma.iiosMessageReceipt.deleteMany();
-  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();
+  await resetDb(prisma);
 }
 
 async function actorIdFor(userId: string): Promise<string> {

packages/iios-service/src/outbox/outbox.relay.ts

@@ -1,4 +1,4 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
 import { PrismaService } from '../prisma/prisma.service';
 import { OutboxBus } from './outbox.bus';
 
@@ -11,14 +11,31 @@ import { OutboxBus } from './outbox.bus';
  * A scheduler (P9) calls this on an interval; tests call it directly.
  */
 @Injectable()
-export class OutboxRelay {
+export class OutboxRelay implements OnModuleInit, OnModuleDestroy {
   private readonly consumer = 'outbox-relay';
+  private readonly logger = new Logger(OutboxRelay.name);
+  private timer?: ReturnType<typeof setInterval>;
 
   constructor(
     private readonly prisma: PrismaService,
     private readonly bus: OutboxBus,
   ) {}
 
+  // Tests construct via `new OutboxRelay(...)`, so onModuleInit never runs there
+  // (no timer). The running app ticks the relay so events flow to consumers.
+  onModuleInit(): void {
+    const ms = Number(process.env.IIOS_RELAY_INTERVAL_MS ?? 500);
+    if (ms > 0) {
+      this.timer = setInterval(() => {
+        void this.relayOnce().catch((err) => this.logger.warn(`relay tick failed: ${(err as Error).message}`));
+      }, ms);
+    }
+  }
+
+  onModuleDestroy(): void {
+    if (this.timer) clearInterval(this.timer);
+  }
+
   async relayOnce(batchSize = 100): Promise<number> {
     const eventIds = await this.prisma.$transaction(async (tx) => {
       const picked = await tx.$queryRaw<Array<{ eventId: string }>>`

packages/iios-service/src/outbox/replay.spec.ts

@@ -1,5 +1,6 @@
 import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
 import { PrismaClient } from '@prisma/client';
+import { resetDb } from '../test-utils/reset-db';
 import { makeFakePorts, portalMessageBasic, replayTwice, isIdempotent } from '@insignia/iios-testkit';
 import { IngestService } from '../interactions/ingest.service';
 import { OutboxRelay } from './outbox.relay';
@@ -11,16 +12,7 @@ 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();
+  await resetDb(prisma);
 }
 
 beforeAll(async () => { await prisma.$connect(); });

packages/iios-service/src/test-utils/reset-db.ts

@@ -0,0 +1,16 @@
+import type { PrismaClient } from '@prisma/client';
+
+/**
+ * FK-safe full reset for DB-backed specs. TRUNCATE … CASCADE clears every kernel
+ * table regardless of relation order, so specs don't each need to know the full
+ * delete order (which broke once Inbox added FK references to interaction/thread/actor).
+ */
+export async function resetDb(prisma: PrismaClient): Promise<void> {
+  await prisma.$executeRawUnsafe(
+    `TRUNCATE TABLE
+      "IiosInboxItemStateHistory","IiosInboxItem","IiosUnreadCounter","IiosMessageReceipt",
+      "IiosOutboxEvent","IiosProcessedEvent","IiosMessagePart","IiosInteraction",
+      "IiosThreadParticipant","IiosThread","IiosActorRef","IiosSourceHandle","IiosChannel","IiosScope"
+      RESTART IDENTITY CASCADE`,
+  );
+}

packages/iios-service/src/threads/threads.spec.ts

@@ -1,5 +1,6 @@
 import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
 import { PrismaClient } from '@prisma/client';
+import { resetDb } from '../test-utils/reset-db';
 import { makeFakePorts, portalMessageBasic } from '@insignia/iios-testkit';
 import { IngestService } from '../interactions/ingest.service';
 import { ThreadsService } from './threads.service';
@@ -10,16 +11,7 @@ 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();
+  await resetDb(prisma);
 }
 
 beforeAll(async () => { await prisma.$connect(); });

packages/iios-service/tsconfig.json

@@ -7,5 +7,5 @@
     "types": ["node"]
   },
   "include": ["src/**/*.ts"],
-  "exclude": ["node_modules", "dist", "src/**/*.spec.ts", "src/**/*.test.ts"]
+  "exclude": ["node_modules", "dist", "src/**/*.spec.ts", "src/**/*.test.ts", "src/test-utils/**"]
 }