feat(cache): TMDB / 豆瓣 / Bangumi 接入统一元数据缓存与策略表
- 策略表「来源 × 资源类型」(metaCachePolicy):搜索 / 识别 7 天(空 1 天)、详情 / 季 / 剧集组 30 天(空 1 天)、 榜单 / 热门 / 每日放送 6 小时(空 1 小时)、人物 / 图片 90 天;取值来自 storage.cacheTtl, 追加 tmdbListHours / tmdbListEmptyHours / imageDays / metaDetailDays / metaMissDays,metaSearchDays 缺省改 7 天 - TMDB 客户端共享缓存端口加 fetch(带请求类别与空结果判定),服务端整条读取交给 MetadataCache; 识别缓存、douban.request / bangumi.request、douban.subject / bangumi.subject、meta.subject_miss、discover.match 都走这一层 - 豆瓣 / bgm 条目只在「不存在」时记查无,网络失败 / 被拦不缓存(本进程冷却 30 分钟) - 关注中作品(isEntityWatched 口径,含选中的剧集组)的 TMDB 条目写入 pinned、命中续期 - 各进程启动 LISTEN 失效通知并每分钟把命中计数写 cache_stat;GET /cache/regions 给元数据 region 带近 7 天命中统计 Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei
This commit is contained in:
rootandClaude Opus 5.5 committed
1 parent
ad472a1a19
commit
1dca95f883
16 files changed
+687
-137
No files matched your search
+191
-29
@@ -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<ServerContext, "db">): PgCache {
|
||||
return pgCacheFor(ctx.db);
|
||||
}
|
||||
|
||||
/** 统一元数据缓存(同一个 db 共用一个实例) */
|
||||
export function metadataCache(ctx: Pick<ServerContext, "db"> & Partial<Pick<ServerContext, "logger">>): 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<ServerContext, "db" | "logger">,
|
||||
connectionString: string,
|
||||
opts: { flushMs?: number } = {},
|
||||
): Promise<() => Promise<void>> {
|
||||
const cache = metadataCache(ctx);
|
||||
const log = ctx.logger.child({ component: "metadata-cache" });
|
||||
let listener: Awaited<ReturnType<typeof listenCacheInvalidation>> | 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<ServerContext, "settings">): Promise<StorageSettings> {
|
||||
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<ServerContext, "settings">): Promise<TmdbCachePolicy> {
|
||||
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:<id>` / `tv:<id>`;不是详情返回 null */
|
||||
/**
|
||||
* 键 → 所属作品:`movie:<id>` / `tv:<id>`(详情、季、剧集、图片、剧集组列表都算);
|
||||
* 剧集组详情 `/tv/episode_group/<gid>` → `group:<gid>`;都不是返回 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<object, { at: number; set: Promise<Set<string>> }>();
|
||||
const WATCHED_TTL_MS = 5 * 60_000;
|
||||
const watchedCache = new WeakMap<object, { at: number; set: Promise<Set<string>> }>();
|
||||
|
||||
export async function trackedTmdbRefs(ctx: Pick<ServerContext, "db">, now = Date.now()): Promise<Set<string>> {
|
||||
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<ServerContext, "db">): void {
|
||||
trackedCache.delete(ctx.db);
|
||||
watchedCache.delete(ctx.db);
|
||||
}
|
||||
|
||||
/** 这个 TMDB 请求键属于关注中的作品(详情 / 季 / 剧集组 / 图片) */
|
||||
export async function isWatchedTmdbKey(ctx: Pick<ServerContext, "db">, key: string): Promise<boolean> {
|
||||
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<ServerContext, "db">,
|
||||
key: string,
|
||||
detailTtlSec: number,
|
||||
): Promise<boolean> {
|
||||
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<ServerContext, "db">, key: string): FetchPolicy<unknown>["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<ServerContext, "db">, days = 7): Promise<Map<string, RegionHitStats>> {
|
||||
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<string, RegionHitStats>();
|
||||
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;
|
||||
}
|
||||
Reference in new issue
Block a user