wmailhub/console/src/services/webhook_delivery.ts

/**
 * Webhook 投递(docs/06、docs/07,v1.1.0 三级模型):
 * - 配置解析优先级:邮箱 -> 域名 -> 全局(settings),命中即止
 * - 每级配置:{ url, headers(自定义请求头 JSON), key }
 *   key 默认以 `Authorization: Bearer <key>` 携带(headers 里显式给了 Authorization 则不覆盖),
 *   并用于投递签名:signature = HMAC-SHA256(key, `${timestamp}.${rawBody}`)
 * - payload 按文档 JSON 结构(event=email.received)
 * - 2xx 成功;失败重试(指数退避);投递行在入队时固化目标配置,重试不跟随后续配置变更
 * - endpoint 每次投递前重过 SSRF 校验(静态 + DNS 解析,本地运行时)
 */

import type { Env } from "../env.ts";
import {
  hmacSha256Hex,
  aesEncrypt,
  aesDecrypt,
  assertWebhookUrlSafe,
  ids,
  type HostResolver,
  WEBHOOK_HEADERS,
  WEBHOOK_TIMEOUT_MS,
  WEBHOOK_MAX_ATTEMPTS,
  WEBHOOK_RETRY_BASE_MS,
} from "../../../shared/src/index.ts";
import { encryptSecret, decryptSecret, getAesKey, getSetting, setSetting } from "../db.ts";
import { nowIso } from "../util.ts";

/** # 成员变量: 单级 Webhook 配置(解析结果) **/
export interface WebhookConfig {
  /** 配置来源层级 **/
  targetType: "mailbox" | "domain" | "global";
  /** 目标 URL **/
  url: string;
  /** 自定义请求头 **/
  headers: Record<string, string>;
  /** key 明文(签名与 Authorization;可能为空串) **/
  key: string;
}

/** # 成员变量: settings 表中全局 Webhook 的存储键 **/
export const GLOBAL_WEBHOOK_SETTING = "global_webhook";

/** # 函数
 * 校验并规整自定义请求头:键名限 [A-Za-z0-9-],值转字符串;非法抛错
 * 参数: headers - 前端传入的对象
 * 返回值: 规整后的头(空对象返回 {})
 */
export function normalizeWebhookHeaders(headers: unknown): Record<string, string> {
  if (headers === undefined || headers === null) return {};
  if (typeof headers !== "object" || Array.isArray(headers)) throw new Error("headers must be an object");
  const out: Record<string, string> = {};
  for (const [k, v] of Object.entries(headers as Record<string, unknown>)) {
    if (!/^[A-Za-z0-9-]{1,64}$/.test(k)) throw new Error(`invalid header name: ${k}`);
    if (v === undefined || v === null) continue;
    if (typeof v !== "string" && typeof v !== "number" && typeof v !== "boolean") {
      throw new Error(`invalid header value for: ${k}`);
    }
    out[k] = String(v);
  }
  return out;
}

function safeParseHeaders(s: string | null): Record<string, string> {
  if (!s) return {};
  try {
    return normalizeWebhookHeaders(JSON.parse(s));
  } catch {
    return {};
  }
}

/** # 成员变量: 邮箱/域名行的 Webhook 列形态 **/
interface WebhookColumns {
  webhook_url: string | null;
  webhook_headers: string | null;
  webhook_key_enc: string | null;
}

/** # 函数
 * 三级 Webhook 解析:邮箱 -> 域名 -> 全局(settings),命中第一级即返回
 * 参数: env - 运行环境;mailboxId - 邮件命中的邮箱(可为 null)
 * 返回值: 命中的配置;三级都未配置时 null(不投递)
 */
export async function resolveWebhookConfig(env: Env, mailboxId: string | null): Promise<WebhookConfig | null> {
  if (mailboxId) {
    const mb = await env.DB.prepare(
      "SELECT webhook_url, webhook_headers, webhook_key_enc, domain_id FROM mailboxes WHERE id = ?",
    )
      .bind(mailboxId)
      .first<WebhookColumns & { domain_id: string }>();
    if (mb?.webhook_url) {
      return {
        targetType: "mailbox",
        url: mb.webhook_url,
        headers: safeParseHeaders(mb.webhook_headers),
        key: mb.webhook_key_enc ? await decryptSecret(env, mb.webhook_key_enc) : "",
      };
    }
    if (mb?.domain_id) {
      const dom = await env.DB.prepare(
        "SELECT webhook_url, webhook_headers, webhook_key_enc FROM domains WHERE id = ?",
      )
        .bind(mb.domain_id)
        .first<WebhookColumns>();
      if (dom?.webhook_url) {
        return {
          targetType: "domain",
          url: dom.webhook_url,
          headers: safeParseHeaders(dom.webhook_headers),
          key: dom.webhook_key_enc ? await decryptSecret(env, dom.webhook_key_enc) : "",
        };
      }
    }
  }
  const raw = await getSetting(env, GLOBAL_WEBHOOK_SETTING);
  if (!raw) return null;
  try {
    const cfg = JSON.parse(raw) as { url?: string; headers?: Record<string, string> };
    if (!cfg.url) return null;
    return { targetType: "global", url: cfg.url, headers: cfg.headers ?? {}, key: await getGlobalWebhookKey(env) };
  } catch {
    return null;
  }
}

/** # 函数
 * 读取全局 Webhook key(密文存 settings)
 */
export async function getGlobalWebhookKey(env: Env): Promise<string> {
  const enc = await getSetting(env, "global_webhook_key_enc");
  return enc ? await decryptSecret(env, enc) : "";
}

/** # 函数
 * 保存全局 Webhook 配置到 settings(url/headers 明文 JSON,key 加密)
 * 参数: cfg.url 为空串表示停用全局 Webhook;key 为 undefined 表示保留旧 key
 */
export async function saveGlobalWebhook(
  env: Env,
  cfg: { url: string; headers?: Record<string, string>; key?: string },
): Promise<void> {
  await setSetting(
    env,
    GLOBAL_WEBHOOK_SETTING,
    JSON.stringify({ url: cfg.url, headers: cfg.headers ?? {} }),
  );
  if (cfg.key !== undefined) {
    if (cfg.key === "") {
      await env.DB.prepare("DELETE FROM settings WHERE key = 'global_webhook_key_enc'").run();
    } else {
      await setSetting(env, "global_webhook_key_enc", await encryptSecret(env, cfg.key));
    }
  }
}

/** # 函数
 * 保存邮箱/域名级 Webhook 配置(url 为空串表示清除该级配置)
 */
export async function saveScopedWebhook(
  env: Env,
  table: "mailboxes" | "domains",
  id: string,
  cfg: { url: string; headers?: Record<string, string>; key?: string },
): Promise<void> {
  if (!["mailboxes", "domains"].includes(table)) throw new Error("bad table");
  const headersJson = JSON.stringify(cfg.headers ?? {});
  const keyEnc = cfg.key !== undefined ? (cfg.key === "" ? null : await encryptSecret(env, cfg.key)) : undefined;
  if (!cfg.url) {
    // 清除该级配置
    await env.DB.prepare(
      `UPDATE ${table} SET webhook_url = NULL, webhook_headers = NULL, webhook_key_enc = NULL, updated_at = ? WHERE id = ?`,
    )
      .bind(nowIso(), id)
      .run();
    return;
  }
  if (keyEnc === undefined) {
    await env.DB.prepare(
      `UPDATE ${table} SET webhook_url = ?, webhook_headers = ?, updated_at = ? WHERE id = ?`,
    )
      .bind(cfg.url, headersJson, nowIso(), id)
      .run();
  } else {
    await env.DB.prepare(
      `UPDATE ${table} SET webhook_url = ?, webhook_headers = ?, webhook_key_enc = ?, updated_at = ? WHERE id = ?`,
    )
      .bind(cfg.url, headersJson, keyEnc, nowIso(), id)
      .run();
  }
}

/** # 函数
 * webhook key 入库密文(供入队时固化)
 */
async function encryptWebhookKey(env: Env, key: string): Promise<string | null> {
  return key ? aesEncrypt(await getAesKey(env), key) : null;
}

/** # 函数
 * 解密投递行固化的 key
 */
async function decryptWebhookKey(env: Env, blob: string | null): Promise<string> {
  return blob ? await aesDecrypt(await getAesKey(env), blob) : "";
}

/** # 成员变量: Webhook 投递 payload 结构(docs/07) **/
export function buildWebhookPayload(email: {
  id: string;
  mailboxAddress: string | null;
  fromAddress: string | null;
  toAddresses: string;
  subject: string | null;
  textBody: string | null;
  receivedAt: string;
}): Record<string, unknown> {
  return {
    event: "email.received",
    id: email.id,
    mailbox: email.mailboxAddress,
    from: email.fromAddress,
    to: safeParseArray(email.toAddresses),
    subject: email.subject,
    text: email.textBody,
    received_at: email.receivedAt,
  };
}

function safeParseArray(s: string | null): string[] {
  if (!s) return [];
  try {
    const v = JSON.parse(s);
    return Array.isArray(v) ? v.map(String) : [];
  } catch {
    return [];
  }
}

/** # 函数
 * 执行一次投递;返回是否成功与错误信息(不修改库)
 * 参数: config - 固化在投递行上的目标配置
 */
export async function attemptDelivery(
  config: { url: string; headers: Record<string, string>; key: string },
  deliveryId: string,
  payload: Record<string, unknown>,
  resolve?: HostResolver,
  allowPrivateHosts = false,
): Promise<{ ok: boolean; status?: number; error?: string }> {
  const unsafe = await assertWebhookUrlSafe(config.url, resolve, allowPrivateHosts);
  if (unsafe) return { ok: false, error: `ssrf guard: ${unsafe}` };

  const body = JSON.stringify(payload);
  const timestamp = String(Math.floor(Date.now() / 1000));

  // 组装请求头:标准头 -> key 的默认 Authorization -> 用户自定义头(可覆盖标准头,但不能去 Content-Type)
  // 未配置 key 时无法协商签名密钥,跳过签名头(仅保留事件标识)
  const headers: Record<string, string> = {
    "Content-Type": "application/json",
    [WEBHOOK_HEADERS.event]: "email.received",
    [WEBHOOK_HEADERS.delivery]: deliveryId,
  };
  if (config.key) {
    headers[WEBHOOK_HEADERS.timestamp] = timestamp;
    headers[WEBHOOK_HEADERS.signature] = await hmacSha256Hex(config.key, `${timestamp}.${body}`);
  }
  const customHasAuth = Object.keys(config.headers).some((k) => k.toLowerCase() === "authorization");
  if (config.key && !customHasAuth) headers["Authorization"] = `Bearer ${config.key}`;
  for (const [k, v] of Object.entries(config.headers)) {
    if (k.toLowerCase() === "content-type") continue;
    headers[k] = v;
  }

  try {
    const resp = await fetch(config.url, {
      method: "POST",
      headers,
      body,
      signal: AbortSignal.timeout(WEBHOOK_TIMEOUT_MS),
    });
    if (resp.ok) return { ok: true, status: resp.status };
    return { ok: false, status: resp.status, error: `endpoint responded ${resp.status}` };
  } catch (err) {
    return { ok: false, error: err instanceof Error ? err.message : String(err) };
  }
}

/** # 函数
 * 计算第 attempt 次失败后的下次重试时间(指数退避)
 */
export function nextRetryIso(attemptCount: number): string {
  const delay = WEBHOOK_RETRY_BASE_MS * 2 ** attemptCount;
  return new Date(Date.now() + delay).toISOString();
}

/** # 成员变量: 投递循环的行结构 **/
interface DueDeliveryRow {
  delivery_id: string;
  url: string;
  headers: string | null;
  key_enc: string | null;
  attempt_count: number;
  email_id: string;
  mailbox_id: string | null;
  from_address: string | null;
  to_addresses: string;
  subject: string | null;
  text_body: string | null;
  received_at: string;
  expires_at: string;
}

/** # 函数
 * Cron 驱动的投递循环:取到期 delivery 逐个投递并落状态。
 * 返回处理统计(供日志与 admin 接口)。
 */
export async function processDueDeliveries(
  env: Env,
  resolve?: HostResolver,
  limit = 20,
  allowPrivateHosts = false,
): Promise<{ processed: number; success: number; failed: number; exhausted: number }> {
  const now = nowIso();
  const due = await env.DB.prepare(
    `SELECT d.id AS delivery_id, d.url, d.headers, d.key_enc, d.attempt_count,
            e.id AS email_id, e.mailbox_id, e.from_address, e.to_addresses, e.subject, e.text_body,
            e.received_at, e.expires_at
     FROM webhook_deliveries d
     JOIN emails e ON e.id = d.email_id
     WHERE d.status IN ('pending', 'failed')
       AND d.attempt_count < ?
       AND (d.next_retry_at IS NULL OR d.next_retry_at <= ?)
     ORDER BY d.created_at
     LIMIT ?`,
  )
    .bind(WEBHOOK_MAX_ATTEMPTS, now, limit)
    .all<DueDeliveryRow>();

  let success = 0;
  let failed = 0;
  let exhausted = 0;
  for (const row of due.results) {
    if (row.expires_at <= now) {
      // 邮件已过期:投递失去意义,直接结束(数据最小化)
      await env.DB.prepare("UPDATE webhook_deliveries SET status = 'exhausted', last_error = 'email expired' WHERE id = ?")
        .bind(row.delivery_id)
        .run();
      exhausted++;
      continue;
    }

    const mailbox = row.mailbox_id
      ? (
          await env.DB.prepare("SELECT address FROM mailboxes WHERE id = ?")
            .bind(row.mailbox_id)
            .first<{ address: string }>()
        )?.address ?? null
      : null;
    const payload = buildWebhookPayload({
      id: row.email_id,
      mailboxAddress: mailbox,
      fromAddress: row.from_address,
      toAddresses: row.to_addresses,
      subject: row.subject,
      textBody: row.text_body,
      receivedAt: row.received_at,
    });
    const result = await attemptDelivery(
      { url: row.url, headers: safeParseHeaders(row.headers), key: await decryptWebhookKey(env, row.key_enc) },
      row.delivery_id,
      payload,
      resolve,
      allowPrivateHosts,
    );
    const attempts = row.attempt_count + 1;
    if (result.ok) {
      await env.DB.prepare(
        "UPDATE webhook_deliveries SET status = 'success', attempt_count = ?, delivered_at = ?, last_error = NULL, next_retry_at = NULL WHERE id = ?",
      )
        .bind(attempts, nowIso(), row.delivery_id)
        .run();
      success++;
    } else if (attempts >= WEBHOOK_MAX_ATTEMPTS) {
      await env.DB.prepare(
        "UPDATE webhook_deliveries SET status = 'exhausted', attempt_count = ?, last_error = ? WHERE id = ?",
      )
        .bind(attempts, result.error ?? "unknown", row.delivery_id)
        .run();
      exhausted++;
    } else {
      await env.DB.prepare(
        "UPDATE webhook_deliveries SET status = 'failed', attempt_count = ?, last_error = ?, next_retry_at = ? WHERE id = ?",
      )
        .bind(attempts, result.error ?? "unknown", nextRetryIso(attempts), row.delivery_id)
        .run();
      failed++;
    }
  }
  return { processed: due.results.length, success, failed, exhausted };
}

/** # 函数
 * 入队:为邮件按三级解析结果创建一条投递记录(未配置任何 Webhook 则不入队)
 * 返回值: 命中的层级或 null
 */
export async function enqueueDeliveryForEmail(
  env: Env,
  emailId: string,
  mailboxId: string | null,
): Promise<WebhookConfig["targetType"] | null> {
  const config = await resolveWebhookConfig(env, mailboxId);
  if (!config) return null;
  await env.DB.prepare(
    `INSERT INTO webhook_deliveries (id, email_id, target_type, url, headers, key_enc, status, attempt_count, next_retry_at, created_at)
     VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?)`,
  )
    .bind(
      ids.delivery(),
      emailId,
      config.targetType,
      config.url,
      JSON.stringify(config.headers),
      await encryptWebhookKey(env, config.key),
      nowIso(),
      nowIso(),
    )
    .run();
  return config.targetType;
}