UHD_Media_Tool/apps/server/test/system/db-retention.test.ts

import { DEFAULT_STORAGE_SETTINGS } from "@umt/contracts";
import { type RawBuilder, sql } from "@umt/db";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import {
  countRules,
  OUTBOX_PURGEABLE_KEYS,
  previewJob,
  RETENTION_JOBS,
  type RetentionRuleContext,
} from "../../src/modules/system/retention.ts";
import { createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts";

// 数据库清理(docs/runbook/storage.md「保留规则」):引擎 run 只删终态且无业务引用 / 父 run 已结束的,
// 元数据跟随 run;流水表各自的保留条件;被引用 / 未结束的不删;dry-run 统计与实际删除一致;分批。
// 另外用 EXPLAIN 确认改写后的查询能走索引(别名 trigram、巡检回填部分索引)。

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

describe.skipIf(!hasPg)("数据库清理(真实 PG)", () => {
  let h: Harness;
  const now = new Date();
  const ago = (days: number) => new Date(now.getTime() - days * DAY);

  beforeAll(async () => {
    h = await createHarness({ engine: true });
  }, 60_000);
  afterAll(async () => {
    await h?.close();
  });

  const job = (target: string) => RETENTION_JOBS.find((x) => x.target === target)!;
  const ctx = (): RetentionRuleContext => ({
    db: h.db,
    engineSchema: h.engineSchema,
    settings: DEFAULT_STORAGE_SETTINGS,
    now,
  });
  const run = (target: string) => job(target).run({ ...ctx(), batch: { batchSize: 2, pauseMs: 0 } });
  const ids = async (q: RawBuilder<{ id: string }>) => (await q.execute(h.db)).rows.map((r) => r.id).sort();
  const ow = (n: string) => sql.raw(`"${h.engineSchema}"."${n}"`);

  it("引擎 run:只删终态、过期、无保留中业务任务引用、父 run 已结束的;步骤级联,元数据跟随 run", async () => {
    const wr = (id: string, status: string, at: Date, parentStep: string | null = null) =>
      sql`insert into ${ow("workflow_runs")} (namespace_id, id, workflow_name, version, status, config, attempts,
            parent_step_attempt_namespace_id, parent_step_attempt_id, created_at, updated_at, finished_at)
          values ('default', ${id}, 'w', 'v1', ${status}, '{}'::jsonb, 1, ${parentStep ? "default" : null}, ${parentStep},
                  ${at}, ${at}, ${status === "running" ? null : at})`.execute(h.db);
    const step = (id: string, runId: string, status: string) =>
      sql`insert into ${ow("step_attempts")} (namespace_id, id, workflow_run_id, step_name, kind, status, config, created_at, updated_at)
          values ('default', ${id}, ${runId}, 's', 'function', ${status}, '{}'::jsonb, ${ago(20)}, ${ago(20)})`.execute(
        h.db,
      );
    const meta = (runId: string, at: Date) =>
      sql`insert into tasks_run_meta (run_id, task_name, task_version, idempotency_key, created_at)
          values (${runId}, 'w', 1, ${runId}, ${at})`.execute(h.db);

    await wr("old-done", "completed", ago(20));
    await step("old-done-s1", "old-done", "completed");
    await step("old-done-s2", "old-done", "completed");
    await meta("old-done", ago(20));
    await wr("old-failed", "failed", ago(15));
    await wr("fresh-done", "completed", ago(3));
    await meta("fresh-done", ago(3));
    await wr("old-running", "running", ago(20));
    await meta("old-running", ago(20));
    // 被保留中的业务任务引用(任务 60 天保留期内)
    await wr("old-task-ref", "completed", ago(20));
    await sql`insert into task (kind, idempotency_key, status, engine_run_id, created_at)
      values ('k', 'ref', 'succeeded', 'old-task-ref', ${ago(20)})`.execute(h.db);
    // 父 run 还在跑:子 run 已结束也保留(父 run 等子 run 时会读它的状态)
    await wr("parent-running", "running", ago(20));
    await step("parent-step", "parent-running", "running");
    await wr("child-of-running", "completed", ago(20), "parent-step");
    // 父 run 已结束的子 run 照常删
    await wr("parent-done", "completed", ago(20));
    await step("parent-done-step", "parent-done", "completed");
    await wr("child-of-done", "completed", ago(20), "parent-done-step");
    // 元数据的 run 早已不在
    await meta("gone", ago(30));

    const preview = await previewJob(job("task_run"), ctx());
    const rows = (label: string) => preview.find((p) => p.label === label)?.rows;
    expect(rows("引擎 run")).toBe(4);
    expect(rows("引擎步骤（随 run 级联）")).toBe(3);
    expect(rows("运行元数据")).toBe(2);

    expect(await run("task_run")).toMatch(/业务任务 0、引擎 run 4、运行元数据 2 行已删除/);
    expect(await ids(sql<{ id: string }>`select id from ${ow("workflow_runs")}`)).toEqual(
      ["child-of-running", "fresh-done", "old-running", "old-task-ref", "parent-running"].sort(),
    );
    expect(await ids(sql<{ id: string }>`select id from ${ow("step_attempts")}`)).toEqual(["parent-step"]);
    expect(await ids(sql<{ id: string }>`select run_id as id from tasks_run_meta`)).toEqual(
      ["fresh-done", "old-running"].sort(),
    );
    // 再看一次:没有可删的了
    expect((await previewJob(job("task_run"), ctx())).filter((p) => p.rows > 0)).toEqual([]);
  });

  it("业务任务:被其它表引用的不删(外键是 set null,删了会断开引用)", async () => {
    const t = await sql<{
      id: string;
      idempotency_key: string;
    }>`insert into task (kind, idempotency_key, status, created_at)
      values ('k', 'old-free', 'succeeded', ${ago(61)}), ('k', 'old-ref', 'failed', ${ago(61)})
      returning id, idempotency_key`.execute(h.db);
    const ref = t.rows.find((r) => r.idempotency_key === "old-ref")!.id;
    await sql`insert into agent_run (trigger, goal, status, task_id) values ('user', 'g', 'succeeded', ${ref})`.execute(
      h.db,
    );
    expect(await run("task_run")).toMatch(/业务任务 1、/);
    const left = await sql<{ k: string }>`select idempotency_key as k from task where kind = 'k' order by 1`.execute(
      h.db,
    );
    expect(left.rows.map((r) => r.k)).toEqual(["old-ref", "ref"]);
  });

  it("流水数据:outbox / 识别尝试 / 已识别条目 / 站点快照各按规则删,dry-run 统计与删除一致,分批", async () => {
    // outbox:已投递 + 过期 + (无键或一次性键)才删
    const ob = (type: string, status: string, days: number, key: string | null) =>
      sql`insert into outbox_event (type, status, at, idempotency_key) values (${type}, ${status}, ${ago(days)}, ${key})`.execute(
        h.db,
      );
    await ob("del-null", "dispatched", 31, null);
    await ob("del-release", "dispatched", 40, "release.observed:r1");
    await ob("del-feed", "dispatched", 40, "feed.due:f1:123");
    await ob("keep-lifecycle", "dispatched", 40, "media.watched/e1/subscribed/s1");
    await ob("keep-decision-updated", "dispatched", 40, "decision.updated:d1");
    await ob("keep-pending", "pending", 40, null);
    await ob("keep-dead", "dead", 40, null);
    await ob("keep-fresh", "dispatched", 29, null);
    expect(OUTBOX_PURGEABLE_KEYS).toContain("release.observed:%");

    const site = (
      await sql<{ id: string }>`insert into site (key, name, adapter, base_url)
        values ('s-retention', 'S', 'nexusphp', 'https://s.example') returning id`.execute(h.db)
    ).rows[0]!.id;
    const entity = (
      await sql<{ id: string }>`insert into media_entity (kind, title) values ('movie', 'M') returning id`.execute(h.db)
    ).rows[0]!.id;
    const release = async (rid: string, bound: boolean) =>
      (
        await sql<{ id: string }>`insert into source_release (site_id, site_release_id, title, media_entity_id)
          values (${site}, ${rid}, ${rid}, ${bound ? entity : null}) returning id`.execute(h.db)
      ).rows[0]!.id;

    // 识别尝试:5 次都过期 → 删最旧的 2 次,留最近 3 次;另一条资源只有 1 次过期的 → 保留
    const r1 = await release("r1", true);
    for (let i = 0; i < 5; i++) {
      await sql`insert into resolution_attempt (source_release_id, outcome, evidence_version, reason, created_at)
        values (${r1}, 'no_match', 'v1', ${`a${i}`}, ${ago(40 + i)})`.execute(h.db);
    }
    const r2 = await release("r2", false);
    await sql`insert into resolution_attempt (source_release_id, outcome, evidence_version, created_at)
      values (${r2}, 'no_match', 'v1', ${ago(90)})`.execute(h.db);

    // 待识别条目:只删已识别 + 资源仍绑定作品 + 无补全记录引用 + 无未结事项的
    const cand = async (rid: string, status: string, bound: boolean, days: number) => {
      const sr = await release(rid, bound);
      return (
        await sql<{
          id: string;
        }>`insert into identity_candidate (source_release_id, status, reason, evidence_version, updated_at)
          values (${sr}, ${status}, 'no_match', 'v1', ${ago(days)}) returning id`.execute(h.db)
      ).rows[0]!.id;
    };
    await cand("c-del", "resolved", true, 61);
    await cand("c-fresh", "resolved", true, 59);
    await cand("c-unbound", "resolved", false, 61);
    await cand("c-dismissed", "dismissed", true, 61);
    await cand("c-archived", "archived", false, 61);
    const enriched = await cand("c-enriched", "resolved", true, 61);
    await sql`insert into metadata_enrichment (identity_candidate_id, normalized_title, query)
      values (${enriched}, 'x', '{}'::jsonb)`.execute(h.db);

    // 站点快照:过期的删,但每个站点最近一次成功的列表快照与运行中的保留
    const snap = (kind: string, status: string, days: number) =>
      sql`insert into site_snapshot (site_id, kind, status, started_at) values (${site}, ${kind}, ${status}, ${ago(days)})`.execute(
        h.db,
      );
    await snap("periodic", "ok", 50); // 删
    await snap("periodic", "ok", 40); // 最近一次成功的列表快照(站点刷新间隔按它判断):留
    await snap("live_search", "ok", 35); // 删
    await snap("periodic", "failed", 33); // 删
    await snap("periodic", "running", 45); // 留
    await snap("live_search", "ok", 10); // 未过期:留(实时补查不算列表快照)

    const preview = await countRules(h.db, job("pipeline").rules!(ctx()));
    expect(Object.fromEntries(preview.map((p) => [p.label, p.rows]))).toEqual({
      已投递事件: 3,
      识别尝试: 2,
      已识别的未识别资源: 1,
      站点抓取记录: 3,
    });
    // 站点快照 3 行、batchSize 2:分两批
    expect(await run("pipeline")).toBe("已投递事件 3、识别尝试 2、已识别的未识别资源 1、站点抓取记录 3 行已删除");

    const types = await sql<{ id: string }>`select type as id from outbox_event`.execute(h.db);
    expect(types.rows.map((r) => r.id).sort()).toEqual(
      ["keep-dead", "keep-decision-updated", "keep-fresh", "keep-lifecycle", "keep-pending"].sort(),
    );
    const attempts = await sql<{ reason: string | null }>`select reason from resolution_attempt
      where source_release_id = ${r1} order by created_at desc`.execute(h.db);
    expect(attempts.rows.map((r) => r.reason)).toEqual(["a0", "a1", "a2"]);
    expect(
      await ids(sql<{ id: string }>`select id from resolution_attempt where source_release_id = ${r2}`),
    ).toHaveLength(1);
    const cands = await sql<{ t: string }>`select r.title as t from identity_candidate c
      join source_release r on r.id = c.source_release_id order by 1`.execute(h.db);
    expect(cands.rows.map((r) => r.t)).toEqual(["c-archived", "c-dismissed", "c-enriched", "c-fresh", "c-unbound"]);
    const snaps = await sql<{ k: string; s: string; d: number }>`select kind as k, status as s,
        round(extract(epoch from ${now}::timestamptz - started_at) / 86400)::int as d
      from site_snapshot order by started_at`.execute(h.db);
    expect(snaps.rows).toEqual([
      { k: "periodic", s: "running", d: 45 },
      { k: "periodic", s: "ok", d: 40 },
      { k: "live_search", s: "ok", d: 10 },
    ]);
    expect((await countRules(h.db, job("pipeline").rules!(ctx()))).every((p) => p.rows === 0)).toBe(true);
  });

  it("站点快照:有了更新的成功列表快照,旧的那次就可以删", async () => {
    const site = (await sql<{ id: string }>`select id from site where key = 's-retention'`.execute(h.db)).rows[0]!.id;
    await sql`insert into site_snapshot (site_id, kind, status, started_at) values (${site}, 'manual', 'partial', ${ago(2)})`.execute(
      h.db,
    );
    const counts = await countRules(h.db, job("pipeline").rules!(ctx()));
    expect(counts.find((p) => p.label === "站点抓取记录")?.rows).toBe(1);
  });

  it("定时任务登记:流水数据清理每天低峰一次(cron)", () => {
    expect(job("pipeline")).toMatchObject({
      jobId: "storage.retention.pipeline",
      schedule: { type: "cron", cron: "40 4 * * *" },
    });
  });

  // ── 改写后的查询能走索引(小表上规划器会选全表扫描,关掉 seqscan 只为确认索引「可用」;
  //    trigram GIN 只能位图扫描,别名用例同时关掉普通索引扫描,免得规划器整扫唯一索引)──

  const plan = async (q: RawBuilder<unknown>, bitmapOnly = false) =>
    await h.db.transaction().execute(async (trx) => {
      await sql`set local enable_seqscan = off`.execute(trx);
      if (bitmapOnly) await sql`set local enable_indexscan = off; set local enable_indexonlyscan = off`.execute(trx);
      const r = await sql<{ "QUERY PLAN": string }>`explain ${q}`.execute(trx);
      return r.rows.map((x) => x["QUERY PLAN"]).join("\n");
    });

  it("别名模糊匹配:`%` 写法走 title_alias_normalized_trgm_idx,只写 similarity() 用不上", async () => {
    const e = (await sql<{ id: string }>`select id from media_entity limit 1`.execute(h.db)).rows[0]!.id;
    await sql`insert into title_alias (media_entity_id, alias, normalized, source)
      select ${e}, 'a' || i, 'alias title ' || i, 'tmdb' from generate_series(1, 50) i`.execute(h.db);
    await sql`analyze title_alias`.execute(h.db);
    const q = "alias title 7";
    const after = await plan(
      sql`select distinct media_entity_id from title_alias
           where normalized = ${sql.lit(q)} or (normalized % ${sql.lit(q)} and similarity(normalized, ${sql.lit(q)}) >= 0.45)`,
      true,
    );
    expect(after).toContain("title_alias_normalized_trgm_idx");
    const before = await plan(
      sql`select distinct media_entity_id from title_alias
           where normalized = ${sql.lit(q)} or similarity(normalized, ${sql.lit(q)}) >= 0.45`,
      true,
    );
    expect(before).not.toContain("title_alias_normalized_trgm_idx");
    // 两种写法结果相同
    const a = await sql<{ n: string }>`select normalized as n from title_alias
      where normalized % ${q} and similarity(normalized, ${q}) >= 0.45 order by 1`.execute(h.db);
    const b = await sql<{ n: string }>`select normalized as n from title_alias
      where similarity(normalized, ${q}) >= 0.45 order by 1`.execute(h.db);
    expect(a.rows).toEqual(b.rows);
    expect(a.rows.length).toBeGreaterThan(0);
  });

  it("巡检回填:走部分索引 scheduled_job_run_harvest_idx", async () => {
    const out = await plan(sql`select r.id from scheduled_job_run r join agent_run a on a.id = r.agent_run_id
      where r.agent_status is null and r.agent_run_id is not null and a.status = any('{succeeded,failed}')`);
    expect(out).toContain("scheduled_job_run_harvest_idx");
  });
});