import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { DiskUsage } from "@umt/adapters";
import { SystemHealth, SystemHealthHistory } from "@umt/contracts";
import { sql } from "@umt/db";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { inferTriggeredBy, triggeredByFromActor } from "../../src/modules/auth-settings/engine-business.ts";
import { diskLevel, processLevel, worst } from "../../src/modules/system/health.ts";
import { HEARTBEAT_STALE_MS, startHeartbeatWriter, writeHeartbeat } from "../../src/modules/system/heartbeat.ts";
import { createSystemModule } from "../../src/modules/system/index.ts";
import { bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts";
import { DEV_KEY, json } from "../helpers.ts";

// W9-I 进程与健康 + 引擎 runs 业务字段(真实 PG;磁盘 / 哨兵检查注入假实现)。

const hasPg = await pgAvailable();

describe("纯函数", () => {
  it("档位合并忽略 unknown;磁盘阈值", () => {
    expect(worst(["ok", "unknown", "warn"])).toBe("warn");
    expect(worst(["unknown"])).toBe("ok");
    expect(worst(["warn", "critical", "ok"])).toBe("critical");
    const t = { warnPercent: 85, criticalPercent: 90 };
    expect(diskLevel(0.5, t)).toBe("ok");
    expect(diskLevel(0.86, t)).toBe("warn");
    expect(diskLevel(0.9, t)).toBe("critical");
  });

  it("进程档位:stale / 无 worker → warn;延迟 ≥5s → critical", () => {
    const p = (kind: "api" | "worker", o: { stale?: boolean; lag?: number } = {}) =>
      ({
        instanceId: `${kind}:h:1`,
        processKind: kind,
        pid: 1,
        host: "h",
        version: "t",
        startedAt: new Date().toISOString(),
        lastBeatAt: new Date().toISOString(),
        beatAgeSec: 0,
        stale: o.stale ?? false,
        eventLoopLagMs: 0,
        eventLoopLagMaxMs: o.lag ?? 0,
        concurrency: {},
        runningRuns: null,
        sseConnections: null,
        extra: {},
      }) as const;
    expect(processLevel([p("api"), p("worker")])).toBe("ok");
    expect(processLevel([p("api")])).toBe("warn");
    expect(processLevel([p("api"), p("worker", { stale: true })])).toBe("warn");
    expect(processLevel([p("api"), p("worker", { lag: 1200 })])).toBe("warn");
    expect(processLevel([p("api"), p("worker", { lag: 6000 })])).toBe("critical");
  });

  it("触发来源推断", () => {
    expect(triggeredByFromActor("api_key:abc")).toBe("api_key");
    expect(triggeredByFromActor("admin_session:u1")).toBe("user");
    expect(triggeredByFromActor("system")).toBeNull();
    const base = {
      kind: "download" as const,
      idempotencyKey: null,
      taskTriggeredBy: null,
      agentTrigger: null,
      agentCreatedBy: null,
      fromOutbox: false,
    };
    expect(inferTriggeredBy({ ...base, kind: "agent", agentTrigger: "subagent" })).toBe("agent");
    expect(inferTriggeredBy({ ...base, kind: "agent", agentTrigger: "user", agentCreatedBy: "api_key:k" })).toBe(
      "api_key",
    );
    expect(inferTriggeredBy({ ...base, taskTriggeredBy: "agent" })).toBe("agent");
    expect(inferTriggeredBy({ ...base, idempotencyKey: "retry:r1:1" })).toBe("user");
    expect(inferTriggeredBy({ ...base, idempotencyKey: "ingestFeed:manual:f:1" })).toBe("user");
    expect(inferTriggeredBy({ ...base, idempotencyKey: "release.observed:x", fromOutbox: true })).toBe("event");
    expect(inferTriggeredBy({ ...base, idempotencyKey: "snapshotSite:s1:123" })).toBe("schedule");
    expect(inferTriggeredBy({ ...base, kind: "schedule", idempotencyKey: "purgeTrash:1" })).toBe("schedule");
    expect(inferTriggeredBy(base)).toBeNull();
  });
});

describe.skipIf(!hasPg)("W9-I 进程与健康(真实 PG)", () => {
  let h: Harness;
  let admin: ReturnType<typeof bearer>;
  const dir = mkdtempSync(join(tmpdir(), "umt-w9i-"));
  const libOk = join(dir, "lib-ok");
  const usage = (path: string): Promise<DiskUsage> =>
    Promise.resolve(
      path.includes("full")
        ? {
            status: "ok",
            path,
            totalBytes: 1000,
            freeBytes: 20,
            availableBytes: 20,
            usedBytes: 980,
            usedRatio: 0.98,
          }
        : path.includes("hang")
          ? new Promise<DiskUsage>(() => {})
          : {
              status: "ok",
              path,
              totalBytes: 1000,
              freeBytes: 500,
              availableBytes: 500,
              usedBytes: 500,
              usedRatio: 0.5,
            },
    );

  beforeAll(async () => {
    h = await createHarness({
      engine: true,
      extraModules: [],
    });
    // harness 在 createHarness 时已建 app;这里再建一个挂了 system 模块的 app(同一 ctx / db)
    const { createApp } = await import("../../src/http/app.ts");
    const { createAuthSettingsModule } = await import("../../src/modules/auth-settings/index.ts");
    h.app = createApp({
      config: h.ctx.config,
      logger: h.ctx.logger,
      bus: h.bus,
      db: h.db,
      role: "all",
      version: "0.0.0-test",
      modules: [
        createAuthSettingsModule({ setupTokenOnStart: false, engineSchema: h.engineSchema }),
        createSystemModule({
          engineSchema: h.engineSchema,
          apiHeartbeat: false,
          healthDeps: { dataDir: dir, diskUsage: usage, ioTimeoutMs: 200 },
        }),
      ],
      ctx: h.ctx,
    }) as Harness["app"];
    admin = bearer(h.app, DEV_KEY);
    await Deno.mkdir(libOk, { recursive: true });
    await Deno.writeTextFile(join(libOk, ".umt-root"), "x");
  }, 60_000);
  afterAll(async () => {
    await h?.close();
    rmSync(dir, { recursive: true, force: true });
  });

  it("心跳写入:upsert 同一 instance_id + 追加采样;超过 45s 标 stale", async () => {
    const w = startHeartbeatWriter({
      db: h.db,
      logger: h.ctx.logger,
      kind: "worker",
      host: "test-host",
      pid: 4242,
      intervalMs: 60_000,
      lag: { take: () => ({ lastMs: 3, windowMaxMs: 12, lifetimeMaxMs: 40 }), stop() {} },
      sample: () => ({ concurrency: { io: 4 }, runningRuns: 2, extra: { group: "io" } }),
    });
    await w.beat();
    await w.beat();
    await w.stop();
    expect(w.instanceId).toBe("worker:test-host:4242");
    const rows = await h.db.selectFrom("process_heartbeat").selectAll().execute();
    expect(rows).toHaveLength(1);
    expect(rows[0]).toMatchObject({
      process_kind: "worker",
      pid: 4242,
      event_loop_lag_ms: 3,
      event_loop_lag_max_ms: 12,
      running_runs: 2,
      concurrency: { io: 4 },
    });
    expect(rows[0]!.extra).toMatchObject({ group: "io", lifetimeMaxLagMs: 40 });
    const samples = await h.db.selectFrom("process_heartbeat_sample").selectAll().execute();
    expect(samples.length).toBeGreaterThanOrEqual(2);

    // 一个 api 进程 60s 前的心跳 → stale
    await writeHeartbeat(h.db, {
      instanceId: "api:old-host:1",
      kind: "api",
      pid: 1,
      host: "old-host",
      version: "0.1.0",
      startedAt: new Date(Date.now() - 3600_000),
      lagMs: 1,
      lagMaxMs: 2,
      sample: { sseConnections: 3 },
      at: new Date(Date.now() - HEARTBEAT_STALE_MS - 15_000),
    });

    const res = await admin("GET", "/system/health");
    expect(res.status).toBe(200);
    const d = SystemHealth.parse((await json(res)).data);
    const byId = new Map(d.processes.map((p) => [p.instanceId, p]));
    expect(byId.get("worker:test-host:4242")).toMatchObject({ stale: false, runningRuns: 2, concurrency: { io: 4 } });
    expect(byId.get("api:old-host:1")).toMatchObject({ stale: true, sseConnections: 3 });
    // stale 进程的 SSE 不计入
    expect(d.sse.connections).toBe(0);
  });

  it("health 各段形状:DB / 磁盘阈值 / 挂载根哨兵 / 引擎 / 定时任务", async () => {
    await sql`insert into library_root (name, local_path, sentinel_file) values
      ('电影', ${libOk}, '.umt-root'), ('剧集(掉线)', ${join(dir, "missing")}, null)`.execute(h.db);
    await sql`insert into downloader_instance (name, base_url) values ('qb1', 'http://qb')`.execute(h.db);
    await sql`insert into path_mapping (view_kind, downloader_instance_id, remote_prefix, local_prefix, priority)
      select 'downloader', id, '/downloads/full', ${join(dir, "full")}, 0 from downloader_instance where name = 'qb1'`.execute(
      h.db,
    );
    const t = (n: string) => sql.raw(`"${h.engineSchema}".${n}`);
    const run = (id: string, status: string, name: string, input: unknown, at: Date, finished: Date | null = null) =>
      sql`insert into ${t("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, context, input, attempts,
          worker_id, available_at, started_at, finished_at, created_at, updated_at)
        values ('default', ${id}, ${name}, 'v1', ${status}, '{}'::jsonb, null, ${JSON.stringify(input)}::jsonb, 1,
          ${status === "running" ? "w1" : null}, ${at}, ${at}, ${finished}, ${at}, ${at})`.execute(h.db);
    const now = Date.now();
    await run("h1", "pending", "ingestFeed", { input: {} }, new Date(now - 600_000));
    await run("h2", "failed", "ingestFeed", { input: {} }, new Date(now - 3600_000), new Date(now - 3500_000));
    await run("h3", "completed", "ingestFeed", { input: {} }, new Date(now - 3600_000), new Date(now - 3500_000));
    await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule,
        last_status, last_error, last_run_at)
      values ('feeds.poll', 'Feed 抓取', 'collect', 'task', true, true, '{"type":"interval","everyMs":60000}'::jsonb,
        '{"type":"interval","everyMs":60000}'::jsonb, 'failed', 'boom', now())`.execute(h.db);
    await sql`insert into scheduled_job_run (job_id, trigger, status, error) values ('feeds.poll', 'schedule', 'failed', 'boom')`.execute(
      h.db,
    );

    const d = SystemHealth.parse((await json(await admin("GET", "/system/health"))).data);
    expect(d.db).toMatchObject({ level: expect.any(String), error: null });
    expect(d.db.latencyMs).toBeGreaterThanOrEqual(0);
    expect(d.db.sizeBytes).toBeGreaterThan(0);
    expect(d.db.connections!.max).toBeGreaterThan(0);

    expect(d.diskThresholds).toEqual({ warnPercent: 85, criticalPercent: 90 });
    const disk = (scope: string, name?: string) => d.disks.find((x) => x.scope === scope && (!name || x.name === name));
    expect(disk("data_dir")).toMatchObject({ path: dir, level: "ok", usedRatio: 0.5 });
    expect(disk("download_dir")).toMatchObject({ name: "qb1 /downloads/full", level: "critical", usedRatio: 0.98 });
    expect(disk("library_root", "电影")).toMatchObject({ level: "ok" });

    const mount = (name: string) => d.mounts.find((m) => m.name === name);
    expect(mount("电影")).toMatchObject({
      kind: "library_root",
      status: "online",
      reason: null,
      sentinelChecked: true,
    });
    expect(mount("剧集(掉线)")).toMatchObject({ status: "offline", reason: "root_missing" });
    expect(mount("qb1 /downloads/full")).toMatchObject({
      kind: "path_mapping",
      status: "offline",
      sentinelChecked: false,
    });

    expect(d.engine).toMatchObject({ pending: 1, finished24h: 2, failed24h: 1, failureRate24h: 0.5, level: "warn" });
    expect(d.engine.oldestPendingAgeSec).toBeGreaterThanOrEqual(590);
    expect(d.schedules).toMatchObject({ level: "warn", enabled: 1, failedRuns24h: 1 });
    expect(d.schedules.failing).toEqual([
      expect.objectContaining({ id: "feeds.poll", name: "Feed 抓取", lastError: "boom" }),
    ]);
    expect(d.level).toBe("critical");
    expect(d.api.version).toMatch(/\d+\.\d+\.\d+/);
  });

  it("磁盘 statfs 卡死 → timeout 标 critical,不拖住请求", async () => {
    await sql`insert into library_root (name, local_path) values ('卡死', ${join(dir, "hang")})`.execute(h.db);
    const t0 = Date.now();
    const d = SystemHealth.parse((await json(await admin("GET", "/system/health"))).data);
    expect(Date.now() - t0).toBeLessThan(3000);
    expect(d.disks.find((x) => x.name === "卡死")).toMatchObject({ level: "critical", error: "timeout" });
    await sql`delete from library_root where name = '卡死'`.execute(h.db);
  });

  it("history:按桶聚合的时间序列;hours 越界 422", async () => {
    const res = await admin("GET", "/system/health/history?hours=1");
    expect(res.status).toBe(200);
    const d = SystemHealthHistory.parse((await json(res)).data);
    expect(d.bucketSec).toBe(15);
    const s = d.series.find((x) => x.instanceId === "worker:test-host:4242");
    expect(s?.processKind).toBe("worker");
    expect(s!.points[0]).toMatchObject({ lagMaxMs: 12, runningRuns: 2 });
    expect((await admin("GET", "/system/health/history?hours=500")).status).toBe(422);
    expect((await json(await admin("GET", "/system/health/history?hours=168"))).data.bucketSec).toBe(1680);
  });

  it("引擎 runs 补 business 字段 + businessKind / businessRefId 筛选", async () => {
    const t = (n: string) => sql.raw(`"${h.engineSchema}".${n}`);
    const at = new Date("2026-09-10T00:00:00Z");
    const run = (id: string, name: string, input: unknown, sec: number) =>
      sql`insert into ${t("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, context, input, attempts,
          available_at, created_at, updated_at)
        values ('default', ${id}, ${name}, 'v1', 'completed', '{}'::jsonb, null, ${JSON.stringify(input)}::jsonb, 1,
          ${at}, ${new Date(at.getTime() + sec * 1000)}, ${at})`.execute(h.db);
    const meta = (runId: string, name: string, key: string) =>
      sql`insert into tasks_run_meta (run_id, task_name, task_version, idempotency_key) values (${runId}, ${name}, 1, ${key})`.execute(
        h.db,
      );

    const feed = await sql<{
      id: string;
    }>`insert into feed_source (name, url_display) values ('HDSky RSS', 'https://x/rss')
      returning id::text`.execute(h.db);
    const feedId = feed.rows[0]!.id;
    const ar = await sql<{ id: string }>`insert into agent_run (trigger, goal, created_by)
      values ('schedule', '巡检:待处理', 'schedule') returning id::text`.execute(h.db);
    const agentId = ar.rows[0]!.id;

    // b1 事件链触发的 Feed 抓取(outbox 幂等键)
    await run("b1", "ingestFeed", { input: { feedSourceId: feedId }, traceId: "t" }, 1);
    await meta("b1", "ingestFeed", "feed.due:x:1");
    await sql`insert into outbox_event (type, payload, idempotency_key) values ('feed.due', '{}'::jsonb, 'feed.due:x:1')`.execute(
      h.db,
    );
    // b2 Agent 会话(定时巡检)
    await run("b2", "runAgent", { input: { agentRunId: agentId } }, 2);
    await meta("b2", "runAgent", `agent_run:${agentId}`);
    // b3 API Key 发起的整理(只有业务任务行能关联)
    await run("b3", "verifyAndImport", { input: { taskId: "00000000-0000-0000-0000-000000000001" } }, 3);
    await meta("b3", "verifyAndImport", "verifyAndImport:files:p1");
    await sql`insert into task (kind, idempotency_key, input, subject_type, subject_id, triggered_by, engine_run_id)
      values ('verifyAndImport', 'files:p1', '{}'::jsonb, 'organization_record', '00000000-0000-0000-0000-0000000000aa', 'api_key:k1', 'b3')`.execute(
      h.db,
    );
    // b4 定时清理回收站(无业务 id)
    await run("b4", "purgeTrash", { input: {} }, 4);
    await meta("b4", "purgeTrash", "purgeTrash:slot-1");
    await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule)
      values ('trash.purge', '清理回收站', 'maintain', 'task', true, true, '{"type":"interval","everyMs":60000}'::jsonb,
        '{"type":"interval","everyMs":60000}'::jsonb)`.execute(h.db);
    // b5 手动立即执行的清理 → 不算 schedule,也没有业务对象
    await run("b5", "purgeTrash", { input: {} }, 5);
    await meta("b5", "purgeTrash", "purgeTrash:manual:r1");

    const list = await json(await admin("GET", "/engine/runs?from=2026-09-10T00:00:00Z&to=2026-09-11T00:00:00Z"));
    const by = new Map(list.data.items.map((r: { id: string; business: unknown }) => [r.id, r.business]));
    expect(by.get("b1")).toEqual({ kind: "feed", refId: feedId, title: "HDSky RSS", triggeredBy: "event" });
    expect(by.get("b2")).toEqual({ kind: "agent", refId: agentId, title: "巡检:待处理", triggeredBy: "schedule" });
    expect(by.get("b3")).toEqual({
      kind: "organization",
      refId: "00000000-0000-0000-0000-0000000000aa",
      title: null,
      triggeredBy: "api_key",
    });
    expect(by.get("b4")).toEqual({
      kind: "schedule",
      refId: "trash.purge",
      title: "清理回收站",
      triggeredBy: "schedule",
    });
    expect(by.get("b5")).toBeNull();

    const f = await json(await admin("GET", "/engine/runs?businessKind=feed"));
    expect(f.data.items.map((r: { id: string }) => r.id)).toEqual(["b1"]);
    const g = await json(await admin("GET", `/engine/runs?businessRefId=${agentId}`));
    expect(g.data.items.map((r: { id: string }) => r.id)).toEqual(["b2"]);
    const s = await json(await admin("GET", "/engine/runs?businessKind=schedule&businessRefId=trash.purge"));
    expect(s.data.items.map((r: { id: string }) => r.id)).toEqual(["b4"]);
    expect((await admin("GET", "/engine/runs?businessKind=bogus")).status).toBe(422);

    // triggeredBy:服务端逐页推断后筛选;多值逗号分隔;limit 截断时游标接着扫
    const range = "from=2026-09-10T00:00:00Z&to=2026-09-11T00:00:00Z";
    const ids = (r: { data: { items: { id: string }[] } }) => r.data.items.map((x) => x.id);
    expect(ids(await json(await admin("GET", `/engine/runs?${range}&triggeredBy=schedule`)))).toEqual(["b4", "b2"]);
    expect(ids(await json(await admin("GET", `/engine/runs?${range}&triggeredBy=event,api_key`)))).toEqual([
      "b3",
      "b1",
    ]);
    const t1 = await json(await admin("GET", `/engine/runs?${range}&triggeredBy=schedule&limit=1`));
    expect(ids(t1)).toEqual(["b4"]);
    expect(t1.data.nextCursor).toBeTruthy();
    const t2 = await json(
      await admin("GET", `/engine/runs?${range}&triggeredBy=schedule&limit=1&cursor=${t1.data.nextCursor}`),
    );
    expect(ids(t2)).toEqual(["b2"]);
    const t3 = await json(
      await admin("GET", `/engine/runs?${range}&triggeredBy=schedule&limit=1&cursor=${t2.data.nextCursor}`),
    );
    expect(ids(t3)).toEqual([]);
    expect(t3.data.nextCursor).toBeNull();
    expect((await admin("GET", "/engine/runs?triggeredBy=bogus")).status).toBe(422);

    const detail = (await json(await admin("GET", "/engine/runs/b2"))).data;
    expect(detail.business).toMatchObject({ kind: "agent", refId: agentId });
    expect(detail.stepAttempts).toEqual([]);
  });
});