UHD_Media_Tool/packages/tasks/src/outbox-relay.ts

import { randomUUID } from "node:crypto";
import { type Kysely, sql } from "kysely";
import { createLogger, type LogSink } from "./logger.ts";
import type { TaskEngine } from "./types.ts";

// biome-ignore lint/suspicious/noExplicitAny: 只用原生 SQL 读写 outbox_event
type AnyDb = Kysely<any>;

export interface OutboxEvent {
  id: string;
  type: string;
  payload: unknown;
  at: Date;
}

export interface OutboxRoute {
  task: string;
  input: unknown;
  /** 缺省用事件 payload.traceId,再缺省生成新的 */
  traceId?: string;
  priority?: number;
}

/** 返回 null 表示该事件无需建任务,直接标记已投递 */
export type OutboxRouteFn = (event: OutboxEvent) => OutboxRoute | null;

export interface OutboxRelayOptions {
  db: AnyDb;
  engine: Pick<TaskEngine, "enqueue">;
  /** 事件 type → 任务映射;只认领 routes 里列出的 type */
  routes: Record<string, OutboxRouteFn>;
  batchSize?: number;
  intervalMs?: number;
  /** 失败达到该次数后置 status='dead'(默认 10);其间按 2^attempts 秒退避,上限 5 分钟 */
  maxAttempts?: number;
  logSink?: LogSink;
  /**
   * 镜像到实时总线(SSE):列出的 type 在投递成功、认领事务提交后逐条交给 publish(按 at, id 顺序)。
   * 只在 mirror 里、没有路由的 type 同样会被认领并直接标记已投递。publish 失败只记日志,不影响投递状态
   * (实时推送是尽力而为,前端断线或错过时靠重新拉快照)。
   */
  mirror?: { types: readonly string[]; publish: (event: OutboxEvent) => unknown };
}

/**
 * Outbox 中继:业务事务写 outbox_event,中继在自己的事务里
 * `SELECT … FOR UPDATE SKIP LOCKED` 认领 status='pending' 且 next_attempt_at 已到的行
 * → enqueue(幂等键 = 行的 idempotency_key,缺省 outbox:<行 id>)→ 标记 dispatched + published_at;
 * 失败则 attempts+1、退避 next_attempt_at、记 last_error,超过 maxAttempts 置 dead(ADR-002 §2.1)。
 * 多个中继进程并发时互不重复认领;enqueue 成功但标记前崩溃,重投时由引擎幂等键去重(24h),
 * 更长的去重由业务唯一约束兜底。
 */
export class OutboxRelay {
  private timer: NodeJS.Timeout | null = null;
  private running: Promise<unknown> | null = null;
  private readonly log;
  readonly id = randomUUID().slice(0, 8);

  constructor(private readonly opts: OutboxRelayOptions) {
    this.log = createLogger(opts.logSink, { component: "tasks.outbox", relay: this.id });
  }

  static idempotencyKey(eventId: string) {
    return `outbox:${eventId}`;
  }

  /** 处理一批;返回本批已投递与失败数 */
  async pollOnce(): Promise<{ published: number; failed: number }> {
    const mirrorTypes = new Set(this.opts.mirror?.types ?? []);
    const types = [...new Set([...Object.keys(this.opts.routes), ...mirrorTypes])];
    if (types.length === 0) return { published: 0, failed: 0 };
    let published = 0;
    let failed = 0;
    const toMirror: OutboxEvent[] = [];
    await this.opts.db.transaction().execute(async (trx) => {
      const rows = await sql<OutboxEvent & { idempotency_key: string | null; attempts: number }>`
        SELECT id, type, payload, at, idempotency_key, attempts FROM outbox_event
        WHERE status = 'pending' AND next_attempt_at <= now() AND type = ANY(${types}::text[])
        ORDER BY at, id
        LIMIT ${this.opts.batchSize ?? 50}
        FOR UPDATE SKIP LOCKED`.execute(trx);
      for (const ev of rows.rows) {
        try {
          const routeFn = this.opts.routes[ev.type];
          const route = routeFn ? routeFn(ev) : null;
          if (route) {
            const payloadTrace = (ev.payload as { traceId?: unknown } | null)?.traceId;
            await this.opts.engine.enqueue(route.task, route.input, {
              idempotencyKey: ev.idempotency_key ?? OutboxRelay.idempotencyKey(ev.id),
              traceId: route.traceId ?? (typeof payloadTrace === "string" ? payloadTrace : randomUUID()),
              ...(route.priority === undefined ? {} : { priority: route.priority }),
            });
          }
          await sql`UPDATE outbox_event SET status = 'dispatched', published_at = now(), attempts = attempts + 1,
            last_error = NULL WHERE id = ${ev.id}`.execute(trx);
          published++;
          if (mirrorTypes.has(ev.type)) toMirror.push({ id: ev.id, type: ev.type, payload: ev.payload, at: ev.at });
        } catch (err) {
          failed++;
          // 退避后重试;超过上限置 dead(需人工处理)
          const attempts = ev.attempts + 1;
          const dead = attempts >= (this.opts.maxAttempts ?? 10);
          const delayS = Math.min(2 ** attempts, 300);
          const lastError = JSON.stringify({ message: err instanceof Error ? err.message : String(err) });
          await sql`UPDATE outbox_event SET attempts = ${attempts}, status = ${dead ? "dead" : "pending"},
            next_attempt_at = now() + make_interval(secs => ${delayS}), last_error = ${lastError}::jsonb
            WHERE id = ${ev.id}`.execute(trx);
          this.log.error("outbox dispatch failed", { eventId: ev.id, type: ev.type, attempts, dead, err });
        }
      }
    });
    // 提交后才推送:回滚的事件不会先被浏览器看到
    for (const ev of toMirror) {
      try {
        await this.opts.mirror!.publish(ev);
      } catch (err) {
        this.log.error("outbox mirror publish failed", { eventId: ev.id, type: ev.type, err });
      }
    }
    return { published, failed };
  }

  start() {
    if (this.timer) return;
    const tick = () => {
      if (this.running) return;
      this.running = this.pollOnce()
        .catch((err) => this.log.error("outbox poll failed", { err }))
        .finally(() => {
          this.running = null;
        });
    };
    this.timer = setInterval(tick, this.opts.intervalMs ?? 1000);
    tick();
  }

  async stop() {
    if (this.timer) clearInterval(this.timer);
    this.timer = null;
    await this.running;
  }
}