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 | Kysely; 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 { 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 { 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(region: string, key: string): Promise { 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; } /** 续期:未过期的条目把 expires_at 推到 now + ttl(只延长不缩短);返回是否有行被续期 */ async renew(region: string, key: string, ttl: number): Promise { if (ttl <= 0) return false; const r = await sql`update kv_cache set expires_at = greatest(expires_at, now() + make_interval(secs => ${ttl})) where region = ${region} and key = ${key} and expires_at > now()`.execute(this.db); return Number(r.numAffectedRows ?? 0) > 0; } async exists(region: string, key: string): Promise { 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 { await sql`delete from kv_cache where region = ${region} and key = ${key}`.execute(this.db); } /** 清一个区,返回删除行数 */ async clear(region: string): Promise { const r = await sql`delete from kv_cache where region = ${region}`.execute(this.db); return Number(r.numAffectedRows ?? 0); } /** MemoryBackend.items:未过期条目(附过期时间与大小,管理界面用),按 key 排序 */ async items( 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 { 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 { 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 { 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( cache: PgCache, key: string, o: CachedOptions, fn: () => Promise, ): Promise { 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(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(); /** 同一个 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; }