UHD_Media_Tool/apps/server/test/system/storage.test.ts

import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { DEFAULT_STORAGE_SETTINGS, SystemHealth, SystemStorage } from "@umt/contracts";
import { type RawBuilder, sql } from "@umt/db";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { scheduleRegistry } from "../../src/modules/schedules/registry.ts";
import { createSystemModule } from "../../src/modules/system/index.ts";
import { CLEANUP_JOB_IDS, deleteInBatches, RETENTION_JOBS } from "../../src/modules/system/retention.ts";
import { budgetLevel, resetDirScans } from "../../src/modules/system/storage.ts";
import { bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts";
import { DEV_KEY, json } from "../helpers.ts";

// CACHE-2 存储预算与保留:保留清理分批与边界、存储面板形状、health 存储档位、手动触发清理。

const hasPg = await pgAvailable();
const DAY = 86_400_000;

describe("预算档位", () => {
  it("≥ 预算 critical,≥ 告警线 warn,取不到 unknown", () => {
    expect(budgetLevel(null, 100, 80)).toBe("unknown");
    expect(budgetLevel(79, 100, 80)).toBe("ok");
    expect(budgetLevel(80, 100, 80)).toBe("warn");
    expect(budgetLevel(100, 100, 80)).toBe("critical");
  });

  it("每个清理目标都对应一个定时任务", () => {
    expect(Object.keys(CLEANUP_JOB_IDS).sort()).toEqual(
      [
        "observation",
        "sse_event",
        "heartbeat",
        "task_run",
        "agent_event",
        "audit",
        "capability_stats",
        "pipeline",
        "identity_pool",
        "analyze",
        "images",
        "kv_cache",
      ].sort(),
    );
  });
});

describe.skipIf(!hasPg)("CACHE-2 存储(真实 PG)", () => {
  let h: Harness;
  let admin: ReturnType<typeof bearer>;
  const dir = mkdtempSync(join(tmpdir(), "umt-cache2-"));

  beforeAll(async () => {
    h = await createHarness({ engine: true });
    const { createApp } = await import("../../src/http/app.ts");
    const { createAuthSettingsModule } = await import("../../src/modules/auth-settings/index.ts");
    const sys = createSystemModule({
      engineSchema: h.engineSchema,
      apiHeartbeat: false,
      storagePaths: { dataDir: dir, sessionDir: join(dir, "agent-sessions") },
      retentionBatch: { batchSize: 3, pauseMs: 0 },
      healthDeps: {
        dataDir: dir,
        diskUsage: async (path) => ({
          status: "ok",
          path,
          totalBytes: 1000,
          freeBytes: 500,
          availableBytes: 500,
          usedBytes: 500,
          usedRatio: 0.5,
        }),
      },
    });
    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 }), sys],
      ctx: h.ctx,
    }) as Harness["app"];
    admin = bearer(h.app, DEV_KEY);
    await sys.start?.(h.ctx);
    // 目录:图片 3 KB、备份 2 份、会话 1 KB、证据 1 KB
    mkdirSync(join(dir, "cache", "images", "t", "p", "w92"), { recursive: true });
    writeFileSync(join(dir, "cache", "images", "t", "p", "w92", "a.jpg"), new Uint8Array(3072));
    for (const b of ["20260901T000000Z", "20260902T000000Z", ".20260903T000000Z.partial"]) {
      mkdirSync(join(dir, "backups", b), { recursive: true });
      writeFileSync(join(dir, "backups", b, "db.dump"), new Uint8Array(1024));
    }
    mkdirSync(join(dir, "agent-sessions"), { recursive: true });
    writeFileSync(join(dir, "agent-sessions", "s.jsonl"), new Uint8Array(1024));
    mkdirSync(join(dir, "tmp", "evidence"), { recursive: true });
    writeFileSync(join(dir, "tmp", "evidence", "e.jpg"), new Uint8Array(1024));
    resetDirScans();
  }, 60_000);
  afterAll(async () => {
    await h?.close();
    rmSync(dir, { recursive: true, force: true });
  });

  const job = (target: string) => {
    const j = RETENTION_JOBS.find((x) => x.target === target)!;
    return (now = new Date(), settings = DEFAULT_STORAGE_SETTINGS) =>
      j.run({ db: h.db, engineSchema: h.engineSchema, settings, now, batch: { batchSize: 3, pauseMs: 0 } });
  };
  const count = async (q: RawBuilder<{ n: number }>) => Number((await q.execute(h.db)).rows[0]!.n);

  it("保留任务登记在维护组,各自一个 key", () => {
    const reg = scheduleRegistry(h.ctx);
    for (const j of RETENTION_JOBS) {
      expect(reg.get(j.jobId)?.defaults, j.jobId).toMatchObject({ group: "maintain" });
    }
    expect(reg.get("storage.analyze")?.defaults.schedule).toMatchObject({ everyMs: 7 * DAY });
    expect(reg.get("system.heartbeat_cleanup")?.defaults.group).toBe("maintain");
  });

  it("deleteInBatches:按批删除,只删满足条件的行;总时限到了停在中途,下次接着删", async () => {
    await sql`insert into sse_event (type, at) select 'x', now() - interval '30 days' from generate_series(1, 10)`.execute(
      h.db,
    );
    await sql`insert into sse_event (type, at) values ('keep', now())`.execute(h.db);
    // 时钟每次调用前进 1 秒,预算 1.5 秒:删完 2 批就停
    let t = 0;
    const r1 = await deleteInBatches(h.db, sql.raw("sse_event"), sql`at < now() - interval '7 days'`, {
      batchSize: 3,
      pauseMs: 0,
      budgetMs: 1500,
      now: () => (t += 1000),
    });
    expect(r1).toEqual({ deleted: 6, batches: 2, truncated: true });
    const r2 = await deleteInBatches(h.db, sql.raw("sse_event"), sql`at < now() - interval '7 days'`, {
      batchSize: 3,
      pauseMs: 0,
    });
    expect(r2).toEqual({ deleted: 4, batches: 2, truncated: false });
    expect(await count(sql<{ n: number }>`select count(*)::int as n from sse_event`)).toBe(1);
  });

  it("每批语句超时:停止本次运行,不抛错", async () => {
    await sql`insert into sse_event (type, at) values ('slow', now() - interval '30 days')`.execute(h.db);
    const r = await deleteInBatches(
      h.db,
      sql.raw("sse_event"),
      sql`at < now() - interval '7 days' and pg_sleep(0.3) is not null`,
      { batchSize: 3, statementTimeoutMs: 50 },
    );
    expect(r).toEqual({ deleted: 0, batches: 0, truncated: true });
    await sql`delete from sse_event where type = 'slow'`.execute(h.db);
  });

  it("观察记录:保留边界(90 天前一刻删,之后保留)", async () => {
    const feed = (
      await sql<{
        id: string;
      }>`insert into feed_source (name, url_display) values ('rss', 'https://x/rss') returning id`.execute(h.db)
    ).rows[0]!.id;
    const now = new Date();
    const ins = (guid: string, at: Date) =>
      sql`insert into observation (feed_id, guid, content_digest, title, created_at)
        values (${feed}, ${guid}, ${guid}, ${guid}, ${at})`.execute(h.db);
    for (let i = 0; i < 5; i++) await ins(`old${i}`, new Date(now.getTime() - 90 * DAY - 60_000 - i));
    await ins("edge", new Date(now.getTime() - 90 * DAY + 60_000));
    await ins("new", now);
    expect(await job("observation")(now)).toMatch(/删除 5 行（2 批）/);
    const left = await sql<{ guid: string }>`select guid from observation order by guid`.execute(h.db);
    expect(left.rows.map((r) => r.guid)).toEqual(["edge", "new"]);
  });

  it("任务运行:只删过期且已结束的业务任务与引擎 run", async () => {
    const now = new Date();
    const old = new Date(now.getTime() - 61 * DAY);
    const ins = (key: string, status: string, at: Date) =>
      sql`insert into task (kind, idempotency_key, status, created_at) values ('k', ${key}, ${status}, ${at})`.execute(
        h.db,
      );
    await ins("t1", "succeeded", old);
    await ins("t2", "failed", old);
    await ins("t3", "running", old);
    await ins("t4", "succeeded", now);
    const t = (n: string) => sql.raw(`"${h.engineSchema}".${n}`);
    const run = (id: string, status: string, at: Date) =>
      sql`insert into ${t("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, attempts, created_at, updated_at)
        values ('default', ${id}, 'w', 'v1', ${status}, '{}'::jsonb, 1, ${at}, ${at})`.execute(h.db);
    await run("r-old-done", "completed", old);
    await run("r-old-running", "running", old);
    await run("r-new", "completed", now);
    expect(await job("task_run")(now)).toMatch(/业务任务 2、引擎 run 1/);
    expect(
      (await sql<{ idempotency_key: string }>`select idempotency_key from task order by 1`.execute(h.db)).rows.map(
        (r) => r.idempotency_key,
      ),
    ).toEqual(["t3", "t4"]);
    expect(
      (await sql<{ id: string }>`select id from ${t("workflow_runs")} order by id`.execute(h.db)).rows.map((r) => r.id),
    ).toEqual(expect.arrayContaining(["r-old-running", "r-new"]));
    expect(
      await count(sql<{ n: number }>`select count(*)::int as n from ${t("workflow_runs")} where id = 'r-old-done'`),
    ).toBe(0);
  });

  it("Agent 事件:只删已结束会话里过期的事件;审计与能力统计分开保留", async () => {
    const now = new Date();
    const runs = await sql<{ id: string; status: string }>`insert into agent_run (trigger, goal, status)
      values ('user', 'g', 'succeeded'), ('user', 'g', 'running') returning id, status`.execute(h.db);
    const done = runs.rows.find((r) => r.status === "succeeded")!.id;
    const live = runs.rows.find((r) => r.status === "running")!.id;
    const old = new Date(now.getTime() - 91 * DAY);
    await sql`insert into agent_run_event (run_id, seq, type, at) values
      (${done}, 1, 'run_started', ${old}), (${done}, 2, 'run_started', ${now}), (${live}, 1, 'run_started', ${old})`.execute(
      h.db,
    );
    expect(await job("agent_event")(now)).toMatch(/智能体事件 1、/);
    expect(await count(sql<{ n: number }>`select count(*)::int as n from agent_run_event`)).toBe(2);

    await sql`delete from audit_event`.execute(h.db);
    const a = (days: number, targetType: string | null) =>
      sql`insert into audit_event (actor_type, action, target_type, at)
        values ('system', 'x', ${targetType}, ${new Date(now.getTime() - days * DAY)})`.execute(h.db);
    await a(181, null);
    await a(179, null);
    await a(100, "capability");
    await a(80, "capability");
    await a(181, "capability");
    expect(await job("audit")(now)).toMatch(/审计事件删除 1 行/);
    expect(await job("capability_stats")(now)).toMatch(/能力审计 2、/);
    const left = await sql<{ t: string | null; d: number }>`select target_type as t,
        round(extract(epoch from now() - at) / 86400)::int as d from audit_event order by d`.execute(h.db);
    expect(left.rows).toEqual([
      { t: "capability", d: 80 },
      { t: null, d: 179 },
    ]);
  });

  it("待识别池:unmatched 超过保留期没有新证据的归档,其余不动", async () => {
    const site = (
      await sql<{ id: string }>`insert into site (key, name, adapter, base_url)
        values ('s-cache2', 'S', 'nexusphp', 'https://s.example') returning id`.execute(h.db)
    ).rows[0]!.id;
    const now = new Date();
    const cand = async (rid: string, reason: string, lastAttemptDays: number) => {
      const sr = (
        await sql<{ id: string }>`insert into source_release (site_id, site_release_id, title)
          values (${site}, ${rid}, ${rid}) returning id`.execute(h.db)
      ).rows[0]!.id;
      await sql`insert into identity_candidate (source_release_id, reason, evidence_version, last_attempt_at)
        values (${sr}, ${reason}, 'v1', ${new Date(now.getTime() - lastAttemptDays * DAY)})`.execute(h.db);
    };
    await cand("r1", "unmatched", 31);
    await cand("r2", "unmatched", 29);
    await cand("r3", "ambiguous", 31);
    expect(await job("identity_pool")(now)).toBe("归档 1 条未识别条目");
    const st = await sql<{ t: string; s: string }>`select sr.title as t, c.status as s from identity_candidate c
      join source_release sr on sr.id = c.source_release_id order by sr.title`.execute(h.db);
    expect(st.rows).toEqual([
      { t: "r1", s: "archived" },
      { t: "r2", s: "pending" },
      { t: "r3", s: "pending" },
    ]);
  });

  it("ANALYZE 任务", async () => {
    expect(await job("analyze")()).toMatch(/ANALYZE 完成/);
  });

  it("GET /system/storage:表体积 top 15、库体积对照预算、kv_cache、目录用量、清理任务", async () => {
    await sql`insert into kv_cache (region, key, value, expires_at) values
      ('tmdb.request', 'a', '{}'::jsonb, now() + interval '1 day'),
      ('tmdb.request', 'b', '{}'::jsonb, now() - interval '1 day')`.execute(h.db);
    const res = await admin("GET", "/system/storage");
    expect(res.status).toBe(200);
    const d = SystemStorage.parse((await json(res)).data);
    expect(d.db.tables.length).toBeGreaterThan(0);
    expect(d.db.tables.length).toBeLessThanOrEqual(15);
    expect(d.db.usedBytes).toBeGreaterThan(0);
    expect(d.db.budgetBytes).toBe(25 * 1024 ** 3);
    expect(d.db.warnBytes).toBe(20 * 1024 ** 3);
    expect(d.kvCache.find((r) => r.region === "tmdb.request")).toMatchObject({ entries: 2, expired: 1 });
    expect(d.images).toMatchObject({ usedBytes: 3072, files: 1, level: "ok", budgetBytes: 20 * 1024 ** 3 });
    expect(d.backups).toMatchObject({ usedBytes: 3072, count: 2, level: "ok" });
    expect(d.sessionsEvidence).toMatchObject({ usedBytes: 2048, level: "ok" });
    expect(d.dataDir.map((x) => x.name)).toEqual(
      expect.arrayContaining(["agent-sessions", "backups", "cache/images", "tmp/evidence"]),
    );
    const obs = d.cleanup.find((c) => c.target === "observation")!;
    expect(obs).toMatchObject({ jobId: "storage.retention.observation", retention: "90 天" });
    expect(d.cleanup.find((c) => c.target === "heartbeat")?.retention).toBe("24 小时");
    expect(d.cleanup.find((c) => c.target === "pipeline")).toMatchObject({ jobId: "storage.retention.pipeline" });
    // 语句统计:测试库没有预加载 pg_stat_statements 时给出原因,不报错
    if (!d.db.statements.enabled) expect(d.db.statements.note).toMatch(/pg_stat_statements/);
    expect(d.level).toBe("ok");
  });

  it("预算超了:存储面板与 health 都报出来", async () => {
    const r = await admin("PATCH", "/settings", {
      storage: { budgets: { pgDataWarnGb: 0.01, pgDataGb: 50 } },
    });
    expect(r.status).toBe(200);
    // 把 PG 告警线压到库体积之下验证 db warn
    const d = SystemStorage.parse((await json(await admin("GET", "/system/storage"))).data);
    expect(d.db.level).toBe("warn");
    expect(d.level).toBe("warn");
    const hl = SystemHealth.parse((await json(await admin("GET", "/system/health"))).data);
    expect(hl.storage.level).toBe("warn");
    expect(hl.storage.alerts.join("")).toMatch(/数据库/);
    expect(["warn", "critical"]).toContain(hl.level);
    await admin("PATCH", "/settings", { storage: { budgets: { pgDataWarnGb: 20, pgDataGb: 25 } } });
  });

  it("POST /system/storage/cleanup/{target}:排入对应定时任务;重复请求合并;未知目标 422", async () => {
    await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule)
      values ('storage.retention.audit', '审计清理', 'maintain', 'task', true, true,
        '{"type":"interval","everyMs":86400000}'::jsonb, '{"type":"interval","everyMs":86400000}'::jsonb)
      on conflict (id) do nothing`.execute(h.db);
    const r1 = await json(await admin("POST", "/system/storage/cleanup/audit"));
    expect(r1.data).toEqual({ target: "audit", jobId: "storage.retention.audit", queued: true });
    const r2 = await json(await admin("POST", "/system/storage/cleanup/audit"));
    expect(r2.data.queued).toBe(false);
    expect((await admin("POST", "/system/storage/cleanup/nope")).status).toBe(422);
    expect((await admin("POST", "/system/storage/cleanup/observation")).status).toBe(404);
  });
});