UHD_Media_Tool/apps/server/src/context.ts

import { type EventType, LIFECYCLE_EVENTS, parseEventPayload } from "@umt/contracts";
import { type DB, type Kysely, sql } from "@umt/db";
import {
  createDefaultRegistry,
  type LogSink,
  OpenWorkflowEngine,
  type OutboxRelay,
  type OutboxRoute,
  type OutboxRouteFn,
  type TaskEngine,
  TaskRegistry,
} from "@umt/tasks";
import type { Logger } from "pino";
import type { Config } from "./config.ts";
import { HookRegistry } from "./hooks/registry.ts";
import type { EventBusPort } from "./http/sse.ts";
import { assertCapabilityCount, createCapabilityRegistry } from "./modules/agent/capability-registry.ts";
import { createSecretStore, createSettingsReader } from "./modules/auth-settings/provided.ts";
import type { OutboxPort, ServerContext, ServerModule, ServerRole } from "./modules/types.ts";
import { configureStorage } from "./storage/registry.ts";

// 进程装配:ServerContext、任务注册表、outbox 路由。api / worker / all 三种角色共用。

/** 引擎在任务注册表建好后才能连接;先给模块一个会转发的句柄 */
export class DeferredTaskEngine implements TaskEngine {
  private inner: TaskEngine | null = null;

  attach(engine: TaskEngine) {
    this.inner = engine;
  }

  get ready(): boolean {
    return this.inner !== null;
  }

  private get e(): TaskEngine {
    if (!this.inner) throw new Error("任务引擎尚未就绪（模块不要在 register* 阶段调用引擎）");
    return this.inner;
  }

  enqueue: TaskEngine["enqueue"] = (...a) => this.e.enqueue(...a);
  signal: TaskEngine["signal"] = (...a) => this.e.signal(...a);
  cancel: TaskEngine["cancel"] = (...a) => this.e.cancel(...a);
  get: TaskEngine["get"] = (...a) => this.e.get(...a);
  list: TaskEngine["list"] = (...a) => this.e.list(...a);
  startWorker: TaskEngine["startWorker"] = (...a) => this.e.startWorker(...a);
  queueCounts: NonNullable<TaskEngine["queueCounts"]> = async (...a) => {
    const q = this.e.queueCounts;
    return q ? q.apply(this.e, a) : a[0].map(() => ({ queued: 0, running: 0 }));
  };
  stop: TaskEngine["stop"] = async () => {
    await this.inner?.stop();
  };
}

export function createOutboxPort(): OutboxPort & { setRelay(r: OutboxRelay): void } {
  let relay: OutboxRelay | null = null;
  return {
    get relay() {
      return relay;
    },
    setRelay(r) {
      relay = r;
    },
    async emit(executor, type, payload, opts = {}) {
      if (parseEventPayload(type as EventType, payload) === null) {
        throw new Error(`outbox 事件 ${type} 的 payload 不符合 contracts 定义`);
      }
      await sql`insert into outbox_event (type, payload, idempotency_key)
        values (${type}, ${JSON.stringify(payload)}::jsonb, ${opts.idempotencyKey ?? null})
        on conflict (idempotency_key) where idempotency_key is not null do nothing`.execute(executor);
    },
  };
}

/** 把 pino logger 接到 @umt/tasks 的 LogSink(tasks 自己已脱敏,这里再经 pino 的脱敏) */
export function tasksLogSink(logger: Logger): LogSink {
  // r.fields 里已带 component(tasks.engine / tasks.outbox)
  return (r) => logger[r.level](r.fields, r.msg);
}

/**
 * 任务注册表:各模块注册自己的任务;没人注册的 §3.1-8 任务用 @umt/tasks 的空壳兜底
 * (outbox 路由到它们时不至于报"未知任务")。同名任务被两个模块注册会在这里报错。
 */
export function buildTaskRegistry(modules: readonly ServerModule[], ctx: ServerContext): TaskRegistry {
  const registry = new TaskRegistry();
  const owner = new Map<string, string>();
  for (const m of modules) {
    const own = new TaskRegistry();
    m.registerTasks(own, ctx);
    for (const name of own.names()) {
      const prev = owner.get(name);
      if (prev) throw new Error(`任务 ${name} 被模块 ${prev} 与 ${m.name} 重复注册`);
      owner.set(name, m.name);
    }
    registry.register(...own.all());
  }
  for (const stub of createDefaultRegistry().all()) if (!registry.has(stub.name)) registry.register(stub);
  return registry;
}

const BROADCAST_TYPES = new Set<string>(LIFECYCLE_EVENTS.map((e) => e.type));

/**
 * 合并各模块的 outbox 路由。生命周期广播事件(LIFECYCLE_EVENTS,ADR-006)可以有多个订阅方,按模块顺序合成一个路由:
 * 各订阅方各自返回任务(或 null),合起来交给中继逐个入队;某个订阅方的路由函数抛错只记日志,不连累别人。
 * 其它事件(内部流程,如 release.observed → resolveObservation)仍只能由一个模块认领。
 */
export function collectOutboxRoutes(
  modules: readonly ServerModule[],
  ctx: ServerContext,
): Record<string, OutboxRouteFn> {
  const subscribers = new Map<string, { module: string; fn: OutboxRouteFn }[]>();
  for (const m of modules) {
    for (const [type, fn] of Object.entries(m.outboxRoutes?.(ctx) ?? {})) {
      const prev = subscribers.get(type) ?? [];
      if (prev.length && !BROADCAST_TYPES.has(type))
        throw new Error(`outbox 事件 ${type} 被模块 ${prev[0]!.module} 与 ${m.name} 重复认领`);
      subscribers.set(type, [...prev, { module: m.name, fn }]);
    }
  }
  const routes: Record<string, OutboxRouteFn> = {};
  for (const [type, subs] of subscribers) {
    if (subs.length === 1) {
      routes[type] = subs[0]!.fn;
      continue;
    }
    routes[type] = (ev) => {
      const out: OutboxRoute[] = [];
      for (const s of subs) {
        let r: ReturnType<OutboxRouteFn>;
        try {
          r = s.fn(ev);
        } catch (err) {
          ctx.logger.warn(
            {
              type,
              module: s.module,
              eventId: ev.id,
              err: { message: err instanceof Error ? err.message : String(err) },
            },
            "outbox 订阅方路由出错，已跳过",
          );
          continue;
        }
        if (r === null) continue;
        out.push(...((Array.isArray(r) ? r : [r]) as OutboxRoute[]));
      }
      return out.length ? out : null;
    };
  }
  return routes;
}

/** 链式钩子注册表:按模块顺序调 registerHooks,执行顺序即注册顺序(ADR-006) */
export function buildHookRegistry(modules: readonly ServerModule[], ctx: ServerContext): HookRegistry {
  const hooks = new HookRegistry(ctx.logger?.child({ component: "hooks" }));
  for (const m of modules) m.registerHooks?.(hooks.forModule(m.name), ctx);
  return hooks;
}

export interface BootstrapOptions {
  role: ServerRole;
  config: Config;
  logger: Logger;
  db: Kysely<DB>;
  events: EventBusPort;
  modules: readonly ServerModule[];
  /** 连接引擎时是否跑引擎迁移;默认只有 all 角色跑(api/worker 分进程时由 --role migrate 预先迁移) */
  engineMigrations?: boolean;
  /** 仅测试覆盖:引擎表 schema(默认 openworkflow) */
  engineSchema?: string;
  /** 仅测试覆盖:不连接真实引擎 */
  engine?: TaskEngine;
}

export interface Bootstrapped {
  ctx: ServerContext;
  tasks: TaskRegistry;
  outbox: ReturnType<typeof createOutboxPort>;
  engine: DeferredTaskEngine;
}

/** 建 ServerContext → 模块注册任务与能力 → 连接引擎。路由与 worker 由调用方按角色启动 */
export async function bootstrap(opts: BootstrapOptions): Promise<Bootstrapped> {
  const { config, logger, db } = opts;
  const engine = new DeferredTaskEngine();
  const outbox = createOutboxPort();
  const settings = createSettingsReader({ db, logger, events: opts.events });
  const ctx: ServerContext = {
    role: opts.role,
    db,
    config,
    logger,
    events: opts.events,
    engine,
    outbox,
    secrets: createSecretStore({ db, masterKey: config.SECRET_MASTER_KEY }),
    settings,
    // 授权(automation_grant / approval + 自动化总开关)、审计(audit_event)、幂等键与操作账本在这里注入
    capabilities: createCapabilityRegistry({ logger, db, settings }),
  };
  configureStorage(ctx);
  const tasks = buildTaskRegistry(opts.modules, ctx);
  for (const m of opts.modules) m.registerCapabilities?.(ctx.capabilities, ctx);
  ctx.hooks = buildHookRegistry(opts.modules, ctx);
  // 装配了 agent 模块(即完整模块清单)时,首批 14 个能力必须全部挂上处理器
  if (opts.modules.some((m) => m.name === "agent")) assertCapabilityCount(ctx.capabilities);
  engine.attach(
    opts.engine ??
      (await OpenWorkflowEngine.connect({
        databaseUrl: config.DATABASE_URL,
        db,
        registry: tasks,
        logSink: tasksLogSink(logger),
        runMigrations: opts.engineMigrations ?? opts.role === "all",
        ...(opts.engineSchema ? { schema: opts.engineSchema } : {}),
      })),
  );
  return { ctx, tasks, outbox, engine };
}