feat(cache): PG kv_cache ported from MP TTLCache, TMDB request and recognize caches

Migration 0042_cache adds UNLOGGED kv_cache. PgCache mirrors MP
runtime/cache.py (region, per-key TTL, TLRU maxsize eviction, @cached as
cachedCall); only the storage moves from memory to PG. TMDB request
responses (MP TMDb.request @cached) and recognize results (MP TmdbCache:
same key, {id:0} negative cache with EMPTY_RESULT_CACHE_TTL, type-conflict
checks, backfill after manual assignment) live in kv_cache, shared by all
processes. Settings cache.tmdbTtlHours / tmdbMaxSize / emptyResultTtlSec
map to META_CACHE_EXPIRE / CONF.tmdb / EMPTY_RESULT_CACHE_TTL. TMDB
details now also store posterPath / backdropPath.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

packages/db/migrations/0042_cache.ts

@@ -0,0 +1,30 @@
+import type { Kysely } from "kysely";
+import { dropTables, execDdl } from "../src/migration-helpers.ts";
+
+// C2-A 缓存:移植 MP app/runtime/cache.py 的 TTLCache(cachetools TLRUCache:按 region 分区、每个 key 独立 TTL、
+// maxsize 超限先删过期再删最久未用),存储从进程内存换成 PG UNLOGGED 表(不写 WAL,崩溃后清空,正合缓存语义),三进程共享。
+// expires_at 过期即视为未命中;accessed_at 为最近读写时间(LRU 依据);定时任务 cache.sweep 每 10 分钟删过期行。
+
+// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型
+export async function up(db: Kysely<any>): Promise<void> {
+  await execDdl(
+    db,
+    `
+create unlogged table kv_cache (
+  region text not null,
+  key text not null,
+  value jsonb not null,
+  expires_at timestamptz not null,
+  updated_at timestamptz not null default now(),
+  accessed_at timestamptz not null default now(),
+  primary key (region, key)
+);
+create index kv_cache_expires_idx on kv_cache (expires_at);
+`,
+  );
+}
+
+// biome-ignore lint/suspicious/noExplicitAny: 同上
+export async function down(db: Kysely<any>): Promise<void> {
+  await execDdl(db, dropTables(["kv_cache"]));
+}

packages/db/src/cache.ts

@@ -0,0 +1,175 @@
+import { type Kysely, sql } from "kysely";
+import type { DB } from "./types.ts";
+
+/**
+ * PG 缓存后端:移植 MP app/runtime/cache.py 的内存后端(MemoryBackend + _MemoryTLRUCache)与 @cached 装饰器语义,
+ * 只把存储从进程内存换成 PG UNLOGGED 表 kv_cache(迁移 0042),三进程共享。行为对照(docs/porting/moviepilot-tmdb-cache.md):
+ * - region 分区;set 时每个 key 独立 TTL(秒),ttl ≤ 0 = 删除该 key;未给 ttl 用后端缺省(DEFAULT_CACHE_TTL 365 天);
+ * - maxsize 按 region 计条数,超出时先删已过期、再删最久未用(TLRU);get 命中刷新"最近使用";
+ * - get 未命中 / 已过期返回 null;clear(region) 清一个区;items(region) 列出未过期条目;
+ * - cached():skipNone / skipEmpty / skipIf 不写缓存,emptyTtl + emptyIf 空结果短 TTL,fresh 跳过读缓存。
+ * 与 MP 的差异只在存储介质:值用 JSON(jsonb)而不是 pickle;过期行由定时任务 sweep() 物理删除。
+ */
+
+/** MP DEFAULT_CACHE_TTL(秒) */
+export const DEFAULT_CACHE_TTL = 365 * 24 * 60 * 60;
+/** MP DEFAULT_CACHE_SIZE */
+export const DEFAULT_CACHE_SIZE = 1024;
+
+// biome-ignore lint/suspicious/noExplicitAny: 测试会传入隔离 schema 的实例
+type AnyDb = Kysely<DB> | Kysely<any>;
+
+export interface SetOptions {
+  /** 秒;缺省 DEFAULT_CACHE_TTL;≤ 0 删除该 key */
+  ttl?: number;
+  /** region 条数上限;缺省 DEFAULT_CACHE_SIZE */
+  maxsize?: number;
+}
+
+export class PgCache {
+  constructor(private readonly db: AnyDb) {}
+
+  /** MemoryBackend.set */
+  async set(region: string, key: string, value: unknown, opts: SetOptions = {}): Promise<void> {
+    const ttl = opts.ttl ?? DEFAULT_CACHE_TTL;
+    if (ttl <= 0) {
+      await this.delete(region, key);
+      return;
+    }
+    await sql`insert into kv_cache (region, key, value, expires_at, updated_at, accessed_at)
+      values (${region}, ${key}, ${JSON.stringify(value ?? null)}::jsonb, now() + make_interval(secs => ${ttl}), now(), now())
+      on conflict (region, key) do update set value = excluded.value, expires_at = excluded.expires_at,
+        updated_at = excluded.updated_at, accessed_at = excluded.accessed_at`.execute(this.db);
+    await this.evict(region, opts.maxsize ?? DEFAULT_CACHE_SIZE);
+  }
+
+  /** TLRUCache 写入时的淘汰:先删已过期,再按最近使用从旧到新删到 maxsize */
+  private async evict(region: string, maxsize: number): Promise<void> {
+    await sql`delete from kv_cache where region = ${region} and expires_at <= now()`.execute(this.db);
+    await sql`delete from kv_cache where region = ${region} and key in (
+        select key from kv_cache where region = ${region}
+         order by accessed_at desc, updated_at desc offset ${Math.max(1, Math.floor(maxsize))})`.execute(this.db);
+  }
+
+  /** MemoryBackend.get:未命中 / 过期返回 null;命中刷新最近使用时间 */
+  async get<T = unknown>(region: string, key: string): Promise<T | null> {
+    const r = await sql<{ value: T }>`update kv_cache set accessed_at = now()
+      where region = ${region} and key = ${key} and expires_at > now() returning value`.execute(this.db);
+    return r.rows.length ? (r.rows[0]!.value as T) : null;
+  }
+
+  async exists(region: string, key: string): Promise<boolean> {
+    const r =
+      await sql`select 1 from kv_cache where region = ${region} and key = ${key} and expires_at > now()`.execute(
+        this.db,
+      );
+    return r.rows.length > 0;
+  }
+
+  async delete(region: string, key: string): Promise<void> {
+    await sql`delete from kv_cache where region = ${region} and key = ${key}`.execute(this.db);
+  }
+
+  /** 清一个区,返回删除行数 */
+  async clear(region: string): Promise<number> {
+    const r = await sql`delete from kv_cache where region = ${region}`.execute(this.db);
+    return Number(r.numAffectedRows ?? 0);
+  }
+
+  /** MemoryBackend.items:未过期条目(附过期时间与大小,管理界面用),按 key 排序 */
+  async items<T = unknown>(
+    region: string,
+    limit = 5000,
+  ): Promise<{ key: string; value: T; expiresAt: Date; updatedAt: Date; bytes: number }[]> {
+    const r = await sql<{ key: string; value: T; expires_at: Date; updated_at: Date; bytes: number }>`
+      select key, value, expires_at, updated_at, pg_column_size(value)::int as bytes
+        from kv_cache where region = ${region} and expires_at > now() order by key limit ${limit}`.execute(this.db);
+    return r.rows.map((x) => ({
+      key: x.key,
+      value: x.value,
+      expiresAt: x.expires_at,
+      updatedAt: x.updated_at,
+      bytes: Number(x.bytes),
+    }));
+  }
+
+  /** 未过期条目数(len(region_cache)) */
+  async count(region: string): Promise<number> {
+    const r = await sql<{ n: number }>`select count(*)::int as n from kv_cache
+      where region = ${region} and expires_at > now()`.execute(this.db);
+    return Number(r.rows[0]?.n ?? 0);
+  }
+
+  /** 删除全部过期行(存储换成 PG 后需要的物理清理;cache.sweep 定时任务) */
+  async sweep(): Promise<number> {
+    const r = await sql`delete from kv_cache where expires_at <= now()`.execute(this.db);
+    return Number(r.numAffectedRows ?? 0);
+  }
+}
+
+/** MP @cached 的参数 */
+export interface CachedOptions<T> {
+  region: string;
+  maxsize?: number;
+  /** 秒 */
+  ttl?: number;
+  /** 缺省 true:结果为 null / undefined 不缓存 */
+  skipNone?: boolean;
+  /** 缺省 false:结果为假值(空数组 / 空对象 / 空串…)不缓存 */
+  skipEmpty?: boolean;
+  /** 返回真值时不缓存(如业务失败的响应) */
+  skipIf?: (v: T) => boolean;
+  /** 空结果的独立 TTL(秒) */
+  emptyTtl?: number;
+  /** 判断空结果;缺省按假值 */
+  emptyIf?: (v: T) => boolean;
+  /** MP fresh():true 时不读缓存(仍写) */
+  fresh?: boolean;
+}
+
+const isFalsy = (v: unknown) =>
+  !v || (Array.isArray(v) && v.length === 0) || (typeof v === "object" && Object.keys(v as object).length === 0);
+
+/** MP @cached 的调用体:命中直接返回;否则执行 fn,按 should_cache / get_cache_ttl 写入 */
+export async function cachedCall<T>(
+  cache: PgCache,
+  key: string,
+  o: CachedOptions<T>,
+  fn: () => Promise<T>,
+): Promise<T> {
+  const should = (v: T | null | undefined) => {
+    if ((o.skipNone ?? true) && (v === null || v === undefined)) return false;
+    if (o.skipEmpty && isFalsy(v)) return false;
+    if (o.skipIf && v !== null && v !== undefined && o.skipIf(v)) return false;
+    return true;
+  };
+  if (!o.fresh) {
+    const hit = await cache.get<T>(o.region, key);
+    if (should(hit)) return hit as T;
+  }
+  const result = await fn();
+  if (!should(result)) return result;
+  let ttl = o.ttl;
+  if (o.emptyTtl !== undefined) {
+    if (result === null || result === undefined) ttl = o.emptyTtl;
+    else if (o.emptyIf) ttl = o.emptyIf(result) ? o.emptyTtl : o.ttl;
+    else ttl = isFalsy(result) ? o.emptyTtl : o.ttl;
+  }
+  await cache.set(o.region, key, result, {
+    ...(ttl !== undefined ? { ttl } : {}),
+    ...(o.maxsize ? { maxsize: o.maxsize } : {}),
+  });
+  return result;
+}
+
+const caches = new WeakMap<object, PgCache>();
+
+/** 同一个 Kysely 实例共用一个 PgCache */
+export function pgCacheFor(db: AnyDb): PgCache {
+  let c = caches.get(db);
+  if (!c) {
+    c = new PgCache(db);
+    caches.set(db, c);
+  }
+  return c;
+}

packages/db/src/index.ts

@@ -1,5 +1,14 @@
 // 统一从本包取 Kysely 与 sql,避免各包各装一份 kysely
 export { type Kysely, sql, type Transaction } from "kysely";
+export {
+  type CachedOptions,
+  cachedCall,
+  DEFAULT_CACHE_SIZE,
+  DEFAULT_CACHE_TTL,
+  PgCache,
+  pgCacheFor,
+  type SetOptions as CacheSetOptions,
+} from "./cache.ts";
 export { type CreateDbOptions, createDb } from "./client.ts";
 export { createMigrator, MIGRATIONS_DIR, type MigratorOptions, migrateToLatest, resetSchema } from "./migrator.ts";
 export {

packages/db/src/types.ts

@@ -827,6 +827,15 @@ export interface IdentityRevisionTable {
   updated_at: GeneratedTimestamptz;
 }
 
+export interface KvCacheTable {
+  region: string;
+  key: string;
+  value: Json;
+  expires_at: Timestamptz;
+  updated_at: GeneratedTimestamptz;
+  accessed_at: GeneratedTimestamptz;
+}
+
 export interface LibraryRootTable {
   id: Generated<string>;
   name: string;
@@ -1642,6 +1651,7 @@ export interface DB {
   identity_candidate: IdentityCandidateTable;
   identity_redirect: IdentityRedirectTable;
   identity_revision: IdentityRevisionTable;
+  kv_cache: KvCacheTable;
   library_root: LibraryRootTable;
   login_attempt: LoginAttemptTable;
   mcp_server_config: McpServerConfigTable;
@@ -1787,6 +1797,9 @@ export type IdentityRedirectUpdate = Updateable<IdentityRedirectTable>;
 export type IdentityRevisionRow = Selectable<IdentityRevisionTable>;
 export type NewIdentityRevision = Insertable<IdentityRevisionTable>;
 export type IdentityRevisionUpdate = Updateable<IdentityRevisionTable>;
+export type KvCacheRow = Selectable<KvCacheTable>;
+export type NewKvCache = Insertable<KvCacheTable>;
+export type KvCacheUpdate = Updateable<KvCacheTable>;
 export type LibraryRootRow = Selectable<LibraryRootTable>;
 export type NewLibraryRoot = Insertable<LibraryRootTable>;
 export type LibraryRootUpdate = Updateable<LibraryRootTable>;

packages/db/test/cache.test.ts

@@ -0,0 +1,110 @@
+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] });
+  });
+});

packages/db/test/schema.test.ts

@@ -117,6 +117,8 @@ const EXPECTED_TABLES = [
   "webauthn_challenge",
   // C2-A 跨进程站点限流(0041)
   "rate_limit_bucket",
+  // C2-A PG 短期缓存(0042,UNLOGGED)
+  "kv_cache",
 ].sort();
 
 /** 不走 entityTable() 的表:追加型/引擎自有表,没有 uuid id + created_at/updated_at 三件套 */
@@ -143,6 +145,8 @@ const NON_ENTITY_TABLES = [
   "webauthn_challenge",
   // 限流桶:key 主键,只有 updated_at
   "rate_limit_bucket",
+  // 缓存:(region, key) 主键
+  "kv_cache",
 ];
 
 describe.skipIf(skip)("schema migrations (real PG)", () => {