JX-BPOS/apps/api/src/app/workers/task-advancer.ts

/**
 * 任务推进器 —— 「过程在后端」的落点(编码规范 §9;语义出处 OS 09)。
 *
 * 「环不依赖浏览器在线 —— 负责人关掉页面,活动照常推进」:
 * 运行中的任务由本 worker 一拍一拍推,直到进入等待(等人 / 等事件)或收敛终态。
 * 预算与停滞熔断在 `task.tick()` 里**环外先判**,这里不重复把关。
 */
import { and, eq, sql } from 'drizzle-orm';
import { taskTable } from '@modules/task';
import type { Container } from '../container.ts';
import { taskMutex } from './task-lock.ts';

export type AdvancerOptions = {
  readonly container: Container;
  /** 一轮最多推几个任务 */
  readonly batchSize?: number;
  /** 单个任务一轮最多推几拍 —— 防止一个任务占满 worker */
  readonly maxTicks?: number;
};

export async function advanceOnce(options: AdvancerOptions): Promise<number> {
  const { container } = options;
  const rows = await container.db
    .select({ id: taskTable.id })
    .from(taskTable)
    .where(and(eq(taskTable.runtime, 'running'), sql`${taskTable.terminal} is null`))
    .limit(options.batchSize ?? 5);

  let advanced = 0;
  for (const row of rows) {
    /*
     * 与派发器共用同一把锁:事件投递也会推进同一个任务。
     * 抢不到就跳过 —— 那边正推着,这边再推就是两条线同时改一份运行时状态。
     */
    const ran = await taskMutex.tryRun(`task:${row.id}`, async () => {
      /** 这一轮之前它是什么样 —— 变了才广播,不然每两秒给在线浏览器发一次噪声 */
      const before = await container.task.get(row.id as never).catch(() => undefined);
      for (let tick = 0; tick < (options.maxTicks ?? 3); tick += 1) {
        try {
          const outcome = await container.task.tick(row.id as never);
          advanced += 1;

          /*
           * 推进器改了状态就**广播**:任务从在办走到终态、或挂起去等人,
           * 都是在线的人该立刻看见的事。
           * 从前只有派发器(事件投递)广播,而任务是被**这里**推到终态的 ——
           * 于是「办完的活不会自己挪到已完成」,得刷页面才认(用户实测报的)。
           */
          if (outcome.task.runtime !== before?.runtime || outcome.task.terminal !== before?.terminal) {
            container.broadcaster.publish({ kind: 'task', id: row.id, reason: 'advanced' });
            if (outcome.task.workflowId !== undefined) {
              container.broadcaster.publish({ kind: 'workflow', id: outcome.task.workflowId, reason: 'task-advanced' });
            }
          }

          // 进入等待或终结就停手 —— 挂起在等的任务由事件唤醒,不空转
          if (outcome.task.runtime !== 'running' || outcome.task.terminal !== undefined) break;
          if (!outcome.progressed) break;
        } catch (error) {
          process.stderr.write(
            `[task-advancer] ${row.id}:${error instanceof Error ? error.message : String(error)}\n`,
          );
          break;
        }
      }
      return true;
    });
    void ran;
  }
  return advanced;
}

export function startTaskAdvancer(options: AdvancerOptions, intervalMs: number): { stop(): void } {
  let stopped = false;
  const timer = setInterval(() => {
    if (stopped) return;
    void advanceOnce(options).catch((error: unknown) => {
      process.stderr.write(`[task-advancer] ${error instanceof Error ? error.message : String(error)}\n`);
    });
  }, intervalMs);
  timer.unref();
  return {
    stop() {
      stopped = true;
      clearInterval(timer);
    },
  };
}