Adds adapter-owned externalEventId extraction to the ChannelAdapter contract (WebhookAdapter→eventId, EmailAdapter→Message-ID) and uses it in InboundService, so email replays dedup on Message-ID (previously only payload.eventId worked). Adds an email-dedup spec + smoke-email.mjs proving inbound → normalized, a reply joins the same email thread, bad signature rejected, and outbound EMAIL SENT. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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);
|
||||
@@ -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();
|
||||
|
||||
@@ -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)) {
|
||||
|
||||
Reference in New Issue
Block a user