feat(service): P5.3 inbound adapter pipeline (ACK-fast webhook + RawEventProjector) + traceId

POST /v1/adapters/:type/webhook verifies HMAC (fail-closed) -> stores raw ->
emits raw.received -> 202. RawEventProjector consumes it off the relay, normalizes
via the adapter, reuses IngestService (traceId threaded raw->interaction).
Deduped on (channelType, externalEventId). rawBody enabled. 48 tests.

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

packages/iios-service/src/adapters/adapter.registry.ts

@@ -0,0 +1,42 @@
+import { Injectable } from '@nestjs/common';
+import type { ScopeVector } from '@insignia/iios-contracts';
+import { WebhookAdapter, type ChannelAdapter } from '@insignia/iios-adapter-sdk';
+
+export interface AdapterConfig {
+  secret: string;
+  scope: ScopeVector;
+}
+
+/**
+ * Registers channel adapters + their per-channel config (secret + scope). In P5
+ * one reference WebhookAdapter backs several channelTypes (WEBHOOK/EMAIL/WHATSAPP)
+ * so the fixtures exercise the same anti-corruption pipeline. Secrets come from
+ * ADAPTER_SECRETS (JSON map); scope defaults to the demo scope.
+ */
+@Injectable()
+export class AdapterRegistry {
+  private readonly adapters = new Map<string, { adapter: ChannelAdapter; config: AdapterConfig }>();
+
+  constructor() {
+    let secrets: Record<string, string> = {};
+    try {
+      secrets = JSON.parse(process.env.ADAPTER_SECRETS ?? '{}');
+    } catch {
+      secrets = {};
+    }
+    const scope: ScopeVector = {
+      orgId: process.env.ADAPTER_ORG ?? 'org_portal-demo',
+      appId: process.env.ADAPTER_APP ?? 'portal-demo',
+    };
+    for (const channelType of ['WEBHOOK', 'EMAIL', 'WHATSAPP']) {
+      this.adapters.set(channelType, {
+        adapter: new WebhookAdapter(channelType),
+        config: { secret: secrets[channelType] ?? 'dev-adapter-secret', scope },
+      });
+    }
+  }
+
+  get(channelType: string): { adapter: ChannelAdapter; config: AdapterConfig } | undefined {
+    return this.adapters.get(channelType);
+  }
+}

packages/iios-service/src/adapters/adapters.controller.ts

@@ -0,0 +1,17 @@
+import { Controller, HttpCode, Param, Post, Req } from '@nestjs/common';
+import type { Request } from 'express';
+import { InboundService } from './inbound.service';
+
+@Controller('v1/adapters')
+export class AdaptersController {
+  constructor(private readonly inbound: InboundService) {}
+
+  /** Provider webhook. Authenticated by HMAC signature (not a session token). */
+  @Post(':channelType/webhook')
+  @HttpCode(202)
+  async webhook(@Param('channelType') channelType: string, @Req() req: Request & { rawBody?: Buffer }) {
+    const rawBody = req.rawBody ? req.rawBody.toString('utf8') : JSON.stringify(req.body ?? {});
+    const headers = req.headers as Record<string, string | undefined>;
+    return this.inbound.receive(channelType, rawBody, headers);
+  }
+}

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

@@ -0,0 +1,15 @@
+import { Module } from '@nestjs/common';
+import { OutboxModule } from '../outbox/outbox.module';
+import { InteractionsModule } from '../interactions/interactions.module';
+import { AdapterRegistry } from './adapter.registry';
+import { InboundService } from './inbound.service';
+import { RawEventProjector } from './raw-event.projector';
+import { AdaptersController } from './adapters.controller';
+
+@Module({
+  imports: [OutboxModule, InteractionsModule],
+  controllers: [AdaptersController],
+  providers: [AdapterRegistry, InboundService, RawEventProjector],
+  exports: [AdapterRegistry, InboundService],
+})
+export class AdaptersModule {}

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

@@ -0,0 +1,75 @@
+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 { signedFixture, emailFixture } from '@insignia/iios-adapter-sdk';
+import { AdapterRegistry } from './adapter.registry';
+import { InboundService } from './inbound.service';
+import { RawEventProjector } from './raw-event.projector';
+import { IngestService } from '../interactions/ingest.service';
+import { ActorResolver } from '../identity/actor.resolver';
+import { OutboxBus } from '../outbox/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;
+const SECRET = 'dev-adapter-secret'; // AdapterRegistry default
+
+const registry = () => new AdapterRegistry();
+const inbound = () => new InboundService(asService, registry());
+const projector = () =>
+  new RawEventProjector(asService, new OutboxBus(), registry(), new IngestService(asService, makeFakePorts(), new ActorResolver(asService)));
+
+async function rawReceivedEvents(): Promise<CloudEvent[]> {
+  const rows = await prisma.iiosOutboxEvent.findMany({ where: { eventType: IIOS_EVENTS.rawReceived }, 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 resetDb(prisma); });
+
+describe('Adapter inbound (P5)', () => {
+  it('signed webhook → raw stored, then projector normalizes → interaction with the text + traceId', async () => {
+    const { body, headers } = signedFixture(emailFixture, SECRET);
+    const ack = await inbound().receive('WEBHOOK', body, headers);
+    expect(ack.status).toBe('RECEIVED');
+
+    const raw = await prisma.iiosInboundRawEvent.findUniqueOrThrow({ where: { id: ack.rawEventId } });
+    expect(raw.signatureStatus).toBe('VERIFIED');
+
+    await projector().onRawReceived((await rawReceivedEvents())[0]!);
+
+    const after = await prisma.iiosInboundRawEvent.findUniqueOrThrow({ where: { id: ack.rawEventId } });
+    expect(after.status).toBe('NORMALIZED');
+    expect(after.interactionId).toBeTruthy();
+
+    const interaction = await prisma.iiosInteraction.findUniqueOrThrow({
+      where: { id: after.interactionId! },
+      include: { parts: true },
+    });
+    expect(interaction.parts[0]?.bodyText).toBe(emailFixture.text);
+    expect(interaction.traceId).toBe(raw.traceId); // trace flows raw → interaction
+  });
+
+  it('bad signature → REJECTED, nothing ingested (fail-closed)', async () => {
+    const { body } = signedFixture(emailFixture, SECRET);
+    await expect(inbound().receive('WEBHOOK', body, { 'x-iios-signature': 'sha256=bad' })).rejects.toThrow();
+    expect(await prisma.iiosInboundRawEvent.count({ where: { status: 'REJECTED' } })).toBe(1);
+    expect(await prisma.iiosInteraction.count()).toBe(0);
+  });
+
+  it('replays: same provider eventId twice → one raw event, one interaction', async () => {
+    const { body, headers } = signedFixture(emailFixture, SECRET);
+    const svc = inbound();
+    await svc.receive('WEBHOOK', body, headers);
+    const second = await svc.receive('WEBHOOK', body, headers);
+    expect(second.deduped).toBe(true);
+
+    await projector().onRawReceived((await rawReceivedEvents())[0]!);
+    expect(await prisma.iiosInboundRawEvent.count()).toBe(1);
+    expect(await prisma.iiosInteraction.count()).toBe(1);
+  });
+});

packages/iios-service/src/adapters/inbound.service.ts

@@ -0,0 +1,100 @@
+import { randomUUID } from 'node:crypto';
+import { Injectable, NotFoundException, UnauthorizedException } from '@nestjs/common';
+import { Prisma } from '@prisma/client';
+import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts';
+import { PrismaService } from '../prisma/prisma.service';
+import { AdapterRegistry } from './adapter.registry';
+
+/**
+ * Inbound edge (ACK-fast). Verifies the provider signature (fail-closed), stores
+ * the raw envelope, emits `raw.received`, and returns immediately (202). The
+ * RawEventProjector does the async normalize→ingest. Deduped on
+ * (channelType, externalEventId) for provider replay/retries.
+ */
+@Injectable()
+export class InboundService {
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly registry: AdapterRegistry,
+  ) {}
+
+  async receive(
+    channelType: string,
+    rawBody: string,
+    headers: Record<string, string | undefined>,
+  ): Promise<{ rawEventId: string; status: string; deduped?: boolean }> {
+    const reg = this.registry.get(channelType);
+    if (!reg) throw new NotFoundException(`unknown channel type: ${channelType}`);
+    const { adapter, config } = reg;
+
+    const payload = this.safeParse(rawBody);
+    const externalEventId = typeof payload?.eventId === 'string' ? payload.eventId : undefined;
+
+    // Fail-closed: a bad/absent signature is recorded and rejected — never ingested.
+    if (!adapter.verifySignature(rawBody, headers, config.secret)) {
+      const rejected = await this.prisma.iiosInboundRawEvent.create({
+        data: {
+          channelType,
+          externalEventId,
+          signatureStatus: 'INVALID',
+          status: 'REJECTED',
+          headers: headers as unknown as Prisma.InputJsonValue,
+          payload: (payload ?? {}) as Prisma.InputJsonValue,
+        },
+      });
+      throw new UnauthorizedException(`invalid signature (raw ${rejected.id})`);
+    }
+
+    // Replay: a repeated provider event id returns the existing raw event.
+    if (externalEventId) {
+      const existing = await this.prisma.iiosInboundRawEvent.findUnique({
+        where: { channelType_externalEventId: { channelType, externalEventId } },
+      });
+      if (existing) return { rawEventId: existing.id, status: existing.status, deduped: true };
+    }
+
+    const traceId = randomUUID();
+    const raw = await this.prisma.iiosInboundRawEvent.create({
+      data: {
+        channelType,
+        externalEventId,
+        signatureStatus: 'VERIFIED',
+        status: 'RECEIVED',
+        headers: headers as unknown as Prisma.InputJsonValue,
+        payload: payload as Prisma.InputJsonValue,
+        traceId,
+      },
+    });
+
+    const event: CloudEvent = {
+      specversion: '1.0',
+      id: `evt_raw_${raw.id}`,
+      type: IIOS_EVENTS.rawReceived,
+      source: `iios/adapter/${channelType}`,
+      subject: `raw_event/${raw.id}`,
+      time: new Date().toISOString(),
+      datacontenttype: 'application/json',
+      insignia: { correlationId: traceId, idempotencyKey: `raw:${raw.id}`, dataClass: 'internal' },
+      data: { rawEventId: raw.id, channelType },
+    };
+    await this.prisma.iiosOutboxEvent.create({
+      data: {
+        aggregateType: 'raw_event',
+        aggregateId: raw.id,
+        eventType: IIOS_EVENTS.rawReceived,
+        cloudEvent: event as unknown as Prisma.InputJsonValue,
+        partitionKey: `${channelType}:${raw.id}`,
+      },
+    });
+
+    return { rawEventId: raw.id, status: 'RECEIVED' };
+  }
+
+  private safeParse(body: string): Record<string, unknown> | undefined {
+    try {
+      return JSON.parse(body) as Record<string, unknown>;
+    } catch {
+      return undefined;
+    }
+  }
+}

packages/iios-service/src/adapters/raw-event.projector.ts

@@ -0,0 +1,61 @@
+import { Injectable, OnModuleInit } from '@nestjs/common';
+import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts';
+import { PrismaService } from '../prisma/prisma.service';
+import { OutboxBus } from '../outbox/outbox.bus';
+import { AdapterRegistry } from './adapter.registry';
+import { IngestService } from '../interactions/ingest.service';
+
+/**
+ * Async normalize→ingest for verified raw events (the "process separately" half
+ * of the ACK-fast webhook). Idempotent via the processed-event ledger; reuses
+ * the kernel IngestService. Carries the raw event's traceId into the interaction.
+ */
+@Injectable()
+export class RawEventProjector implements OnModuleInit {
+  private readonly consumer = 'raw-event-projector';
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly bus: OutboxBus,
+    private readonly registry: AdapterRegistry,
+    private readonly ingest: IngestService,
+  ) {}
+
+  onModuleInit(): void {
+    this.bus.on(IIOS_EVENTS.rawReceived, (p) => void this.onRawReceived(p as CloudEvent).catch(() => {}));
+  }
+
+  async onRawReceived(event: CloudEvent): Promise<void> {
+    if (!(await this.claim(event.id))) return;
+    const { rawEventId } = event.data as { rawEventId: string };
+    const raw = await this.prisma.iiosInboundRawEvent.findUnique({ where: { id: rawEventId } });
+    if (!raw || raw.status !== 'RECEIVED') return;
+
+    const reg = this.registry.get(raw.channelType);
+    if (!reg) {
+      await this.prisma.iiosInboundRawEvent.update({ where: { id: rawEventId }, data: { status: 'FAILED' } });
+      return;
+    }
+
+    try {
+      const req = reg.adapter.normalize(raw.payload, { scope: reg.config.scope });
+      req.traceId = raw.traceId ?? undefined;
+      const key = `${raw.channelType}:${raw.externalEventId ?? raw.id}`;
+      const res = await this.ingest.ingest(req, key);
+      await this.prisma.iiosInboundRawEvent.update({
+        where: { id: rawEventId },
+        data: { status: 'NORMALIZED', interactionId: res.interactionId },
+      });
+    } catch {
+      await this.prisma.iiosInboundRawEvent.update({ where: { id: rawEventId }, data: { status: 'FAILED' } });
+    }
+  }
+
+  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/app.module.ts

@@ -8,6 +8,7 @@ import { ThreadsModule } from './threads/threads.module';
 import { MessageModule } from './messaging/message.module';
 import { InboxModule } from './inbox/inbox.module';
 import { SupportModule } from './support/support.module';
+import { AdaptersModule } from './adapters/adapters.module';
 import { HealthController } from './health.controller';
 import { DevController } from './dev/dev.controller';
 
@@ -22,6 +23,7 @@ import { DevController } from './dev/dev.controller';
     MessageModule,
     InboxModule,
     SupportModule,
+    AdaptersModule,
   ],
   controllers: [HealthController, DevController],
 })

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

@@ -163,6 +163,7 @@ export class IngestService {
             externalId: req.providerEventId,
             occurredAt: new Date(req.occurredAt),
             status: 'NORMALIZED',
+            traceId: req.traceId,
             policyDecisionRef: decision.decisionRef,
             metadata: (req.metadata ?? undefined) as Prisma.InputJsonValue | undefined,
           },

packages/iios-service/src/main.ts

@@ -5,7 +5,7 @@ import { ValidationPipe } from '@nestjs/common';
 import { AppModule } from './app.module';
 
 async function bootstrap(): Promise<void> {
-  const app = await NestFactory.create(AppModule);
+  const app = await NestFactory.create(AppModule, { rawBody: true });
   app.useGlobalPipes(
     new ValidationPipe({ whitelist: true, forbidNonWhitelisted: true, transform: true }),
   );