diff --git a/packages/iios-adapter-sdk/src/email-adapter.ts b/packages/iios-adapter-sdk/src/email-adapter.ts index 99713e0..ca06a0c 100644 --- a/packages/iios-adapter-sdk/src/email-adapter.ts +++ b/packages/iios-adapter-sdk/src/email-adapter.ts @@ -43,6 +43,11 @@ export class EmailAdapter implements ChannelAdapter { return hmacVerify(rawBody, secret, typeof sig === 'string' ? sig : undefined); } + externalEventId(payload: unknown): string | undefined { + const id = (payload as EmailInboundPayload)?.messageId; + return typeof id === 'string' ? id : undefined; + } + normalize(payload: unknown, ctx: NormalizeContext): IngestInteractionRequest { const p = payload as EmailInboundPayload; // A reply's thread root is the first Message-ID in its chain → same IIOS thread as the original. diff --git a/packages/iios-adapter-sdk/src/types.ts b/packages/iios-adapter-sdk/src/types.ts index e1499db..d478f4a 100644 --- a/packages/iios-adapter-sdk/src/types.ts +++ b/packages/iios-adapter-sdk/src/types.ts @@ -34,6 +34,8 @@ export interface ChannelAdapter { channelType: string; capability: AdapterCapability; verifySignature(rawBody: string, headers: Record, secret: string): boolean; + /** Provider-native id used to dedup replays/retries at the inbound edge (e.g. eventId, Message-ID). */ + externalEventId?(payload: unknown): string | undefined; normalize(payload: unknown, ctx: NormalizeContext): IngestInteractionRequest; fetch?(ref: string): Promise; send(cmd: OutboundCommandInput): Promise; diff --git a/packages/iios-adapter-sdk/src/webhook-adapter.ts b/packages/iios-adapter-sdk/src/webhook-adapter.ts index 283ac12..86944ff 100644 --- a/packages/iios-adapter-sdk/src/webhook-adapter.ts +++ b/packages/iios-adapter-sdk/src/webhook-adapter.ts @@ -37,6 +37,11 @@ export class WebhookAdapter implements ChannelAdapter { return hmacVerify(rawBody, secret, typeof sig === 'string' ? sig : undefined); } + externalEventId(payload: unknown): string | undefined { + const id = (payload as WebhookPayload)?.eventId; + return typeof id === 'string' ? id : undefined; + } + normalize(payload: unknown, ctx: NormalizeContext): IngestInteractionRequest { const p = payload as WebhookPayload; return { diff --git a/packages/iios-service/scripts/smoke-email.mjs b/packages/iios-service/scripts/smoke-email.mjs new file mode 100644 index 0000000..e6fe59e --- /dev/null +++ b/packages/iios-service/scripts/smoke-email.mjs @@ -0,0 +1,88 @@ +// Email adapter smoke: a signed inbound email → normalized interaction; a reply +// (In-Reply-To) lands on the SAME email thread; outbound EMAIL → SENT (sandbox). +// Requires the service running with IIOS_DEV_TOKENS=1 (relay timer on). No real network. +import 'dotenv/config'; +import crypto from 'node:crypto'; +import { randomUUID } from 'node:crypto'; +import jwt from 'jsonwebtoken'; +import { PrismaClient } from '@prisma/client'; + +const SERVICE = process.env.SMOKE_URL ?? 'http://localhost:3200'; +const APP_ID = 'portal-demo'; +const APP_SECRET = JSON.parse(process.env.APP_SECRETS ?? '{"portal-demo":"dev-secret"}')[APP_ID]; +const EMAIL_SECRET = (() => { + try { return JSON.parse(process.env.ADAPTER_SECRETS ?? '{}').EMAIL ?? 'dev-adapter-secret'; } catch { return 'dev-adapter-secret'; } +})(); +const prisma = new PrismaClient(); + +const token = jwt.sign({ sub: 'inspector', appId: APP_ID, orgId: `org_${APP_ID}` }, APP_SECRET, { algorithm: 'HS256', expiresIn: '1h' }); +const sign = (body) => 'sha256=' + crypto.createHmac('sha256', EMAIL_SECRET).update(body).digest('hex'); +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); +const assert = (c, m) => { if (!c) { console.error('✗', m); process.exit(1); } console.log('✓', m); }; + +async function postEmail(payload) { + const body = JSON.stringify(payload); + return fetch(`${SERVICE}/v1/adapters/EMAIL/webhook`, { + method: 'POST', + headers: { 'content-type': 'application/json', 'x-iios-signature': sign(body) }, + body, + }); +} +async function inboundList() { + const r = await fetch(`${SERVICE}/v1/adapters/inbound`, { headers: { authorization: `Bearer ${token}` } }); + if (!r.ok) throw new Error(`GET /v1/adapters/inbound ${r.status}`); + return r.json(); +} +async function waitNormalized(messageId) { + for (let i = 0; i < 40; i++) { + const e = (await inboundList()).find((x) => x.externalEventId === messageId); + if (e && e.status === 'NORMALIZED' && e.interactionId) return e.interactionId; + await sleep(300); + } + return null; +} + +// Unique Message-IDs per run so re-runs don't dedup against earlier ones. +const run = randomUUID().slice(0, 8); +const mid1 = `<${run}-1@smoke>`; +const mid2 = `<${run}-2@smoke>`; + +// 1) A first inbound email → normalized interaction. +const r1 = await postEmail({ messageId: mid1, from: 'buyer@smoke', fromName: 'Buyer', subject: 'Need help with my order', text: 'my order is late' }); +assert(r1.status === 202, 'signed email accepted (202)'); +const i1 = await waitNormalized(mid1); +assert(i1, 'first email normalized into an interaction'); + +// 2) A reply (In-Reply-To) → same email thread. +const r2 = await postEmail({ messageId: mid2, from: 'buyer@smoke', subject: 'Re: Need help with my order', text: 'any update?', inReplyTo: mid1, references: [mid1] }); +assert(r2.status === 202, 'reply email accepted (202)'); +const i2 = await waitNormalized(mid2); +assert(i2, 'reply email normalized into an interaction'); + +const [it1, it2] = await Promise.all([ + prisma.iiosInteraction.findUniqueOrThrow({ where: { id: i1 }, include: { thread: true } }), + prisma.iiosInteraction.findUniqueOrThrow({ where: { id: i2 }, include: { thread: true } }), +]); +assert(it1.threadId === it2.threadId, 'reply joined the SAME email thread as the original'); +assert(it1.thread?.externalThreadRef === `email:${mid1}`, `thread rooted at the original Message-ID (${it1.thread?.externalThreadRef})`); + +// 3) Bad signature → rejected. +const bad = await fetch(`${SERVICE}/v1/adapters/EMAIL/webhook`, { + method: 'POST', + headers: { 'content-type': 'application/json', 'x-iios-signature': 'sha256=bad' }, + body: JSON.stringify({ messageId: `<${run}-bad@smoke>`, from: 'x', text: 'x' }), +}); +assert(bad.status === 401, 'bad signature rejected (401)'); + +// 4) Outbound EMAIL → governed egress → SENT (sandbox). +const sendRes = await fetch(`${SERVICE}/v1/adapters/EMAIL/send`, { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${token}` }, + body: JSON.stringify({ target: 'buyer@smoke', payload: { subject: 'Re: Need help', text: 'thanks, looking into it' } }), +}); +const cmd = await sendRes.json(); +assert(cmd.status === 'SENT', `outbound email SENT via broker (status ${cmd.status})`); + +console.log('\nEmail adapter smoke: PASS'); +await prisma.$disconnect(); +process.exit(0); diff --git a/packages/iios-service/src/adapters/adapters.spec.ts b/packages/iios-service/src/adapters/adapters.spec.ts index 8c4b154..00ec66d 100644 --- a/packages/iios-service/src/adapters/adapters.spec.ts +++ b/packages/iios-service/src/adapters/adapters.spec.ts @@ -113,6 +113,17 @@ describe('Email adapter — inbound + threading', () => { expect(await prisma.iiosThread.count()).toBe(1); // one email conversation }); + it('redelivered email (same Message-ID) → one raw event, one interaction (dedup)', async () => { + const svc = inbound(); + const { body, headers } = signedFixture(emailInboundFixture, SECRET); + await svc.receive('EMAIL', body, headers); + const second = await svc.receive('EMAIL', body, headers); + expect(second.deduped).toBe(true); + await normalizeAll(); + expect(await prisma.iiosInboundRawEvent.count()).toBe(1); + expect(await prisma.iiosInteraction.count()).toBe(1); + }); + 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(); diff --git a/packages/iios-service/src/adapters/inbound.service.ts b/packages/iios-service/src/adapters/inbound.service.ts index 8b219ff..08f8b5d 100644 --- a/packages/iios-service/src/adapters/inbound.service.ts +++ b/packages/iios-service/src/adapters/inbound.service.ts @@ -28,7 +28,8 @@ export class InboundService { const { adapter, config } = reg; const payload = this.safeParse(rawBody); - const externalEventId = typeof payload?.eventId === 'string' ? payload.eventId : undefined; + // The adapter extracts its provider-native id (eventId, Message-ID, …) for replay dedup. + const externalEventId = adapter.externalEventId?.(payload) ?? (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)) {