diff --git a/packages/iios-service/src/adapters/adapter.registry.ts b/packages/iios-service/src/adapters/adapter.registry.ts index 5ce5e93..8d8dd62 100644 --- a/packages/iios-service/src/adapters/adapter.registry.ts +++ b/packages/iios-service/src/adapters/adapter.registry.ts @@ -1,6 +1,6 @@ import { Injectable } from '@nestjs/common'; import type { ScopeVector } from '@insignia/iios-contracts'; -import { WebhookAdapter, type ChannelAdapter } from '@insignia/iios-adapter-sdk'; +import { WebhookAdapter, EmailAdapter, type ChannelAdapter } from '@insignia/iios-adapter-sdk'; export interface AdapterConfig { secret: string; @@ -8,10 +8,9 @@ export interface AdapterConfig { } /** - * 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. + * Registers channel adapters + their per-channel config (secret + scope). EMAIL uses + * the email-native EmailAdapter (RFC threading); the rest use the reference + * WebhookAdapter. Secrets come from ADAPTER_SECRETS (JSON map); scope defaults to demo. */ @Injectable() export class AdapterRegistry { @@ -29,8 +28,9 @@ export class AdapterRegistry { appId: process.env.ADAPTER_APP ?? 'portal-demo', }; for (const channelType of ['WEBHOOK', 'EMAIL', 'WHATSAPP', 'PORTAL']) { + const adapter: ChannelAdapter = channelType === 'EMAIL' ? new EmailAdapter() : new WebhookAdapter(channelType); this.adapters.set(channelType, { - adapter: new WebhookAdapter(channelType), + adapter, config: { secret: secrets[channelType] ?? 'dev-adapter-secret', scope }, }); } diff --git a/packages/iios-service/src/adapters/adapters.spec.ts b/packages/iios-service/src/adapters/adapters.spec.ts index 9c793b5..8c4b154 100644 --- a/packages/iios-service/src/adapters/adapters.spec.ts +++ b/packages/iios-service/src/adapters/adapters.spec.ts @@ -6,7 +6,7 @@ import { resetDb } from '../test-utils/reset-db'; import { ProjectionCursorService } from '../projection/projection-cursor.service'; import { makeFakePorts } from '@insignia/iios-testkit'; import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts'; -import { signedFixture, emailFixture } from '@insignia/iios-adapter-sdk'; +import { signedFixture, emailFixture, emailInboundFixture, emailReplyFixture } from '@insignia/iios-adapter-sdk'; import { AdapterRegistry } from './adapter.registry'; import { InboundService } from './inbound.service'; import { OutboundService } from './outbound.service'; @@ -81,6 +81,46 @@ describe('Adapter inbound (P5)', () => { }); }); +describe('Email adapter — inbound + threading', () => { + const receiveEmail = async (payload: object) => { + const { body, headers } = signedFixture(payload, SECRET); + return inbound().receive('EMAIL', body, headers); + }; + const normalizeAll = async () => { + for (const ev of await rawReceivedEvents()) await projector().onRawReceived(ev); + }; + + it('a signed inbound email → a normalized interaction with the body on the email thread', async () => { + const ack = await receiveEmail(emailInboundFixture); + expect(ack.status).toBe('RECEIVED'); + await normalizeAll(); + + const raw = await prisma.iiosInboundRawEvent.findUniqueOrThrow({ where: { id: ack.rawEventId } }); + expect(raw.status).toBe('NORMALIZED'); + const interaction = await prisma.iiosInteraction.findUniqueOrThrow({ where: { id: raw.interactionId! }, include: { parts: true, thread: true } }); + expect(interaction.parts[0]?.bodyText).toBe(emailInboundFixture.text); + expect(interaction.thread?.externalThreadRef).toBe('email:'); + }); + + it('a reply email (In-Reply-To) lands on the SAME email thread as the original', async () => { + await receiveEmail(emailInboundFixture); + await receiveEmail(emailReplyFixture); + await normalizeAll(); + + const interactions = await prisma.iiosInteraction.findMany({ orderBy: { occurredAt: 'asc' } }); + expect(interactions).toHaveLength(2); + expect(interactions[0]!.threadId).toBe(interactions[1]!.threadId); // reply joined the thread + expect(await prisma.iiosThread.count()).toBe(1); // one email conversation + }); + + it('bad signature on EMAIL → REJECTED, nothing ingested (fail-closed)', async () => { + const { body } = signedFixture(emailInboundFixture, SECRET); + await expect(inbound().receive('EMAIL', 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); + }); +}); + describe('Adapter outbound (P5)', () => { const outbound = (limit?: number): OutboundService => { const s = new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService)); diff --git a/packages/iios-service/src/capability/capability.registry.ts b/packages/iios-service/src/capability/capability.registry.ts index ed780d4..4e4161d 100644 --- a/packages/iios-service/src/capability/capability.registry.ts +++ b/packages/iios-service/src/capability/capability.registry.ts @@ -2,6 +2,7 @@ import { Injectable, NotFoundException } from '@nestjs/common'; import type { CapabilityProvider } from '@insignia/iios-contracts'; import { SandboxProvider } from './sandbox.provider'; import { HttpProvider } from './http.provider'; +import { EmailProvider } from './email.provider'; const DEFAULT_CHANNELS = ['WEBHOOK', 'EMAIL', 'WHATSAPP', 'PORTAL']; @@ -19,7 +20,9 @@ export class CapabilityProviderRegistry { for (const ch of DEFAULT_CHANNELS) this.register(new SandboxProvider([ch])); for (const ch of DEFAULT_CHANNELS) { const url = process.env[`IIOS_PROVIDER_URL_${ch}`]; - if (url) this.register(new HttpProvider(ch, url)); + if (!url) continue; + // EMAIL gets an email-shaped envelope provider; other channels use the generic HTTP one. + this.register(ch === 'EMAIL' ? new EmailProvider(url) : new HttpProvider(ch, url)); } } diff --git a/packages/iios-service/src/capability/email.provider.ts b/packages/iios-service/src/capability/email.provider.ts new file mode 100644 index 0000000..d4fe99a --- /dev/null +++ b/packages/iios-service/src/capability/email.provider.ts @@ -0,0 +1,40 @@ +import type { CapabilityProvider, CapabilityRequest, ProviderResult } from '@insignia/iios-contracts'; + +interface EmailPayload { + subject?: string; + text?: string; + html?: string; + inReplyTo?: string; +} + +/** + * Email egress provider: shapes the outbound command into an email envelope + * ({to, subject, text, html, inReplyTo}) and POSTs it to a configured send endpoint + * (a vendor mail API / SMTP relay behind a URL). Activated only when + * IIOS_PROVIDER_URL_EMAIL is set; a transport failure is surfaced as FAILED, never thrown. + */ +export class EmailProvider implements CapabilityProvider { + readonly name = 'email-http'; + readonly channelTypes = ['EMAIL']; + readonly capabilities = { canSend: true }; + + constructor(private readonly url: string) {} + + async send(req: CapabilityRequest): Promise { + const started = Date.now(); + const p = (req.payload ?? {}) as EmailPayload; + const email = { to: req.target, subject: p.subject ?? '(no subject)', text: p.text, html: p.html, inReplyTo: p.inReplyTo }; + try { + const res = await fetch(this.url, { + method: 'POST', + headers: { 'content-type': 'application/json', 'idempotency-key': req.idempotencyKey }, + body: JSON.stringify(email), + }); + const latencyMs = Date.now() - started; + if (!res.ok) return { providerRef: `email-${res.status}`, outcome: 'FAILED', errorCode: `HTTP_${res.status}`, latencyMs }; + return { providerRef: `email-${req.idempotencyKey}`, outcome: 'SENT', latencyMs }; + } catch (err) { + return { providerRef: 'email-error', outcome: 'FAILED', errorCode: (err as Error).message.slice(0, 60), latencyMs: Date.now() - started }; + } + } +}