From 61f585a70752e9db7fb0ed5fe6a2c32e8a672a47 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 21:51:32 +0800 Subject: [PATCH 1/3] feat(sites): cross-process site rate limit via PG token bucket Migration 0041_rate_limit adds rate_limit_bucket. PgTokenBucket (@umt/db) refills and takes tokens under SELECT ... FOR UPDATE on the DB clock and polls with jitter within a wait budget; penalize() carries Retry-After. Site adapters get createSharedSiteRateLimiter (key site:), shared by api / worker / worker:agent, with an in-process serial queue in front and a local fallback when the bucket is unavailable. Co-Authored-By: Claude Opus 5.5 --- .../src/modules/sources-identity/runtime.ts | 23 +++- apps/server/test/sources-identity/api.test.ts | 13 ++ packages/adapters/src/sites/index.ts | 9 +- .../adapters/src/sites/rate-limit.test.ts | 48 ++++++- packages/adapters/src/sites/rate-limit.ts | 78 +++++++++++ packages/db/migrations/0041_rate_limit.ts | 30 +++++ packages/db/src/index.ts | 7 + packages/db/src/rate-limit.ts | 127 ++++++++++++++++++ packages/db/src/types.ts | 13 ++ packages/db/test/fixtures/bucket-worker.ts | 21 +++ packages/db/test/rate-limit.test.ts | 98 ++++++++++++++ packages/db/test/schema.test.ts | 4 + 12 files changed, 466 insertions(+), 5 deletions(-) create mode 100644 packages/db/migrations/0041_rate_limit.ts create mode 100644 packages/db/src/rate-limit.ts create mode 100644 packages/db/test/fixtures/bucket-worker.ts create mode 100644 packages/db/test/rate-limit.test.ts diff --git a/apps/server/src/modules/sources-identity/runtime.ts b/apps/server/src/modules/sources-identity/runtime.ts index b4062a73..a62e0021 100644 --- a/apps/server/src/modules/sources-identity/runtime.ts +++ b/apps/server/src/modules/sources-identity/runtime.ts @@ -1,5 +1,6 @@ import { createRssFetcher, + createSharedSiteRateLimiter, createSiteAdapter, createSiteRateLimiter, createTmdbClient, @@ -12,10 +13,10 @@ import { type TmdbClient, } from "@umt/adapters"; import { DEFAULT_SETTINGS, type SitesSettings } from "@umt/contracts"; -import type { SiteRow } from "@umt/db"; +import { pgTokenBucketFor, type SiteRow } from "@umt/db"; import type { ServerContext } from "../types.ts"; -// 模块运行时依赖:站点适配器与限流器(每站一个实例)、RSS 抓取器、TMDB 客户端、时钟。 +// 模块运行时依赖:站点适配器与限流器(每站一个实例,令牌在 PG 跨进程共享)、RSS 抓取器、TMDB 客户端、时钟。 // 测试通过 createSourcesIdentityModule(overrides) 注入 fetch / TMDB / 时钟。 export interface RuntimeOverrides { @@ -28,6 +29,8 @@ export interface RuntimeOverrides { rss?: Parameters[0]; /** 调度 tick 间隔(毫秒),默认 30s */ schedulerIntervalMs?: number; + /** false = 站点限流只用进程内令牌桶(缺省走 PG 跨进程令牌桶 rate_limit_bucket) */ + sharedRateLimit?: boolean; } interface CachedAdapter { @@ -158,7 +161,21 @@ export class SourcesRuntime { ]); const hit = this.adapters.get(site.id); if (hit && hit.sig === sig) return hit.adapter; - const limiter = createSiteRateLimiter(limiterOptions(def, site.rate_limit, defaults.defaultRateLimit)); + const lo = limiterOptions(def, site.rate_limit, defaults.defaultRateLimit); + // 跨进程:api / worker / worker:agent 对同一站点共用 PG 令牌桶(key = site:);进程内先串行排队 + const limiter = + this.overrides.sharedRateLimit === false + ? createSiteRateLimiter(lo) + : createSharedSiteRateLimiter({ + ...lo, + key: `site:${site.id}`, + bucket: pgTokenBucketFor(this.ctx.db), + onError: (err) => + this.ctx.logger.warn( + { siteId: site.id, err: { message: (err as Error).message } }, + "站点共享限流不可用，本次退回进程内限流", + ), + }); const cfg = (site.config ?? {}) as { apiBaseUrl?: string }; const baseUrl = def.umt.kind === "mteam" ? cfg.apiBaseUrl : sameUrl(site.base_url, def.domain) ? undefined : site.base_url; diff --git a/apps/server/test/sources-identity/api.test.ts b/apps/server/test/sources-identity/api.test.ts index 8e4693d8..7b099684 100644 --- a/apps/server/test/sources-identity/api.test.ts +++ b/apps/server/test/sources-identity/api.test.ts @@ -131,6 +131,19 @@ describe.skipIf(!up)("来源管理与站点工作台 API", () => { expect(observed.length).toBe(r.newCount); const site = await json(await h.req(`/sites/${siteId}`)); expect(site.body.data.lastSnapshotAt).not.toBeNull(); + // 限流间隔 > 0 时令牌在 PG 跨进程共享(rate_limit_bucket,key = site:);0 = 不限流只走本地 + await h.db + .updateTable("site") + .set({ rate_limit: JSON.stringify({ minIntervalMs: 5, concurrency: 5 }) }) + .where("id", "=", siteId) + .execute(); + expect((await snapshotSite(h.rt, { siteId, kind: "manual" })).status).toBe("ok"); + const bucket = await h.db + .selectFrom("rate_limit_bucket") + .selectAll() + .where("key", "=", `site:${siteId}`) + .executeTakeFirst(); + expect(bucket).toMatchObject({ capacity: 5, refill_per_sec: 200 }); }); it("工作台列表:快照优先,标注来源与更新时间;筛选与分页", async () => { diff --git a/packages/adapters/src/sites/index.ts b/packages/adapters/src/sites/index.ts index ff32bdcd..f6ff55a7 100644 --- a/packages/adapters/src/sites/index.ts +++ b/packages/adapters/src/sites/index.ts @@ -38,7 +38,14 @@ export { mteamSearchParams, parseMteamResult, } from "./mteam/spider.ts"; -export { createSiteRateLimiter, type RateLimiterOptions, type SiteRateLimiter } from "./rate-limit.ts"; +export { + createSharedSiteRateLimiter, + createSiteRateLimiter, + type RateLimiterOptions, + type SharedBucketPort, + type SharedRateLimiterOptions, + type SiteRateLimiter, +} from "./rate-limit.ts"; export { createSiteAdapter, getSiteAdapter, diff --git a/packages/adapters/src/sites/rate-limit.test.ts b/packages/adapters/src/sites/rate-limit.test.ts index a92be361..fb5d98b7 100644 --- a/packages/adapters/src/sites/rate-limit.test.ts +++ b/packages/adapters/src/sites/rate-limit.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { createSiteRateLimiter } from "./rate-limit.ts"; +import { createSharedSiteRateLimiter, createSiteRateLimiter, type SharedBucketPort } from "./rate-limit.ts"; import { parseSizeMp as parseSize } from "./spider/spider.ts"; import { parseCount, parseSiteTime } from "./util.ts"; @@ -34,6 +34,52 @@ describe("站点令牌桶", () => { }); }); +describe("跨进程站点限流器", () => { + it("同进程串行去共享桶扣令牌,key 与速率按站点参数;Retry-After 同时顺延共享桶", async () => { + const calls: { key: string; capacity: number; refillPerSec: number }[] = []; + let inflight = 0; + let maxInflight = 0; + const penalties: number[] = []; + const bucket: SharedBucketPort = { + async acquire(key, _cost, o) { + inflight++; + maxInflight = Math.max(maxInflight, inflight); + await new Promise((r) => setTimeout(r, 5)); + inflight--; + calls.push({ key, capacity: o.capacity, refillPerSec: o.refillPerSec }); + return { ok: true, waitedMs: 5 }; + }, + async penalize(_k, ms) { + penalties.push(ms); + }, + }; + const l = createSharedSiteRateLimiter({ key: "site:x", bucket, minIntervalMs: 500, burst: 2 }); + await Promise.all([l.acquire(), l.acquire(), l.acquire()]); + expect(maxInflight).toBe(1); + expect(calls).toEqual(Array(3).fill({ key: "site:x", capacity: 2, refillPerSec: 2 })); + expect(l.stats.granted).toBe(3); + l.penalize(3000); + expect(penalties).toEqual([3000]); + }); + + it("共享桶不可用时退回进程内令牌桶", async () => { + const errors: unknown[] = []; + const bucket: SharedBucketPort = { + acquire: () => Promise.reject(new Error("db down")), + penalize: () => Promise.reject(new Error("db down")), + }; + const l = createSharedSiteRateLimiter({ + key: "site:y", + bucket, + minIntervalMs: 10, + onError: (e) => errors.push(e), + }); + await l.acquire(); + expect(l.stats.granted).toBe(1); + expect(errors).toHaveLength(1); + }); +}); + describe("解析辅助", () => { it("parseSize/parseCount/parseSiteTime", () => { expect(parseSize("78.42 GB")).toBe(84202833838); diff --git a/packages/adapters/src/sites/rate-limit.ts b/packages/adapters/src/sites/rate-limit.ts index 2ae5f81b..0cb2b4c4 100644 --- a/packages/adapters/src/sites/rate-limit.ts +++ b/packages/adapters/src/sites/rate-limit.ts @@ -93,3 +93,81 @@ export function createSiteRateLimiter(opts: RateLimiterOptions): SiteRateLimiter }, }; } + +/** 跨进程令牌桶端口(@umt/db 的 PgTokenBucket 结构上满足;adapters 不依赖 db 包) */ +export interface SharedBucketPort { + acquire( + key: string, + cost: number, + opts: { capacity: number; refillPerSec: number; waitMs?: number; signal?: AbortSignal }, + ): Promise<{ ok: boolean; waitedMs: number }>; + penalize(key: string, ms: number, spec: { capacity: number; refillPerSec: number }): Promise; +} + +export interface SharedRateLimiterOptions extends RateLimiterOptions { + /** 桶 key,站点用 `site:` */ + key: string; + bucket: SharedBucketPort; + /** 单轮等待预算,缺省 60s;超出后重新排队(直到 signal 中止) */ + waitMs?: number; + /** 共享桶出错(DB 不可达等)时回调;本次退回进程内令牌桶,不让站点请求失败 */ + onError?: (err: unknown) => void; +} + +/** + * 跨进程站点限流器:进程内先串行排队(同进程并发不去 PG 抢锁),再到共享桶扣令牌; + * 429 Retry-After 同时顺延本地与共享桶。minIntervalMs = 0(不限流)时只走本地。 + */ +export function createSharedSiteRateLimiter(opts: SharedRateLimiterOptions): SiteRateLimiter { + const local = createSiteRateLimiter(opts); + const interval = Math.max(0, opts.minIntervalMs); + const spec = { capacity: Math.max(1, opts.burst ?? 1), refillPerSec: interval > 0 ? 1000 / interval : 0 }; + const stats = { granted: 0, waitedMs: 0 }; + let queue: Promise = Promise.resolve(); + const onError = opts.onError ?? (() => {}); + + const viaLocal = async (signal?: AbortSignal) => { + const before = local.stats.waitedMs; + await local.acquire(signal); + stats.waitedMs += local.stats.waitedMs - before; + }; + + return { + stats, + tryAcquire() { + if (!local.tryAcquire()) return false; + stats.granted++; + return true; + }, + penalize(ms) { + local.penalize(ms); + if (interval > 0) opts.bucket.penalize(opts.key, ms, spec).catch(onError); + }, + acquire(signal) { + const run = queue.then(async () => { + if (interval === 0) { + await viaLocal(signal); + } else { + try { + for (;;) { + const r = await opts.bucket.acquire(opts.key, 1, { + ...spec, + waitMs: opts.waitMs ?? 60_000, + ...(signal ? { signal } : {}), + }); + stats.waitedMs += r.waitedMs; + if (r.ok) break; + } + } catch (e) { + if (signal?.aborted) throw e; + onError(e); + await viaLocal(signal); + } + } + stats.granted++; + }); + queue = run.catch(() => {}); + return run; + }, + }; +} diff --git a/packages/db/migrations/0041_rate_limit.ts b/packages/db/migrations/0041_rate_limit.ts new file mode 100644 index 00000000..854def31 --- /dev/null +++ b/packages/db/migrations/0041_rate_limit.ts @@ -0,0 +1,30 @@ +import type { Kysely } from "kysely"; +import { dropTables, execDdl } from "../src/migration-helpers.ts"; + +// C2-A 跨进程站点限流:PG 令牌桶(api / worker / worker:agent 共用)。 +// - key:`site:<站点 id>` 等;tokens 为上次结算时的令牌数(可为负,表示被 Retry-After 顺延); +// - capacity / refill_per_sec 由调用方每次扣令牌时带上(站点限流设置变化即时生效); +// - blocked_until:站点要求暂停(429 Retry-After)时在此之前不发放令牌。 +// 扣令牌走 `select … for update`,补充按 DB 时钟(clock_timestamp)结算,多进程时钟不一致也不影响。 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + ` +create table rate_limit_bucket ( + key text primary key, + tokens double precision not null, + capacity double precision not null check (capacity > 0), + refill_per_sec double precision not null check (refill_per_sec > 0), + blocked_until timestamptz, + updated_at timestamptz not null default now() +); +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl(db, dropTables(["rate_limit_bucket"])); +} diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index fb8141c8..05b34b56 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -2,4 +2,11 @@ export { type Kysely, sql, type Transaction } from "kysely"; export { type CreateDbOptions, createDb } from "./client.ts"; export { createMigrator, MIGRATIONS_DIR, type MigratorOptions, migrateToLatest, resetSchema } from "./migrator.ts"; +export { + type AcquireOptions, + type AcquireResult, + type BucketSpec, + PgTokenBucket, + pgTokenBucketFor, +} from "./rate-limit.ts"; export type * from "./types.ts"; diff --git a/packages/db/src/rate-limit.ts b/packages/db/src/rate-limit.ts new file mode 100644 index 00000000..e49ef0e2 --- /dev/null +++ b/packages/db/src/rate-limit.ts @@ -0,0 +1,127 @@ +import { type Kysely, sql } from "kysely"; +import type { DB } from "./types.ts"; + +/** + * 跨进程令牌桶(rate_limit_bucket,迁移 0041)。api / worker / worker:agent 三个进程对同一 key 共用一个桶: + * 每次扣令牌在事务里 `select … for update` 锁住该行,按 DB 时钟补充后扣减;拿不到则算出还差多久, + * 调用方在 waitMs 预算内睡到那时(带抖动)再试。进程内的突发由调用方的本地串行队列挡住(见 adapters 的站点限流器)。 + */ + +export interface BucketSpec { + /** 桶容量(突发上限),≥ 1 */ + capacity: number; + /** 每秒补充的令牌数,> 0 */ + refillPerSec: number; +} + +export interface AcquireOptions extends BucketSpec { + /** 最多等多久(毫秒);0 = 只试一次。缺省 60s */ + waitMs?: number; + signal?: AbortSignal; +} + +export interface AcquireResult { + ok: boolean; + /** 本次累计等待(毫秒) */ + waitedMs: number; +} + +// biome-ignore lint/suspicious/noExplicitAny: 测试会传入隔离 schema 的实例 +type AnyDb = Kysely | Kysely; + +function sleep(ms: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + if (signal?.aborted) return reject(signal.reason); + const t = setTimeout(() => { + signal?.removeEventListener("abort", onAbort); + resolve(); + }, ms); + const onAbort = () => { + clearTimeout(t); + reject(signal?.reason); + }; + signal?.addEventListener("abort", onAbort, { once: true }); + }); +} + +export class PgTokenBucket { + constructor( + private readonly db: AnyDb, + private readonly opts: { jitterMs?: number; random?: () => number } = {}, + ) {} + + /** + * 试扣一次:成功返回 0,否则返回还需等待的毫秒数(被 penalize 顺延时为到期时间)。 + * 行不存在时按满桶创建;容量 / 速率以本次传入为准(设置改了即时生效)。 + */ + async tryTake(key: string, cost: number, spec: BucketSpec): Promise { + const capacity = Math.max(1, spec.capacity); + const rate = Math.max(1e-6, spec.refillPerSec); + const need = Math.min(cost, capacity); + return await this.db.transaction().execute(async (trx) => { + await sql`insert into rate_limit_bucket (key, tokens, capacity, refill_per_sec) + values (${key}, ${capacity}, ${capacity}, ${rate}) on conflict (key) do nothing`.execute(trx); + const row = ( + await sql<{ tokens: number; elapsed: number; blocked_ms: number | null }>` + select tokens, + greatest(0, extract(epoch from clock_timestamp() - updated_at))::float8 as elapsed, + (extract(epoch from blocked_until - clock_timestamp()) * 1000)::float8 as blocked_ms + from rate_limit_bucket where key = ${key} for update`.execute(trx) + ).rows[0]; + if (!row) return 1000; + let tokens = Math.min(capacity, Number(row.tokens) + Number(row.elapsed) * rate); + const blocked = row.blocked_ms !== null && Number(row.blocked_ms) > 0 ? Math.ceil(Number(row.blocked_ms)) : 0; + let wait = 0; + if (blocked > 0) wait = blocked; + else if (tokens >= need) tokens -= need; + else wait = Math.ceil(((need - tokens) / rate) * 1000); + await sql`update rate_limit_bucket set tokens = ${tokens}, capacity = ${capacity}, refill_per_sec = ${rate}, + updated_at = clock_timestamp(), + blocked_until = case when blocked_until <= clock_timestamp() then null else blocked_until end + where key = ${key}`.execute(trx); + return wait; + }); + } + + /** 扣 cost 个令牌;waitMs 内拿不到返回 ok:false。signal 中止时抛出其 reason。 */ + async acquire(key: string, cost: number, opts: AcquireOptions): Promise { + const budget = Math.max(0, opts.waitMs ?? 60_000); + const jitterMax = this.opts.jitterMs ?? 50; + const random = this.opts.random ?? Math.random; + const t0 = Date.now(); + for (;;) { + opts.signal?.throwIfAborted(); + const wait = await this.tryTake(key, cost, opts); + const spent = Date.now() - t0; + if (wait <= 0) return { ok: true, waitedMs: spent }; + if (spent + wait > budget) return { ok: false, waitedMs: spent }; + // 抖动:多个进程同时醒来时错开,避免一起撞锁再一起落空 + const jitter = Math.floor(random() * Math.min(jitterMax, Math.max(1, wait * 0.2))); + await sleep(wait + jitter, opts.signal); + } + } + + /** 站点要求暂停(Retry-After):ms 内不发放令牌,已有令牌清零 */ + async penalize(key: string, ms: number, spec: BucketSpec): Promise { + const capacity = Math.max(1, spec.capacity); + const rate = Math.max(1e-6, spec.refillPerSec); + const until = sql`clock_timestamp() + make_interval(secs => ${Math.max(0, ms) / 1000})`; + await sql`insert into rate_limit_bucket (key, tokens, capacity, refill_per_sec, blocked_until, updated_at) + values (${key}, 0, ${capacity}, ${rate}, ${until}, clock_timestamp()) + on conflict (key) do update set tokens = least(rate_limit_bucket.tokens, 0), + blocked_until = greatest(coalesce(rate_limit_bucket.blocked_until, clock_timestamp()), ${until}), + updated_at = clock_timestamp()`.execute(this.db); + } +} + +const buckets = new WeakMap(); + +/** 同一个 Kysely 实例共用一个 PgTokenBucket */ +export function pgTokenBucketFor(db: AnyDb): PgTokenBucket { + let b = buckets.get(db); + if (!b) { + b = new PgTokenBucket(db); + buckets.set(db, b); + } + return b; +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index 985e168f..f76d35e3 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -1156,6 +1156,15 @@ export interface ProcessHeartbeatSampleTable { sse_connections: number | null; } +export interface RateLimitBucketTable { + key: string; + tokens: number; + capacity: number; + refill_per_sec: number; + blocked_until: Timestamptz | null; + updated_at: GeneratedTimestamptz; +} + export interface ReconcileRunTable { id: Generated; kind: ReconcileRunKind; @@ -1653,6 +1662,7 @@ export interface DB { preference_tag: PreferenceTagTable; process_heartbeat: ProcessHeartbeatTable; process_heartbeat_sample: ProcessHeartbeatSampleTable; + rate_limit_bucket: RateLimitBucketTable; reconcile_run: ReconcileRunTable; resolution_attempt: ResolutionAttemptTable; rule_atom: RuleAtomTable; @@ -1837,6 +1847,9 @@ export type ProcessHeartbeatUpdate = Updateable; export type ProcessHeartbeatSampleRow = Selectable; export type NewProcessHeartbeatSample = Insertable; export type ProcessHeartbeatSampleUpdate = Updateable; +export type RateLimitBucketRow = Selectable; +export type NewRateLimitBucket = Insertable; +export type RateLimitBucketUpdate = Updateable; export type ReconcileRunRow = Selectable; export type NewReconcileRun = Insertable; export type ReconcileRunUpdate = Updateable; diff --git a/packages/db/test/fixtures/bucket-worker.ts b/packages/db/test/fixtures/bucket-worker.ts new file mode 100644 index 00000000..06168fed --- /dev/null +++ b/packages/db/test/fixtures/bucket-worker.ts @@ -0,0 +1,21 @@ +// 子进程:对同一 key 连续扣 N 个令牌,每次成功输出一行时间戳(毫秒)。由 rate-limit.test.ts 以 Deno 子进程启动。 +import { Kysely, PostgresDialect } from "kysely"; +import pg from "pg"; +import { PgTokenBucket } from "../../src/rate-limit.ts"; + +const [url, schema, key, n, capacity, rate] = Deno.args as [string, string, string, string, string, string]; +const db = new Kysely({ + dialect: new PostgresDialect({ + pool: new pg.Pool({ connectionString: url, max: 1, options: `-c search_path=${schema},public` }), + }), +}); +const bucket = new PgTokenBucket(db); +try { + for (let i = 0; i < Number(n); i++) { + const r = await bucket.acquire(key, 1, { capacity: Number(capacity), refillPerSec: Number(rate), waitMs: 30_000 }); + if (!r.ok) throw new Error("等待超时"); + console.log(Date.now()); + } +} finally { + await db.destroy(); +} diff --git a/packages/db/test/rate-limit.test.ts b/packages/db/test/rate-limit.test.ts new file mode 100644 index 00000000..4156fd8f --- /dev/null +++ b/packages/db/test/rate-limit.test.ts @@ -0,0 +1,98 @@ +import { fileURLToPath } from "node:url"; +import { Kysely, PostgresDialect, sql } from "kysely"; +import pg from "pg"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { type DB, migrateToLatest, PgTokenBucket } from "../src/index.ts"; + +// 跨进程令牌桶(0041):单进程语义 + 三个 Deno 子进程并发对同一 key 的总速率不超限。 +const url = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt"; +const skip = process.env.UMT_SKIP_DB_TESTS === "1"; +const schema = `umt_rl_${process.pid}_${Date.now().toString(36)}`; + +describe.skipIf(skip)("PgTokenBucket(真实 PG)", () => { + let admin: Kysely; + let db: Kysely; + + 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({ + 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("容量内立即发放,之后按速率补充;waitMs=0 拿不到即返回", async () => { + const b = new PgTokenBucket(db); + const spec = { capacity: 2, refillPerSec: 10 }; + expect(await b.tryTake("t:a", 1, spec)).toBe(0); + expect(await b.tryTake("t:a", 1, spec)).toBe(0); + const wait = await b.tryTake("t:a", 1, spec); + expect(wait).toBeGreaterThan(0); + expect(wait).toBeLessThanOrEqual(100); + expect((await b.acquire("t:a", 1, { ...spec, waitMs: 0 })).ok).toBe(false); + const r = await b.acquire("t:a", 1, { ...spec, waitMs: 2000 }); + expect(r.ok).toBe(true); + }); + + it("penalize 顺延:期间不发放令牌", async () => { + const b = new PgTokenBucket(db); + const spec = { capacity: 5, refillPerSec: 100 }; + await b.penalize("t:p", 400, spec); + const wait = await b.tryTake("t:p", 1, spec); + expect(wait).toBeGreaterThan(250); + const t0 = Date.now(); + expect((await b.acquire("t:p", 1, { ...spec, waitMs: 3000 })).ok).toBe(true); + expect(Date.now() - t0).toBeGreaterThanOrEqual(200); + }); + + it("取消等待", async () => { + const b = new PgTokenBucket(db); + const spec = { capacity: 1, refillPerSec: 0.1 }; + await b.tryTake("t:c", 1, spec); + const ac = new AbortController(); + const p = b.acquire("t:c", 1, { ...spec, waitMs: 60_000, signal: ac.signal }); + setTimeout(() => ac.abort(new Error("cancelled")), 50); + await expect(p).rejects.toThrow("cancelled"); + }); + + it("三个子进程并发对同一 key:总速率不超过 容量 + 速率 × 时长", async () => { + const script = fileURLToPath(new URL("./fixtures/bucket-worker.ts", import.meta.url)); + const capacity = 2; + const rate = 20; + const perProc = 12; + const procs = [0, 1, 2].map(() => + new Deno.Command(Deno.execPath(), { + args: ["run", "-A", script, url, schema, "t:procs", String(perProc), String(capacity), String(rate)], + stdout: "piped", + stderr: "piped", + }).output(), + ); + const outs = await Promise.all(procs); + const stamps: number[] = []; + for (const o of outs) { + expect(o.code, new TextDecoder().decode(o.stderr)).toBe(0); + for (const line of new TextDecoder().decode(o.stdout).split("\n")) if (line.trim()) stamps.push(Number(line)); + } + stamps.sort((a, b) => a - b); + expect(stamps).toHaveLength(perProc * 3); + // 36 个令牌:首批 capacity 个立即发放,其余按 20/s → 至少 (36-2)/20 = 1.7s + const span = stamps.at(-1)! - stamps[0]!; + expect(span).toBeGreaterThanOrEqual(((perProc * 3 - capacity) / rate) * 1000 - 150); + // 任意 500ms 窗口内发放数 ≤ 容量 + 速率 × 0.5s(+1 容忍子进程打印时间戳的误差) + const win = 500; + for (let i = 0; i < stamps.length; i++) { + let j = i; + while (j < stamps.length && stamps[j]! - stamps[i]! < win) j++; + expect(j - i).toBeLessThanOrEqual(capacity + (rate * win) / 1000 + 1); + } + }, 60_000); +}); diff --git a/packages/db/test/schema.test.ts b/packages/db/test/schema.test.ts index 1962e803..6c733c36 100644 --- a/packages/db/test/schema.test.ts +++ b/packages/db/test/schema.test.ts @@ -115,6 +115,8 @@ const EXPECTED_TABLES = [ "admin_recovery_code", "admin_passkey", "webauthn_challenge", + // C2-A 跨进程站点限流(0041) + "rate_limit_bucket", ].sort(); /** 不走 entityTable() 的表:追加型/引擎自有表,没有 uuid id + created_at/updated_at 三件套 */ @@ -139,6 +141,8 @@ const NON_ENTITY_TABLES = [ // 一次性凭据:只记创建 / 用掉的时间 "admin_recovery_code", "webauthn_challenge", + // 限流桶:key 主键,只有 updated_at + "rate_limit_bucket", ]; describe.skipIf(skip)("schema migrations (real PG)", () => { From 946fca5e55f6480186384887657e57089c4b51f1 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 22:26:29 +0800 Subject: [PATCH 2/3] 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 --- apps/server/src/modules/index.ts | 2 + apps/server/src/modules/search/media.ts | 1 + .../src/modules/sources-identity/runtime.ts | 23 +++ .../service/identity-admin.ts | 5 +- .../sources-identity/service/metadata.ts | 3 + .../sources-identity/service/resolve.ts | 40 +++- .../service/tmdb-recognize-cache.ts | 154 +++++++++++++++ apps/server/src/shared-cache.ts | 36 ++++ apps/server/test/modules.test.ts | 3 +- .../test/sources-identity/resolve.test.ts | 36 ++++ packages/adapters/src/tmdb/client.ts | 42 ++++- packages/adapters/src/tmdb/tmdb.test.ts | 25 +++ packages/contracts/src/content.ts | 1 + packages/contracts/src/search.ts | 2 + packages/contracts/src/settings.ts | 40 +++- packages/contracts/test/db-sync.test.ts | 3 + packages/db/migrations/0042_cache.ts | 30 +++ packages/db/src/cache.ts | 175 ++++++++++++++++++ packages/db/src/index.ts | 9 + packages/db/src/types.ts | 13 ++ packages/db/test/cache.test.ts | 110 +++++++++++ packages/db/test/schema.test.ts | 4 + 22 files changed, 743 insertions(+), 14 deletions(-) create mode 100644 apps/server/src/modules/sources-identity/service/tmdb-recognize-cache.ts create mode 100644 apps/server/src/shared-cache.ts create mode 100644 packages/db/migrations/0042_cache.ts create mode 100644 packages/db/src/cache.ts create mode 100644 packages/db/test/cache.test.ts diff --git a/apps/server/src/modules/index.ts b/apps/server/src/modules/index.ts index 7654a110..c199e88c 100644 --- a/apps/server/src/modules/index.ts +++ b/apps/server/src/modules/index.ts @@ -2,6 +2,7 @@ import { acquisitionOrganizeModule } from "./acquisition-organize/index.ts"; import { agentModule } from "./agent/index.ts"; import { authSettingsModule } from "./auth-settings/index.ts"; import { browserModule } from "./browser/index.ts"; +import { cacheModule } from "./cache/index.ts"; import { fileManagerModule } from "./file-manager/index.ts"; import { mcpModule } from "./mcp/index.ts"; import { piModule } from "./pi/index.ts"; @@ -28,5 +29,6 @@ export const SERVER_MODULES: readonly ServerModule[] = [ searchModule, browserModule, mcpModule, + cacheModule, systemModule, ]; diff --git a/apps/server/src/modules/search/media.ts b/apps/server/src/modules/search/media.ts index a9206da3..21e980a6 100644 --- a/apps/server/src/modules/search/media.ts +++ b/apps/server/src/modules/search/media.ts @@ -163,6 +163,7 @@ export class TmdbSearch { year: m.year, overview: m.overview ? (m.overview.length > 300 ? `${m.overview.slice(0, 300)}…` : m.overview) : null, popularity: m.popularity, + ...(m.posterPath ? { posterPath: m.posterPath } : {}), })), ) .then((list) => { diff --git a/apps/server/src/modules/sources-identity/runtime.ts b/apps/server/src/modules/sources-identity/runtime.ts index a62e0021..40d6d8f6 100644 --- a/apps/server/src/modules/sources-identity/runtime.ts +++ b/apps/server/src/modules/sources-identity/runtime.ts @@ -11,9 +11,11 @@ import { type SiteDefinition, type SiteRateLimiter, type TmdbClient, + type TmdbSharedCache, } from "@umt/adapters"; import { DEFAULT_SETTINGS, type SitesSettings } from "@umt/contracts"; import { pgTokenBucketFor, type SiteRow } from "@umt/db"; +import { CACHE_REGIONS, sharedCache, type TmdbCachePolicy, tmdbCachePolicy } from "../../shared-cache.ts"; import type { ServerContext } from "../types.ts"; // 模块运行时依赖:站点适配器与限流器(每站一个实例,令牌在 PG 跨进程共享)、RSS 抓取器、TMDB 客户端、时钟。 @@ -71,12 +73,14 @@ export class SourcesRuntime { return this.tmdbClient; } const meta = await this.ctx.settings.get("metadata").catch(() => null); + const policy = await tmdbCachePolicy(this.ctx); const env = (k: string) => Deno.env.get(k); const ref = meta?.tmdbApiKeySecretRef ?? null; const baseUrl = meta?.tmdbBaseUrl ?? env("TMDB_BASE_URL"); const sig = JSON.stringify([ ref, baseUrl ?? null, + policy, ref ? null : [!!env("TMDB_API_KEY"), !!env("TMDB_ACCESS_TOKEN")], ]); if (this.tmdbClient !== undefined && this.tmdbSig === sig) return this.tmdbClient; @@ -103,6 +107,9 @@ export class SourcesRuntime { ...(accessToken ? { accessToken } : {}), ...(baseUrl ? { baseUrl } : {}), ...(this.overrides.fetch ? { fetch: this.overrides.fetch } : {}), + // 请求响应进 PG 缓存(region tmdb.request),三进程共享;TTL / 条数上限 / 空结果短 TTL 取设置(对照 MP) + cache: { ttlMs: policy.ttl * 1000, emptyTtlMs: policy.emptyTtl * 1000 }, + sharedCache: tmdbSharedCache(this.ctx, policy), }) : null; return this.tmdbClient; @@ -233,6 +240,22 @@ export class SourcesRuntime { } } +/** TMDB 请求响应缓存:PG kv_cache region=tmdb.request;PG 出错按未命中 / 不写处理,不让 TMDB 请求失败 */ +function tmdbSharedCache(ctx: ServerContext, policy: TmdbCachePolicy): TmdbSharedCache { + const cache = sharedCache(ctx); + const warn = (err: unknown) => + ctx.logger.warn({ err: { message: (err as Error).message } }, "TMDB 缓存读写失败，按未命中处理"); + return { + get: (key) => + cache.get(CACHE_REGIONS.tmdbRequest, key).catch((e) => { + warn(e); + return null; + }), + set: (key, value, ttlMs) => + cache.set(CACHE_REGIONS.tmdbRequest, key, value, { ttl: ttlMs / 1000, maxsize: policy.maxsize }).catch(warn), + }; +} + function sameUrl(a: string, b: string): boolean { return a.replace(/\/+$/, "") === b.replace(/\/+$/, ""); } diff --git a/apps/server/src/modules/sources-identity/service/identity-admin.ts b/apps/server/src/modules/sources-identity/service/identity-admin.ts index 1409915b..f555448a 100644 --- a/apps/server/src/modules/sources-identity/service/identity-admin.ts +++ b/apps/server/src/modules/sources-identity/service/identity-admin.ts @@ -23,6 +23,7 @@ import { resolveEntityId, } from "./entities.ts"; import { recordAttempt } from "./resolve.ts"; +import { backfillRecognizeCache } from "./tmdb-recognize-cache.ts"; // 人工纠错:指定来源资源身份(加锁)、改作品外部映射(新增修订、旧映射退役)、合并作品(旧 ID 跳转)。 // 所有变更留 identity_revision / resolution_attempt / audit_event,不直接覆盖历史。 @@ -85,7 +86,7 @@ export async function assignReleaseIdentity( for (let attempt = 0; ; attempt++) { try { - return await ctx.db.transaction().execute(async (trx) => { + const done = await ctx.db.transaction().execute(async (trx) => { let entityId: string; if (localId) { const r = await resolveEntityId(trx, localId); @@ -190,6 +191,8 @@ export async function assignReleaseIdentity( }); return { mediaEntityId: entityId, contentUnitIds: unitIds }; }); + await backfillRecognizeCache(rt, parsed, done.mediaEntityId); + return done; } catch (e) { if (e instanceof IdentityConflictError && attempt < 2) continue; throw e; diff --git a/apps/server/src/modules/sources-identity/service/metadata.ts b/apps/server/src/modules/sources-identity/service/metadata.ts index 5335d7b0..9b355a87 100644 --- a/apps/server/src/modules/sources-identity/service/metadata.ts +++ b/apps/server/src/modules/sources-identity/service/metadata.ts @@ -32,6 +32,9 @@ function tmdbFields(m: MetadataCandidate): Partial { +/** + * TMDB 召回。先查识别缓存(移植 MP TmdbCache,见 tmdb-recognize-cache.ts):命中且有标题 → 按 id 取详情作唯一远端候选; + * 命中负缓存 → 不再搜索(视为 TMDB 无结果)。强制复查(force)不读缓存但照常回写(MP recognize_media 的 cache=False)。 + */ +async function tmdbCandidates( + rt: SourcesRuntime, + parsed: ParsedRelease, + useCache: boolean, +): Promise<{ remote: MetadataCandidate[]; cacheHit: boolean; meta: RecognizeMeta | null }> { const tmdb = await rt.tmdb(); - if (!tmdb) return []; + if (!tmdb) return { remote: [], cacheHit: false, meta: null }; + const meta = recognizeMetaOf(parsed); + if (meta && useCache) { + const hit = await loadRecognizeCache(rt, meta); + if (hit) { + if (hit.id && hit.title && (hit.type === "movie" || hit.type === "tv")) { + return { remote: [await tmdb.details(hit.type, hit.id)], cacheHit: true, meta }; + } + return { remote: [], cacheHit: true, meta }; + } + } const queries = [parsed.titleCn, parsed.titleEn, parsed.title].filter((t): t is string => !!t); const seen = new Map(); for (const q of [...new Set(queries)].slice(0, 2)) { @@ -166,7 +185,7 @@ async function tmdbCandidates(rt: SourcesRuntime, parsed: ParsedRelease): Promis if (!seen.has(k)) seen.set(k, m); } } - return [...seen.values()].slice(0, 20); + return { remote: [...seen.values()].slice(0, 20), cacheHit: false, meta }; } async function loadRelease(rt: SourcesRuntime, input: ResolveInput) { @@ -240,14 +259,27 @@ export async function resolveRelease(rt: SourcesRuntime, input: ResolveInput): P const local = await localCandidates(ctx.db, parsed, release); let remote: MetadataCandidate[] = []; let lookupError: string | null = null; + let recognize: { cacheHit: boolean; meta: RecognizeMeta | null } = { cacheHit: false, meta: null }; try { - remote = await tmdbCandidates(rt, parsed); + const r = await tmdbCandidates(rt, parsed, !input.force); + remote = r.remote; + recognize = r; } catch (e) { if (!(e instanceof TmdbError)) throw e; lookupError = `${e.kind}: ${e.message}`; } const candidates = [...local, ...remote.map(tmdbToCandidate)]; const resolution = resolveIdentity(parsed, candidates); + // 识别缓存回写(MP TmdbCache.update):识别到 TMDB 条目写正常条目,确认无匹配写负缓存;TMDB 出错 / 有歧义不写 + if (recognize.meta && !recognize.cacheHit && !lookupError) { + const top = resolution.status === "auto" ? resolution.best?.candidate : undefined; + const hit = + top?.provider === "tmdb" + ? remote.find((m) => m.externalId === top.externalId && m.entityType === top.entityType) + : undefined; + const info = hit ?? (resolution.status === "no_match" ? null : undefined); + await saveRecognizeCache(rt, recognize.meta, info); + } const version = evidenceVersion(parsed, resolution.ranked); // 已识别且证据未变:幂等短路(强制复查时) diff --git a/apps/server/src/modules/sources-identity/service/tmdb-recognize-cache.ts b/apps/server/src/modules/sources-identity/service/tmdb-recognize-cache.ts new file mode 100644 index 00000000..40a92775 --- /dev/null +++ b/apps/server/src/modules/sources-identity/service/tmdb-recognize-cache.ts @@ -0,0 +1,154 @@ +import type { MetadataCandidate } from "@umt/adapters"; +import type { ParsedRelease } from "@umt/domain"; +import { CACHE_REGIONS, sharedCache, tmdbCachePolicy } from "../../../shared-cache.ts"; +import type { SourcesRuntime } from "../runtime.ts"; + +// TMDB 识别缓存:移植 MP app/modules/themoviedb/cache.py 的 TmdbCache(对照表见 docs/porting/moviepilot-tmdb-cache.md)。 +// 存储从 内存 TTLCache + pickle 文件持久化 换成 PG kv_cache(region tmdb.recognize),每条独立 expires_at, +// 三进程共享;"持久化与恢复"等价于表本身。语义照抄: +// - 键:[类型][语言]{tmdbid 或 名称}-{年份}-{开始季}(Python 的 None 原样写作 "None"); +// - 值:{id, type, year, title, poster_path, backdrop_path};负缓存 {id: 0}; +// - TTL:正常条目 = META_CACHE_EXPIRE(cache.tmdbTtlHours);负缓存 = EMPTY_RESULT_CACHE_TTL(cache.emptyResultTtlSec); +// - 条数上限 = CONF.tmdb(cache.tmdbMaxSize),超出删最旧; +// - 网络错误(None)不写;查无结果({})写负缓存; +// - 读 / 写时"元数据是电视剧、缓存是电影"视为类型冲突:读时丢弃该条,写时拒绝写入。 + +/** MP MediaType 的值(键里的类型段,与 MP 一致用中文) */ +export type MpMediaType = "电影" | "电视剧" | "未知"; + +export interface RecognizeCacheEntry { + /** TMDB id;0 = 负缓存(未识别) */ + id: number; + type?: "movie" | "tv" | null; + year?: string | null; + title?: string | null; + poster_path?: string | null; + backdrop_path?: string | null; +} + +/** 键所需的元数据(对照 MP MetaBase:type / name / year / begin_season / media_source+media_id) */ +export interface RecognizeMeta { + type: MpMediaType; + name: string; + year: number | null; + beginSeason: number | null; + /** 已知 TMDB id 时键用它代替名称(MP:media_source == TMDB 时取 media_id) */ + tmdbId?: string | null; +} + +const isAllChinese = (s: string) => /^[㐀-鿿豈-﫿]+$/.test(s); + +/** 解析结果 → 识别元数据(MP MetaBase.name:中文名全是汉字优先,否则英文名,再退回中文名) */ +export function recognizeMetaOf(p: ParsedRelease): RecognizeMeta | null { + const cn = p.titleCn ?? null; + const en = p.titleEn ?? null; + const name = cn && isAllChinese(cn) ? cn : (en ?? cn ?? p.title ?? ""); + if (!name) return null; + // MP MetaVideo:识别出季或集即为电视剧,否则类型未知 + const tv = p.season !== undefined || p.episodes !== undefined; + return { type: tv ? "电视剧" : "未知", name, year: p.year ?? null, beginSeason: p.season ?? null }; +} + +/** TmdbCache.__get_key */ +export function recognizeKey(meta: RecognizeMeta, locale: string): string { + const py = (v: number | null) => (v === null ? "None" : String(v)); + return `[${meta.type}][${locale}]${meta.tmdbId || meta.name}-${py(meta.year)}-${py(meta.beginSeason)}`; +} + +/** TmdbCache.__is_type_conflicted:只有"元数据是电视剧、结果是电影"算冲突;负缓存不带类型不算 */ +function typeConflicted(meta: RecognizeMeta, type: unknown, id: unknown): boolean { + if (meta.type !== "电视剧" || !id) return false; + return type === "movie"; +} + +function localeOf(tmdb: unknown): string { + return (tmdb as { language?: string } | null)?.language ?? "zh-CN"; +} + +/** TmdbCache.get:未命中返回 null;类型冲突的脏条目删除后按未命中处理 */ +export async function loadRecognizeCache(rt: SourcesRuntime, meta: RecognizeMeta): Promise { + const tmdb = await rt.tmdb(); + const key = recognizeKey(meta, localeOf(tmdb)); + const cache = sharedCache(rt.ctx); + const v = await cache.get(CACHE_REGIONS.tmdbRecognize, key).catch(() => null); + if (!v || typeof v !== "object") return null; + if (typeConflicted(meta, v.type, v.id)) { + rt.ctx.logger.warn({ key, title: v.title, type: v.type }, "识别缓存类型与元数据冲突，已丢弃并重新识别"); + await cache.delete(CACHE_REGIONS.tmdbRecognize, key); + return null; + } + return v; +} + +/** + * TmdbCache.update:info 为识别结果写正常条目;info 为 null(查无此项)写负缓存 {id: 0}; + * info 为 undefined(网络错误 / 没有结论)不写,允许下次重新请求。 + */ +export async function saveRecognizeCache( + rt: SourcesRuntime, + meta: RecognizeMeta, + info: + | Pick + | null + | undefined, +): Promise { + if (info === undefined) return; + const tmdb = await rt.tmdb(); + const key = recognizeKey(meta, localeOf(tmdb)); + const policy = await tmdbCachePolicy(rt.ctx); + const cache = sharedCache(rt.ctx); + if (info) { + if (typeConflicted(meta, info.entityType, info.externalId)) { + rt.ctx.logger.warn({ key, title: info.title, type: info.entityType }, "识别结果类型与元数据冲突，不写入识别缓存"); + return; + } + const entry: RecognizeCacheEntry = { + id: Number(info.externalId), + type: info.entityType, + year: info.year === null ? null : String(info.year), + title: info.title, + poster_path: info.posterPath ?? null, + backdrop_path: info.backdropPath ?? null, + }; + await cache.set(CACHE_REGIONS.tmdbRecognize, key, entry, { ttl: policy.ttl, maxsize: policy.maxsize }); + return; + } + // 负识别缓存用独立短 TTL(EMPTY_RESULT_CACHE_TTL):故障期间的空结果不被固化整个有效期 + await cache.set(CACHE_REGIONS.tmdbRecognize, key, { id: 0 }, { ttl: policy.emptyTtl, maxsize: policy.maxsize }); +} + +/** + * 人工认定后回填识别缓存(MP update_recognize_cache:覆盖名称负缓存,避免同名资源再次回查)。 + * 作品没有已接受的 TMDB 身份时不写。失败只记日志。 + */ +export async function backfillRecognizeCache( + rt: SourcesRuntime, + parsed: ParsedRelease, + entityId: string, +): Promise { + try { + const meta = recognizeMetaOf(parsed); + if (!meta || !(await rt.tmdb())) return; + const row = await rt.ctx.db + .selectFrom("external_identity as i") + .innerJoin("media_entity as e", "e.id", "i.media_entity_id") + .select(["i.entity_type", "i.external_id", "e.title", "e.year", "e.metadata"]) + .where("i.media_entity_id", "=", entityId) + .where("i.provider", "=", "tmdb") + .where("i.status", "=", "accepted") + .executeTakeFirst(); + if (!row || (row.entity_type !== "movie" && row.entity_type !== "tv")) return; + const md = (row.metadata ?? {}) as Record; + const str = (v: unknown) => (typeof v === "string" && v ? v : null); + await saveRecognizeCache(rt, meta, { + externalId: row.external_id, + entityType: row.entity_type, + title: row.title, + year: row.year, + posterPath: str(md.posterPath?.value), + backdropPath: str(md.backdropPath?.value), + }); + } catch (e) { + rt.ctx.logger.warn({ entityId, err: { message: (e as Error).message } }, "识别缓存回填失败"); + } +} diff --git a/apps/server/src/shared-cache.ts b/apps/server/src/shared-cache.ts new file mode 100644 index 00000000..193a9aad --- /dev/null +++ b/apps/server/src/shared-cache.ts @@ -0,0 +1,36 @@ +import { DEFAULT_SETTINGS } from "@umt/contracts"; +import { type PgCache, pgCacheFor } from "@umt/db"; +import type { ServerContext } from "./modules/types.ts"; + +// 进程级共享缓存入口:PG kv_cache(移植 MP app/runtime/cache.py 的 TTLCache,存储换成 PG,三进程共享)。 +// region 与 TTL 规则见 docs/runbook/cache.md 与 docs/porting/moviepilot-tmdb-cache.md。 + +export const CACHE_REGIONS = { + /** TMDB 请求响应(MP TMDb.request 的 @cached):key = path?query */ + tmdbRequest: "tmdb.request", + /** TMDB 识别结果(MP TmdbCache):key = [类型][语言]{tmdbid 或名称}-{年份}-{季} */ + tmdbRecognize: "tmdb.recognize", +} as const; + +export function sharedCache(ctx: Pick): PgCache { + return pgCacheFor(ctx.db); +} + +/** TMDB 缓存策略(MP:CONF.meta = META_CACHE_EXPIRE 小时、CONF.tmdb 条数、EMPTY_RESULT_CACHE_TTL 秒) */ +export interface TmdbCachePolicy { + /** 秒 */ + ttl: number; + maxsize: number; + /** 秒 */ + emptyTtl: number; +} + +export async function tmdbCachePolicy(ctx: Pick): Promise { + const c = await ctx.settings.get("cache").catch(() => DEFAULT_SETTINGS.cache); + const d = DEFAULT_SETTINGS.cache; + return { + ttl: (c.tmdbTtlHours ?? d.tmdbTtlHours) * 3600, + maxsize: c.tmdbMaxSize ?? d.tmdbMaxSize, + emptyTtl: c.emptyResultTtlSec ?? d.emptyResultTtlSec, + }; +} diff --git a/apps/server/test/modules.test.ts b/apps/server/test/modules.test.ts index efb4bb85..c927632c 100644 --- a/apps/server/test/modules.test.ts +++ b/apps/server/test/modules.test.ts @@ -43,7 +43,7 @@ const pingModule: ServerModule = { }; describe("ServerModule 骨架", () => { - it("清单里是四个 W2 模块 + 文件管理 + Pi 上游 + 定时任务 + 全局搜索 + 浏览器 + MCP + 进程与健康(末尾),名字唯一", () => { + it("清单里是四个 W2 模块 + 文件管理 + Pi 上游 + 定时任务 + 全局搜索 + 浏览器 + MCP + 缓存 + 进程与健康(末尾),名字唯一", () => { expect(SERVER_MODULES.map((m) => m.name)).toEqual([ "auth-settings", "sources-identity", @@ -55,6 +55,7 @@ describe("ServerModule 骨架", () => { "search", "browser", "mcp", + "cache", "system", ]); }); diff --git a/apps/server/test/sources-identity/resolve.test.ts b/apps/server/test/sources-identity/resolve.test.ts index 353d5e01..ea6e4109 100644 --- a/apps/server/test/sources-identity/resolve.test.ts +++ b/apps/server/test/sources-identity/resolve.test.ts @@ -1,4 +1,5 @@ import { CapabilityRegistry } from "@umt/agent"; +import { sql } from "@umt/db"; import { createMockTmdb, type MockTmdb, TMDB_FIXTURES, type TmdbFixture } from "@umt/testkit"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { refreshMetadata } from "../../src/modules/sources-identity/service/metadata.ts"; @@ -453,4 +454,39 @@ describe.skipIf(!up)("resolveObservation / 身份与元数据", () => { expect(search.items.length).toBeGreaterThan(0); await expect(reg.invoke("identity.resolve", { sourceReleaseId: twinsRelease }, c)).rejects.toThrow(); }); + it("识别缓存(移植 MP TmdbCache):键 [类型][语言]名称-年份-季;命中不再搜索;查无写 {id:0} 短 TTL;force 不读缓存", async () => { + const rows = async () => + ( + await sql<{ key: string; value: { id: number; type?: string; title?: string }; minutes: number }>` + select key, value, extract(epoch from expires_at - now()) / 60 as minutes + from kv_cache where region = 'tmdb.recognize'`.execute(h.db) + ).rows; + const opp = (await rows()).find((r) => r.key === "[未知][zh-CN]Oppenheimer-2023-None"); + expect(opp?.value).toMatchObject({ id: 872585, type: "movie", title: "奥本海默" }); + expect(Number(opp!.minutes)).toBeGreaterThan(23 * 60); + // 同名同年的另一条资源:命中识别缓存,不再调 TMDB 搜索 + const searches = tmdb.callsTo(/\/3\/search\//).length; + const again = await release("Oppenheimer.2023.1080p.BluRay.x264-OTHER"); + await resolveRelease(h.rt, { sourceReleaseId: again.id }); + expect(tmdb.callsTo(/\/3\/search\//).length).toBe(searches); + // 查无此项 → 负缓存 {id: 0},TTL = 空结果短 TTL(30 分钟) + const none = await release("Zzqxv.Nonexistent.Title.2019.1080p.WEB-DL-NOGRP"); + await resolveRelease(h.rt, { sourceReleaseId: none.id }); + const neg = (await rows()).find((r) => r.key === "[未知][zh-CN]Zzqxv Nonexistent Title-2019-None"); + expect(neg?.value).toEqual({ id: 0 }); + expect(Number(neg!.minutes)).toBeLessThanOrEqual(30); + const s2 = tmdb.callsTo(/\/3\/search\//).length; + await resolveRelease(h.rt, { sourceReleaseId: none.id }); + expect(tmdb.callsTo(/\/3\/search\//).length).toBe(s2); + // force:不读识别缓存(回源结果由请求层缓存兜住),照常回写识别缓存 + const at = async () => + ( + await sql<{ t: Date }>`select updated_at as t from kv_cache + where region = 'tmdb.recognize' and key = ${neg!.key}`.execute(h.db) + ).rows[0]!.t.getTime(); + const before = await at(); + await new Promise((r) => setTimeout(r, 10)); + await resolveRelease(h.rt, { sourceReleaseId: none.id, force: true }); + expect(await at()).toBeGreaterThan(before); + }); }); diff --git a/packages/adapters/src/tmdb/client.ts b/packages/adapters/src/tmdb/client.ts index 6d8f7924..f14ac4f7 100644 --- a/packages/adapters/src/tmdb/client.ts +++ b/packages/adapters/src/tmdb/client.ts @@ -59,6 +59,8 @@ export interface MetadataCandidate { overview: string | null; /** TMDB poster_path(如 `/abc.jpg`),拼图片 CDN 用;没有为 null */ posterPath: string | null; + /** TMDB backdrop_path;没有为 null(旧数据 / 测试构造可缺省) */ + backdropPath?: string | null; seasons?: CandidateSeason[]; externalIds?: { imdb?: string; tvdb?: string }; /** 是否已拉过详情(别名/季信息只在详情中完整) */ @@ -98,12 +100,30 @@ export interface TmdbConfig { minIntervalMs?: number; /** 默认 max 1024、ttl 24h;空结果(搜索 0 条)用 emptyTtlMs,默认 30min(对照 MP EMPTY_RESULT_CACHE_TTL) */ cache?: { max?: number; ttlMs?: number; emptyTtlMs?: number }; + /** + * 跨进程共享的请求缓存(服务端传 PG 缓存);给了就不再用进程内 LRU。 + * 语义对照 MP @cached(skip_none、业务失败不缓存、results 为空按 emptyTtlMs)。 + */ + sharedCache?: TmdbSharedCache; fetch?: typeof fetch; now?: () => number; } type Json = Record; +/** + * 共享请求缓存端口(对照 MP TMDb.request 的 @cached;服务端用 PG 实现):key 为 `path?query`(不含凭证); + * get 未命中返回 null;set 的 ttlMs 由客户端按 ttl / 空结果短 TTL 决定。 + */ +export interface TmdbSharedCache { + get(key: string): Promise; + set(key: string, value: unknown, ttlMs: number): Promise; +} + +/** 空结果(搜索 0 条)用短 TTL(对照 MP EMPTY_RESULT_CACHE_TTL) */ +const isEmptyResult = (v: unknown) => + Array.isArray((v as Json | null)?.results) && ((v as Json).results as unknown[]).length === 0; + const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); const s = (v: unknown): string | null => (typeof v === "string" && v !== "" ? v : null); const n = (v: unknown): number | null => (typeof v === "number" && Number.isFinite(v) ? v : null); @@ -130,6 +150,11 @@ export class TmdbClient { return redactUrl(text, [this.cfg.apiKey, this.cfg.accessToken]); } + /** 首选查询语言(识别缓存键里的语言段,对照 MP TMDB_LOCALE) */ + get language(): string { + return this.langs[0]!; + } + cacheStats() { return { size: this.cache.size, hits: this.cache.hits, misses: this.cache.misses, coalesced: this.cache.coalesced }; } @@ -139,13 +164,21 @@ export class TmdbClient { for (const k of Object.keys(params).sort()) if (params[k] !== undefined) qs.set(k, params[k]!); const key = `${path}?${qs}`; const emptyTtl = this.cfg.cache?.emptyTtlMs ?? 30 * 60_000; + if (this.cfg.sharedCache) { + const shared = this.cfg.sharedCache; + const hit = await shared.get(key); + if (hit !== null && hit !== undefined) return hit as T; + // 网络错误 / 业务失败在 fetchWithRetry 抛出,不会走到写缓存(MP skip_none / skip_if) + const v = await this.limiter.run(() => this.fetchWithRetry(path, qs)); + if (v !== null && v !== undefined) { + await shared.set(key, v, isEmptyResult(v) ? emptyTtl : (this.cfg.cache?.ttlMs ?? 24 * 3600_000)); + } + return v as T; + } return this.cache.getOrLoad( key, () => this.limiter.run(() => this.fetchWithRetry(path, qs)), - (v) => - Array.isArray((v as Json | null)?.results) && ((v as Json).results as unknown[]).length === 0 - ? emptyTtl - : undefined, + (v) => (isEmptyResult(v) ? emptyTtl : undefined), ) as Promise; } @@ -269,6 +302,7 @@ export class TmdbClient { popularity: n(r.popularity), overview: s(r.overview), posterPath: s(r.poster_path), + backdropPath: s(r.backdrop_path), detailed: false, }; } diff --git a/packages/adapters/src/tmdb/tmdb.test.ts b/packages/adapters/src/tmdb/tmdb.test.ts index 210d7597..83fa3cf8 100644 --- a/packages/adapters/src/tmdb/tmdb.test.ts +++ b/packages/adapters/src/tmdb/tmdb.test.ts @@ -110,6 +110,31 @@ describe("TmdbClient 缓存/限流/错误分类", () => { expect(mock.calls.length).toBe(n1 * 2); }); + it("共享响应缓存:给了 sharedCache 就走它,key 不含凭证,空结果 30 分钟,其余按 ttlMs", async () => { + const store = new Map(); + const { mock, tmdb } = await setup({ + cache: { ttlMs: 7 * 3600_000 }, + sharedCache: { + get: async (key) => store.get(key)?.v ?? null, + set: async (key, v, ttl) => { + store.set(key, { v, ttl }); + }, + }, + }); + await tmdb.details("movie", 872585); + const n1 = mock.calls.length; + await tmdb.details("movie", 872585); + expect(mock.calls.length).toBe(n1); + await tmdb.search("不存在的片名xyz", { type: "movie" }); + const keys = [...store.keys()]; + expect(keys.some((k) => k.includes("test-tmdb-key"))).toBe(false); + const detailKey = keys.find((k) => k.startsWith("/movie/872585?"))!; + expect(store.get(detailKey)!.ttl).toBe(7 * 3600_000); + const emptyKey = keys.find((k) => k.startsWith("/search/movie?"))!; + expect(store.get(emptyKey)!.ttl).toBe(30 * 60_000); + expect(tmdb.cacheStats().size).toBe(0); + }); + it("并发合并:同一请求并发只发一次", async () => { const { mock, tmdb } = await setup(); await Promise.all(Array.from({ length: 5 }, () => tmdb.season(100088, 1))); diff --git a/packages/contracts/src/content.ts b/packages/contracts/src/content.ts index 55009e3e..36df07a7 100644 --- a/packages/contracts/src/content.ts +++ b/packages/contracts/src/content.ts @@ -189,6 +189,7 @@ export const EntityMetadataField = z.enum([ "originalLanguage", "originCountry", "posterPath", + "backdropPath", ]); export type EntityMetadataField = z.infer; diff --git a/packages/contracts/src/search.ts b/packages/contracts/src/search.ts index b78b344e..08676531 100644 --- a/packages/contracts/src/search.ts +++ b/packages/contracts/src/search.ts @@ -52,6 +52,8 @@ export const TmdbCandidate = z.object({ year: z.number().int().nullable(), overview: z.string().nullable(), popularity: z.number().nullable(), + /** TMDB poster_path;前端经图片代理显示海报 */ + posterPath: z.string().nullable().optional(), /** 已有本地作品接受了这个 TMDB 身份时给出(跟随合并跳转),前端显示"查看"而不是"订阅" */ localEntityId: Id.nullable(), }); diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 7c362bc9..ea337df3 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -24,6 +24,16 @@ const CacheShape = z.object({ checkIntervalSec: z.number().int().min(10).max(3600), /** 缓存清理的目标使用率(低水位) */ cleanupTargetPercent: Percent, + /** TMDB 缓存(请求响应 + 识别结果)保留小时数(对照 MP META_CACHE_EXPIRE,缺省 24) */ + tmdbTtlHours: z.number().int().min(1).max(720), + /** TMDB 缓存每类条数上限,超出删最旧(对照 MP CONF.tmdb,缺省 256;大内存模式 1024) */ + tmdbMaxSize: z.number().int().min(16).max(100_000), + /** 空结果 / 未识别的短缓存秒数(对照 MP EMPTY_RESULT_CACHE_TTL,缺省 1800) */ + emptyResultTtlSec: z.number().int().min(0).max(86_400), + /** 图片代理落盘缓存(对照 MP GLOBAL_IMAGE_CACHE,缺省关闭;关闭时仍代理,只是不落盘) */ + globalImageCache: z.boolean(), + /** 图片缓存保留天数,超过即清理(对照 MP GLOBAL_IMAGE_CACHE_DAYS,缺省 7) */ + globalImageCacheDays: z.number().int().min(1).max(3650), }); /** 下载缓存卷硬阈值与恢复水位(REQ §8.4.1) */ export const CacheSettings = CacheShape.refine((v) => v.resumePercent < v.hardThresholdPercent, { @@ -330,7 +340,17 @@ export type SettingsKey = z.infer; /** 缺省设置(settings 表缺 key 或缺字段时使用) */ export const DEFAULT_SETTINGS: GlobalSettings = { automation: { enabled: true, readOnlyReconcile: false }, - cache: { hardThresholdPercent: 90, resumePercent: 85, checkIntervalSec: 60, cleanupTargetPercent: 80 }, + cache: { + hardThresholdPercent: 90, + resumePercent: 85, + checkIntervalSec: 60, + cleanupTargetPercent: 80, + tmdbTtlHours: 24, + tmdbMaxSize: 256, + emptyResultTtlSec: 1800, + globalImageCache: false, + globalImageCacheDays: 7, + }, organize: { defaultMode: "copy", allowCrossVolumeMove: true, minSizeBytes: 50 * 1024 * 1024 }, retention: { seedRetentionDays: 7, logRetentionDays: 180 }, pi: { @@ -587,10 +607,22 @@ export const SETTINGS_SECTIONS: readonly SettingsSectionDef[] = [ { id: "metadata", label: "元数据", - description: "TMDB API Key 与地址", + description: "TMDB API Key 与地址、TMDB 缓存、图片代理", legacyTabs: [], - anchors: [{ id: "tmdb", label: "TMDB" }], - fields: ["metadata.tmdbApiKeySecretRef", "metadata.tmdbBaseUrl"], + anchors: [ + { id: "tmdb", label: "TMDB" }, + { id: "tmdb-cache", label: "TMDB 缓存" }, + { id: "images", label: "图片代理与缓存" }, + ], + fields: [ + "metadata.tmdbApiKeySecretRef", + "metadata.tmdbBaseUrl", + "cache.tmdbTtlHours", + "cache.tmdbMaxSize", + "cache.emptyResultTtlSec", + "cache.globalImageCache", + "cache.globalImageCacheDays", + ], resources: [], }, { diff --git a/packages/contracts/test/db-sync.test.ts b/packages/contracts/test/db-sync.test.ts index 324fb3a4..eba5d075 100644 --- a/packages/contracts/test/db-sync.test.ts +++ b/packages/contracts/test/db-sync.test.ts @@ -130,6 +130,9 @@ const INTERNAL_TABLES = [ "SseEvent", // W9-C 命名模板版本登记(0032):服务端内部,经 /naming/* 暴露 "NamingTemplateVersion", + // C2-A 跨进程限流桶(0041)与 PG 缓存(0042):服务端内部,缓存经 /cache/* 的专用 DTO 暴露 + "RateLimitBucket", + "KvCache", // W9-I 进程心跳(0034):服务端内部,经 /system/health 的专用 DTO 暴露 "ProcessHeartbeat", "ProcessHeartbeatSample", diff --git a/packages/db/migrations/0042_cache.ts b/packages/db/migrations/0042_cache.ts new file mode 100644 index 00000000..748c88fd --- /dev/null +++ b/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): Promise { + 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): Promise { + await execDdl(db, dropTables(["kv_cache"])); +} diff --git a/packages/db/src/cache.ts b/packages/db/src/cache.ts new file mode 100644 index 00000000..fdcff6a6 --- /dev/null +++ b/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 | 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; + } + + 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; +} diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 05b34b56..b2642d60 100644 --- a/packages/db/src/index.ts +++ b/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 { diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index f76d35e3..8515f3af 100644 --- a/packages/db/src/types.ts +++ b/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; 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; export type IdentityRevisionRow = Selectable; export type NewIdentityRevision = Insertable; export type IdentityRevisionUpdate = Updateable; +export type KvCacheRow = Selectable; +export type NewKvCache = Insertable; +export type KvCacheUpdate = Updateable; export type LibraryRootRow = Selectable; export type NewLibraryRoot = Insertable; export type LibraryRootUpdate = Updateable; diff --git a/packages/db/test/cache.test.ts b/packages/db/test/cache.test.ts new file mode 100644 index 00000000..e0a0822a --- /dev/null +++ b/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; + let db: Kysely; + + 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({ + 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] }); + }); +}); diff --git a/packages/db/test/schema.test.ts b/packages/db/test/schema.test.ts index 6c733c36..684f653a 100644 --- a/packages/db/test/schema.test.ts +++ b/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)", () => { From 78625739449cc85462c7b445a8570c0f8c3bff1d Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 22:26:29 +0800 Subject: [PATCH 3/3] feat(cache): TMDB image proxy, posters, cache admin API and settings cards GET /img/tmdb/{size}/{file} ported from MP ImageHelper / fetch_image (sanitized cache path under DATA_DIR/cache/images, GLOBAL_IMAGE_CACHE read switch, md5 ETag / 304, day-based cleanup job). Cache module adds cache.sweep / cache.images_clean jobs and /cache/tmdb, /cache/images list and clear endpoints. Library list/detail and the search palette show posters with the neutral placeholder as fallback; metadata settings get TMDB cache and image cache cards. Runbook docs/runbook/cache.md and porting notes docs/porting/moviepilot-tmdb-cache.md. Co-Authored-By: Claude Opus 5.5 --- .../acquisition-organize/routes/library.ts | 2 + apps/server/src/modules/cache/images.ts | 189 ++++++++++++++++ apps/server/src/modules/cache/index.ts | 210 ++++++++++++++++++ apps/server/test/cache/cache.test.ts | 146 ++++++++++++ apps/web/app/api/cache.ts | 37 +++ apps/web/app/components/library/parts.tsx | 29 ++- .../app/components/search/command-palette.tsx | 24 +- apps/web/app/components/settings/metadata.tsx | 127 ++++++++++- apps/web/app/routes/console/library.$id.tsx | 18 +- apps/web/app/routes/console/library.tsx | 8 +- docs/porting/moviepilot-tmdb-cache.md | 72 ++++++ docs/runbook/cache.md | 61 +++++ packages/contracts/src/api.ts | 34 +++ packages/contracts/src/cache.ts | 77 +++++++ packages/contracts/src/index.ts | 1 + packages/contracts/src/organization.ts | 2 + 16 files changed, 1028 insertions(+), 9 deletions(-) create mode 100644 apps/server/src/modules/cache/images.ts create mode 100644 apps/server/src/modules/cache/index.ts create mode 100644 apps/server/test/cache/cache.test.ts create mode 100644 apps/web/app/api/cache.ts create mode 100644 docs/porting/moviepilot-tmdb-cache.md create mode 100644 docs/runbook/cache.md create mode 100644 packages/contracts/src/cache.ts diff --git a/apps/server/src/modules/acquisition-organize/routes/library.ts b/apps/server/src/modules/acquisition-organize/routes/library.ts index 8816e67f..baf07399 100644 --- a/apps/server/src/modules/acquisition-organize/routes/library.ts +++ b/apps/server/src/modules/acquisition-organize/routes/library.ts @@ -150,6 +150,7 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) { where (s.media_entity_id = e.id or sr.media_entity_id = e.id) and dt.removed_at is null and dt.status in ('queued','submitting','downloading','stalled','paused'))::int`.as("active_downloads"), ) + .select(sql`e.metadata #>> '{posterPath,value}'`.as("poster_path")) .select( sql`(select max(r.finished_at) from organization_record r where r.media_entity_id = e.id and r.status in ('succeeded','partial'))`.as( "last_org", @@ -220,6 +221,7 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) { slotCounts: { ...emptyCounts(), ...(r.slot_counts ?? {}) }, activeDownloads: r.active_downloads, lastOrganizedAt: r.last_org ? new Date(r.last_org).toISOString() : null, + posterPath: r.poster_path ?? null, })); return { items, nextCursor: p.nextCursor }; }, diff --git a/apps/server/src/modules/cache/images.ts b/apps/server/src/modules/cache/images.ts new file mode 100644 index 00000000..d3219010 --- /dev/null +++ b/apps/server/src/modules/cache/images.ts @@ -0,0 +1,189 @@ +import { createHash } from "node:crypto"; +import { promises as fs } from "node:fs"; +import { dirname, extname, join } from "node:path"; +import { TMDB_IMAGE_SIZES } from "@umt/contracts"; + +// TMDB 图片代理与文件缓存,移植 MP app/application/image.py(ImageHelper)+ api/endpoints/system.py(fetch_image): +// - 缓存路径 = SecurityUtils.sanitize_url_path(url):URL 路径去掉开头 "/" 后百分号编码,超过 120 字符换成 +// compressed_<扩展名>;没有扩展名补 .jpg;落在 DATA_DIR/cache/images/ 下(MP CACHE_PATH/images); +// - 读缓存只在 use_cache(GLOBAL_IMAGE_CACHE)开启时;拉到的合法图片总是写盘(与 MP 一致); +// - 合法图片按内容识别 MIME,拒绝非图片与 SVG;ETag = md5(内容),Cache-Control: public, max-age=7 天,If-None-Match 相同回 304; +// - 清理:删除 mtime 早于 GLOBAL_IMAGE_CACHE_DAYS 天的文件与空目录(MP clear_temp → SystemUtils.clear)。 +// 我们只代理 TMDB 图片域(MP SECURITY_IMAGE_DOMAINS 的 TMDB 项),上游失败回 502。 + +export const TMDB_IMAGE_HOST = "image.tmdb.org"; +const FILE_RE = /^[A-Za-z0-9_-]+\.(jpg|jpeg|png|webp)$/; + +export class ImageProxyError extends Error { + constructor( + readonly status: 400 | 502, + message: string, + ) { + super(message); + } +} + +/** SecurityUtils.sanitize_url_path + ImageHelper._prepare_cache_path */ +export function imageCachePath(url: string, maxLength = 120): string { + const path = new URL(url).pathname.replace(/^\/+/, ""); + let safe = path + .split("/") + .map((s) => encodeURIComponent(decodeURIComponent(s))) + .join("/"); + if (safe.length > maxLength) { + const hash = createHash("sha256").update(safe).digest("hex").slice(0, 16); + safe = `compressed_${hash}${extname(safe).toLowerCase()}`; + } + return extname(safe) ? safe : `${safe}.jpg`; +} + +/** ImageHelper.get_image_mime_type(按文件头识别;非图片 / SVG 返回 null) */ +export function imageMimeType(b: Uint8Array): string | null { + if (b.length >= 3 && b[0] === 0xff && b[1] === 0xd8 && b[2] === 0xff) return "image/jpeg"; + if (b.length >= 8 && b[0] === 0x89 && b[1] === 0x50 && b[2] === 0x4e && b[3] === 0x47) return "image/png"; + if ( + b.length >= 12 && + String.fromCharCode(...b.slice(0, 4)) === "RIFF" && + String.fromCharCode(...b.slice(8, 12)) === "WEBP" + ) + return "image/webp"; + if (b.length >= 6 && String.fromCharCode(...b.slice(0, 4)) === "GIF8") return "image/gif"; + return null; +} + +export interface ImageCacheOptions { + /** DATA_DIR/cache/images */ + dir: string; + fetch?: typeof globalThis.fetch; + timeoutMs?: number; + /** 上游基址(测试指向 mock);缺省 https://image.tmdb.org/t/p */ + upstreamBase?: string; +} + +export class ImageCache { + private readonly fetchImpl: typeof globalThis.fetch; + + constructor(private readonly opts: ImageCacheOptions) { + this.fetchImpl = opts.fetch ?? globalThis.fetch; + } + + get dir(): string { + return this.opts.dir; + } + + /** 尺寸白名单 + 文件名校验后拼上游 URL(只允许 TMDB 图片域) */ + upstream(size: string, file: string): string { + if (!(TMDB_IMAGE_SIZES as readonly string[]).includes(size)) throw new ImageProxyError(400, "不支持的图片尺寸"); + if (!FILE_RE.test(file)) throw new ImageProxyError(400, "图片路径不合法"); + const base = (this.opts.upstreamBase ?? `https://${TMDB_IMAGE_HOST}/t/p`).replace(/\/+$/, ""); + const url = `${base}/${size}/${file}`; + if (!this.opts.upstreamBase && new URL(url).hostname !== TMDB_IMAGE_HOST) { + throw new ImageProxyError(400, "只代理 TMDB 图片"); + } + return url; + } + + /** fetch_image:取图(可读缓存)→ ETag / 304 */ + async serve(size: string, file: string, useCache: boolean, ifNoneMatch: string | null): Promise { + const url = this.upstream(size, file); + const { content, mime } = await this.fetchImage(url, useCache); + const etag = createHash("md5").update(content).digest("hex"); + const headers = new Headers({ + etag, + "cache-control": `public, max-age=${86400 * 7}`, + "content-type": mime, + "x-content-type-options": "nosniff", + }); + if (ifNoneMatch === etag) return new Response(null, { status: 304, headers }); + return new Response(content as BodyInit, { status: 200, headers }); + } + + /** ImageHelper.fetch_image_with_mime_type */ + private async fetchImage(url: string, useCache: boolean): Promise<{ content: Uint8Array; mime: string }> { + const path = join(this.opts.dir, imageCachePath(url)); + if (useCache) { + const cached = await fs.readFile(path).catch(() => null); + const mime = cached?.length ? imageMimeType(cached) : null; + if (cached && mime) return { content: cached, mime }; + } + let res: Response; + try { + res = await this.fetchImpl(url, { + signal: AbortSignal.timeout(this.opts.timeoutMs ?? 20_000), + headers: { accept: "image/avif,image/webp,image/apng,*/*" }, + }); + } catch (e) { + const reason = (e as Error).name === "TimeoutError" ? "超时" : (e as Error).message; + throw new ImageProxyError(502, `拉取 TMDB 图片失败：${reason}`); + } + if (res.status !== 200) { + await res.body?.cancel().catch(() => {}); + throw new ImageProxyError(502, `TMDB 图片返回 HTTP ${res.status}`); + } + const content = new Uint8Array(await res.arrayBuffer()); + const mime = imageMimeType(content); + if (!mime) throw new ImageProxyError(502, "TMDB 返回的不是图片"); + await fs.mkdir(dirname(path), { recursive: true }); + const tmp = `${path}.${process.pid}.tmp`; + await fs.writeFile(tmp, content); + await fs.rename(tmp, path); + return { content, mime }; + } + + private async files(): Promise<{ path: string; size: number; mtimeMs: number }[]> { + const out: { path: string; size: number; mtimeMs: number }[] = []; + const walk = async (d: string) => { + let names: string[]; + try { + names = await fs.readdir(d); + } catch { + return; + } + for (const n of names) { + const p = join(d, n); + const st = await fs.stat(p).catch(() => null); + if (!st) continue; + if (st.isDirectory()) await walk(p); + else out.push({ path: p, size: st.size, mtimeMs: st.mtimeMs }); + } + }; + await walk(this.opts.dir); + return out; + } + + async usage(): Promise<{ bytes: number; files: number }> { + const f = await this.files(); + return { bytes: f.reduce((a, x) => a + x.size, 0), files: f.length }; + } + + /** SystemUtils.clear(images, days):删 mtime 早于 days 天的文件,再删空目录;返回删除数 */ + async clean(days: number, now = Date.now()): Promise { + let deleted = 0; + for (const f of await this.files()) { + if (Math.floor((now - f.mtimeMs) / 86_400_000) > days) { + await fs.rm(f.path, { force: true }); + deleted++; + } + } + const prune = async (d: string): Promise => { + const names = await fs.readdir(d).catch(() => null); + if (!names) return false; + let empty = true; + for (const n of names) { + const p = join(d, n); + const st = await fs.stat(p).catch(() => null); + if (st?.isDirectory() && (await prune(p))) await fs.rmdir(p).catch(() => {}); + else empty = false; + } + return empty; + }; + await prune(this.opts.dir); + return deleted; + } + + async clear(): Promise { + const n = (await this.files()).length; + await fs.rm(this.opts.dir, { recursive: true, force: true }); + return n; + } +} diff --git a/apps/server/src/modules/cache/index.ts b/apps/server/src/modules/cache/index.ts new file mode 100644 index 00000000..59233861 --- /dev/null +++ b/apps/server/src/modules/cache/index.ts @@ -0,0 +1,210 @@ +import { join } from "node:path"; +import { createRoute, type RouteConfig } from "@hono/zod-openapi"; +import { + Api, + apiResponse, + DEFAULT_SETTINGS, + type EndpointDef, + type TmdbCacheItem, + type TmdbCacheList, +} from "@umt/contracts"; +import type { z } from "zod"; +import { dataPaths } from "../../config.ts"; +import { requireScope } from "../../http/auth.ts"; +import { errorBody } from "../../http/errors.ts"; +import { CACHE_REGIONS, sharedCache, tmdbCachePolicy } from "../../shared-cache.ts"; +import { registerScheduledJob } from "../schedules/registry.ts"; +import type { ModuleApp, ServerContext, ServerModule } from "../types.ts"; +import { ImageCache, ImageProxyError } from "./images.ts"; + +/** + * C2-A 缓存(docs/runbook/cache.md): + * - PG 缓存 kv_cache(UNLOGGED,移植 MP TTLCache)的过期行物理清理任务 cache.sweep(每 10 分钟); + * - TMDB 缓存管理(对照 MP tmdb_cache_items / tmdb_cache_clear):GET /cache/tmdb、POST /cache/tmdb/clear; + * - 图片代理 GET /img/tmdb/{size}/{file}(会话或 API Key read,移植 MP ImageHelper / fetch_image)+ DATA_DIR/cache/images 落盘, + * 用量 / 清空 GET /cache/images、POST /cache/images/clear,按保留天数清理的任务 cache.images_clean(每天)。 + */ +export interface CacheModuleOptions { + /** 图片上游 fetch(测试注入) */ + imageFetch?: typeof globalThis.fetch; + /** 图片上游基址(测试指向 mock) */ + imageUpstreamBase?: string; + /** 图片缓存目录;缺省 DATA_DIR/cache/images */ + imageDir?: string; +} + +type Infer = T extends z.ZodType ? z.infer : Record; + +function route( + app: ModuleApp, + def: D, + handler: (i: { query: Infer }) => Promise, +) { + const request: RouteConfig["request"] = {}; + if (def.query) request.query = def.query as never; + const r = createRoute({ + method: def.method, + path: def.path, + summary: def.summary, + tags: ["cache"], + middleware: [requireScope(def.scope)] as never, + request, + responses: { + 200: { description: def.summary, content: { "application/json": { schema: apiResponse(def.response) } } }, + }, + }); + // biome-ignore lint/suspicious/noExplicitAny: 路由泛型在封装里擦除,入参已由 zod-openapi 校验 + app.openapi(r, (async (c: any) => { + const data = def.response.parse(await handler({ query: def.query ? c.req.valid("query") : {} } as never)); + return c.json({ ok: true, data }, 200); + // biome-ignore lint/suspicious/noExplicitAny: 同上 + }) as any); +} + +const str = (v: unknown) => (typeof v === "string" && v ? v : null); + +export function createCacheModule(opts: CacheModuleOptions = {}): ServerModule & { + images(ctx: ServerContext): ImageCache; +} { + const imageCaches = new WeakMap(); + const images = (ctx: ServerContext) => { + let c = imageCaches.get(ctx); + if (!c) { + c = new ImageCache({ + dir: opts.imageDir ?? join(dataPaths(ctx.config).dataDir, "cache", "images"), + ...(opts.imageFetch ? { fetch: opts.imageFetch } : {}), + ...(opts.imageUpstreamBase ? { upstreamBase: opts.imageUpstreamBase } : {}), + }); + imageCaches.set(ctx, c); + } + return c; + }; + const cacheSettings = (ctx: ServerContext) => ctx.settings.get("cache").catch(() => DEFAULT_SETTINGS.cache); + + return { + name: "cache", + images, + registerRoutes(app, ctx) { + // 图片代理:不进 OpenAPI(二进制响应); 带 cookie 会话即可,API Key 需 read + app.get("/img/tmdb/:size/:file", requireScope("read"), async (c) => { + const s = await cacheSettings(ctx); + try { + // MP cache_img:GLOBAL_IMAGE_CACHE 开启才读磁盘缓存 + return await images(ctx).serve( + c.req.param("size"), + c.req.param("file"), + s.globalImageCache ?? false, + c.req.header("if-none-match") ?? null, + ); + } catch (e) { + if (e instanceof ImageProxyError) { + return c.json(errorBody(e.status === 400 ? "bad_request" : "upstream_unavailable", e.message), e.status); + } + throw e; + } + }); + + route(app, Api.cache.tmdb, async ({ query }) => { + const cache = sharedCache(ctx); + const policy = await tmdbCachePolicy(ctx); + const want = (k: "request" | "recognize") => !query.kind || query.kind === k; + const items: TmdbCacheItem[] = []; + if (want("recognize")) { + for (const r of await cache.items>(CACHE_REGIONS.tmdbRecognize, query.limit)) { + const v = r.value ?? {}; + const t = v.type === "movie" || v.type === "tv" ? v.type : "unknown"; + items.push({ + kind: "recognize", + key: r.key, + expiresAt: r.expiresAt.toISOString(), + updatedAt: r.updatedAt.toISOString(), + bytes: r.bytes, + tmdbId: typeof v.id === "number" ? v.id : Number(v.id) || 0, + title: str(v.title), + year: str(v.year), + mediaType: t, + posterPath: str(v.poster_path), + }); + } + } + if (want("request")) { + for (const r of await cache.items(CACHE_REGIONS.tmdbRequest, query.limit)) { + items.push({ + kind: "request", + key: r.key, + expiresAt: r.expiresAt.toISOString(), + updatedAt: r.updatedAt.toISOString(), + bytes: r.bytes, + tmdbId: null, + title: null, + year: null, + mediaType: null, + posterPath: null, + }); + } + } + const out: TmdbCacheList = { + items, + counts: { + request: await cache.count(CACHE_REGIONS.tmdbRequest), + recognize: await cache.count(CACHE_REGIONS.tmdbRecognize), + }, + maxSize: policy.maxsize, + ttlHours: Math.round(policy.ttl / 3600), + emptyResultTtlSec: policy.emptyTtl, + }; + return out; + }); + route(app, Api.cache.tmdbClear, async () => { + const cache = sharedCache(ctx); + const deleted = + (await cache.clear(CACHE_REGIONS.tmdbRecognize)) + (await cache.clear(CACHE_REGIONS.tmdbRequest)); + ctx.logger.info({ deleted }, "TMDB 缓存已清空"); + return { deleted }; + }); + route(app, Api.cache.images, async () => { + const s = await cacheSettings(ctx); + const img = images(ctx); + const u = await img.usage(); + return { + enabled: s.globalImageCache ?? false, + days: s.globalImageCacheDays ?? DEFAULT_SETTINGS.cache.globalImageCacheDays, + usedBytes: u.bytes, + files: u.files, + dir: img.dir, + }; + }); + route(app, Api.cache.imagesClear, async () => ({ deleted: await images(ctx).clear() })); + }, + registerTasks() {}, + start(ctx: ServerContext) { + registerScheduledJob(ctx, { + key: "cache.sweep", + defaults: { + name: "缓存清理", + description: "删除 PG 缓存里已过期的条目（TMDB 请求与识别缓存）", + group: "maintain", + schedule: { type: "interval", everyMs: 10 * 60_000 }, + firstRun: "after_interval", + }, + run: async () => ({ summary: `删除 ${await sharedCache(ctx).sweep()} 条过期缓存` }), + }); + registerScheduledJob(ctx, { + key: "cache.images_clean", + defaults: { + name: "图片缓存清理", + description: "删除超过保留天数的图片代理缓存文件（对照 MP 启动时的 clear_temp）", + group: "maintain", + schedule: { type: "interval", everyMs: 24 * 3_600_000 }, + }, + run: async () => { + const s = await cacheSettings(ctx); + const n = await images(ctx).clean(s.globalImageCacheDays ?? DEFAULT_SETTINGS.cache.globalImageCacheDays); + return { summary: `删除 ${n} 张过期图片` }; + }, + }); + }, + }; +} + +export const cacheModule = createCacheModule(); diff --git a/apps/server/test/cache/cache.test.ts b/apps/server/test/cache/cache.test.ts new file mode 100644 index 00000000..fc8fac34 --- /dev/null +++ b/apps/server/test/cache/cache.test.ts @@ -0,0 +1,146 @@ +import { mkdtempSync, rmSync, utimesSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { PgCache, sql } from "@umt/db"; +import { createMockTmdb, type MockTmdb } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { imageCachePath } from "../../src/modules/cache/images.ts"; +import { createCacheModule } from "../../src/modules/cache/index.ts"; +import { scheduleRegistry } from "../../src/modules/schedules/registry.ts"; +import { SourcesRuntime } from "../../src/modules/sources-identity/runtime.ts"; +import { sharedCache } from "../../src/shared-cache.ts"; +import { bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts"; +import { DEV_KEY, json } from "../helpers.ts"; + +// C2-A 缓存:TMDB 请求缓存进 kv_cache(TTL / maxsize 取设置,对照 MP)、/cache/tmdb 列表与清空、cache.sweep、 +// 图片代理(对照 MP ImageHelper:缓存路径、GLOBAL_IMAGE_CACHE 才读盘、md5 ETag / 304、按天清理)。 + +const hasPg = await pgAvailable(); +const JPEG = new Uint8Array([0xff, 0xd8, 0xff, 0xe0, 1, 2, 3, 4]); + +describe("图片缓存路径(MP sanitize_url_path)", () => { + it("URL 路径去掉开头 /;过长换 compressed_;无扩展名补 .jpg", () => { + expect(imageCachePath("https://image.tmdb.org/t/p/w342/abc.jpg")).toBe("t/p/w342/abc.jpg"); + expect(imageCachePath(`https://x.org/${"a".repeat(130)}.png`)).toMatch(/^compressed_[0-9a-f]{16}\.png$/); + expect(imageCachePath("https://x.org/t/p/w92/noext")).toBe("t/p/w92/noext.jpg"); + }); +}); + +describe.skipIf(!hasPg)("C2-A 缓存(真实 PG)", () => { + let h: Harness; + let admin: ReturnType; + let tmdb: MockTmdb; + const dir = mkdtempSync(join(tmpdir(), "umt-c2a-img-")); + let upstreamHits = 0; + const mod = createCacheModule({ + imageDir: dir, + imageUpstreamBase: "https://image.tmdb.org/t/p", + imageFetch: async (input) => { + upstreamHits++; + if (String(input).includes("broken")) return new Response("", { headers: { "content-type": "text/html" } }); + return new Response(JPEG, { headers: { "content-type": "image/jpeg" } }); + }, + }); + + beforeAll(async () => { + h = await createHarness({ extraModules: [mod] }); + admin = bearer(h.app, DEV_KEY); + tmdb = await createMockTmdb({ apiKey: "0123456789abcdef0123456789abcdef" }); + }); + afterAll(async () => { + await tmdb?.close(); + await h?.close(); + rmSync(dir, { recursive: true, force: true }); + }); + + it("TMDB 请求响应进 kv_cache(tmdb.request,TTL = cache.tmdbTtlHours,条数上限 = cache.tmdbMaxSize),另一个进程直接命中", async () => { + const r = await admin("PATCH", "/settings", { + metadata: { tmdbApiKey: "0123456789abcdef0123456789abcdef", tmdbBaseUrl: `${tmdb.url}/3` }, + cache: { tmdbTtlHours: 6, tmdbMaxSize: 16 }, + }); + expect(r.status).toBe(200); + await (await new SourcesRuntime(h.ctx).tmdb())!.details("movie", 872585); + const calls = tmdb.calls.length; + const rows = ( + await sql<{ key: string; hours: number }>`select key, extract(epoch from expires_at - now()) / 3600 as hours + from kv_cache where region = 'tmdb.request'`.execute(h.db) + ).rows; + const key = rows.find((x) => x.key.startsWith("/movie/872585?"))!.key; + expect(rows.every((x) => !x.key.includes("0123456789abcdef"))).toBe(true); + expect(Number(rows[0]!.hours)).toBeGreaterThan(5.9); + expect(await new PgCache(h.db).get("tmdb.request", key)).toMatchObject({ id: 872585 }); + await (await new SourcesRuntime(h.ctx).tmdb())!.details("movie", 872585); + expect(tmdb.calls.length).toBe(calls); + // 空搜索结果按 EMPTY_RESULT_CACHE_TTL(30 分钟) + await (await new SourcesRuntime(h.ctx).tmdb())!.search("完全不存在的片名c2a", { type: "movie" }); + const empty = ( + await sql<{ m: number }>`select extract(epoch from expires_at - now()) / 60 as m + from kv_cache where region = 'tmdb.request' and key like '/search/movie?%'`.execute(h.db) + ).rows[0]; + expect(Number(empty!.m)).toBeLessThanOrEqual(30); + }); + + it("GET /cache/tmdb 列出条目(键 / 过期 / 大小)与上限;POST /cache/tmdb/clear 清空", async () => { + await sharedCache(h.ctx).set( + "tmdb.recognize", + "[未知][zh-CN]Oppenheimer-2023-None", + { id: 872585, type: "movie", year: "2023", title: "奥本海默", poster_path: "/p.jpg" }, + { ttl: 3600 }, + ); + const list = await json(await admin("GET", "/cache/tmdb")); + expect(list.data).toMatchObject({ maxSize: 16, ttlHours: 6, emptyResultTtlSec: 1800 }); + expect(list.data.counts.request).toBeGreaterThan(0); + const rec = list.data.items.find((i: { kind: string }) => i.kind === "recognize"); + expect(rec).toMatchObject({ tmdbId: 872585, title: "奥本海默", mediaType: "movie", posterPath: "/p.jpg" }); + expect(rec.bytes).toBeGreaterThan(0); + const cleared = await json(await admin("POST", "/cache/tmdb/clear")); + expect(cleared.data.deleted).toBeGreaterThan(1); + expect((await json(await admin("GET", "/cache/tmdb"))).data.items).toEqual([]); + }); + + it("cache.sweep / cache.images_clean 登记为维护任务;sweep 删除过期行", async () => { + await mod.start?.(h.ctx); + const reg = scheduleRegistry(h.ctx); + expect(reg.get("cache.sweep")?.defaults).toMatchObject({ group: "maintain", schedule: { everyMs: 600_000 } }); + expect(reg.get("cache.images_clean")?.defaults.group).toBe("maintain"); + await sharedCache(h.ctx).set("t", "gone", 1, { ttl: 60 }); + await sql`update kv_cache set expires_at = now() - interval '1 second' where region = 't'`.execute(h.db); + const run = { + jobId: "cache.sweep", + runId: "r", + trigger: "manual" as const, + now: new Date(), + config: {}, + slot: "1", + }; + expect((await reg.get("cache.sweep")!.run(run))?.summary).toMatch(/删除 [1-9]\d* 条过期缓存/); + }); + + it("图片代理:只收 TMDB 尺寸 / 文件名;GLOBAL_IMAGE_CACHE 关闭时每次回源(仍落盘),开启后读盘;md5 ETag / 304;非图片 502;按天清理", async () => { + const get = (p: string, headers: Record = {}) => + h.app.request(`/api/v1${p}`, { headers: { authorization: `Bearer ${DEV_KEY}`, ...headers } }); + expect((await get("/img/tmdb/w9999/a.jpg")).status).toBe(400); + expect((await get("/img/tmdb/w342/..%2Fetc.jpg")).status).toBe(400); + expect((await h.app.request("/api/v1/img/tmdb/w342/a.jpg")).status).toBe(401); + const r1 = await get("/img/tmdb/w342/a.jpg"); + expect(r1.status).toBe(200); + expect(r1.headers.get("content-type")).toBe("image/jpeg"); + expect(r1.headers.get("cache-control")).toBe("public, max-age=604800"); + const etag = r1.headers.get("etag")!; + expect(etag).toMatch(/^[0-9a-f]{32}$/); + await get("/img/tmdb/w342/a.jpg"); + expect(upstreamHits).toBe(2); + await admin("PATCH", "/settings", { cache: { globalImageCache: true } }); + const r3 = await get("/img/tmdb/w342/a.jpg", { "if-none-match": etag }); + expect(r3.status).toBe(304); + expect(upstreamHits).toBe(2); + expect((await get("/img/tmdb/w342/broken.jpg")).status).toBe(502); + const usage = await json(await admin("GET", "/cache/images")); + expect(usage.data).toMatchObject({ enabled: true, days: 7, files: 1 }); + const old = new Date(Date.now() - 9 * 86_400_000); + utimesSync(join(dir, "t/p/w342/a.jpg"), old, old); + expect(await mod.images(h.ctx).clean(7)).toBe(1); + expect((await json(await admin("GET", "/cache/images"))).data.files).toBe(0); + expect((await json(await admin("POST", "/cache/images/clear"))).data.deleted).toBe(0); + }); +}); diff --git a/apps/web/app/api/cache.ts b/apps/web/app/api/cache.ts new file mode 100644 index 00000000..d7784407 --- /dev/null +++ b/apps/web/app/api/cache.ts @@ -0,0 +1,37 @@ +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import type { CacheClearResult, ImageCacheStatus, TmdbCacheList } from "@umt/contracts"; +import { apiData } from "~/lib/api.ts"; + +// C2-A 缓存管理:TMDB 缓存(条数 / 上限 / TTL / 清空)、图片代理缓存(用量 / 清空) + +const keys = { tmdb: ["cache", "tmdb"] as const, images: ["cache", "images"] as const }; + +export function useTmdbCache() { + return useQuery({ + queryKey: keys.tmdb, + queryFn: ({ signal }) => apiData("/cache/tmdb?limit=1", { signal }), + }); +} + +export function useClearTmdbCache() { + const qc = useQueryClient(); + return useMutation({ + mutationFn: () => apiData("/cache/tmdb/clear", { method: "POST" }), + onSuccess: () => qc.invalidateQueries({ queryKey: keys.tmdb }), + }); +} + +export function useImageCache() { + return useQuery({ + queryKey: keys.images, + queryFn: ({ signal }) => apiData("/cache/images", { signal }), + }); +} + +export function useClearImageCache() { + const qc = useQueryClient(); + return useMutation({ + mutationFn: () => apiData("/cache/images/clear", { method: "POST" }), + onSuccess: () => qc.invalidateQueries({ queryKey: keys.images }), + }); +} diff --git a/apps/web/app/components/library/parts.tsx b/apps/web/app/components/library/parts.tsx index 07981858..71eee40f 100644 --- a/apps/web/app/components/library/parts.tsx +++ b/apps/web/app/components/library/parts.tsx @@ -1,6 +1,7 @@ +import { tmdbImageUrl } from "@umt/contracts"; import type { LucideIcon } from "lucide-react"; import { Clapperboard, Film, Layers, Tv } from "lucide-react"; -import type { ReactNode } from "react"; +import { type ReactNode, useState } from "react"; import { StatusBadge, type StatusTone } from "~/components/console/status-badge.tsx"; import { cn } from "~/lib/utils.ts"; @@ -117,19 +118,43 @@ export function ChipToggleGroup({ const KIND_ICON: Record = { movie: Film, series: Tv, variety: Clapperboard, collection: Layers }; -/** 封面占位:无彩材质 + 类型图标 + 首字(不引入海报色彩) */ +/** media_entity.metadata 里的最终海报路径(metadata.posterPath.value) */ +export function posterPathOf(metadata: unknown): string | null { + const v = (metadata as { posterPath?: { value?: unknown } } | null)?.posterPath?.value; + return typeof v === "string" && v ? v : null; +} + +/** + * 封面:有 TMDB 海报路径时经图片代理显示海报,加载失败或没有海报时回退无彩占位(类型图标 + 首字)。 + */ export function CoverPlaceholder({ title, kind, small = false, className, + posterPath, }: { title: string; kind: string; /** 列表缩略:只显示首字,不叠类型图标 */ small?: boolean; className?: string; + /** TMDB poster_path */ + posterPath?: string | null | undefined; }) { + const [failed, setFailed] = useState(false); + const src = failed ? null : tmdbImageUrl(posterPath, small ? "w154" : "w342"); + if (src) { + return ( + setFailed(true)} + className={cn("shrink-0 rounded-lg border border-border bg-foreground/[0.05] object-cover", className)} + /> + ); + } const Icon = KIND_ICON[kind] ?? Film; const glyph = title.replace(/^w32-/, "").trim().charAt(0) || "?"; return ( diff --git a/apps/web/app/components/search/command-palette.tsx b/apps/web/app/components/search/command-palette.tsx index 21bac055..33597c53 100644 --- a/apps/web/app/components/search/command-palette.tsx +++ b/apps/web/app/components/search/command-palette.tsx @@ -1,4 +1,4 @@ -import type { LocalMediaHit, SearchAction, TmdbCandidate } from "@umt/contracts"; +import { type LocalMediaHit, type SearchAction, type TmdbCandidate, tmdbImageUrl } from "@umt/contracts"; import { ArrowRight, CornerDownLeft, Film, Globe, LoaderCircle, type LucideIcon, Plus, Search } from "lucide-react"; import { type ReactNode, useEffect, useId, useMemo, useRef, useState } from "react"; import { Link, useNavigate } from "react-router"; @@ -33,6 +33,8 @@ type Item = { key: string; group: string; icon: LucideIcon; + /** TMDB 海报路径:有则以小海报代替图标(失败回退图标) */ + posterPath?: string | null | undefined; title: ReactNode; meta?: ReactNode; /** 右侧动作提示 */ @@ -216,7 +218,7 @@ function PaletteBody({ initial }: { initial: string }) { i === active ? "bg-foreground/[0.06]" : "", )} > - +
{it.title}
{it.meta ?
{it.meta}
: null} @@ -287,12 +289,30 @@ function localItem(h: LocalMediaHit, run: () => void): Item { }; } +function ItemIcon({ it }: { it: Item }) { + const [failed, setFailed] = useState(false); + const src = failed ? null : tmdbImageUrl(it.posterPath, "w92"); + if (src) { + return ( + setFailed(true)} + className="h-9 w-6 shrink-0 rounded-tight border border-border object-cover" + /> + ); + } + return ; +} + function tmdbItem(c: TmdbCandidate, view: () => void, subscribe: (c: TmdbCandidate) => void): Item { const local = !!c.localEntityId; return { key: `tmdb-${c.entityType}-${c.externalId}`, group: "TMDB", icon: local ? Globe : Plus, + posterPath: c.posterPath, title: ( {c.title} diff --git a/apps/web/app/components/settings/metadata.tsx b/apps/web/app/components/settings/metadata.tsx index ae78c5fc..091a8e5c 100644 --- a/apps/web/app/components/settings/metadata.tsx +++ b/apps/web/app/components/settings/metadata.tsx @@ -1,10 +1,11 @@ import { toast } from "sonner"; +import { useClearImageCache, useClearTmdbCache, useImageCache, useTmdbCache } from "~/api/cache.ts"; import { useUpdateSettings } from "~/api/settings.ts"; import { StatusBadge } from "~/components/console/status-badge.tsx"; import { Button } from "~/components/ui/button.tsx"; import { errorMessage } from "~/lib/api.ts"; import { str } from "./resource-section.tsx"; -import { type CardSpec, SettingsCard, SettingsGate } from "./settings-card.tsx"; +import { type CardSpec, n, SettingsCard, SettingsGate } from "./settings-card.tsx"; // 元数据:TMDB API Key(只写,存 secret)与 Base URL @@ -67,6 +68,128 @@ function TmdbKeyStatus({ saved }: { saved: boolean }) { ); } +// TMDB 缓存(对照 MP:META_CACHE_EXPIRE / CONF.tmdb / EMPTY_RESULT_CACHE_TTL) +const TMDB_CACHE_CARD: CardSpec = { + anchor: "tmdb-cache", + title: "TMDB 缓存", + description: "请求响应与识别结果存在数据库里，三个进程共用。超过条数上限时删最久未用的条目。", + fields: [ + { name: "cache.tmdbTtlHours", label: "有效期（小时）", kind: "number" }, + { name: "cache.tmdbMaxSize", label: "条数上限", kind: "number", hint: "请求缓存和识别缓存各自计数" }, + { + name: "cache.emptyResultTtlSec", + label: "空结果有效期（秒）", + kind: "number", + hint: "搜索无结果、未识别时的短缓存", + }, + ], + fromSettings: (s) => ({ + "cache.tmdbTtlHours": n(s.cache.tmdbTtlHours), + "cache.tmdbMaxSize": n(s.cache.tmdbMaxSize), + "cache.emptyResultTtlSec": n(s.cache.emptyResultTtlSec), + }), + toPatch: (v) => ({ + cache: { + tmdbTtlHours: Number(v["cache.tmdbTtlHours"]), + tmdbMaxSize: Number(v["cache.tmdbMaxSize"]), + emptyResultTtlSec: Number(v["cache.emptyResultTtlSec"]), + }, + }), + extra: () => , +}; + +function TmdbCacheStatus() { + const q = useTmdbCache(); + const clear = useClearTmdbCache(); + const c = q.data?.counts; + return ( +
+ + {c ? `识别 ${c.recognize} 条 · 请求 ${c.request} 条 · 上限各 ${q.data?.maxSize} 条` : "读取中…"} + + +
+ ); +} + +// 图片代理与缓存(对照 MP GLOBAL_IMAGE_CACHE / GLOBAL_IMAGE_CACHE_DAYS) +const IMAGES_CARD: CardSpec = { + anchor: "images", + title: "图片代理与缓存", + description: "海报经服务端从 TMDB 取回。开启缓存后保存到数据目录，下次直接读取。", + fields: [ + { name: "cache.globalImageCache", label: "缓存图片", kind: "switch" }, + { name: "cache.globalImageCacheDays", label: "保留天数", kind: "number", hint: "超过天数的图片每天清理一次" }, + ], + fromSettings: (s) => ({ + "cache.globalImageCache": s.cache.globalImageCache, + "cache.globalImageCacheDays": n(s.cache.globalImageCacheDays), + }), + toPatch: (v) => ({ + cache: { + globalImageCache: v["cache.globalImageCache"] === true, + globalImageCacheDays: Number(v["cache.globalImageCacheDays"]), + }, + }), + extra: () => , +}; + +function ImageCacheStatus() { + const q = useImageCache(); + const clear = useClearImageCache(); + const d = q.data; + return ( +
+ + {d ? `${d.files} 张 · ${(d.usedBytes / 1024 / 1024).toFixed(1)} MiB` : "读取中…"} + + +
+ ); +} + export function MetadataSection() { - return {(s) => }; + return ( + + {(s) => ( + <> + + + + + )} + + ); } diff --git a/apps/web/app/routes/console/library.$id.tsx b/apps/web/app/routes/console/library.$id.tsx index b8e332d8..569102ac 100644 --- a/apps/web/app/routes/console/library.$id.tsx +++ b/apps/web/app/routes/console/library.$id.tsx @@ -51,7 +51,16 @@ import { UNIT_KIND, VERIFY_STATUS, } from "~/components/library/labels.ts"; -import { Chip, ChipList, CoverPlaceholder, Panel, ProgressLine, relTime, Tone } from "~/components/library/parts.tsx"; +import { + Chip, + ChipList, + CoverPlaceholder, + Panel, + ProgressLine, + posterPathOf, + relTime, + Tone, +} from "~/components/library/parts.tsx"; import { SlotCard } from "~/components/library/slot-card.tsx"; import { EntityNamingButton } from "~/components/naming/entity-naming-dialog.tsx"; import { Button } from "~/components/ui/button.tsx"; @@ -617,7 +626,12 @@ export default function LibraryDetailPage() {

旧 ID {redirected} 已跳转到当前作品。

) : null}
- +

diff --git a/apps/web/app/routes/console/library.tsx b/apps/web/app/routes/console/library.tsx index 37d527a1..e4ff4a9e 100644 --- a/apps/web/app/routes/console/library.tsx +++ b/apps/web/app/routes/console/library.tsx @@ -128,7 +128,13 @@ function WorksView() { header: "作品", cell: (r) => (
- +
`,无后缀补 `.jpg`;放在 `CACHE_PATH/images/` | `imageCachePath`,放在 `DATA_DIR/cache/images/` | +| `use_cache` 才读缓存;拉到的合法图片总会 `file_cache.set` | 同 | +| `/cache/image`:`use_cache = GLOBAL_IMAGE_CACHE`(缺省 False) | 设置 `cache.globalImageCache`(缺省关闭) | +| `get_image_mime_type`:Pillow 识别,拒绝非图片与 SVG | 按文件头识别 jpeg / png / webp / gif,其余拒绝 | +| `ETag = md5(content)`,`Cache-Control: public, max-age=604800`,If-None-Match 相同回 304 | 同 | +| `SECURITY_IMAGE_DOMAINS` 白名单 | 只代理 `image.tmdb.org`,尺寸与文件名走白名单 | +| `clear_temp`:删除 `CACHE_PATH/images` 下 mtime 超过 `GLOBAL_IMAGE_CACHE_DAYS`(7)天的文件与空目录 | 设置 `cache.globalImageCacheDays`(缺省 7);定时任务 `cache.images_clean` 每天执行 | + +偏离:MP 失败时返回空响应,我们回 502(前端回退占位);MP 按配置走代理服务器,我们直连(TMDB API 也未走代理); +MP 只在启动时清理,我们改为每天一次的定时任务。 diff --git a/docs/runbook/cache.md b/docs/runbook/cache.md new file mode 100644 index 00000000..3549c615 --- /dev/null +++ b/docs/runbook/cache.md @@ -0,0 +1,61 @@ +# 缓存与限流运行手册(C2-A) + +不引入 Redis / Deno KV。跨进程共享的状态都放 PG:站点限流令牌桶、TMDB 缓存。缓存语义照搬 MoviePilot, +对照与偏离见 [docs/porting/moviepilot-tmdb-cache.md](../porting/moviepilot-tmdb-cache.md)。 + +## 跨进程站点限流 + +- 表 `rate_limit_bucket`(迁移 `0041_rate_limit`):key `site:<站点 id>`,tokens / capacity / refill_per_sec / blocked_until。 +- `@umt/db` 的 `PgTokenBucket.acquire(key, cost, {capacity, refillPerSec, waitMs})`:事务里 `select … for update`, + 按 DB 时钟补令牌后扣减;拿不到按需要等待的时间加抖动轮询,超过 waitMs 返回 `ok:false`(站点限流器会重新排队)。 +- 站点适配器用 `createSharedSiteRateLimiter`:进程内先串行排队,再到 PG 扣令牌;429 的 Retry-After 写 `blocked_until`。 + 站点限流间隔为 0(不限流)时只走本地;PG 出错时本次退回进程内令牌桶并记 warn。 +- 容量 = 站点并发(burst),速率 = 1000 / 最小间隔毫秒。改站点限流设置即时生效(每次扣令牌带上最新参数)。 + +排查:`select * from rate_limit_bucket where key = 'site:'`;`blocked_until` 不为空说明站点要求暂停。 + +## PG 缓存(kv_cache) + +- 表 `kv_cache`(迁移 `0042_cache`,`UNLOGGED`:不写 WAL,崩溃后清空):(region, key) 主键,value jsonb, + expires_at 每条独立,accessed_at 为最近读写时间。 +- `@umt/db` 的 `PgCache`(对照 MP `TTLCache`):set 带 ttl(秒,≤ 0 删除)与 maxsize(按 region 计条数,超出先删过期再删最久未用); + get 命中刷新最近使用;`cachedCall` 对照 MP `@cached`(skip_none / skip_if / empty_ttl / fresh)。 +- 定时任务 `cache.sweep`(维护组,每 10 分钟)物理删除过期行。 + +| region | 内容 | 键 | TTL | 条数上限 | +|---|---|---|---|---| +| `tmdb.request` | TMDB 请求响应(MP `TMDb.request`) | `path?query`(不含凭证) | `cache.tmdbTtlHours`;`results` 为空按 `cache.emptyResultTtlSec` | `cache.tmdbMaxSize` | +| `tmdb.recognize` | TMDB 识别结果(MP `TmdbCache`) | `[类型][语言]{tmdbid 或名称}-{年份}-{季}` | 同上;未识别 `{id:0}` 按空结果 TTL | `cache.tmdbMaxSize` | + +设置(设置 → 元数据 → TMDB 缓存):`cache.tmdbTtlHours`(缺省 24,MP `META_CACHE_EXPIRE`)、`cache.tmdbMaxSize`(缺省 256,MP `CONF.tmdb`)、 +`cache.emptyResultTtlSec`(缺省 1800,MP `EMPTY_RESULT_CACHE_TTL`)。 + +API:`GET /api/v1/cache/tmdb?kind=recognize|request&limit=` 列出条目(键 / 过期 / 大小,识别条目带 tmdbId / 标题 / 海报); +`POST /api/v1/cache/tmdb/clear` 清空两类(仅 admin)。 + +识别缓存何时读写(`sources-identity/service/tmdb-recognize-cache.ts`): +- 识别资源时先读:命中且有标题 → 按 id 取 TMDB 详情作为唯一远端候选;命中 `{id:0}` → 不再搜索;强制复查(force)不读。 +- 写:自动识别到 TMDB 条目写正常条目;确认无匹配写 `{id:0}`;TMDB 出错或有歧义不写。 +- 人工认定资源身份后回填(MP `update_recognize_cache`),覆盖名称负缓存。 +- "元数据是剧集、缓存是电影"视为类型冲突:读时丢弃,写时拒绝。 + +## 图片代理与海报 + +- `GET /api/v1/img/tmdb/{size}/{file}`(会话或 API Key read):只代理 `image.tmdb.org`,size 走白名单(w92…original), + file 只允许 `[A-Za-z0-9_-]+.(jpg|jpeg|png|webp)`。 +- 缓存路径对照 MP `sanitize_url_path`:`DATA_DIR/cache/images/t/p//`。 +- `cache.globalImageCache`(缺省关闭,MP `GLOBAL_IMAGE_CACHE`)开启才读磁盘缓存;拉到的合法图片总会写盘。 +- 按文件头识别图片类型,非图片 / SVG 拒绝;ETag = md5(内容),`Cache-Control: public, max-age=604800`,If-None-Match 相同回 304。 +- 上游超时(20s)/ 非 200 / 非图片 → 502,前端回退无彩占位。 +- 清理:定时任务 `cache.images_clean`(每天)删除 mtime 早于 `cache.globalImageCacheDays`(缺省 7)天的文件。 +- 用量 / 清空:`GET /api/v1/cache/images`、`POST /api/v1/cache/images/clear`(仅 admin);设置 → 元数据 → 图片代理与缓存。 + +前端:作品库列表、作品详情、搜索面板的 TMDB 候选在有 `posterPath`(metadata.posterPath / TMDB 候选)时显示海报, +加载失败回退原来的无彩占位。TMDB 详情刷新会把 posterPath / backdropPath 写进 metadata_field。 + +## 常见问题 + +- TMDB 缓存不生效:确认 `select region, count(*) from kv_cache group by 1` 有 `tmdb.request`;PG 读写失败时按未命中处理并记 + `TMDB 缓存读写失败，按未命中处理`。 +- 改了 TMDB 缓存设置:新写入的条目按新 TTL / 上限;已有条目保持原过期时间,必要时清空。 +- 数据库重启后缓存为空:UNLOGGED 表的正常行为。 diff --git a/packages/contracts/src/api.ts b/packages/contracts/src/api.ts index 665641e7..d5efdc4e 100644 --- a/packages/contracts/src/api.ts +++ b/packages/contracts/src/api.ts @@ -3,6 +3,7 @@ import * as agent from "./agent.ts"; import * as assets from "./assets.ts"; import * as auth from "./auth.ts"; import * as browser from "./browser.ts"; +import * as cache from "./cache.ts"; import { CapabilityDescriptor, KnownCapabilityId } from "./capability.ts"; import { type ApiScope, CursorQuery, cursorPage, HealthStatus, Id, IdParam } from "./common.ts"; import * as content from "./content.ts"; @@ -1611,6 +1612,39 @@ export const Api = { }), }, + // ── 缓存(C2-A,docs/runbook/cache.md) ── + cache: { + tmdb: ep({ + method: "get", + path: "/cache/tmdb", + scope: "admin", + summary: "TMDB 缓存条目（请求响应与识别结果：键、过期时间、大小）与上限 / TTL", + query: cache.TmdbCacheQuery, + response: cache.TmdbCacheList, + }), + tmdbClear: ep({ + method: "post", + path: "/cache/tmdb/clear", + scope: "admin", + summary: "清空 TMDB 缓存（请求响应与识别结果）", + response: cache.CacheClearResult, + }), + images: ep({ + method: "get", + path: "/cache/images", + scope: "admin", + summary: "图片代理缓存用量", + response: cache.ImageCacheStatus, + }), + imagesClear: ep({ + method: "post", + path: "/cache/images/clear", + scope: "admin", + summary: "清空图片代理缓存", + response: cache.CacheClearResult, + }), + }, + // ── 设置 ── settings: { get: ep({ method: "get", path: "/settings", scope: "read", summary: "全局设置", response: GlobalSettings }), diff --git a/packages/contracts/src/cache.ts b/packages/contracts/src/cache.ts new file mode 100644 index 00000000..0e81f501 --- /dev/null +++ b/packages/contracts/src/cache.ts @@ -0,0 +1,77 @@ +import { z } from "zod"; +import { NonNegInt, Timestamp } from "./common.ts"; + +// C2-A 缓存管理(docs/runbook/cache.md):TMDB 缓存(请求响应 + 识别结果,对照 MP TmdbCache)、图片代理缓存。 + +export const TmdbCacheKind = z.enum(["request", "recognize"]); +export type TmdbCacheKind = z.infer; + +/** 一条 TMDB 缓存(对照 MP list_items:识别条目带 tmdbId / 标题 / 年份 / 类型 / 海报) */ +export const TmdbCacheItem = z.object({ + kind: TmdbCacheKind, + key: z.string(), + expiresAt: Timestamp, + updatedAt: Timestamp, + /** jsonb 存储字节数 */ + bytes: NonNegInt, + /** 识别条目:TMDB id,0 = 负缓存(未识别) */ + tmdbId: z.number().int().nullable(), + title: z.string().nullable(), + year: z.string().nullable(), + mediaType: z.enum(["movie", "tv", "unknown"]).nullable(), + posterPath: z.string().nullable(), +}); +export type TmdbCacheItem = z.infer; + +export const TmdbCacheQuery = z.object({ + kind: TmdbCacheKind.optional(), + limit: z.coerce.number().int().min(1).max(5000).default(500), +}); + +export const TmdbCacheList = z.object({ + items: z.array(TmdbCacheItem), + counts: z.object({ request: NonNegInt, recognize: NonNegInt }), + /** 每类条数上限(cache.tmdbMaxSize) */ + maxSize: NonNegInt, + ttlHours: NonNegInt, + emptyResultTtlSec: NonNegInt, +}); +export type TmdbCacheList = z.infer; + +export const CacheClearResult = z.object({ deleted: NonNegInt }); +export type CacheClearResult = z.infer; + +/** 图片代理缓存用量 */ +export const ImageCacheStatus = z.object({ + /** cache.globalImageCache */ + enabled: z.boolean(), + /** cache.globalImageCacheDays */ + days: NonNegInt, + usedBytes: NonNegInt, + files: NonNegInt, + dir: z.string(), +}); +export type ImageCacheStatus = z.infer; + +/** TMDB 图片尺寸白名单(与 TMDB configuration.images 一致) */ +export const TMDB_IMAGE_SIZES = [ + "w92", + "w154", + "w185", + "w300", + "w342", + "w500", + "w780", + "w1280", + "h632", + "original", +] as const; +export type TmdbImageSize = (typeof TMDB_IMAGE_SIZES)[number]; + +/** 海报 / 背景图路径 → 图片代理 URL(相对 API_PREFIX 之外的完整路径,前端直接用作 img src) */ +export function tmdbImageUrl(path: string | null | undefined, size: TmdbImageSize = "w342"): string | null { + if (!path) return null; + const p = path.startsWith("/") ? path.slice(1) : path; + if (!/^[A-Za-z0-9_-]+\.(jpg|jpeg|png|webp)$/.test(p)) return null; + return `/api/v1/img/tmdb/${size}/${p}`; +} diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index be807ee5..0fccabdb 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -4,6 +4,7 @@ export * from "./api.ts"; export * from "./assets.ts"; export * from "./auth.ts"; export * from "./browser.ts"; +export * from "./cache.ts"; export * from "./capabilities-io.ts"; export * from "./capability.ts"; export * from "./common.ts"; diff --git a/packages/contracts/src/organization.ts b/packages/contracts/src/organization.ts index a6983d34..3f8d0010 100644 --- a/packages/contracts/src/organization.ts +++ b/packages/contracts/src/organization.ts @@ -340,6 +340,8 @@ export const LibraryItem = z.object({ slotCounts: SlotStatusCounts, activeDownloads: NonNegInt, lastOrganizedAt: Timestamp.nullable(), + /** TMDB 海报路径(metadata.posterPath);前端经图片代理显示,没有时显示无彩占位 */ + posterPath: z.string().nullable().optional(), }); export type LibraryItem = z.infer;