BPOS-TMP-REPO/apps/api/src/app/workers/dispatcher.ts

/**
 * 投递派发器 —— ADR-008 #2。
 *
 * 按 `state='pending' and next_attempt_at <= now()` 取投递记录,
 * 对**目标 ID 取 advisory lock** 串行执行 —— 这正是「同任务按序」的实现(§7.1)。
 * 消费幂等由 `sys.event_consumption(event_id, consumer)` 唯一约束承担。
 *
 * 消费者(任务 / 触发器)随 M4 / M5 落地;在此之前派发器把投递标记为已送达并留痕,
 * 保证投递链路本身可测、不积压(docs/debt.md · M1-3)。
 */
import { sql } from 'drizzle-orm';
import type { Delivery } from '@modules/event';
import { claimPendingDeliveries } from '@modules/event';
import type { Container } from '../container.ts';
import { taskMutex } from './task-lock.ts';
import { claimTask, releaseTask } from './task-claim.ts';

/** 消费者:按目标类型分派 */
export type Consumer = (delivery: Delivery, container: Container) => Promise<void>;

export type DispatcherDeps = {
  readonly container: Container;
  readonly consumers: ReadonlyMap<Delivery['target']['kind'], Consumer>;
  readonly batchSize?: number;
};

export async function dispatchOnce(deps: DispatcherDeps): Promise<number> {
  const { container } = deps;
  const batch = await claimPendingDeliveries(container.db, deps.batchSize ?? 20);
  let handled = 0;

  for (const delivery of batch) {
    // 同任务按序:取到锁才处理,取不到说明同目标的另一条正在处理,留待下一轮
    await taskMutex.tryRun(`${delivery.target.kind}:${String(delivery.target.id)}`, async () => {
      /*
       * **投递也会推一拍**(consumer 里 `task.deliver` 末尾就是 `tick`)——
       * 所以它同样要抢库里那份租约:上面那把锁只在本进程内算数,而两个后端连同一个库时
       * (2026-08-24 实测)投递与另一台的推进器会并发写同一份 `loop_state`。
       * 抢不到就先不处理,这一条留给下一轮 —— 它还在 pending,不会丢。
       */
      const taskId = delivery.target.kind === 'task' ? String(delivery.target.id) : undefined;
      if (taskId !== undefined && !(await claimTask(container.db, taskId))) return;
      try {
        const consumer = deps.consumers.get(delivery.target.kind);
        if (consumer !== undefined) {
          const claimed = await container.event.claimConsumption(
            delivery.eventId,
            `${delivery.target.kind}:${delivery.target.id}`,
          );
          // 至少一次 + 幂等:已消费过的重复投递直接标记完成,不再执行副作用
          if (claimed) await consumer(delivery, container);
        }
        await container.db.execute(
          sql`update sys.event_delivery set state = 'delivered' where id = ${delivery.id}`,
        );
        // 已落库、已送达之后才广播 —— 广播不是投递,只是告诉在线浏览器「这个目标动了」
        container.broadcaster.publish({
          kind: delivery.target.kind === 'message-position' ? 'staff' : 'task',
          id: String(delivery.target.id),
          reason: delivery.target.kind,
        });
        handled += 1;
      } catch (error) {
        await container.event.reportFailure(delivery.id, error instanceof Error ? error.message : String(error));
      } finally {
        // 推完就还租约 —— 不还也不会锁死(会过期),还了下一条投递立刻接得上
        if (taskId !== undefined) await releaseTask(container.db, taskId);
      }
    });
  }
  return handled;
}

export function startDispatcher(deps: DispatcherDeps, intervalMs: number): { stop(): void } {
  let stopped = false;
  const tick = async (): Promise<void> => {
    if (stopped) return;
    try {
      await dispatchOnce(deps);
    } catch (error) {
      process.stderr.write(`[dispatcher] ${error instanceof Error ? error.message : String(error)}\n`);
    }
  };
  const timer = setInterval(() => void tick(), intervalMs);
  timer.unref();
  return {
    stop() {
      stopped = true;
      clearInterval(timer);
    },
  };
}