UHD_Media_Tool/packages/tasks/src/op-ledger.ts

import { randomUUID } from "node:crypto";
import { type Kysely, sql, type Transaction } from "kysely";
import { FencingError } from "./errors.ts";
import type { JsonValue } from "./types.ts";

// biome-ignore lint/suspicious/noExplicitAny: 本包只用原生 SQL 访问自有表,不依赖业务库类型
type AnyDb = Kysely<any>;
// biome-ignore lint/suspicious/noExplicitAny: 同上
type AnyTrx = Transaction<any>;

export interface AttemptInfo {
  opKey: string;
  attemptToken: string;
  /** 第几次尝试(1 起) */
  attempt: number;
  /** 之前有 attempt 但没有结果:先对账外部状态(qB 按 hash、文件按路径+大小/hash),再决定收尾还是重做 */
  resumed: boolean;
}

export interface WithAttemptTokenOptions<T> {
  runId?: string;
  /** 缺省新生成;每次调用都是一个新 attempt,会覆盖旧 token */
  attemptToken?: string;
  /** 在同一事务里、fencing 校验通过后提交业务状态;抛错则整体回滚 */
  commit?: (trx: AnyTrx, result: T, info: AttemptInfo) => Promise<void>;
}

/**
 * 操作账本 fencing(ADR-002 提醒 3):
 * 1. 已有 succeeded 结果 → 直接返回,不重做;
 * 2. 否则写入新 attempt_token(旧 attempt 自此失效)并执行 fn;
 * 3. 提交时 `SELECT … FOR UPDATE` 校验 token 仍是自己,不是则抛 FencingError(租约已丢,拒绝提交)。
 */
export async function withAttemptToken<T extends JsonValue>(
  db: AnyDb,
  opKey: string,
  fn: (info: AttemptInfo) => Promise<T>,
  opts: WithAttemptTokenOptions<T> = {},
): Promise<T> {
  const token = opts.attemptToken ?? randomUUID();
  const claimed = await sql<{ attempts: number }>`
    INSERT INTO tasks_op_ledger (op_key, run_id, attempt_token, attempts, status)
    VALUES (${opKey}, ${opts.runId ?? null}, ${token}, 1, 'started')
    ON CONFLICT (op_key) DO UPDATE
      SET attempt_token = EXCLUDED.attempt_token, attempts = tasks_op_ledger.attempts + 1,
          started_at = now(), run_id = COALESCE(EXCLUDED.run_id, tasks_op_ledger.run_id)
      WHERE tasks_op_ledger.status <> 'succeeded'
    RETURNING attempts`.execute(db);

  const row = claimed.rows[0];
  if (!row) {
    const done = await sql<{ result: T }>`SELECT result FROM tasks_op_ledger WHERE op_key = ${opKey}`.execute(db);
    return done.rows[0]!.result;
  }

  const info: AttemptInfo = { opKey, attemptToken: token, attempt: row.attempts, resumed: row.attempts > 1 };
  let result: T;
  try {
    result = await fn(info);
  } catch (err) {
    await sql`UPDATE tasks_op_ledger SET last_error = ${String((err as Error)?.message ?? err)}
      WHERE op_key = ${opKey} AND attempt_token = ${token}`.execute(db);
    throw err;
  }

  await db.transaction().execute(async (trx) => {
    await assertAttemptToken(trx, opKey, token);
    if (opts.commit) await opts.commit(trx, result, info);
    await sql`UPDATE tasks_op_ledger
      SET status = 'succeeded', result = ${JSON.stringify(result)}::jsonb, completed_at = now(), last_error = NULL
      WHERE op_key = ${opKey}`.execute(trx);
  });
  return result;
}

/** 在业务事务里校验 token 并锁住账本行;用于 withAttemptToken 之外的业务提交 */
export async function assertAttemptToken(trx: AnyTrx, opKey: string, attemptToken: string): Promise<void> {
  const cur = await sql<{ attempt_token: string; status: string }>`
    SELECT attempt_token, status FROM tasks_op_ledger WHERE op_key = ${opKey} FOR UPDATE`.execute(trx);
  const r = cur.rows[0];
  if (!r || r.attempt_token !== attemptToken || r.status === "succeeded") throw new FencingError(opKey);
}

export interface OpLedgerRow {
  opKey: string;
  runId: string | null;
  attemptToken: string;
  attempts: number;
  status: "started" | "succeeded";
  result: unknown;
  lastError: string | null;
}

export async function getOperation(db: AnyDb, opKey: string): Promise<OpLedgerRow | null> {
  const r = await sql<{
    op_key: string;
    run_id: string | null;
    attempt_token: string;
    attempts: number;
    status: "started" | "succeeded";
    result: unknown;
    last_error: string | null;
  }>`SELECT * FROM tasks_op_ledger WHERE op_key = ${opKey}`.execute(db);
  const row = r.rows[0];
  return row
    ? {
        opKey: row.op_key,
        runId: row.run_id,
        attemptToken: row.attempt_token,
        attempts: row.attempts,
        status: row.status,
        result: row.result,
        lastError: row.last_error,
      }
    : null;
}