UHD_Media_Tool/apps/server/src/shared-cache.ts

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";

// 进程级共享缓存入口(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)。
 * 按请求类别取 storage.cacheTtl:搜索 / find 与识别结果用 searchTtl,详情(/movie/{id}、/tv/{id}、季)用 ttl,
 * 空结果与未识别负缓存用 emptyTtl;列表 / 图片见 metaCachePolicy。
 */
export interface TmdbCachePolicy {
  /** 详情 TTL(秒) */
  ttl: number;
  /** 搜索 / find / 识别结果 TTL(秒) */
  searchTtl: number;
  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,
  };
}

/** TMDB 请求键是否是搜索类(按 searchTtl 缓存) */
export function isTmdbSearchPath(key: string): boolean {
  return /^\/(search|find)\//.test(key);
}

/**
 * 键 → 所属作品:`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;
}

// ── 关注中的作品:条目 pinned(LRU 不淘汰)、命中续期 ──
// 关注口径同 service/watched.ts isEntityWatched;集合 = 关注中作品的 TMDB 身份 + 选中的剧集组,按进程缓存 5 分钟。

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 = 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 {
  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 详情缓存命中后的续期(旧接口,保留给直接用 PgCache 的地方):键属于关注中的作品时把过期时间推到
 * now + 详情 TTL(只延长不缩短)并标 pinned。返回是否续期。
 */
export async function renewTrackedTmdbDetail(
  ctx: Pick<ServerContext, "db">,
  key: string,
  detailTtlSec: number,
): Promise<boolean> {
  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;
}