wmailhub/console/src/services/email_ingest.ts

/**
 * 邮件接收管线(docs/06、docs/11):
 * Console 收到 Email Worker 提交的 raw MIME 后:
 *   校验大小限制 -> postal-mime 解析 -> 关联 Mailbox(精确匹配,否则 Domain catch-all)->
 *   raw 写 R2、附件写 R2 -> emails/email_attachments 入库 -> 计算 expires_at ->
 *   为启用的 Webhook 创建 delivery(投递失败不影响邮件入库)。
 *
 * malformed mail 不拒绝:尽量解析,失败时仍保存 raw 并以 fallback 字段入库(可观测)。
 */

import PostalMime from "postal-mime";
import type { Env } from "../env.ts";
import {
  ids,
  DEFAULT_RETENTION_SECONDS,
  MAX_EMAIL_BYTES,
  MAX_ATTACHMENTS,
  MAX_ATTACHMENT_BYTES,
} from "../../../shared/src/index.ts";
import { nowIso } from "../util.ts";
import { enqueueDeliveryForEmail } from "./webhook_delivery.ts";

/** # 成员变量: postal-mime 解析出的结构(仅本文件用到的字段) **/
interface ParsedMail {
  messageId?: string;
  from?: string;
  to?: { address?: string }[];
  cc?: { address?: string }[];
  subject?: string;
  text?: string;
  html?: string;
  attachments?: {
    filename?: string;
    mimeType?: string;
    content: Uint8Array;
    contentId?: string;
  }[];
}

/** # 成员变量: 入站信封信息 **/
export interface IngestInput {
  /** raw MIME 原文 **/
  raw: Uint8Array;
  /** SMTP 信封发件人(Email Worker 传递) **/
  mailFrom: string;
  /** SMTP 信封收件人(Email Worker 传递) **/
  rcptTo: string;
  /** 归属 Console 账号 id(Worker credential 所属账号) **/
  accountId: string;
}

/** # 成员变量: 入库结果 **/
export interface IngestResult {
  id: string;
  mailboxId: string | null;
  expiresAt: string;
  attachmentCount: number;
  size: number;
}

/** # 函数
 * 存储路径:emails/{yyyy}/{mm}/{dd}/{email_id}/raw.eml(docs/09)
 */
export function rawStorageKey(emailId: string, date = new Date()): string {
  const yyyy = date.getUTCFullYear();
  const mm = String(date.getUTCMonth() + 1).padStart(2, "0");
  const dd = String(date.getUTCDate()).padStart(2, "0");
  return `emails/${yyyy}/${mm}/${dd}/${emailId}/raw.eml`;
}

/** # 函数
 * 附件路径:emails/{...}/{email_id}/attachments/{attachment_id}
 */
export function attachmentStorageKey(emailId: string, attachmentId: string, date = new Date()): string {
  return rawStorageKey(emailId, date).replace(/raw\.eml$/, `attachments/${attachmentId}`);
}

/** # 函数
 * 邮件入库主流程;大小/数量超限抛错(由 internal 路由转为 413)
 */
export async function ingestEmail(env: Env, input: IngestInput): Promise<IngestResult> {
  if (input.raw.byteLength > MAX_EMAIL_BYTES) {
    throw new IngestError(`email exceeds max size ${MAX_EMAIL_BYTES}`, 413);
  }

  const receivedAt = nowIso();
  const receivedDate = new Date(receivedAt);

  // MIME 解析;malformed 邮件降级入库
  let parsed: ParsedMail | null = null;
  try {
    parsed = (await PostalMime.parse(input.raw)) as unknown as ParsedMail;
  } catch {
    parsed = null;
  }

  const toAddresses = (parsed?.to ?? [])
    .map((t) => t.address ?? "")
    .filter(Boolean)
    .concat(parsed ? [] : input.rcptTo ? [input.rcptTo] : []);
  const ccAddresses = (parsed?.cc ?? []).map((t) => t.address ?? "").filter(Boolean);
  // postal-mime 的 from 是 { address, name } 对象(也可能是数组)
  const fromAddress =
    typeof parsed?.from === "string"
      ? parsed.from
      : Array.isArray(parsed?.from)
        ? ((parsed.from[0] as { address?: string } | undefined)?.address ?? input.mailFrom)
        : ((parsed?.from as { address?: string } | undefined)?.address ?? input.mailFrom);

  // Mailbox 关联:任意收件地址精确匹配,否则该域名 catch-all,否则 NULL
  const emailId = ids.email();
  const mailboxId = await resolveMailbox(env, [
    ...toAddresses,
    ...(toAddresses.length === 0 && input.rcptTo ? [input.rcptTo] : []),
  ]);

  // 保留策略:mailbox 覆盖 > 全局默认
  let retention = DEFAULT_RETENTION_SECONDS;
  if (mailboxId) {
    const mb = await env.DB.prepare("SELECT retention_seconds FROM mailboxes WHERE id = ?")
      .bind(mailboxId)
      .first<{ retention_seconds: number | null }>();
    if (mb?.retention_seconds) retention = mb.retention_seconds;
  }
  const expiresAt = new Date(receivedDate.getTime() + retention * 1000).toISOString();

  // R2:raw MIME
  const rawKey = rawStorageKey(emailId, receivedDate);
  await env.R2.put(rawKey, input.raw, {
    httpMetadata: { "Content-Type": "message/rfc822" },
  });

  // 附件(数量/单文件限额,超限附件跳过并继续)
  const attachments = (parsed?.attachments ?? []).slice(0, MAX_ATTACHMENTS);
  const attachmentRows: {
    id: string;
    email_id: string;
    filename: string;
    content_type: string;
    size: number;
    storage_key: string;
    content_id: string | null;
  }[] = [];
  for (const att of attachments) {
    if (att.content.byteLength > MAX_ATTACHMENT_BYTES) continue;
    const attId = ids.attachment();
    const key = attachmentStorageKey(emailId, attId, receivedDate);
    await env.R2.put(key, att.content, {
      httpMetadata: { "Content-Type": att.mimeType || "application/octet-stream" },
    });
    attachmentRows.push({
      id: attId,
      email_id: emailId,
      filename: att.filename || "unnamed",
      content_type: att.mimeType || "application/octet-stream",
      size: att.content.byteLength,
      storage_key: key,
      content_id: att.contentId ?? null,
    });
  }

  // 入库
  await env.DB.prepare(
    `INSERT INTO emails (id, account_id, mailbox_id, message_id, from_address, to_addresses, cc_addresses,
       subject, text_body, html_body, raw_storage_key, size, received_at, expires_at, created_at)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
  )
    .bind(
      emailId,
      input.accountId,
      mailboxId,
      parsed?.messageId ?? null,
      fromAddress,
      JSON.stringify(toAddresses),
      JSON.stringify(ccAddresses),
      parsed?.subject ?? "(unparsable mail)",
      parsed?.text ?? null,
      parsed?.html ?? null,
      rawKey,
      input.raw.byteLength,
      receivedAt,
      expiresAt,
      receivedAt,
    )
    .run();

  for (const row of attachmentRows) {
    await env.DB.prepare(
      "INSERT INTO email_attachments (id, email_id, filename, content_type, size, storage_key, content_id) VALUES (?, ?, ?, ?, ?, ?, ?)",
    )
      .bind(row.id, row.email_id, row.filename, row.content_type, row.size, row.storage_key, row.content_id)
      .run();
  }

  // Webhook 投递(按 邮箱->域名->全局 解析;未配置则不入队;投递失败不影响入库)
  await enqueueDeliveryForEmail(env, emailId, mailboxId);

  return {
    id: emailId,
    mailboxId,
    expiresAt,
    attachmentCount: attachmentRows.length,
    size: input.raw.byteLength,
  };
}

/** # 成员变量: 带自定义 payload 的入站错误(HTTP 状态码) **/
export class IngestError extends Error {
  constructor(
    message: string,
    public status: number,
  ) {
    super(message);
  }
}

/** # 函数
 * 收件地址 -> mailbox 解析:精确匹配优先,其次域名 catch-all(仅 active)
 */
export async function resolveMailbox(env: Env, addresses: string[]): Promise<string | null> {
  for (const addr of addresses) {
    const a = addr.toLowerCase();
    const hit = await env.DB.prepare("SELECT id FROM mailboxes WHERE address = ? AND status = 'active'")
      .bind(a)
      .first<{ id: string }>();
    if (hit) return hit.id;
  }
  for (const addr of addresses) {
    const a = addr.toLowerCase();
    const domain = a.includes("@") ? a.split("@")[1] : null;
    if (!domain) continue;
    const catchAll = await env.DB.prepare(
      `SELECT m.id FROM mailboxes m JOIN domains d ON m.domain_id = d.id
       WHERE m.is_catch_all = 1 AND m.status = 'active' AND d.name = ?`,
    )
      .bind(domain)
      .first<{ id: string }>();
    if (catchAll) return catchAll.id;
  }
  return null;
}