wmailhub/tests/pipeline.test.ts

/** 邮件管线测试(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<Record<string, unknown>>();
    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);
  });
});