UHD_Media_Tool/apps/server/src/modules/system/index.ts

import { createRoute, type RouteConfig } from "@hono/zod-openapi";
import {
  type Actor,
  Api,
  apiResponse,
  DEFAULT_SETTINGS,
  type EndpointDef,
  type StorageCleanupTarget,
} from "@umt/contracts";
import { sql } from "@umt/db";
import type { z } from "zod";
import { dataPaths } from "../../config.ts";
import { requireScope } from "../../http/auth.ts";
import { ApiException } from "../../http/errors.ts";
import { sseConnectionCount } from "../../http/sse.ts";
import { storageSettings } from "../../shared-cache.ts";
import { registerScheduledJob } from "../schedules/registry.ts";
import type { ModuleApp, ServerContext, ServerModule } from "../types.ts";
import { buildHealth, type HealthDeps, loadHistory } from "./health.ts";
import {
  type HeartbeatWriter,
  instanceIdOf,
  pruneHeartbeats,
  SERVER_VERSION,
  safeHostname,
  startHeartbeatWriter,
} from "./heartbeat.ts";
import { createResourceSampler, type ResourceSampler } from "./resources.ts";
import { type BatchOptions, CLEANUP_JOB_IDS, RETENTION_JOBS } from "./retention.ts";
import { buildStorage, type StoragePaths, storageHealth } from "./storage.ts";

export { HEARTBEAT_STALE_MS, HEARTBEAT_WRITE_MS, startHeartbeatWriter } from "./heartbeat.ts";

/**
 * W9-I 进程与健康(docs/runbook/ops.md):
 * - GET /system/health:进程心跳 / 事件循环延迟 / SSE 连接 / DB / 磁盘 / 挂载根 / 引擎 / 定时任务,仅 admin;
 * - GET /system/health/history?hours=:心跳采样时间序列。
 * 心跳写入:worker 进程在 worker/index.ts 里起写入器;纯 api 进程(role=api)在注册路由时起(role=all 由 worker 写 kind=all)。
 * 清理:定时任务 system.heartbeat_cleanup(维护组,1 小时)删 24 小时没心跳的进程行与超过
 * storage.retentionDays.heartbeatHours 的采样。
 *
 * CACHE-2 存储(docs/runbook/storage.md):
 * - GET /system/storage:表体积 top 15、库体积、kv_cache 各 region、图片 / 备份 / 会话与证据对照预算、DATA_DIR 分目录、清理任务;
 * - POST /system/storage/cleanup/{target}:把对应的维护任务排入立即执行(同 schedules run_now 语义);
 * - 保留清理任务 storage.retention.* 与每周 storage.analyze(维护组,各自开关与周期在定时任务看板改)。
 */
export interface SystemModuleOptions {
  /** 引擎 schema(测试覆盖);缺省 openworkflow */
  engineSchema?: string;
  /** 测试注入:替换 health 取数依赖 */
  healthDeps?: Partial<Pick<HealthDeps, "diskUsage" | "checkRoot" | "now" | "ioTimeoutMs" | "dataDir">>;
  /** 缺省 true;false 时 role=api 也不起心跳写入器(测试) */
  apiHeartbeat?: boolean;
  /** 定时任务 system.heartbeat_cleanup 的默认周期,默认 1 小时;<=0 不登记 */
  cleanupIntervalMs?: number;
  /** 存储面板路径(测试覆盖);缺省按 DATA_DIR / AGENT_SESSION_DIR */
  storagePaths?: StoragePaths;
  /** false 时不登记保留清理任务(测试);缺省 true */
  retentionJobs?: boolean;
  /** 保留清理的分批参数(测试) */
  retentionBatch?: BatchOptions;
  /** 资源采样器(测试注入);false 不起;缺省 api / all 角色注册路由时起一个每 5 秒读 /proc 的采样器 */
  resourceSampler?: ResourceSampler | false;
}

type Infer<T> = T extends z.ZodType ? z.infer<T> : Record<string, never>;

function route<D extends EndpointDef>(
  app: ModuleApp,
  def: D,
  handler: (i: { query: Infer<D["query"]>; params: Infer<D["params"]>; actor: Actor | undefined }) => Promise<unknown>,
) {
  const request: RouteConfig["request"] = {};
  if (def.query) request.query = def.query as never;
  if (def.params) request.params = def.params as never;
  const r = createRoute({
    method: def.method,
    path: def.path,
    summary: def.summary,
    tags: ["system"],
    middleware: [requireScope(def.scope)] as never,
    request,
    responses: {
      200: { description: def.summary, content: { "application/json": { schema: apiResponse(def.response) } } },
    },
  });
  // biome-ignore lint/suspicious/noExplicitAny: 路由泛型在封装里擦除,入参已由 zod-openapi 校验
  app.openapi(r, (async (c: any) => {
    const data = def.response.parse(
      await handler({
        query: def.query ? c.req.valid("query") : {},
        params: def.params ? c.req.valid("param") : {},
        actor: c.get("actor"),
      } as never),
    );
    return c.json({ ok: true, data }, 200);
    // biome-ignore lint/suspicious/noExplicitAny: 同上
  }) as any);
}

export function createSystemModule(opts: SystemModuleOptions = {}): ServerModule & {
  apiHeartbeat(): HeartbeatWriter | null;
} {
  const startedAt = new Date();
  let writer: HeartbeatWriter | null = null;
  let sampler: ResourceSampler | null = null;
  const ownSampler = !opts.resourceSampler;
  return {
    name: "system",
    apiHeartbeat: () => writer,
    registerRoutes(app, ctx: ServerContext) {
      if (ctx.role === "api" && opts.apiHeartbeat !== false && !writer) {
        writer = startHeartbeatWriter({
          db: ctx.db,
          logger: ctx.logger,
          kind: "api",
          startedAt,
          sample: () => ({ sseConnections: sseConnectionCount() }),
        });
      }
      if (opts.resourceSampler) sampler = opts.resourceSampler;
      else if (opts.resourceSampler !== false && !sampler) sampler = createResourceSampler();
      const selfKind = ctx.role === "all" ? "all" : ctx.role === "api" ? "api" : null;
      const selfId = selfKind ? instanceIdOf(selfKind, safeHostname(), Deno.pid) : undefined;
      const paths = (): StoragePaths => {
        if (opts.storagePaths) return opts.storagePaths;
        const p = dataPaths(ctx.config);
        return { dataDir: opts.healthDeps?.dataDir ?? p.dataDir, sessionDir: p.sessionDir };
      };

      route(app, Api.system.healthDetail, async () => {
        const cache = await ctx.settings.get("cache").catch(() => DEFAULT_SETTINGS.cache);
        return await buildHealth({
          db: ctx.db,
          engineSchema: opts.engineSchema ?? "openworkflow",
          dataDir: opts.healthDeps?.dataDir ?? dataPaths(ctx.config).dataDir,
          self: {
            version: SERVER_VERSION,
            startedAt,
            sseConnections: sseConnectionCount(),
            ...(selfId ? { instanceId: selfId } : {}),
          },
          thresholds: { warnPercent: cache.resumePercent, criticalPercent: cache.hardThresholdPercent },
          storage: async () =>
            await storageHealth({ db: ctx.db, paths: paths(), settings: await storageSettings(ctx) }),
          ...(opts.healthDeps ?? {}),
        });
      });
      route(app, Api.system.healthHistory, async ({ query }) => await loadHistory(ctx.db, query.hours));
      route(app, Api.system.resources, async ({ query }) => {
        // 采样器只在注册路由的进程里起;测试关掉时临时起一个不自动采样的
        sampler ??= createResourceSampler({ autoStart: false });
        return await sampler.snapshot(query.minutes);
      });
      route(app, Api.system.storage, async () =>
        buildStorage({
          db: ctx.db,
          paths: paths(),
          settings: await storageSettings(ctx),
          engineSchema: opts.engineSchema ?? "openworkflow",
        }),
      );
      route(app, Api.system.storageCleanup, async ({ params, actor }) => {
        const target = params.target as StorageCleanupTarget;
        const jobId = CLEANUP_JOB_IDS[target];
        const job = await ctx.db
          .selectFrom("scheduled_job")
          .select(["id", "available"])
          .where("id", "=", jobId)
          .executeTakeFirst();
        if (!job) throw new ApiException("not_found", `清理任务 ${jobId} 还没登记（调度器未同步）`);
        if (!job.available) throw new ApiException("conflict", "任务未配置，不能执行");
        // 同 schedules 的立即执行:只置 run_requested_at,已在排队或运行中则合并
        const r = await sql`update scheduled_job set run_requested_at = now()
          where id = ${jobId} and run_requested_at is null and running_since is null`.execute(ctx.db);
        const queued = Number(r.numAffectedRows ?? 0) > 0;
        if (queued) {
          await ctx.db
            .insertInto("audit_event")
            .values({
              actor_type: actor?.type === "api_key" ? "api_key" : actor ? "user" : "system",
              actor_id: actor?.id ?? null,
              action: "storage.cleanup",
              target_type: "scheduled_job",
              target_id: jobId,
              payload: JSON.stringify({ target }),
            })
            .execute()
            .catch((err) => ctx.logger.warn({ err }, "审计写入失败"));
        }
        return { target, jobId, queued };
      });
    },
    registerTasks() {},
    start(ctx) {
      if (opts.retentionJobs !== false) {
        for (const j of RETENTION_JOBS) {
          registerScheduledJob(ctx, {
            key: j.jobId,
            lane: "maintenance",
            defaults: {
              name: j.name,
              description: j.description,
              group: "maintain",
              schedule: { type: "interval", everyMs: j.everyMs },
              firstRun: "after_interval",
            },
            run: async ({ now }) => ({
              summary: await j.run({
                db: ctx.db,
                engineSchema: opts.engineSchema ?? "openworkflow",
                settings: await storageSettings(ctx),
                now,
                ...(opts.retentionBatch ? { batch: opts.retentionBatch } : {}),
              }),
            }),
          });
        }
      }
      const everyMs = opts.cleanupIntervalMs ?? 3600_000;
      if (everyMs <= 0) return;
      registerScheduledJob(ctx, {
        key: "system.heartbeat_cleanup",
        lane: "maintenance",
        defaults: {
          name: "进程心跳清理",
          description: "删除过期的进程状态记录",
          group: "maintain",
          schedule: { type: "interval", everyMs },
        },
        run: async ({ now }) => {
          const sampleHours = (await storageSettings(ctx)).retentionDays.heartbeatHours;
          const r = await pruneHeartbeats(ctx.db, { now, sampleHours });
          return { summary: `删除 ${r.rows} 个进程行、${r.samples} 条采样` };
        },
      });
    },
    async stop() {
      await writer?.stop();
      writer = null;
      if (ownSampler) sampler?.stop();
      sampler = null;
    },
  };
}

export const systemModule = createSystemModule();