diff --git a/packages/iios-contracts/src/ingest.ts b/packages/iios-contracts/src/ingest.ts index 5ddc988..5c5577e 100644 --- a/packages/iios-contracts/src/ingest.ts +++ b/packages/iios-contracts/src/ingest.ts @@ -30,6 +30,8 @@ export interface IngestInteractionRequest { }>; occurredAt: string; providerEventId?: string; + /** Optional trace id to carry onto the created interaction (e.g. from an adapter). */ + traceId?: string; metadata?: Record; } diff --git a/packages/iios-service/package.json b/packages/iios-service/package.json index 54a22e8..a5c36fd 100644 --- a/packages/iios-service/package.json +++ b/packages/iios-service/package.json @@ -13,6 +13,7 @@ }, "dependencies": { "@insignia/iios-contracts": "workspace:*", + "@insignia/iios-adapter-sdk": "workspace:*", "@nestjs/common": "^11.1.27", "@nestjs/core": "^11.1.27", "@nestjs/platform-express": "^11.1.27", diff --git a/packages/iios-service/src/adapters/adapter.registry.ts b/packages/iios-service/src/adapters/adapter.registry.ts new file mode 100644 index 0000000..5afecb6 --- /dev/null +++ b/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(); + + constructor() { + let secrets: Record = {}; + 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); + } +} diff --git a/packages/iios-service/src/adapters/adapters.controller.ts b/packages/iios-service/src/adapters/adapters.controller.ts new file mode 100644 index 0000000..1a24b83 --- /dev/null +++ b/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; + return this.inbound.receive(channelType, rawBody, headers); + } +} diff --git a/packages/iios-service/src/adapters/adapters.module.ts b/packages/iios-service/src/adapters/adapters.module.ts new file mode 100644 index 0000000..bf5e0c0 --- /dev/null +++ b/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 {} diff --git a/packages/iios-service/src/adapters/adapters.spec.ts b/packages/iios-service/src/adapters/adapters.spec.ts new file mode 100644 index 0000000..0130881 --- /dev/null +++ b/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 { + 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); + }); +}); diff --git a/packages/iios-service/src/adapters/inbound.service.ts b/packages/iios-service/src/adapters/inbound.service.ts new file mode 100644 index 0000000..8b219ff --- /dev/null +++ b/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, + ): 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 | undefined { + try { + return JSON.parse(body) as Record; + } catch { + return undefined; + } + } +} diff --git a/packages/iios-service/src/adapters/raw-event.projector.ts b/packages/iios-service/src/adapters/raw-event.projector.ts new file mode 100644 index 0000000..bbfb2bb --- /dev/null +++ b/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 { + 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 { + 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/app.module.ts b/packages/iios-service/src/app.module.ts index cb6feed..65a26ab 100644 --- a/packages/iios-service/src/app.module.ts +++ b/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], }) diff --git a/packages/iios-service/src/interactions/ingest.service.ts b/packages/iios-service/src/interactions/ingest.service.ts index 0641b26..3fd9061 100644 --- a/packages/iios-service/src/interactions/ingest.service.ts +++ b/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, }, diff --git a/packages/iios-service/src/main.ts b/packages/iios-service/src/main.ts index 3430fde..f1d7926 100644 --- a/packages/iios-service/src/main.ts +++ b/packages/iios-service/src/main.ts @@ -5,7 +5,7 @@ import { ValidationPipe } from '@nestjs/common'; import { AppModule } from './app.module'; async function bootstrap(): Promise { - const app = await NestFactory.create(AppModule); + const app = await NestFactory.create(AppModule, { rawBody: true }); app.useGlobalPipes( new ValidationPipe({ whitelist: true, forbidNonWhitelisted: true, transform: true }), ); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 0a90ba9..0558772 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -153,6 +153,9 @@ importers: packages/iios-service: dependencies: + '@insignia/iios-adapter-sdk': + specifier: workspace:* + version: link:../iios-adapter-sdk '@insignia/iios-contracts': specifier: workspace:* version: link:../iios-contracts