wmailhub/console/src/services/provision.ts

/**
 * 账号接入管线(docs/11,Connect Cloudflare Account):
 *   添加账号(建 cloudflare_accounts 记录)-> Cloudflare OAuth -> 授权回调 ->
 *   Account/Zone 发现 -> 同步账号下全部域名(pending)+ catch-all 邮箱 ->
 *   Email Worker 部署 + WorkerCredential 签发(账号级)。
 *
 * Email Routing/catch-all 规则不随 OAuth 自动执行,而是在域名列表中按域名单独触发
 * (setupDomainEmail);一个账号只部署一个 Email Worker(幂等,重复执行不产生重复资源)。
 * 账号级步骤状态写入 cloudflare_accounts.provision_steps。
 */

import type { Env } from "../env.ts";
import { getCFApi, type CFApi } from "./cfapi.ts";
import { encryptSecret, decryptSecret, decryptConnectionToken } from "../db.ts";
import { ids, sha256Hex, DEFAULT_RETENTION_SECONDS } from "../../../shared/src/index.ts";
import { nowIso } from "../util.ts";

/** # 成员变量: 账号接入步骤定义(顺序即展示顺序) **/
export const PROVISION_STEPS = [
  { key: "oauth", label: "Cloudflare OAuth 授权" },
  { key: "discover", label: "Account / Zone 发现" },
  { key: "domains", label: "域名与 catch-all 邮箱同步" },
  { key: "worker", label: "Email Worker 部署与 Worker 凭据" },
] as const;

/** # 成员变量: 步骤状态 **/
export type StepStatus = "pending" | "running" | "done" | "error";

/** # 函数
 * 初始化步骤状态 JSON
 */
export function initialSteps(): Record<string, { status: StepStatus; error?: string }> {
  const out: Record<string, { status: StepStatus; error?: string }> = {};
  for (const s of PROVISION_STEPS) out[s.key] = { status: "pending" };
  return out;
}

/** # 函数
 * 账号的 Email Worker 脚本名(幂等:同一账号固定同名;同时作为 Worker -> Console 的身份 ID)
 */
export function emailWorkerNameFor(accountId: string): string {
  return `wmailhub-email-${accountId.replace(/[^a-z0-9-]/gi, "-").toLowerCase()}`;
}

/** # 函数
 * 添加账号:创建 cloudflare_accounts 记录(等待 OAuth 授权)。
 */
export async function createAccount(
  env: Env,
  opts: { name: string },
): Promise<{ accountId: string }> {
  const accountId = ids.account();
  const now = nowIso();
  await env.DB.prepare(
    `INSERT INTO cloudflare_accounts (id, name, cloudflare_account_id, worker_id, status,
       provision_steps, created_at, updated_at)
     VALUES (?, ?, NULL, NULL, 'pending_oauth', ?, ?, ?)`,
  )
    .bind(accountId, opts.name, JSON.stringify(initialSteps()), now, now)
    .run();
  return { accountId };
}

/** # 函数
 * 确保账号存在可用的 Worker credential:
 * 已有 active credential 直接复用(不泄露 secret);否则签发全新 credential。
 * 返回 secret 明文仅供部署注入(本地测试/重新部署 Worker 时使用)。
 */
export async function ensureWorkerCredential(
  env: Env,
  accountId: string,
  workerId: string,
): Promise<{ credentialId: string; secret: string | null }> {
  const now = nowIso();
  const existing = await env.DB.prepare(
    "SELECT id FROM worker_credentials WHERE cloudflare_account_id = ? AND worker_id = ? AND status = 'active'",
  )
    .bind(accountId, workerId)
    .first<{ id: string }>();
  if (existing) return { credentialId: existing.id, secret: null };

  const secret = "whs_" + crypto.randomUUID().replace(/-/g, "") + crypto.randomUUID().replace(/-/g, "");
  const credentialId = ids.credential();
  await env.DB.prepare(
    "INSERT INTO worker_credentials (id, cloudflare_account_id, worker_id, secret_hash, secret_enc, secret_last4, status, created_at) VALUES (?, ?, ?, ?, ?, ?, 'active', ?)",
  )
    .bind(
      credentialId,
      accountId,
      workerId,
      await sha256Hex(secret),
      await encryptSecret(env, secret),
      secret.slice(-4),
      now,
    )
    .run();
  return { credentialId, secret };
}

/** # 成员变量: 授权完成后的完整接入流程参数 **/
export interface ProvisionInput {
  accountId: string;
  accessToken: string;
  cfAccountId?: string;
}

/** # 函数
 * OAuth 授权成功后的自动接入主流程:发现账号 + 同步全部域名(pending)+ catch-all 邮箱 +
 * 部署账号级 Email Worker + 签发 Worker credential。幂等,可重复调用;返回是否全部成功。
 */
export async function provisionAccount(env: Env, input: ProvisionInput): Promise<{ ok: boolean; error?: string }> {
  const cf: CFApi = getCFApi(env);
  const account = await env.DB.prepare("SELECT id, name, status FROM cloudflare_accounts WHERE id = ?")
    .bind(input.accountId)
    .first<{ id: string; name: string; status: string }>();
  if (!account || account.status === "revoked") return { ok: false, error: "account not found" };

  const steps = await loadSteps(env, input.accountId);
  const mark = async (key: string, status: StepStatus, error?: string) => {
    steps[key] = { status, error };
    await env.DB.prepare("UPDATE cloudflare_accounts SET provision_steps = ?, updated_at = ? WHERE id = ?")
      .bind(JSON.stringify(steps), nowIso(), input.accountId)
      .run();
  };

  try {
    await env.DB.prepare("UPDATE cloudflare_accounts SET status = 'provisioning', updated_at = ? WHERE id = ?")
      .bind(nowIso(), input.accountId)
      .run();

    // 1. OAuth 授权
    await mark("oauth", "done");

    // 2. Account/Zone 发现
    await mark("discover", "running");
    const accounts = await cf.listAccounts(input.accessToken);
    if (accounts.length === 0) throw new Error("no cloudflare account accessible");
    const cfAccount = input.cfAccountId
      ? accounts.find((a) => a.id === input.cfAccountId) ?? accounts[0]
      : accounts[0];
    // 自动创建的账号以 Cloudflare 账户名命名
    const newName = account.name.startsWith("Cloudflare 账号") ? cfAccount.name : account.name;
    await env.DB.prepare(
      "UPDATE cloudflare_accounts SET cloudflare_account_id = ?, name = ?, updated_at = ? WHERE id = ?",
    )
      .bind(cfAccount.id, newName, nowIso(), input.accountId)
      .run();
    const zones = await cf.listZones(input.accessToken, cfAccount.id);
    await mark("discover", "done");

    // 3. 同步全部域名(pending,等待逐域名「设置 Email Workers」)+ catch-all 邮箱
    await mark("domains", "running");
    for (const zone of zones) {
      await upsertDomain(env, {
        cloudflareAccountId: cfAccount.id,
        accountId: input.accountId,
        name: zone.name,
        zoneId: zone.id,
        workerId: null,
      });
    }
    const domainRows = await env.DB.prepare("SELECT id, name FROM domains WHERE account_id = ?")
      .bind(input.accountId)
      .all<{ id: string; name: string }>();
    for (const d of domainRows.results) {
      await upsertMailbox(env, {
        domainId: d.id,
        address: `*@${d.name}`,
        isCatchAll: true,
      });
    }
    await mark("domains", "done");

    // 4. 部署账号级 Email Worker + 签发 Worker credential(docs/11 §14)
    await mark("worker", "running");
    await deployAccountWorker(env, input.accountId, input.accessToken);
    await mark("worker", "done");

    await env.DB.prepare("UPDATE cloudflare_accounts SET status = 'active', updated_at = ? WHERE id = ?")
      .bind(nowIso(), input.accountId)
      .run();
    return { ok: true };
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    // 标记第一个未完成步骤为 error
    for (const s of PROVISION_STEPS) {
      if (steps[s.key]?.status === "running") {
        await mark(s.key, "error", message);
        break;
      }
    }
    await env.DB.prepare("UPDATE cloudflare_accounts SET status = 'error', updated_at = ? WHERE id = ?")
      .bind(nowIso(), input.accountId)
      .run();
    return { ok: false, error: message };
  }
}

/** # 函数
 * 部署(或更新)账号级 Email Worker:
 * 确保 Worker credential -> 渲染脚本(Console 地址 + Worker ID + secret)-> 上传 Cloudflare。
 * credential 已存在时复用旧 secret(重新上传脚本不轮换密钥);secret 仅在首次签发时用于部署。
 */
export async function deployAccountWorker(env: Env, accountId: string, accessToken: string): Promise<string> {
  const account = await env.DB.prepare("SELECT id, cloudflare_account_id FROM cloudflare_accounts WHERE id = ?")
    .bind(accountId)
    .first<{ id: string; cloudflare_account_id: string | null }>();
  if (!account?.cloudflare_account_id) throw new Error("account not connected via OAuth");

  const workerId = await getOrCreateWorkerId(env, accountId);
  const { secret } = await ensureWorkerCredential(env, accountId, workerId);
  if (!secret) {
    // 已有 credential:重新上传脚本时需要原 secret,先解密
    const row = await env.DB.prepare(
      "SELECT secret_enc FROM worker_credentials WHERE cloudflare_account_id = ? AND worker_id = ? AND status = 'active' ORDER BY created_at DESC LIMIT 1",
    )
      .bind(accountId, workerId)
      .first<{ secret_enc: string }>();
    const decrypted = row ? await decryptSecret(env, row.secret_enc) : null;
    await putWorkerScript(env, accessToken, account.cloudflare_account_id, workerId, decrypted ?? "");
    return workerId;
  }
  await putWorkerScript(env, accessToken, account.cloudflare_account_id, workerId, secret);
  return workerId;
}

/** # 函数
 * 账号的 worker_id(幂等:无则生成并写入 cloudflare_accounts.worker_id)
 */
async function getOrCreateWorkerId(env: Env, accountId: string): Promise<string> {
  const row = await env.DB.prepare("SELECT worker_id FROM cloudflare_accounts WHERE id = ?")
    .bind(accountId)
    .first<{ worker_id: string | null }>();
  if (row?.worker_id) return row.worker_id;
  const workerId = emailWorkerNameFor(accountId);
  await env.DB.prepare("UPDATE cloudflare_accounts SET worker_id = ?, updated_at = ? WHERE id = ?")
    .bind(workerId, nowIso(), accountId)
    .run();
  return workerId;
}

/** # 函数
 * 上传 Email Worker 脚本(Console 地址取 PUBLIC_BASE_URL,与 OAuth redirect 同源)
 */
async function putWorkerScript(
  env: Env,
  accessToken: string,
  cfAccountId: string,
  workerId: string,
  workerSecret: string,
): Promise<void> {
  const cf: CFApi = getCFApi(env);
  const consoleUrl = env.PUBLIC_BASE_URL ?? "";
  if (!consoleUrl) throw new Error("console public url unknown (PUBLIC_BASE_URL)");
  await cf.putEmailWorker({ accessToken, accountId: cfAccountId, scriptName: workerId, consoleUrl, workerId, workerSecret });
}

/** # 函数
 * 为单个域名配置 Email Workers:确保账号级 Worker 部署 -> 开启 Email Routing ->
 * catch-all 规则指向 Worker -> 域名置 active。幂等,失败信息写入 domains.setup_error。
 */
export async function setupDomainEmail(env: Env, domainId: string): Promise<{ ok: boolean; error?: string }> {
  const domain = await env.DB.prepare(
    "SELECT id, name, zone_id, account_id FROM domains WHERE id = ?",
  )
    .bind(domainId)
    .first<{ id: string; name: string; zone_id: string | null; account_id: string | null }>();
  if (!domain) return { ok: false, error: "domain not found" };
  if (!domain.account_id || !domain.zone_id) return { ok: false, error: "domain not bound to a Cloudflare zone" };

  const account = await env.DB.prepare(
    "SELECT id, cloudflare_account_id FROM cloudflare_accounts WHERE id = ? AND status != 'revoked'",
  )
    .bind(domain.account_id)
    .first<{ id: string; cloudflare_account_id: string | null }>();
  if (!account?.cloudflare_account_id) return { ok: false, error: "account not connected via OAuth" };

  const accessToken = await decryptConnectionToken(env);
  if (!accessToken) return { ok: false, error: "no active Cloudflare OAuth connection" };

  const cf: CFApi = getCFApi(env);
  try {
    const workerId = await deployAccountWorker(env, account.id, accessToken);
    await cf.enableEmailRouting({ accessToken, zoneId: domain.zone_id });
    await cf.setCatchAllRule({ accessToken, zoneId: domain.zone_id, workerName: workerId });
    await env.DB.prepare(
      "UPDATE domains SET worker_id = ?, status = 'active', setup_error = NULL, updated_at = ? WHERE id = ?",
    )
      .bind(workerId, nowIso(), domainId)
      .run();
    return { ok: true };
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    await env.DB.prepare("UPDATE domains SET status = 'error', setup_error = ?, updated_at = ? WHERE id = ?")
      .bind(message, nowIso(), domainId)
      .run();
    return { ok: false, error: message };
  }
}

/** # 函数
 * 读取并解析账号的步骤状态
 */
export async function loadSteps(env: Env, accountId: string) {
  const row = await env.DB.prepare("SELECT provision_steps FROM cloudflare_accounts WHERE id = ?")
    .bind(accountId)
    .first<{ provision_steps: string }>();
  if (!row?.provision_steps) return initialSteps();
  try {
    return JSON.parse(row.provision_steps) as Record<string, { status: StepStatus; error?: string }>;
  } catch {
    return initialSteps();
  }
}

/** # 函数
 * upsert Domain(按 cf account+name 幂等)。无 workerId 时保持/置为 pending(待配置 Email Workers)。
 */
export async function upsertDomain(
  env: Env,
  d: {
    cloudflareAccountId: string;
    accountId: string;
    name: string;
    zoneId: string;
    workerId: string | null;
  },
): Promise<string> {
  const existing = await env.DB.prepare(
    "SELECT id, worker_id FROM domains WHERE cloudflare_account_id = ? AND name = ?",
  )
    .bind(d.cloudflareAccountId, d.name)
    .first<{ id: string; worker_id: string | null }>();
  const now = nowIso();
  if (existing) {
    // 已配置过 Worker 的域名不回退状态;仅当新配置带 worker 时更新
    if (existing.worker_id || !d.workerId) {
      await env.DB.prepare("UPDATE domains SET account_id = ?, zone_id = ?, updated_at = ? WHERE id = ?")
        .bind(d.accountId, d.zoneId, now, existing.id)
        .run();
    } else {
      await env.DB.prepare(
        "UPDATE domains SET account_id = ?, zone_id = ?, worker_id = ?, status = 'active', updated_at = ? WHERE id = ?",
      )
        .bind(d.accountId, d.zoneId, d.workerId, now, existing.id)
        .run();
    }
    return existing.id;
  }
  const id = ids.domain();
  await env.DB.prepare(
    "INSERT INTO domains (id, cloudflare_account_id, account_id, name, status, worker_id, zone_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
  )
    .bind(id, d.cloudflareAccountId, d.accountId, d.name, d.workerId ? "active" : "pending", d.workerId, d.zoneId, now, now)
    .run();
  return id;
}

/** # 函数
 * upsert Mailbox(按 address 幂等)
 */
export async function upsertMailbox(
  env: Env,
  m: { domainId: string; address: string; isCatchAll?: boolean; retentionSeconds?: number },
): Promise<string> {
  const existing = await env.DB.prepare("SELECT id FROM mailboxes WHERE address = ?")
    .bind(m.address.toLowerCase())
    .first<{ id: string }>();
  const now = nowIso();
  if (existing) return existing.id;
  const id = ids.mailbox();
  await env.DB.prepare(
    "INSERT INTO mailboxes (id, domain_id, address, status, retention_seconds, is_catch_all, created_at, updated_at) VALUES (?, ?, ?, 'active', ?, ?, ?, ?)",
  )
    .bind(
      id,
      m.domainId,
      m.address.toLowerCase(),
      m.retentionSeconds ?? DEFAULT_RETENTION_SECONDS,
      m.isCatchAll ? 1 : 0,
      now,
      now,
    )
    .run();
  return id;
}