UHD_Media_Tool/packages/db/test/cache.test.ts

import { Kysely, PostgresDialect, sql } from "kysely";
import pg from "pg";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { cachedCall, type DB, migrateToLatest, PgCache } from "../src/index.ts";

// PG 缓存后端(0042 kv_cache,UNLOGGED;移植 MP runtime/cache.py 的 TTLCache / @cached 语义)。
const url = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
const skip = process.env.UMT_SKIP_DB_TESTS === "1";
const schema = `umt_cache_${process.pid}_${Date.now().toString(36)}`;

describe.skipIf(skip)("PgCache(真实 PG)", () => {
  let admin: Kysely<unknown>;
  let db: Kysely<DB>;

  beforeAll(async () => {
    admin = new Kysely({ dialect: new PostgresDialect({ pool: new pg.Pool({ connectionString: url, max: 1 }) }) });
    await sql.raw(`create schema ${schema}`).execute(admin);
    db = new Kysely<DB>({
      dialect: new PostgresDialect({
        pool: new pg.Pool({ connectionString: url, max: 4, options: `-c search_path=${schema},public` }),
      }),
    });
    await migrateToLatest(db, { migrationTableSchema: schema });
  });

  afterAll(async () => {
    await db?.destroy();
    await sql.raw(`drop schema if exists ${schema} cascade`).execute(admin);
    await admin?.destroy();
  });

  it("kv_cache 是 UNLOGGED 表", async () => {
    const r = await sql<{ p: string }>`select relpersistence as p from pg_class c
      join pg_namespace n on n.oid = c.relnamespace where n.nspname = ${schema} and c.relname = 'kv_cache'`.execute(db);
    expect(r.rows[0]?.p).toBe("u");
  });

  it("set/get 跨实例共享;每个 key 独立 TTL,过期即未命中;ttl ≤ 0 删除;clear / items / count / sweep", async () => {
    const a = new PgCache(db);
    const b = new PgCache(db);
    await a.set("t", "k1", { x: 1, list: [1, 2] }, { ttl: 60 });
    expect(await b.get("t", "k1")).toEqual({ x: 1, list: [1, 2] });
    expect(await b.get("t", "nope")).toBeNull();
    await a.set("t", "short", "v", { ttl: 0.05 });
    await new Promise((r) => setTimeout(r, 120));
    expect(await b.get("t", "short")).toBeNull();
    expect(await a.exists("t", "k1")).toBe(true);
    await a.set("t", "k1", "x", { ttl: 0 });
    expect(await a.exists("t", "k1")).toBe(false);
    await a.set("t", "k2", 2, { ttl: 60 });
    expect((await a.items("t")).map((i) => i.key)).toEqual(["k2"]);
    expect(await a.count("t")).toBe(1);
    await sql`update kv_cache set expires_at = now() - interval '1 second' where region = 't'`.execute(db);
    expect(await a.sweep()).toBe(1);
    await a.set("t", "k3", 3, { ttl: 60 });
    expect(await a.clear("t")).toBe(1);
  });

  it("maxsize:超限先删过期,再删最久未用(get 命中刷新最近使用,对照 TLRUCache)", async () => {
    const c = new PgCache(db);
    const tick = () => new Promise((r) => setTimeout(r, 5));
    await c.set("cap", "expired", 0, { ttl: 0.01, maxsize: 3 });
    await tick();
    await tick();
    for (const k of ["a", "b", "c"]) {
      await c.set("cap", k, k, { ttl: 60, maxsize: 3 });
      await tick();
    }
    await c.get("cap", "a"); // a 最近使用
    await tick();
    await c.set("cap", "d", "d", { ttl: 60, maxsize: 3 });
    expect((await c.items("cap")).map((i) => i.key)).toEqual(["a", "c", "d"]);
    const all = await sql<{ n: number }>`select count(*)::int as n from kv_cache where region = 'cap'`.execute(db);
    expect(all.rows[0]!.n).toBe(3);
  });

  it("cachedCall(@cached):skipNone / skipIf 不写;空结果按 emptyTtl;fresh 跳过读", async () => {
    const c = new PgCache(db);
    let n = 0;
    const o = {
      region: "cc",
      ttl: 3600,
      emptyTtl: 1800,
      emptyIf: (v: { results: unknown[] }) => v.results.length === 0,
    };
    const full = () =>
      cachedCall(c, "full", o, async () => {
        n++;
        return { results: [1] };
      });
    await full();
    await full();
    expect(n).toBe(1);
    await cachedCall(c, "empty", o, async () => ({ results: [] }));
    const ttl = await sql<{ k: string; s: number }>`select key as k, extract(epoch from expires_at - now()) as s
      from kv_cache where region = 'cc' order by key`.execute(db);
    expect(Number(ttl.rows.find((r) => r.k === "empty")!.s)).toBeLessThanOrEqual(1800);
    expect(Number(ttl.rows.find((r) => r.k === "full")!.s)).toBeGreaterThan(1800);
    await cachedCall(c, "none", o, async () => null as unknown as { results: unknown[] });
    await cachedCall(c, "fail", { ...o, skipIf: () => true }, async () => ({ results: [1] }));
    expect(await c.exists("cc", "none")).toBe(false);
    expect(await c.exists("cc", "fail")).toBe(false);
    await cachedCall(c, "full", { ...o, fresh: true }, async () => {
      n++;
      return { results: [2] };
    });
    expect(n).toBe(2);
    expect(await c.get("cc", "full")).toEqual({ results: [2] });
  });
});