/** 邮件管线测试(docs/11 两阶段):ingest 入库/附件/降级/限额/TTL/orphan/凭据撤销 */ import { describe, expect, test } from "bun:test"; import { ingestEmail, resolveMailbox } from "../console/src/services/email_ingest.ts"; import { runCron } from "../console/src/services/cleanup.ts"; import { processDueDeliveries } from "../console/src/services/webhook_delivery.ts"; import { MAX_EMAIL_BYTES } from "../shared/src/index.ts"; import { makeTestEnv, signedIngestRequest, TEST_MAIL } from "./helpers.ts"; describe("邮件接收管线(POST /internal/email/ingest)", () => { test("完整解析:subject/from/to/cc/text/html/附件/邮箱关联/expires_at", async () => { const t = await makeTestEnv(); const body = new TextEncoder().encode(TEST_MAIL); const resp = await t.app.fetch(await signedIngestRequest(t, body), t.env); expect(resp.status).toBe(202); const data = (await resp.json()) as { id: string; mailbox_id: string; expires_at: string }; expect(data.mailbox_id).toBe(t.mailboxId); // catch-all 匹配 const email = (await t.env.DB.prepare("SELECT * FROM emails WHERE id = ?").bind(data.id).first()) as Record< string, unknown >; expect(email.subject).toBe("Test Mail"); expect(email.from_address).toBe("alice@example.org"); expect(JSON.parse(email.to_addresses as string)).toEqual(["hello@example.test"]); expect(JSON.parse(email.cc_addresses as string)).toEqual(["bob@example.net"]); expect(email.text_body).toContain("plain body"); expect(email.html_body).toContain("html body"); expect(email.account_id).toBe(t.accountId); // expires_at ≈ received_at + 86400s const delta = (new Date(email.expires_at as string).getTime() - new Date(email.received_at as string).getTime()) / 1000; expect(delta).toBe(86_400); const atts = await t.env.DB.prepare("SELECT * FROM email_attachments WHERE email_id = ?").bind(data.id).all(); expect(atts.results.length).toBe(1); expect(atts.results[0].filename).toBe("a.txt"); // raw 已入 R2 const obj = await t.env.R2.get(email.raw_storage_key as string); expect(obj).not.toBeNull(); const attObj = await t.env.R2.get(atts.results[0].storage_key as string); expect(await attObj?.text()).toContain("attachment-content"); }); test("malformed 邮件不拒收:降级入库,raw 可取回", async () => { const t = await makeTestEnv(); const garbage = new TextEncoder().encode("\x00\x01not-a-mime-at-all\r\n###"); const resp = await t.app.fetch(await signedIngestRequest(t, garbage), t.env); expect(resp.status).toBe(202); const count = await t.env.DB.prepare("SELECT COUNT(*) AS n FROM emails").first<{ n: number }>(); expect(count?.n).toBe(1); const email = await t.env.DB.prepare("SELECT * FROM emails").first>(); expect(email?.subject).toBe("(unparsable mail)"); const obj = await t.env.R2.get(email?.raw_storage_key as string); expect(obj).not.toBeNull(); }); test("超大附件被跳过,邮件本身照常入库(限额防护)", async () => { const t = await makeTestEnv(); const { MAX_ATTACHMENT_BYTES } = await import("../shared/src/index.ts"); const bigB64 = Buffer.from(new Uint8Array(MAX_ATTACHMENT_BYTES + 1)).toString("base64"); const mail = [ "From: a@b.c", "To: hello@example.test", "Subject: big att", "MIME-Version: 1.0", "Content-Type: multipart/mixed; boundary=\"BB\"", "", "--BB", "Content-Type: application/octet-stream; name=\"big.bin\"", "Content-Disposition: attachment; filename=\"big.bin\"", "Content-Transfer-Encoding: base64", "", bigB64, "--BB", "Content-Type: text/plain", "", "body", "--BB--", ].join("\r\n"); const result = await ingestEmail(t.env, { raw: new TextEncoder().encode(mail), mailFrom: "a@b.c", rcptTo: "hello@example.test", accountId: t.accountId, }); const atts = await t.env.DB.prepare("SELECT id FROM email_attachments WHERE email_id = ?") .bind(result.id) .all(); expect(atts.results.length).toBe(0); // 超限附件跳过 const row = await t.env.DB.prepare("SELECT id FROM emails WHERE id = ?").bind(result.id).first(); expect(row).not.toBeNull(); // 邮件入库 }); test("超大邮件返回 413(资源耗尽防护)", async () => { const t = await makeTestEnv(); const big = new Uint8Array(MAX_EMAIL_BYTES + 1); const resp = await t.app.fetch(await signedIngestRequest(t, big), t.env); expect(resp.status).toBe(413); }); test("精确地址匹配优先于 catch-all", async () => { const t = await makeTestEnv(); const now = new Date().toISOString(); await t.env.DB.prepare( "INSERT INTO mailboxes (id, domain_id, address, status, is_catch_all, created_at, updated_at) VALUES ('mbx_exact', 'dom_test1', 'hello@example.test', 'active', 0, ?, ?)", ) .bind(now, now) .run(); const hit = await resolveMailbox(t.env, ["hello@example.test"]); expect(hit).toBe("mbx_exact"); const other = await resolveMailbox(t.env, ["nobody@example.test"]); expect(other).toBe(t.mailboxId); // catch-all const miss = await resolveMailbox(t.env, ["x@other.example"]); expect(miss).toBeNull(); }); test("mailbox 自定义 retention 覆盖默认 1 天", async () => { const t = await makeTestEnv(); const now = new Date().toISOString(); await t.env.DB.prepare( "UPDATE mailboxes SET retention_seconds = 300 WHERE id = ?", ) .bind(t.mailboxId) .run(); const result = await ingestEmail(t.env, { raw: new TextEncoder().encode(TEST_MAIL), mailFrom: "a@b.c", rcptTo: "hello@example.test", accountId: t.accountId, }); const delta = (new Date(result.expiresAt).getTime() - new Date(now).getTime()) / 1000; expect(delta).toBeLessThan(400); }); test("收件人已不可接收时 ingest 被拒(409,authorize 与 ingest 之间状态变化)", async () => { const t = await makeTestEnv(); // 禁用 catch-all 邮箱 -> authorize 逻辑不再接受该收件人 await t.env.DB.prepare("UPDATE mailboxes SET status = 'disabled' WHERE id = ?") .bind(t.mailboxId) .run(); const body = new TextEncoder().encode(TEST_MAIL); const resp = await t.app.fetch(await signedIngestRequest(t, body), t.env); expect(resp.status).toBe(409); const data = (await resp.json()) as { code: string }; expect(data.code).toBe("MAILBOX_DISABLED"); const count = await t.env.DB.prepare("SELECT COUNT(*) AS n FROM emails").first<{ n: number }>(); expect(count?.n).toBe(0); }); test("域名归属其它账号的 ingest 请求被拒(DOMAIN_NOT_FOUND)", async () => { const t = await makeTestEnv(); // 收件人域名不在 credential 所属账号下 const body = new TextEncoder().encode(TEST_MAIL); const resp = await t.app.fetch( await signedIngestRequest(t, body, { recipient: "hello@other-domain.test" }), t.env, ); expect(resp.status).toBe(409); const data = (await resp.json()) as { code: string }; expect(data.code).toBe("DOMAIN_NOT_FOUND"); }); }); describe("TTL 清理与 orphan", () => { test("过期邮件被清理(R2 + DB),未过期保留", async () => { const t = await makeTestEnv(); const result = await ingestEmail(t.env, { raw: new TextEncoder().encode(TEST_MAIL), mailFrom: "a@b.c", rcptTo: "hello@example.test", accountId: t.accountId, }); // 人为把 expires_at 改到过去 await t.env.DB.prepare("UPDATE emails SET expires_at = '2000-01-01T00:00:00Z' WHERE id = ?") .bind(result.id) .run(); const stats = await runCron(t.env, () => processDueDeliveries(t.env)); expect(stats.expiredEmails).toBe(1); const row = await t.env.DB.prepare("SELECT id FROM emails WHERE id = ?").bind(result.id).first(); expect(row).toBeNull(); const atts = await t.env.DB.prepare("SELECT id FROM email_attachments WHERE email_id = ?") .bind(result.id) .all(); expect(atts.results.length).toBe(0); const obj = await t.env.R2.get(result.id ? `emails/2000/01/01/${result.id}/raw.eml` : "nonexist"); expect(obj).toBeNull(); }); test("orphan 附件与投递记录被清理", async () => { const t = await makeTestEnv(); const now = new Date().toISOString(); await t.env.DB.prepare( "INSERT INTO email_attachments (id, email_id, filename, size) VALUES ('att_orphan', 'msg_gone', 'x.txt', 1)", ) .run(); await t.env.DB.prepare( "INSERT INTO webhook_deliveries (id, email_id, target_type, url, status, attempt_count, created_at) VALUES ('dlv_orphan', 'msg_gone', 'global', 'https://hooks.example.com/x', 'pending', 0, ?)", ) .bind(now) .run(); const stats = await runCron(t.env, () => processDueDeliveries(t.env)); expect(stats.orphanAttachments).toBe(1); expect(stats.orphanDeliveries).toBe(1); }); }); describe("Worker credential 撤销", () => { test("撤销后 Worker 请求被 401,且不因重试恢复", async () => { const t = await makeTestEnv(); await t.env.DB.prepare("UPDATE worker_credentials SET status='revoked', revoked_at=? WHERE id='wcred1'") .bind(new Date().toISOString()) .run(); const body = new TextEncoder().encode(TEST_MAIL); const resp = await t.app.fetch(await signedIngestRequest(t, body), t.env); expect(resp.status).toBe(401); const data = (await resp.json()) as { error: string }; expect(data.error).toContain("revoked"); }); test("账号 revoked 后 credential 立即失效", async () => { const t = await makeTestEnv(); await t.env.DB.prepare("UPDATE cloudflare_accounts SET status = 'revoked' WHERE id = ?") .bind(t.accountId) .run(); const req = await signedIngestRequest(t, new TextEncoder().encode(TEST_MAIL)); const resp = await t.app.fetch(req, t.env); expect(resp.status).toBe(401); }); });