diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts index a2b206cb..de1fd89f 100644 --- a/apps/server/src/main.ts +++ b/apps/server/src/main.ts @@ -84,6 +84,7 @@ const { createApp } = await import("./http/app.ts"); const { PgEventBus } = await import("./http/sse-pg.ts"); const { createLogger } = await import("./logger.ts"); const { SERVER_MODULES } = await import("./modules/index.ts"); +const { startMetadataCache } = await import("./shared-cache.ts"); const { diffCounts, enableReadOnlyReconcile, exportMappings, writeMappingExport } = await import("./ops.ts"); const { auditStorageWarnings, checkStorage, loadMountedPaths, runPreflight } = await import("./preflight.ts"); const { startWorker } = await import("./worker/index.ts"); @@ -219,6 +220,8 @@ try { Deno.exit(1); } const { ctx, outbox, engine } = booted; +// 统一元数据缓存:LISTEN 跨进程失效(在线才启用进程内一级缓存)+ 命中统计定期落库 +stops.push(await startMetadataCache(ctx, config.DATABASE_URL)); if (role === "api" || role === "all") { const app = createApp({ diff --git a/apps/server/src/modules/cache/index.ts b/apps/server/src/modules/cache/index.ts index 097133d9..5f38f5ab 100644 --- a/apps/server/src/modules/cache/index.ts +++ b/apps/server/src/modules/cache/index.ts @@ -14,7 +14,16 @@ import { dataPaths } from "../../config.ts"; import { externalFetch } from "../../egress.ts"; import { requireScope } from "../../http/auth.ts"; import { errorBody } from "../../http/errors.ts"; -import { CACHE_REGIONS, sharedCache, storageSettings, tmdbCachePolicy } from "../../shared-cache.ts"; +import { + CACHE_REGIONS, + METADATA_REGIONS, + metadataCache, + type RegionHitStats, + regionHitStats, + sharedCache, + storageSettings, + 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"; @@ -72,6 +81,13 @@ function route( const str = (v: unknown) => (typeof v === "string" && v ? v : null); +/** 命中统计 → 接口形状;命中率 = (L1 + L2 + 旧值) / (L1 + L2 + 旧值 + 未命中) */ +function hitView(h: RegionHitStats | undefined) { + const v = h ?? { l1Hits: 0, l2Hits: 0, staleHits: 0, misses: 0, failures: 0, refreshes: 0 }; + const hit = v.l1Hits + v.l2Hits + v.staleHits; + return { ...v, hitRate: hit + v.misses > 0 ? hit / (hit + v.misses) : null }; +} + export function createCacheModule(opts: CacheModuleOptions = {}): ServerModule & { images(ctx: ServerContext): ImageCache; } { @@ -166,7 +182,7 @@ export function createCacheModule(opts: CacheModuleOptions = {}): ServerModule & return out; }); route(app, Api.cache.tmdbClear, async () => { - const cache = sharedCache(ctx); + const cache = metadataCache(ctx); const deleted = (await cache.clear(CACHE_REGIONS.tmdbRecognize)) + (await cache.clear(CACHE_REGIONS.tmdbRequest)); ctx.logger.info({ deleted }, "TMDB 缓存已清空"); @@ -194,18 +210,30 @@ export function createCacheModule(opts: CacheModuleOptions = {}): ServerModule & coalesce(sum(pg_column_size(value) + pg_column_size(key)), 0)::text as bytes, max(updated_at) as last from kv_cache group by region order by region`.execute(ctx.db); - return { - items: r.rows.map((x) => ({ - region: x.region, - entries: Number(x.entries), - expired: Number(x.expired), - bytes: Number(x.bytes), - lastWriteAt: x.last ? new Date(x.last).toISOString() : null, - })), + // 统一元数据缓存的 region 带最近 7 天命中统计(口径见 contracts CacheHitStats) + const hits = await regionHitStats(ctx).catch(() => new Map()); + const withHits = (region: string) => { + const h = hits.get(region); + if (!h) return METADATA_REGIONS.includes(region) ? { hits: hitView(undefined) } : {}; + return { hits: hitView(h) }; }; + const items = r.rows.map((x) => ({ + region: x.region, + entries: Number(x.entries), + expired: Number(x.expired), + bytes: Number(x.bytes), + lastWriteAt: x.last ? new Date(x.last).toISOString() : null, + ...withHits(x.region), + })); + // 条目已清空、但最近有读取统计的元数据 region 也列出来 + for (const region of hits.keys()) + if (!items.some((i) => i.region === region)) + items.push({ region, entries: 0, expired: 0, bytes: 0, lastWriteAt: null, ...withHits(region) }); + items.sort((a, b) => a.region.localeCompare(b.region)); + return { items }; }); route(app, Api.cache.regionClear, async ({ params }) => { - const deleted = await sharedCache(ctx).clear(params.region); + const deleted = await metadataCache(ctx).clear(params.region); ctx.logger.info({ region: params.region, deleted }, "缓存区已清空"); return { deleted }; }); @@ -217,7 +245,7 @@ export function createCacheModule(opts: CacheModuleOptions = {}): ServerModule & lane: "maintenance", defaults: { name: "缓存清理", - description: "删除过期的 TMDB 与识别缓存", + description: "删除过期的元数据与其它缓存", group: "maintain", schedule: { type: "interval", everyMs: 10 * 60_000 }, firstRun: "after_interval", diff --git a/apps/server/src/modules/search/discover.ts b/apps/server/src/modules/search/discover.ts index 6e72634c..3126b7f4 100644 --- a/apps/server/src/modules/search/discover.ts +++ b/apps/server/src/modules/search/discover.ts @@ -24,10 +24,10 @@ import { type DiscoverTmdbQuery, type InventoryEntry, } from "@umt/contracts"; -import { pgCacheFor } from "@umt/db"; import { type IdentityCandidate, parseRelease, resolveIdentity } from "@umt/domain"; import { ApiException } from "../../http/errors.ts"; import { runSandbox } from "../../sandbox/client.ts"; +import { CACHE_REGIONS, metaCachePolicy, metadataCache, storageSettings } from "../../shared-cache.ts"; import type { SourcesRuntime } from "../sources-identity/runtime.ts"; import { subjectRecall } from "../sources-identity/service/id-escalation.ts"; import { metaClients, toCached } from "../sources-identity/service/meta-subjects.ts"; @@ -41,9 +41,8 @@ import { localEntities } from "./media.ts"; // - 每条结果按 TMDB 身份标本地状态(已订阅 / 媒体库已有 / 下载中);豆瓣 / bgm 条目匹配过 TMDB 的才知道(discover.match)。 // 本地状态每次现查,不进缓存。 -/** 豆瓣 / bgm 条目 → TMDB 作品的确定匹配(点开时写入,列表据此标本地状态) */ -const MATCH_REGION = "discover.match"; -const MATCH_TTL_SEC = 30 * 86_400; +/** 豆瓣 / bgm 条目 → TMDB 作品的确定匹配(点开时写入,列表据此标本地状态);TTL 按「豆瓣 / bgm × 详情」 */ +const MATCH_REGION = CACHE_REGIONS.discoverMatch; const MATCH_MAX = 20_000; export type InventoryLookup = (ids: string[]) => Promise>; @@ -108,12 +107,12 @@ const fromTmdb = (t: TmdbListItem): DiscoverItem => ({ /** 豆瓣 / bgm 条目补上记住的 TMDB 匹配,再按 TMDB 身份查本地作品与库存 */ export async function annotate(rt: SourcesRuntime, items: DiscoverItem[], inventory: InventoryLookup) { - const cache = pgCacheFor(rt.ctx.db); + const cache = metadataCache(rt.ctx); await Promise.all( items .filter((it) => !it.tmdb && it.source !== "tmdb") .map(async (it) => { - const hit = await cache.get(MATCH_REGION, `${it.source}:${it.id}`).catch(() => null); + const hit = await cache.get(MATCH_REGION, `${it.source}:${it.id}`); if (hit) it.tmdb = { type: hit.type, id: hit.id }; }), ); @@ -257,9 +256,9 @@ const view = (m: MetadataCandidate): DiscoverMatchCandidate => ({ }); export async function matchTmdb(rt: SourcesRuntime, req: DiscoverMatchRequest): Promise { - const cache = pgCacheFor(rt.ctx.db); + const cache = metadataCache(rt.ctx); const memoKey = `${req.source}:${req.id}`; - const memo = await cache.get(MATCH_REGION, memoKey).catch(() => null); + const memo = await cache.get(MATCH_REGION, memoKey); if (memo) return { match: memo, candidates: [memo] }; const subject = toCached({ ...req, cover: null, rating: null, date: null, episodes: null, url: "" }, rt.now()); const base = parseRelease(req.title); @@ -278,6 +277,9 @@ export async function matchTmdb(rt: SourcesRuntime, req: DiscoverMatchRequest): const best = res.status === "auto" && res.best ? byCand.get(res.best.candidate) : undefined; const match = best ? view(best) : null; // 只记确定的匹配;不确定 / 失败不记,下次重算 - if (match) await cache.set(MATCH_REGION, memoKey, match, { ttl: MATCH_TTL_SEC, maxsize: MATCH_MAX }).catch(() => {}); + if (match) { + const ttl = metaCachePolicy((await storageSettings(rt.ctx)).cacheTtl, req.source, "detail").ttl; + await cache.set(MATCH_REGION, memoKey, match, { ttl, maxsize: MATCH_MAX }).catch(() => {}); + } return DiscoverMatchResult.parse({ match, candidates: ranked.map(view) }); } diff --git a/apps/server/src/modules/sources-identity/runtime.ts b/apps/server/src/modules/sources-identity/runtime.ts index 64854cec..cef5cb14 100644 --- a/apps/server/src/modules/sources-identity/runtime.ts +++ b/apps/server/src/modules/sources-identity/runtime.ts @@ -11,19 +11,13 @@ import { type SiteDefinition, type SiteRateLimiter, type TmdbClient, + type TmdbRequestKind, type TmdbSharedCache, } from "@umt/adapters"; import { DEFAULT_SETTINGS, type SitesSettings } from "@umt/contracts"; import { pgTokenBucketFor, type SiteRow, sql } from "@umt/db"; import { egressKey, externalFetch, siteEgress, siteFetchFor } from "../../egress.ts"; -import { - CACHE_REGIONS, - renewTrackedTmdbDetail, - sharedCache, - type TmdbCachePolicy, - tmdbCachePolicy, - tmdbDetailRef, -} from "../../shared-cache.ts"; +import { CACHE_REGIONS, metadataCache, type TmdbCachePolicy, tmdbCachePolicy, watchedPin } from "../../shared-cache.ts"; import type { ServerContext } from "../types.ts"; import { CloudflareGate, withClearance } from "./service/cloudflare.ts"; import { failureSampleSink } from "./service/failure-sample.ts"; @@ -165,6 +159,8 @@ export class SourcesRuntime { ttlMs: policy.ttl * 1000, searchTtlMs: policy.searchTtl * 1000, emptyTtlMs: policy.emptyTtl * 1000, + listTtlMs: policy.listTtl * 1000, + imageTtlMs: policy.imageTtl * 1000, }, sharedCache: tmdbSharedCache(this.ctx, policy), } @@ -420,28 +416,34 @@ export class SourcesRuntime { } /** - * TMDB 请求响应缓存:PG kv_cache region=tmdb.request;PG 出错按未命中 / 不写处理,不让 TMDB 请求失败。 - * 详情命中且属于追踪中的作品时续期(CACHE-2),续期失败只记 debug。 + * TMDB 请求响应缓存:统一元数据缓存 region=tmdb.request(L1 进程内存 + L2 kv_cache,过期先用旧值后台刷新、同请求合并)。 + * TTL 按请求类别取策略表(搜索 / 详情 / 列表 / 图片);关注中作品的条目写入 pinned、命中续期。 + * PG 出错按未命中 / 不写处理,不让 TMDB 请求失败。get / set 保留给不走 fetch 的旧调用方。 */ -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 缓存读写失败,按未命中处理"); +export function tmdbSharedCache(ctx: ServerContext, policy: TmdbCachePolicy): TmdbSharedCache { + const cache = metadataCache(ctx); + const region = CACHE_REGIONS.tmdbRequest; + const ttlOf = (kind: TmdbRequestKind) => + kind === "search" + ? { ttl: policy.searchTtl, emptyTtl: policy.emptyTtl } + : kind === "list" + ? { ttl: policy.listTtl, emptyTtl: policy.listEmptyTtl } + : kind === "image" + ? { ttl: policy.imageTtl, emptyTtl: policy.emptyTtl } + : { ttl: policy.ttl, emptyTtl: policy.emptyTtl }; return { - get: async (key) => { - const v = await cache.get(CACHE_REGIONS.tmdbRequest, key).catch((e) => { - warn(e); - return null; - }); - if (v !== null && tmdbDetailRef(key)) { - await renewTrackedTmdbDetail(ctx, key, policy.ttl).catch((e) => - ctx.logger.debug({ err: { message: (e as Error).message } }, "TMDB 详情缓存续期失败"), - ); - } - return v; - }, + fetch: (req, load) => + cache.fetch( + region, + req.key, + { ...ttlOf(req.kind), isEmpty: req.isEmpty, maxsize: policy.maxsize, pin: watchedPin(ctx, req.key) }, + load, + ), + get: (key) => cache.get(region, key), set: (key, value, ttlMs) => - cache.set(CACHE_REGIONS.tmdbRequest, key, value, { ttl: ttlMs / 1000, maxsize: policy.maxsize }).catch(warn), + cache + .set(region, key, value, { ttl: ttlMs / 1000, maxsize: policy.maxsize }) + .catch((e) => ctx.logger.warn({ err: { message: (e as Error).message } }, "TMDB 缓存读写失败,按未命中处理")), }; } diff --git a/apps/server/src/modules/sources-identity/service/douban-subject.ts b/apps/server/src/modules/sources-identity/service/douban-subject.ts index 1e12d1b4..56b32153 100644 --- a/apps/server/src/modules/sources-identity/service/douban-subject.ts +++ b/apps/server/src/modules/sources-identity/service/douban-subject.ts @@ -1,5 +1,6 @@ import { formatDetailPage } from "@umt/adapters"; -import { pgCacheFor } from "@umt/db"; +import { DEFAULT_STORAGE_SETTINGS } from "@umt/contracts"; +import { metaCachePolicy, metadataCache, storageSettings } from "../../../shared-cache.ts"; import type { ServerContext } from "../../types.ts"; // ID-3 豆瓣条目缓存(kv_cache region douban.subject,key = 豆瓣 subject id):只存标题 / 别名 / 年份,供「详情页只有豆瓣 ID」时 @@ -10,8 +11,8 @@ import type { ServerContext } from "../../types.ts"; // - douban_api:豆瓣移动版接口(识别升级时缓存没有才查)。bgm.tv 条目同形,存 region bangumi.subject。 export const DOUBAN_SUBJECT_REGION = "douban.subject"; -const TTL_SEC = 30 * 86_400; -const MAX_ENTRIES = 20_000; +/** 条目 TTL 取策略表「豆瓣 / bgm × 详情」(storage.cacheTtl.metaDetailDays,缺省 30 天) */ +export const META_SUBJECT_MAX_ENTRIES = 20_000; export interface DoubanSubject { id: string; @@ -92,16 +93,24 @@ export function doubanSubjectFromSnippet( }; } -type Db = Pick; +type Db = Pick & Partial>; -/** 写入缓存;站点详情来源优先(已有 site_detail 条目时不用搜索摘要覆盖) */ +/** 条目 TTL(秒):有设置读设置,否则缺省 */ +export async function metaSubjectTtl(ctx: Db): Promise { + const t = ctx.settings + ? (await storageSettings({ settings: ctx.settings })).cacheTtl + : DEFAULT_STORAGE_SETTINGS.cacheTtl; + return metaCachePolicy(t, "douban", "detail").ttl; +} + +/** 写入缓存(统一元数据缓存,两级);站点详情来源优先(已有 site_detail 条目时不用搜索摘要覆盖) */ export async function saveDoubanSubject(ctx: Db, s: DoubanSubject, region = DOUBAN_SUBJECT_REGION): Promise { - const cache = pgCacheFor(ctx.db); + const cache = metadataCache(ctx); if (s.source === "web_snippet") { const prev = await cache.get(region, s.id); if (prev?.source === "site_detail") return; } - await cache.set(region, s.id, s, { ttl: TTL_SEC, maxsize: MAX_ENTRIES }); + await cache.set(region, s.id, s, { ttl: await metaSubjectTtl(ctx), maxsize: META_SUBJECT_MAX_ENTRIES }); } export async function loadDoubanSubject( @@ -110,7 +119,7 @@ export async function loadDoubanSubject( region = DOUBAN_SUBJECT_REGION, ): Promise { try { - const v = await pgCacheFor(ctx.db).get(region, id); + const v = await metadataCache(ctx).get(region, id); return v && typeof v.title === "string" && v.title ? v : null; } catch { return null; diff --git a/apps/server/src/modules/sources-identity/service/meta-subjects.ts b/apps/server/src/modules/sources-identity/service/meta-subjects.ts index efe038b1..4f7df5ce 100644 --- a/apps/server/src/modules/sources-identity/service/meta-subjects.ts +++ b/apps/server/src/modules/sources-identity/service/meta-subjects.ts @@ -4,51 +4,56 @@ import { createDoubanClient, type DoubanClient, type MetaRequestCache, + MetaSourceError, type MetaSubject, RequestLimiter, } from "@umt/adapters"; -import { cachedCall, pgCacheFor } from "@umt/db"; import { normalizeTitle, toSimplified } from "@umt/domain"; -import { storageSettings } from "../../../shared-cache.ts"; +import { CACHE_REGIONS, metaCachePolicy, metadataCache, storageSettings } from "../../../shared-cache.ts"; import type { SourcesRuntime } from "../runtime.ts"; -import { DOUBAN_SUBJECT_REGION, type DoubanSubject, loadDoubanSubject, saveDoubanSubject } from "./douban-subject.ts"; +import { + DOUBAN_SUBJECT_REGION, + type DoubanSubject, + loadDoubanSubject, + META_SUBJECT_MAX_ENTRIES, +} from "./douban-subject.ts"; // 豆瓣 / bgm.tv 作为辅助信息源(对照 MP:豆瓣、Bangumi 条目 → 名称 + 年份 → TMDB 投影;身份仍以 TMDB 为准)。 // 条目先查缓存(30 天),没有才在线查一次;查不到 / 网络失败记 6 小时负缓存,不反复打外站。 -export const BANGUMI_SUBJECT_REGION = "bangumi.subject"; -const MISS_REGION = "meta.subject_miss"; -const MISS_TTL_SEC = 6 * 3600; +export const BANGUMI_SUBJECT_REGION = CACHE_REGIONS.bangumiSubject; +const MISS_REGION = CACHE_REGIONS.subjectMiss; +/** 网络失败 / 被拦后同一条目在本进程里冷却多久不再在线查(不写缓存,只是不连着打外站) */ +const FAIL_COOLDOWN_MS = 30 * 60_000; +const failedAt = new Map(); export type MetaSource = "douban" | "bangumi"; const REGION: Record = { douban: DOUBAN_SUBJECT_REGION, bangumi: BANGUMI_SUBJECT_REGION }; /** 豆瓣 / bgm 请求缓存 region(对照 MP DoubanApi.__invoke / BangumiApi 的 @cached);key 见 metaRequestKey */ -export const META_REQUEST_REGIONS = { douban: "douban.request", bangumi: "bangumi.request" } as const; +export const META_REQUEST_REGIONS = { + douban: CACHE_REGIONS.doubanRequest, + bangumi: CACHE_REGIONS.bangumiRequest, +} as const; const META_REQUEST_MAXSIZE = 5000; /** 每源限流(进程内):豆瓣风控严,间隔大一些;缓存命中不占额度 */ const META_LIMITERS = { douban: new RequestLimiter(2, 300), bangumi: new RequestLimiter(3, 100) }; /** - * 请求缓存:搜索按 storage.cacheTtl.metaSearchDays,列表(榜单 / 热门 / 每日放送 / 趋势页)按 metaListHours, - * 空结果按 metaEmptyHours(不超过正常 TTL);失败(网络、403、风控页)在适配器里抛出,cachedCall 不写。 - * 条目详情不走这里:douban.subject / bangumi.subject 条目级缓存(30 天)已经存了,避免同一份数据存两遍。 + * 请求缓存(统一元数据缓存):按策略表「来源 × 资源类型」取 TTL——搜索 metaSearchDays / 查无 metaMissDays, + * 列表(榜单 / 热门 / 每日放送 / 趋势页)metaListHours / 空 metaEmptyHours;过期先用旧值后台刷新。 + * 失败(网络、403、风控页)在适配器里抛出,不写。 + * 条目详情不走这里:douban.subject / bangumi.subject 条目级缓存已经存了,避免同一份数据存两遍。 */ export function metaRequestCache(ctx: SourcesRuntime["ctx"]): MetaRequestCache { return { async cached(req, load) { const t = (await storageSettings(ctx)).cacheTtl; - const ttl = req.kind === "search" ? t.metaSearchDays * 86_400 : t.metaListHours * 3600; - return cachedCall( - pgCacheFor(ctx.db), + const p = metaCachePolicy(t, req.source, req.kind); + return metadataCache(ctx).fetch( + META_REQUEST_REGIONS[req.source], req.key, - { - region: META_REQUEST_REGIONS[req.source], - ttl, - emptyTtl: Math.min(ttl, t.metaEmptyHours * 3600), - emptyIf: req.empty, - maxsize: META_REQUEST_MAXSIZE, - }, + { ...p, isEmpty: req.empty, maxsize: META_REQUEST_MAXSIZE }, load, ); }, @@ -77,23 +82,38 @@ export function toCached(s: MetaSubject, now: Date): DoubanSubject { }; } -/** 条目(缓存 → 在线);拿不到为 null */ +/** + * 条目(缓存 → 在线,统一元数据缓存:过期先用旧值后台刷新);拿不到为 null。 + * 条目不存在记查无(metaMissDays);网络失败 / 被拦不缓存,同一条目本进程冷却 30 分钟再查。 + */ export async function loadSubject(rt: SourcesRuntime, source: MetaSource, id: string): Promise { - const hit = await loadDoubanSubject(rt.ctx, id, REGION[source]); - if (hit) return hit; const clients = metaClients(rt); - if (!clients) return null; - const cache = pgCacheFor(rt.ctx.db); + if (!clients) return loadDoubanSubject(rt.ctx, id, REGION[source]); + const cache = metadataCache(rt.ctx); const missKey = `${source}:${id}`; - if (await cache.get(MISS_REGION, missKey).catch(() => null)) return null; + const t = (await storageSettings(rt.ctx)).cacheTtl; + const p = metaCachePolicy(t, source, "detail"); try { - const s = source === "douban" ? await clients.douban.subject(id) : await clients.bangumi.subject(id); - const cached = toCached(s, rt.now()); - await saveDoubanSubject(rt.ctx, cached, REGION[source]); - return cached; + const v = await cache.fetch( + REGION[source], + id, + { ttl: p.ttl, maxsize: META_SUBJECT_MAX_ENTRIES, skipIf: (x) => !x?.title }, + async () => { + if ((failedAt.get(missKey) ?? 0) > Date.now() - FAIL_COOLDOWN_MS) return null; + if (await cache.get(MISS_REGION, missKey)) return null; + const s = source === "douban" ? await clients.douban.subject(id) : await clients.bangumi.subject(id); + return toCached(s, rt.now()); + }, + ); + return v && typeof v.title === "string" && v.title ? v : null; } catch (e) { rt.ctx.logger.info({ source, id, err: (e as Error).message }, "辅助信息源条目查询失败"); - await cache.set(MISS_REGION, missKey, true, { ttl: MISS_TTL_SEC, maxsize: 5000 }).catch(() => {}); + if (e instanceof MetaSourceError && e.kind === "not_found") + await cache.set(MISS_REGION, missKey, true, { ttl: p.emptyTtl, maxsize: 5000 }).catch(() => {}); + else { + if (failedAt.size > 5000) failedAt.clear(); + failedAt.set(missKey, Date.now()); + } return null; } } 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 index c7f2d18d..c9413e84 100644 --- a/apps/server/src/modules/sources-identity/service/tmdb-recognize-cache.ts +++ b/apps/server/src/modules/sources-identity/service/tmdb-recognize-cache.ts @@ -1,6 +1,6 @@ import type { MetadataCandidate } from "@umt/adapters"; import type { ParsedRelease } from "@umt/domain"; -import { CACHE_REGIONS, sharedCache, tmdbCachePolicy } from "../../../shared-cache.ts"; +import { CACHE_REGIONS, metadataCache, 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)。 @@ -92,7 +92,7 @@ export async function loadRecognizeCache( ): Promise { const tmdb = await rt.tmdb(); const key = recognizeKey(meta, localeOf(tmdb)); - const cache = sharedCache(rt.ctx); + const cache = metadataCache(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)) { @@ -120,7 +120,7 @@ export async function saveRecognizeCache( const tmdb = await rt.tmdb(); const key = recognizeKey(meta, localeOf(tmdb)); const policy = await tmdbCachePolicy(rt.ctx); - const cache = sharedCache(rt.ctx); + const cache = metadataCache(rt.ctx); if (info) { if (typeConflicted(meta, info.entityType, info.externalId)) { rt.ctx.logger.warn({ key, title: info.title, type: info.entityType }, "识别结果类型与元数据冲突,不写入识别缓存"); diff --git a/apps/server/src/modules/sources-identity/service/watched.ts b/apps/server/src/modules/sources-identity/service/watched.ts index 13cb7e67..b3f0f7e5 100644 --- a/apps/server/src/modules/sources-identity/service/watched.ts +++ b/apps/server/src/modules/sources-identity/service/watched.ts @@ -1,4 +1,4 @@ -import { sql } from "@umt/db"; +import { type RawBuilder, sql } from "@umt/db"; import type { ServerContext } from "../../types.ts"; import type { Exec } from "./common.ts"; @@ -17,9 +17,8 @@ import type { Exec } from "./common.ts"; const OPEN_SUBSCRIPTION = ["active", "pending"]; const DOWNLOAD_DONE = ["imported", "removed"]; -export async function isEntityWatched(exec: Exec, entityId: string): Promise { - const r = await sql<{ watched: boolean }>` - select ( +/** 作品 e(SQL 片段:作品 id 表达式)是否关注中;isEntityWatched 与 watchedTmdbRefs 共用一个口径 */ +const watchedExpr = (entityId: RawBuilder) => sql`( exists (select 1 from subscription s where s.media_entity_id = ${entityId} and s.status = any(${OPEN_SUBSCRIPTION})) or exists (select 1 from media_asset a where a.media_entity_id = ${entityId} and a.status = 'active') or exists (select 1 from external_identity x @@ -27,7 +26,7 @@ export async function isEntityWatched(exec: Exec, entityId: string): Promise>'lookup' in ('search', 'manual')) or exists (select 1 from acquisition_plan p left join acquisition_slot sl on sl.id = p.slot_id - where (sl.media_entity_id = ${entityId} or p.evaluation->>'mediaEntityId' = ${entityId}::text) + where (sl.media_entity_id = ${entityId} or p.evaluation->>'mediaEntityId' = (${entityId})::text) and (p.evaluation->>'manual')::boolean is true) or exists (select 1 from download_task t left join source_release r on r.id = t.source_release_id @@ -37,7 +36,28 @@ export async function isEntityWatched(exec: Exec, entityId: string): Promise` / `tv:` 与选中的剧集组 `group:`。 + * 口径同 isEntityWatched。 + */ +export async function watchedTmdbRefs(exec: Exec): Promise> { + const r = await sql<{ ref: string; grp: string | null }>` + select x.entity_type || ':' || x.external_id as ref, e.episode_group_id as grp + from external_identity x join media_entity e on e.id = x.media_entity_id + where x.provider = 'tmdb' and x.status = 'accepted' and x.entity_type in ('movie', 'tv') + and e.status <> 'merged' and ${watchedExpr(sql`e.id`)}`.execute(exec); + const out = new Set(); + for (const x of r.rows) { + out.add(x.ref); + if (x.grp) out.add(`group:${x.grp}`); + } + return out; +} + +export async function isEntityWatched(exec: Exec, entityId: string): Promise { + const r = await sql<{ watched: boolean }>`select ${watchedExpr(sql`${entityId}::uuid`)} as watched`.execute(exec); return r.rows[0]?.watched === true; } diff --git a/apps/server/src/shared-cache.ts b/apps/server/src/shared-cache.ts index 6cf2099c..5a2425ec 100644 --- a/apps/server/src/shared-cache.ts +++ b/apps/server/src/shared-cache.ts @@ -1,33 +1,131 @@ -import { DEFAULT_SETTINGS, type StorageSettings, withStorageDefaults } from "@umt/contracts"; -import { type PgCache, pgCacheFor, sql } from "@umt/db"; +import { DEFAULT_SETTINGS, type StorageCacheTtl, type StorageSettings, withStorageDefaults } from "@umt/contracts"; +import { + type FetchPolicy, + listenCacheInvalidation, + type MetadataCache, + metadataCacheFor, + type PgCache, + pgCacheFor, + sql, +} from "@umt/db"; +import { watchedTmdbRefs } from "./modules/sources-identity/service/watched.ts"; 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; -// CACHE-2 起各 region 的 TTL 取值来源是 settings.storage.cacheTtl(docs/runbook/storage.md),语义不变。 +// 进程级共享缓存入口(docs/runbook/cache.md): +// - PgCache:PG kv_cache(移植 MP app/runtime/cache.py 的 TTLCache,存储换成 PG,三进程共享),非元数据的缓存仍直接用; +// - MetadataCache:统一元数据缓存,TMDB / 豆瓣 / Bangumi 共用。L1 进程内存 + L2 kv_cache、过期先用旧值后台刷新、 +// 同请求合并、按 region 计命中。TTL 按「来源 × 资源类型」取 settings.storage.cacheTtl(metaCachePolicy)。 export const CACHE_REGIONS = { /** TMDB 请求响应(MP TMDb.request 的 @cached):key = path?query */ tmdbRequest: "tmdb.request", /** TMDB 识别结果(MP TmdbCache):key = [类型][语言]{tmdbid 或名称}-{年份}-{季} */ tmdbRecognize: "tmdb.recognize", + doubanRequest: "douban.request", + bangumiRequest: "bangumi.request", + doubanSubject: "douban.subject", + bangumiSubject: "bangumi.subject", + /** 豆瓣 / bgm 条目查无(只记「不存在」,网络失败 / 被拦不记) */ + subjectMiss: "meta.subject_miss", + /** 豆瓣 / bgm 条目 → TMDB 作品的确定匹配 */ + discoverMatch: "discover.match", } as const; +/** 命中统计只看这些 region(统一元数据缓存);其它 region 不经 MetadataCache */ +export const METADATA_REGIONS: readonly string[] = Object.values(CACHE_REGIONS); + const DAY = 86_400; +const HOUR = 3_600; export function sharedCache(ctx: Pick): PgCache { return pgCacheFor(ctx.db); } +/** 统一元数据缓存(同一个 db 共用一个实例) */ +export function metadataCache(ctx: Pick & Partial>): MetadataCache { + return metadataCacheFor(ctx.db, ctx.logger ? { logger: ctx.logger } : {}); +} + +/** + * 进程启动时调用:LISTEN umt_cache_invalidate(在线才启用 L1)+ 每分钟把命中计数写进 cache_stat。 + * 返回 stop(停监听、最后写一次统计)。监听连不上时只记警告,L1 保持停用,照常走 L2。 + */ +export async function startMetadataCache( + ctx: Pick, + connectionString: string, + opts: { flushMs?: number } = {}, +): Promise<() => Promise> { + const cache = metadataCache(ctx); + const log = ctx.logger.child({ component: "metadata-cache" }); + let listener: Awaited> | null = null; + try { + listener = await listenCacheInvalidation(cache, { connectionString, logger: log }); + } catch (err) { + log.warn({ err: { message: (err as Error).message } }, "缓存失效监听启动失败,进程内一级缓存停用"); + } + const flush = () => + cache.flushStats().catch((err) => log.debug({ err: { message: (err as Error).message } }, "缓存统计写入失败")); + const timer = setInterval(flush, opts.flushMs ?? 60_000); + return async () => { + clearInterval(timer); + await listener?.stop(); + await flush(); + }; +} + /** settings.storage(读失败用缺省) */ export async function storageSettings(ctx: Pick): Promise { return withStorageDefaults(await ctx.settings.get("storage").catch(() => undefined)); } +// ── 策略表:来源 × 资源类型 → TTL ── + +export type MetaCacheSource = "tmdb" | "douban" | "bangumi"; +/** search:搜索 / 识别;detail:详情 / 季 / 剧集组 / 条目;list:榜单 / 热门 / 每日放送;image:人物 / 图片 */ +export type MetaResourceKind = "search" | "detail" | "list" | "image"; + +export interface MetaCachePolicy { + /** 秒 */ + ttl: number; + /** 空结果 / 查无(秒) */ + emptyTtl: number; +} + +/** + * 策略表(docs/runbook/cache.md)。宽限期统一按 TTL 的 50%(上限 7 天,见 @umt/db defaultGrace);失败不缓存。 + * + * | 来源 | search | detail | list | image | + * | TMDB | tmdbSearchDays / tmdbMissDays | tmdbDetailDays / tmdbMissDays | tmdbListHours / tmdbListEmptyHours | imageDays / tmdbMissDays | + * | 豆瓣 / bgm | metaSearchDays / metaMissDays | metaDetailDays / metaMissDays | metaListHours / metaEmptyHours | (无) | + */ +export function metaCachePolicy(t: StorageCacheTtl, source: MetaCacheSource, kind: MetaResourceKind): MetaCachePolicy { + const cap = (ttl: number, empty: number) => ({ ttl, emptyTtl: Math.min(ttl, empty) }); + if (source === "tmdb") { + switch (kind) { + case "search": + return cap(t.tmdbSearchDays * DAY, t.tmdbMissDays * DAY); + case "list": + return cap(t.tmdbListHours * HOUR, t.tmdbListEmptyHours * HOUR); + case "image": + return cap(t.imageDays * DAY, t.tmdbMissDays * DAY); + default: + return cap(t.tmdbDetailDays * DAY, t.tmdbMissDays * DAY); + } + } + switch (kind) { + case "search": + return cap(t.metaSearchDays * DAY, t.metaMissDays * DAY); + case "list": + return cap(t.metaListHours * HOUR, t.metaEmptyHours * HOUR); + default: + return cap(t.metaDetailDays * DAY, t.metaMissDays * DAY); + } +} + /** * TMDB 缓存策略(MP:CONF.meta = META_CACHE_EXPIRE、CONF.tmdb 条数、EMPTY_RESULT_CACHE_TTL)。 - * CACHE-2:MP 只有一个 CONF.meta,这里按请求类别取 storage.cacheTtl:搜索 / find 与识别结果用 searchTtl, - * 详情(/movie/{id}、/tv/{id}、季)用 ttl(= tmdbDetailDays),空结果与未识别负缓存用 emptyTtl。 + * 按请求类别取 storage.cacheTtl:搜索 / find 与识别结果用 searchTtl,详情(/movie/{id}、/tv/{id}、季)用 ttl, + * 空结果与未识别负缓存用 emptyTtl;列表 / 图片见 metaCachePolicy。 */ export interface TmdbCachePolicy { /** 详情 TTL(秒) */ @@ -37,16 +135,25 @@ export interface TmdbCachePolicy { maxsize: number; /** 秒 */ emptyTtl: number; + /** 列表(秒)与列表空结果(秒) */ + listTtl: number; + listEmptyTtl: number; + /** 人物 / 图片(秒) */ + imageTtl: number; } export async function tmdbCachePolicy(ctx: Pick): Promise { const c = await ctx.settings.get("cache").catch(() => DEFAULT_SETTINGS.cache); const t = (await storageSettings(ctx)).cacheTtl; + const list = metaCachePolicy(t, "tmdb", "list"); return { ttl: t.tmdbDetailDays * DAY, searchTtl: t.tmdbSearchDays * DAY, maxsize: c.tmdbMaxSize ?? DEFAULT_SETTINGS.cache.tmdbMaxSize, emptyTtl: t.tmdbMissDays * DAY, + listTtl: list.ttl, + listEmptyTtl: list.emptyTtl, + imageTtl: t.imageDays * DAY, }; } @@ -55,47 +162,102 @@ export function isTmdbSearchPath(key: string): boolean { return /^\/(search|find)\//.test(key); } -/** 详情键 → `movie:` / `tv:`;不是详情返回 null */ +/** + * 键 → 所属作品:`movie:` / `tv:`(详情、季、剧集、图片、剧集组列表都算); + * 剧集组详情 `/tv/episode_group/` → `group:`;都不是返回 null。 + */ export function tmdbDetailRef(key: string): string | null { + const g = /^\/tv\/episode_group\/([^/?]+)/.exec(key); + if (g) return `group:${decodeURIComponent(g[1]!)}`; const m = /^\/(movie|tv)\/(\d+)(?:[/?]|$)/.exec(key); return m ? `${m[1]}:${m[2]}` : null; } -// ── 追踪中的作品:详情缓存命中时续期 ── -// 追踪 = 作品有未结束的订阅(active / paused / pending)且有已接受的 TMDB 身份。集合按进程缓存 5 分钟。 +// ── 关注中的作品:条目 pinned(LRU 不淘汰)、命中续期 ── +// 关注口径同 service/watched.ts isEntityWatched;集合 = 关注中作品的 TMDB 身份 + 选中的剧集组,按进程缓存 5 分钟。 -const TRACKED_TTL_MS = 5 * 60_000; -const trackedCache = new WeakMap> }>(); +const WATCHED_TTL_MS = 5 * 60_000; +const watchedCache = new WeakMap> }>(); export async function trackedTmdbRefs(ctx: Pick, now = Date.now()): Promise> { - const hit = trackedCache.get(ctx.db); - if (hit && now - hit.at < TRACKED_TTL_MS) return hit.set; - const set = sql<{ ref: string }>`select distinct x.entity_type || ':' || x.external_id as ref - from subscription s join external_identity x on x.media_entity_id = s.media_entity_id - where s.status in ('active', 'paused', 'pending') and x.provider = 'tmdb' and x.status = 'accepted' - and x.entity_type in ('movie', 'tv')` - .execute(ctx.db) - .then((r) => new Set(r.rows.map((x) => x.ref))); - trackedCache.set(ctx.db, { at: now, set }); - set.catch(() => trackedCache.delete(ctx.db)); + const hit = watchedCache.get(ctx.db); + if (hit && now - hit.at < WATCHED_TTL_MS) return hit.set; + const set = watchedTmdbRefs(ctx.db); + watchedCache.set(ctx.db, { at: now, set }); + set.catch(() => watchedCache.delete(ctx.db)); return set; } -/** 测试用:清掉追踪集合缓存 */ +/** 清掉关注集合缓存(测试;进入关注时预热前也清,让新关注的作品立即 pinned) */ export function resetTrackedTmdbRefs(ctx: Pick): void { - trackedCache.delete(ctx.db); + watchedCache.delete(ctx.db); +} + +/** 这个 TMDB 请求键属于关注中的作品(详情 / 季 / 剧集组 / 图片) */ +export async function isWatchedTmdbKey(ctx: Pick, key: string): Promise { + const ref = tmdbDetailRef(key); + return !!ref && (await trackedTmdbRefs(ctx)).has(ref); } /** - * TMDB 详情缓存命中后的续期:键属于追踪中的作品时把过期时间推到 now + 详情 TTL(只延长不缩短)。 - * 返回是否续期。失败不影响命中结果。 + * TMDB 详情缓存命中后的续期(旧接口,保留给直接用 PgCache 的地方):键属于关注中的作品时把过期时间推到 + * now + 详情 TTL(只延长不缩短)并标 pinned。返回是否续期。 */ export async function renewTrackedTmdbDetail( ctx: Pick, key: string, detailTtlSec: number, ): Promise { - const ref = tmdbDetailRef(key); - if (!ref || !(await trackedTmdbRefs(ctx)).has(ref)) return false; - return await sharedCache(ctx).renew(CACHE_REGIONS.tmdbRequest, key, detailTtlSec); + if (!(await isWatchedTmdbKey(ctx, key))) return false; + return await sharedCache(ctx).renew(CACHE_REGIONS.tmdbRequest, key, detailTtlSec, { pin: true }); +} + +/** 关注中作品的 TMDB 条目:写入 pinned、命中续期(FetchPolicy.pin) */ +export function watchedPin(ctx: Pick, key: string): FetchPolicy["pin"] { + return tmdbDetailRef(key) ? () => isWatchedTmdbKey(ctx, key) : undefined; +} + +// ── 命中统计(GET /cache/regions 用) ── + +export interface RegionHitStats { + l1Hits: number; + l2Hits: number; + staleHits: number; + misses: number; + failures: number; + refreshes: number; +} + +/** + * 最近 days 天各 region 的命中统计:cache_stat(各进程每分钟写入的增量之和)+ 本进程还没写入的计数。 + * 其它进程最近不到一分钟的计数不在内。 + */ +export async function regionHitStats(ctx: Pick, days = 7): Promise> { + const r = await sql<{ + region: string; + l1: number; + l2: number; + stale: number; + miss: number; + fail: number; + refresh: number; + }>`select region, sum(l1_hits)::int8 as l1, sum(l2_hits)::int8 as l2, sum(stale_hits)::int8 as stale, + sum(misses)::int8 as miss, sum(failures)::int8 as fail, sum(refreshes)::int8 as refresh + from cache_stat where day > current_date - ${days}::int group by region`.execute(ctx.db); + const out = new Map(); + for (const x of r.rows) + out.set(x.region, { + l1Hits: Number(x.l1), + l2Hits: Number(x.l2), + staleHits: Number(x.stale), + misses: Number(x.miss), + failures: Number(x.fail), + refreshes: Number(x.refresh), + }); + for (const [region, c] of metadataCache(ctx).pendingStats()) { + const o = out.get(region) ?? { l1Hits: 0, l2Hits: 0, staleHits: 0, misses: 0, failures: 0, refreshes: 0 }; + for (const k of Object.keys(o) as (keyof RegionHitStats)[]) o[k] += c[k]; + out.set(region, o); + } + return out; } diff --git a/apps/server/test/cache/cache.test.ts b/apps/server/test/cache/cache.test.ts index 5736b683..02a4c9ee 100644 --- a/apps/server/test/cache/cache.test.ts +++ b/apps/server/test/cache/cache.test.ts @@ -67,7 +67,10 @@ describe.skipIf(!hasPg)("C2-A 缓存(真实 PG)", () => { await (await new SourcesRuntime(h.ctx).tmdb())!.details("movie", 872585); const calls = tmdb.calls.length; const rows = ( - await sql<{ key: string; days: number }>`select key, extract(epoch from expires_at - now()) / 86400 as days + await sql<{ + key: string; + days: number; + }>`select key, extract(epoch from coalesce(fresh_until, expires_at) - now()) / 86400 as days from kv_cache where region = 'tmdb.request'`.execute(h.db) ).rows; const detail = rows.find((x) => x.key.startsWith("/movie/872585?"))!; @@ -80,7 +83,7 @@ describe.skipIf(!hasPg)("C2-A 缓存(真实 PG)", () => { const days = async (like: string) => Number( ( - await sql<{ d: number }>`select extract(epoch from expires_at - now()) / 86400 as d + await sql<{ d: number }>`select extract(epoch from coalesce(fresh_until, expires_at) - now()) / 86400 as d from kv_cache where region = 'tmdb.request' and key like ${like}`.execute(h.db) ).rows[0]!.d, ); @@ -96,7 +99,7 @@ describe.skipIf(!hasPg)("C2-A 缓存(真实 PG)", () => { expect(empty).toBeLessThanOrEqual(2); }); - it("追踪中的作品:详情缓存命中时续期到 now + 详情 TTL;未追踪的不动", async () => { + it("关注中的作品:详情缓存命中时续期到 now + 详情 TTL 并标 pinned;未关注的不动", async () => { const ent = ( await sql<{ id: string; @@ -111,19 +114,24 @@ describe.skipIf(!hasPg)("C2-A 缓存(真实 PG)", () => { const client = (await new SourcesRuntime(h.ctx).tmdb())!; await client.details("movie", 872585); await client.details("movie", 693134); - await sql`update kv_cache set expires_at = now() + interval '1 day' + await sql`update kv_cache set fresh_until = now() + interval '1 day', expires_at = now() + interval '2 days', + pinned = false where region = 'tmdb.request' and (key like '/movie/872585?%' or key like '/movie/693134?%')`.execute(h.db); await client.details("movie", 872585); await client.details("movie", 693134); const left = async (like: string) => Number( ( - await sql<{ d: number }>`select extract(epoch from expires_at - now()) / 86400 as d + await sql<{ d: number }>`select extract(epoch from fresh_until - now()) / 86400 as d from kv_cache where region = 'tmdb.request' and key like ${like}`.execute(h.db) ).rows[0]!.d, ); expect(await left("/movie/872585?%")).toBeGreaterThan(19.9); expect(await left("/movie/693134?%")).toBeLessThanOrEqual(1); + const pinned = await sql<{ k: string; p: boolean }>`select key as k, pinned as p from kv_cache + where region = 'tmdb.request' and (key like '/movie/872585?%' or key like '/movie/693134?%')`.execute(h.db); + expect(pinned.rows.filter((r) => r.p).every((r) => r.k.startsWith("/movie/872585?"))).toBe(true); + expect(pinned.rows.some((r) => r.p)).toBe(true); }); it("GET /cache/tmdb 列出条目(键 / 过期 / 大小)与上限;POST /cache/tmdb/clear 清空", async () => { diff --git a/apps/server/test/cache/metadata-cache.test.ts b/apps/server/test/cache/metadata-cache.test.ts new file mode 100644 index 00000000..581df51c --- /dev/null +++ b/apps/server/test/cache/metadata-cache.test.ts @@ -0,0 +1,183 @@ +import { metadataCacheFor, PgCache, sql } from "@umt/db"; +import { createMockTmdb, type MockTmdb } from "@umt/testkit"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { createCacheModule } from "../../src/modules/cache/index.ts"; +import { SourcesRuntime } from "../../src/modules/sources-identity/runtime.ts"; +import { metaRequestCache } from "../../src/modules/sources-identity/service/meta-subjects.ts"; +import { metaCachePolicy, regionHitStats, resetTrackedTmdbRefs } from "../../src/shared-cache.ts"; +import { bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts"; +import { DEV_KEY, json } from "../helpers.ts"; + +// 统一元数据缓存接到 TMDB / 豆瓣 / bgm:策略表、过期先用旧值后台刷新、同请求合并、失败不缓存、 +// 关注中作品 pinned 不被淘汰、按 region 统计命中(GET /cache/regions)。 + +const hasPg = await pgAvailable(); +const KEY = "0123456789abcdef0123456789abcdef"; + +const until = async (cond: () => boolean | Promise, ms = 5000) => { + const end = Date.now() + ms; + while (!(await cond())) { + if (Date.now() > end) throw new Error("等待超时"); + await new Promise((r) => setTimeout(r, 20)); + } +}; + +describe("策略表(来源 × 资源类型)", () => { + it("缺省:搜索 7 天(空 1 天)、详情 30 天(空 1 天)、列表 6 小时(空 1 小时)、图片 90 天", async () => { + const { DEFAULT_STORAGE_SETTINGS } = await import("@umt/contracts"); + const t = DEFAULT_STORAGE_SETTINGS.cacheTtl; + const day = 86_400; + for (const s of ["tmdb", "douban", "bangumi"] as const) { + expect(metaCachePolicy(t, s, "search")).toEqual({ ttl: 7 * day, emptyTtl: day }); + expect(metaCachePolicy(t, s, "detail")).toEqual({ ttl: 30 * day, emptyTtl: day }); + expect(metaCachePolicy(t, s, "list")).toEqual({ ttl: 6 * 3600, emptyTtl: 3600 }); + } + expect(metaCachePolicy(t, "tmdb", "image")).toEqual({ ttl: 90 * day, emptyTtl: day }); + }); +}); + +describe.skipIf(!hasPg)("统一元数据缓存(真实 PG)", () => { + let h: Harness; + let admin: ReturnType; + let tmdb: MockTmdb; + + beforeAll(async () => { + h = await createHarness({ extraModules: [createCacheModule()] }); + admin = bearer(h.app, DEV_KEY); + tmdb = await createMockTmdb({ apiKey: KEY }); + const r = await admin("PATCH", "/settings", { + metadata: { tmdbApiKey: KEY, tmdbBaseUrl: `${tmdb.url}/3` }, + cache: { tmdbMaxSize: 16 }, + }); + expect(r.status).toBe(200); + }); + afterAll(async () => { + await tmdb?.close(); + await h?.close(); + }); + + const client = async () => (await new SourcesRuntime(h.ctx).tmdb())!; + const row = async (like: string) => + ( + await sql<{ key: string; fresh: number; pinned: boolean }>`select key, + extract(epoch from fresh_until - now()) / 86400 as fresh, pinned + from kv_cache where region = 'tmdb.request' and key like ${like}`.execute(h.db) + ).rows[0]; + + it("TMDB 详情过期在宽限期内:先返回旧值,后台只刷一次;刷新后是新值", async () => { + const c = await client(); + await c.details("movie", 872585); + const before = tmdb.callsTo("/3/movie/872585").length; + // 改成「已过新鲜期、还在宽限期」并在值里做个记号(热度 12345) + await sql`update kv_cache set fresh_until = now() - interval '1 second', expires_at = now() + interval '1 day', + value = jsonb_set(value, '{popularity}', '12345') + where region = 'tmdb.request' and key like '/movie/872585?%'`.execute(h.db); + const keys = ( + await sql<{ + key: string; + }>`select key from kv_cache where region = 'tmdb.request' and key like '/movie/872585?%'`.execute(h.db) + ).rows.map((r) => r.key); + expect(keys.length).toBeGreaterThan(0); + // 回源慢一点,让两个调用方都在刷新完成前读到 + tmdb.fault("/3/movie/872585", { delayMs: 300, times: 10 }); + const [a, b] = await Promise.all([c.details("movie", 872585), c.details("movie", 872585)]); + expect(a.popularity).toBe(12345); + expect(b.popularity).toBe(12345); + // 后台刷新:两个调用方同时拿到旧值,每个过期键只回源一次;刷新后新鲜且是新值 + await until(async () => { + const r = await sql<{ n: number }>`select count(*)::int as n from kv_cache where region = 'tmdb.request' + and key like '/movie/872585?%' and fresh_until > now() + interval '29 days'`.execute(h.db); + return r.rows[0]!.n === keys.length; + }); + expect(tmdb.callsTo("/3/movie/872585").length - before).toBe(keys.length); + tmdb.clearFaults(); + expect((await c.details("movie", 872585)).popularity).not.toBe(12345); + }); + + it("同请求合并:并发同一详情只回源一次;失败(5xx)不缓存", async () => { + const c = await client(); + const n0 = tmdb.callsTo("/3/tv/95396/season/1").length; + await Promise.all(Array.from({ length: 5 }, () => c.getJson("/tv/95396/season/1", { language: "zh-CN" }))); + expect(tmdb.callsTo("/3/tv/95396/season/1").length - n0).toBe(1); + // 5xx 重试用尽后抛错(客户端缺省重试 3 次,退避约 3.5 秒),不写缓存 + tmdb.fault("/3/tv/100088/season/1", { status: 503, times: 10 }); + await expect(c.getJson("/tv/100088/season/1", { language: "zh-CN" })).rejects.toThrow(); + expect(await row("/tv/100088/season/1?%")).toBeUndefined(); + tmdb.clearFaults(); + }); + + it("关注中的作品:条目 pinned,条数上限淘汰时不被清掉", async () => { + const ent = ( + await sql<{ + id: string; + }>`insert into media_entity (kind, title) values ('series', '人生切割术') returning id`.execute(h.db) + ).rows[0]!.id; + await sql`insert into external_identity (media_entity_id, provider, entity_type, external_id, status) + values (${ent}, 'tmdb', 'tv', '95396', 'accepted')`.execute(h.db); + await sql`insert into subscription (media_entity_id, media_kind, status) values (${ent}, 'series', 'active')`.execute( + h.db, + ); + resetTrackedTmdbRefs(h.ctx); + await sql`delete from kv_cache where region = 'tmdb.request'`.execute(h.db); + const c = await client(); + await c.getJson("/tv/95396", { language: "zh-CN" }); + expect((await row("/tv/95396?%"))?.pinned).toBe(true); + // 上限 16:再写一批别的请求,关注作品的条目留着 + for (let i = 0; i < 24; i++) await c.searchRaw("movie", `不存在${i}`); + expect(await row("/tv/95396?%")).toBeDefined(); + const n = await sql<{ n: number }>`select count(*)::int as n from kv_cache where region = 'tmdb.request'`.execute( + h.db, + ); + expect(n.rows[0]!.n).toBeLessThanOrEqual(16); + }); + + it("豆瓣 / bgm 请求缓存按策略表:搜索 7 天、列表 6 小时、空结果按查无 / 列表空;失败不写", async () => { + const cache = metaRequestCache(h.ctx); + await cache.cached({ source: "douban", key: "GET s1", kind: "search", empty: () => false }, async () => [1]); + await cache.cached({ source: "douban", key: "GET s0", kind: "search", empty: () => true }, async () => []); + await cache.cached({ source: "bangumi", key: "GET l1", kind: "list", empty: () => false }, async () => ({ + data: [1], + })); + await expect( + cache.cached({ source: "bangumi", key: "GET bad", kind: "list", empty: () => false }, async () => { + throw new Error("HTTP 403"); + }), + ).rejects.toThrow("403"); + const ttl = async (region: string, key: string) => + Number( + ( + await sql<{ s: number }>`select extract(epoch from fresh_until - now()) as s from kv_cache + where region = ${region} and key = ${key}`.execute(h.db) + ).rows[0]?.s ?? -1, + ); + expect(await ttl("douban.request", "GET s1")).toBeGreaterThan(7 * 86400 - 60); + expect(await ttl("douban.request", "GET s0")).toBeLessThanOrEqual(86400); + expect(await ttl("bangumi.request", "GET l1")).toBeLessThanOrEqual(6 * 3600); + expect(await ttl("bangumi.request", "GET l1")).toBeGreaterThan(6 * 3600 - 60); + expect(await ttl("bangumi.request", "GET bad")).toBe(-1); + }); + + it("命中统计:各 region 计 L1 / L2 / 旧值 / 未命中 / 失败 / 后台刷新;写入 cache_stat 后接口合计并给命中率", async () => { + const mc = metadataCacheFor(h.db); + const pending = mc.pendingStats().get("tmdb.request"); + expect(pending?.misses).toBeGreaterThan(0); + expect(pending?.staleHits).toBeGreaterThan(0); + expect(pending?.refreshes).toBeGreaterThan(0); + expect(pending?.failures).toBeGreaterThan(0); + await mc.flushStats(); + expect(mc.pendingStats().size).toBe(0); + // 再来一次 L2 命中(未写入 cache_stat 的也算进接口) + await (await client()).getJson("/tv/95396", { language: "zh-CN" }); + const stats = await regionHitStats(h.ctx); + expect(stats.get("tmdb.request")!.l2Hits).toBeGreaterThan(0); + const list = await json(await admin("GET", "/cache/regions")); + const tmdbRow = list.data.items.find((i: { region: string }) => i.region === "tmdb.request"); + expect(tmdbRow.hits.hitRate).toBeGreaterThan(0); + expect(tmdbRow.hits.hitRate).toBeLessThan(1); + expect(tmdbRow.hits).toMatchObject({ misses: expect.any(Number), l2Hits: expect.any(Number) }); + // 非元数据 region 没有命中统计 + await new PgCache(h.db).set("site.detail", "x", 1, { ttl: 60 }); + const again = await json(await admin("GET", "/cache/regions")); + expect(again.data.items.find((i: { region: string }) => i.region === "site.detail").hits).toBeUndefined(); + }); +}); diff --git a/apps/server/test/search/discover.test.ts b/apps/server/test/search/discover.test.ts index 45859a60..cdbaad60 100644 --- a/apps/server/test/search/discover.test.ts +++ b/apps/server/test/search/discover.test.ts @@ -218,13 +218,13 @@ describe.skipIf(!up)("发现页接口", () => { expect(meta.filter((m) => m.includes("movie_showing"))).toHaveLength(3); expect(meta.at(-1)).toContain("start=20&count=20"); - // 空结果:写入但用空结果 TTL(缺省 1 小时),正常 6 小时 + // 空结果:写入但用空结果 TTL(缺省 1 小时),正常 6 小时(看新鲜期;过期后还有 50% 宽限期先用旧值) doubanReply = () => json({ subject_collection_items: [], total: 0 }); await req("/discover/douban?list=tv_korean"); const rows = await cache.items(META_REQUEST_REGIONS.douban); const ttlOf = (part: string) => { const row = rows.find((x) => x.key.includes(part))!; - return Math.round((new Date(row.expiresAt!).getTime() - Date.now()) / 60_000); + return Math.round((new Date(row.freshUntil).getTime() - Date.now()) / 60_000); }; expect(ttlOf("tv_korean")).toBeLessThanOrEqual(60); expect(ttlOf("tv_korean")).toBeGreaterThan(55); diff --git a/packages/adapters/src/tmdb/client.ts b/packages/adapters/src/tmdb/client.ts index d94c76fc..1a2b2f47 100644 --- a/packages/adapters/src/tmdb/client.ts +++ b/packages/adapters/src/tmdb/client.ts @@ -105,7 +105,15 @@ export interface TmdbConfig { * listTtlMs:流行趋势 / 热门 / 正在上映等列表与 /discover/ 的 TTL(缺省 6 小时且不超过 ttlMs;列表会变,不能按详情缓存 30 天), * 这类请求的空结果 TTL 也不超过它。 */ - cache?: { max?: number; ttlMs?: number; searchTtlMs?: number; emptyTtlMs?: number; listTtlMs?: number }; + cache?: { + max?: number; + ttlMs?: number; + searchTtlMs?: number; + emptyTtlMs?: number; + listTtlMs?: number; + /** 人物 / 图片列表(/person/*、/images);缺省同 ttlMs */ + imageTtlMs?: number; + }; /** * 跨进程共享的请求缓存(服务端传 PG 缓存);给了就不再用进程内 LRU。 * 语义对照 MP @cached(skip_none、业务失败不缓存、results 为空按 emptyTtlMs)。 @@ -124,6 +132,27 @@ type Json = Record; export interface TmdbSharedCache { get(key: string): Promise; set(key: string, value: unknown, ttlMs: number): Promise; + /** + * 给了就由它负责整条读取路径(服务端统一元数据缓存:两级、过期先用旧值、同请求合并、统计、关注作品续期), + * TTL 按 req.kind 取服务端策略表;load 抛错(网络、鉴权、限流、5xx)不缓存。没给时走 get / set。 + */ + fetch?( + req: { key: string; kind: TmdbRequestKind; isEmpty: (v: unknown) => boolean }, + load: () => Promise, + ): Promise; +} + +/** 请求类别(缓存 TTL 的依据):搜索 / find、列表、人物 / 图片、其余为详情(含季、剧集组) */ +export type TmdbRequestKind = "search" | "list" | "image" | "detail"; + +/** 人物 / 图片类请求:/person/*、…/images */ +export const isTmdbImagePath = (path: string) => /^\/person\//.test(path) || /\/images$/.test(path); + +export function tmdbRequestKind(path: string): TmdbRequestKind { + if (/^\/(search|find)\//.test(path)) return "search"; + if (isTmdbListPath(path)) return "list"; + if (isTmdbImagePath(path)) return "image"; + return "detail"; } function requestQuery(params: Record): URLSearchParams { @@ -186,10 +215,16 @@ export class TmdbClient { const key = `${path}?${qs}`; let emptyTtl = this.cfg.cache?.emptyTtlMs ?? 30 * 60_000; const baseTtl = this.cfg.cache?.ttlMs ?? 24 * 3600_000; - let ttl = /^\/(search|find)\//.test(path) ? (this.cfg.cache?.searchTtlMs ?? baseTtl) : baseTtl; - if (isTmdbListPath(path)) { + const kind = tmdbRequestKind(path); + let ttl = kind === "search" ? (this.cfg.cache?.searchTtlMs ?? baseTtl) : baseTtl; + if (kind === "list") { ttl = this.cfg.cache?.listTtlMs ?? Math.min(baseTtl, 6 * 3600_000); emptyTtl = Math.min(emptyTtl, ttl); + } else if (kind === "image") ttl = this.cfg.cache?.imageTtlMs ?? baseTtl; + if (this.cfg.sharedCache?.fetch) { + return (await this.cfg.sharedCache.fetch({ key, kind, isEmpty: isEmptyResult }, () => + this.limiter.run(() => this.fetchWithRetry(path, qs)), + )) as T; } if (this.cfg.sharedCache) { const shared = this.cfg.sharedCache; diff --git a/packages/adapters/src/tmdb/tmdb.test.ts b/packages/adapters/src/tmdb/tmdb.test.ts index 83fa3cf8..a976843a 100644 --- a/packages/adapters/src/tmdb/tmdb.test.ts +++ b/packages/adapters/src/tmdb/tmdb.test.ts @@ -7,6 +7,7 @@ import { matchTmdbMulti, type TmdbConfig, TmdbError, + tmdbRequestKind, yearRange, } from "./index.ts"; @@ -135,6 +136,44 @@ describe("TmdbClient 缓存/限流/错误分类", () => { expect(tmdb.cacheStats().size).toBe(0); }); + it("共享缓存给了 fetch:整条读取交给它,带请求类别(搜索 / 详情 / 列表 / 图片)与空结果判定;失败不进缓存", async () => { + const seen: { key: string; kind: string; empty: boolean }[] = []; + const store = new Map(); + const { mock, tmdb } = await setup({ + sharedCache: { + get: async () => null, + set: async () => {}, + fetch: async (req, load) => { + if (store.has(req.key)) return store.get(req.key); + const v = await load(); + seen.push({ key: req.key, kind: req.kind, empty: req.isEmpty(v) }); + store.set(req.key, v); + return v; + }, + }, + }); + await tmdb.details("movie", 872585); + await tmdb.details("movie", 872585); + await tmdb.search("不存在的片名xyz", { type: "movie" }); + await tmdb.getJson("/movie/872585/images", {}); + await tmdb.getJson("/trending/all/week", {}); + await expect(tmdb.getJson("/movie/999999999", {})).rejects.toThrow(); + const byPath = new Map(seen.map((x) => [x.key.split("?")[0], [x.kind, x.empty]])); + expect(Object.fromEntries(byPath)).toEqual({ + "/movie/872585": ["detail", false], + "/search/movie": ["search", true], + "/movie/872585/images": ["image", false], + "/trending/all/week": ["list", false], + }); + const detailCalls = mock.callsTo("/3/movie/872585").length; + await tmdb.details("movie", 872585); + expect(mock.callsTo("/3/movie/872585")).toHaveLength(detailCalls); + expect(seen.some((x) => x.key.startsWith("/movie/999999999"))).toBe(false); + expect(tmdbRequestKind("/person/1")).toBe("image"); + expect(tmdbRequestKind("/tv/1/season/2")).toBe("detail"); + expect(tmdbRequestKind("/find/tt1")).toBe("search"); + }); + it("并发合并:同一请求并发只发一次", async () => { const { mock, tmdb } = await setup(); await Promise.all(Array.from({ length: 5 }, () => tmdb.season(100088, 1))); diff --git a/packages/contracts/src/cache.ts b/packages/contracts/src/cache.ts index 2b748e57..4ff11ce8 100644 --- a/packages/contracts/src/cache.ts +++ b/packages/contracts/src/cache.ts @@ -82,6 +82,27 @@ export function tmdbImageUrl(path: string | null | undefined, size: TmdbImageSiz export const CacheRegionName = z.string().regex(/^[a-z0-9][a-z0-9._-]{0,63}$/, "缓存区名不合法"); export const CacheRegionParam = z.object({ region: CacheRegionName }); +/** + * 命中统计口径(docs/runbook/cache.md):各进程在内存里计数,每分钟把增量加进 cache_stat(按天 × region), + * 接口返回最近 7 天各进程之和 + 当前进程还没写入的部分(别的进程最近不到一分钟的计数不在内)。 + * hitRate = (l1Hits + l2Hits + staleHits) / (l1Hits + l2Hits + staleHits + misses);没有读取时为 null。 + */ +export const CacheHitStats = z.object({ + /** 进程内存(一级)命中 */ + l1Hits: NonNegInt, + /** PG(二级)命中 */ + l2Hits: NonNegInt, + /** 过期在宽限期内、先返回旧值的次数 */ + staleHits: NonNegInt, + misses: NonNegInt, + /** 回源失败(网络、被拦、5xx) */ + failures: NonNegInt, + /** 后台刷新次数 */ + refreshes: NonNegInt, + hitRate: z.number().min(0).max(1).nullable(), +}); +export type CacheHitStats = z.infer; + export const CacheRegionStat = z.object({ region: z.string(), /** 未过期条目数 */ @@ -92,6 +113,8 @@ export const CacheRegionStat = z.object({ bytes: NonNegInt, /** 最近一次写入 */ lastWriteAt: Timestamp.nullable(), + /** 命中统计(统一元数据缓存的 region 才有):最近 7 天,见 CacheHitStats */ + hits: CacheHitStats.optional(), }); export type CacheRegionStat = z.infer; diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index df4c843f..28a91969 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -578,8 +578,19 @@ export const StorageCacheTtl = z.object({ metaSearchDays: Days, /** 豆瓣 / bgm 榜单、热门、每日放送等列表请求(小时) */ metaListHours: z.number().int().min(1).max(720), - /** 豆瓣 / bgm 空结果(小时,对照 MP empty_ttl) */ + /** 豆瓣 / bgm 列表空结果(小时,对照 MP empty_ttl);搜索空结果改按 metaMissDays */ metaEmptyHours: z.number().int().min(1).max(720), + // ── 统一元数据缓存(2026-10-06,docs/runbook/cache.md 策略表);旧库缺这些字段时按缺省补 ── + /** 豆瓣 / bgm 条目详情(douban.subject / bangumi.subject)与发现页条目 → TMDB 匹配(discover.match) */ + metaDetailDays: Days, + /** 豆瓣 / bgm 查无结果:搜索空结果、条目不存在(meta.subject_miss) */ + metaMissDays: Days, + /** TMDB 榜单 / 热门 / 趋势 / 按条件筛选(小时) */ + tmdbListHours: z.number().int().min(1).max(720), + /** TMDB 列表空结果(小时) */ + tmdbListEmptyHours: z.number().int().min(1).max(720), + /** TMDB 人物 / 图片列表(/person/*、/images) */ + imageDays: Days, }); export type StorageCacheTtl = z.infer; @@ -608,9 +619,14 @@ export const DEFAULT_STORAGE_SETTINGS: StorageSettings = { tmdbMissDays: 1, siteDetailDays: 7, webSearchDays: 1, - metaSearchDays: 1, + metaSearchDays: 7, metaListHours: 6, metaEmptyHours: 1, + metaDetailDays: 30, + metaMissDays: 1, + tmdbListHours: 6, + tmdbListEmptyHours: 1, + imageDays: 90, }, };