feat(metadata): one-shot metadata.backfill_languages maintenance job (FIX-5)

Co-Authored-By: Claude Opus 5.5 <[email protected]>

apps/server/src/modules/sources-identity/index.ts

@@ -9,6 +9,7 @@ import { workbenchRoutes } from "./routes/workbench.ts";
 import { type RuntimeOverrides, SourcesRuntime } from "./runtime.ts";
 import { readOnlyReconcile, tickFeeds, tickReviews, tickSnapshots } from "./scheduler.ts";
 import { tickSiteHealth, tickSiteUserData } from "./service/account.ts";
+import { BACKFILL_LANGUAGES_JOB, backfillLanguages } from "./service/backfill-languages.ts";
 import { registerSourcesTasks, sourcesOutboxRoutes } from "./tasks.ts";
 
 /**
@@ -89,6 +90,23 @@ export function createSourcesIdentityModule(overrides: RuntimeOverrides = {}): S
         },
         run: async () => ({ summary: `${await tickSiteHealth(rt)} 个站点` }),
       });
+      // FIX-5 存量元数据回刷:部署后首次启动即运行,跑完自动停用;可在计划页手动再执行
+      registerScheduledJob(ctx, {
+        key: BACKFILL_LANGUAGES_JOB,
+        defaults: {
+          name: "元数据语言回刷",
+          description:
+            "对已确认 TMDB 身份的作品按语言回退链完整刷新一次（标题、简介、每集信息），TMDB 每秒最多 2 次请求；跑完自动停用，可手动再执行",
+          group: "maintain",
+          schedule: { type: "interval", everyMs: 24 * 3_600_000 },
+          firstRun: "now",
+        },
+        run: async ({ runId, jobId }) => {
+          if (await readOnlyReconcile(rt)) return { status: "skipped", summary: "只读对账模式" };
+          const r = await backfillLanguages(rt, { runId, jobId, disableWhenDone: true });
+          return { status: r.status, summary: r.summary };
+        },
+      });
     },
   };
 }

apps/server/src/modules/sources-identity/runtime.ts

@@ -79,6 +79,38 @@ export class SourcesRuntime {
       if (this.tmdbClient === undefined) this.tmdbClient = this.overrides.tmdb();
       return this.tmdbClient;
     }
+    const { sig, cfg } = await this.tmdbConfig();
+    if (this.tmdbClient !== undefined && this.tmdbSig === sig) return this.tmdbClient;
+    this.tmdbSig = sig;
+    this.tmdbClient = cfg ? createTmdbClient(cfg) : null;
+    return this.tmdbClient;
+  }
+
+  /**
+   * 批量任务用的独立 TMDB 客户端(FIX-5 元数据回刷):串行、网络请求间隔 ≥ minIntervalMs(缓存命中不占额度),
+   * 同样走 PG 请求缓存;requests() 返回实际发出的网络请求数。测试注入的客户端原样返回(requests 为 null)。
+   */
+  async tmdbForBatch(opts: {
+    minIntervalMs: number;
+  }): Promise<{ client: TmdbClient; requests: () => number | null } | null> {
+    if (this.overrides.tmdb) {
+      const client = await this.tmdb();
+      return client ? { client, requests: () => null } : null;
+    }
+    const { cfg } = await this.tmdbConfig();
+    if (!cfg) return null;
+    let n = 0;
+    const base = cfg.fetch ?? globalThis.fetch;
+    const counted: typeof fetch = (input, init) => {
+      n++;
+      return base(input, init);
+    };
+    const client = createTmdbClient({ ...cfg, fetch: counted, maxConcurrent: 1, minIntervalMs: opts.minIntervalMs });
+    return { client, requests: () => n };
+  }
+
+  /** TMDB 客户端配置(未配置凭证时 cfg 为 null)与签名(设置变化时重建客户端) */
+  private async tmdbConfig(): Promise<{ sig: string; cfg: Parameters<typeof createTmdbClient>[0] | null }> {
     const meta = await this.ctx.settings.get("metadata").catch(() => null);
     const policy = await tmdbCachePolicy(this.ctx);
     const env = (k: string) => Deno.env.get(k);
@@ -90,7 +122,6 @@ export class SourcesRuntime {
       policy,
       ref ? null : [!!env("TMDB_API_KEY"), !!env("TMDB_ACCESS_TOKEN")],
     ]);
-    if (this.tmdbClient !== undefined && this.tmdbSig === sig) return this.tmdbClient;
     let apiKey: string | undefined;
     let accessToken: string | undefined;
     if (ref) {
@@ -106,10 +137,9 @@ export class SourcesRuntime {
       apiKey = env("TMDB_API_KEY");
       accessToken = env("TMDB_ACCESS_TOKEN");
     }
-    this.tmdbSig = sig;
-    this.tmdbClient =
+    const cfg =
       apiKey || accessToken
-        ? createTmdbClient({
+        ? {
             ...(apiKey ? { apiKey } : {}),
             ...(accessToken ? { accessToken } : {}),
             ...(baseUrl ? { baseUrl } : {}),
@@ -121,9 +151,9 @@ export class SourcesRuntime {
               emptyTtlMs: policy.emptyTtl * 1000,
             },
             sharedCache: tmdbSharedCache(this.ctx, policy),
-          })
+          }
         : null;
-    return this.tmdbClient;
+    return { sig, cfg };
   }
 
   /** 站点默认(settings.sites),读库并刷新缓存;读失败用上次值 */

apps/server/src/modules/sources-identity/service/backfill-languages.ts

@@ -0,0 +1,90 @@
+import { sql } from "@umt/db";
+import type { SourcesRuntime } from "../runtime.ts";
+import { refreshMetadata } from "./metadata.ts";
+
+// FIX-5 存量元数据回刷(维护任务 metadata.backfill_languages):MD-1 语言回退链上线前建档的作品,标题 / 简介 / 每集信息
+// 仍是旧语言。对所有已确认 TMDB 身份的作品按当前语言链跑一次 full 刷新(同 POST /media-entities/{id}/refresh?full=1)。
+// - 限速:独立 TMDB 客户端串行、网络请求间隔 ≥ 500 ms(每秒最多 2 次);请求走 PG 请求缓存(对照 MP @cached),
+//   命中不发请求、不占额度,中断后重跑已取过的作品几乎不再请求 TMDB。
+// - 进度写本次运行记录的摘要(scheduled_job_run.summary),结束时写最终统计。
+// - 跑完(不论个别失败)把任务停用(customized,之后同步定义不会再打开);手动「立即执行」可再跑一次。
+
+export const BACKFILL_LANGUAGES_JOB = "metadata.backfill_languages";
+/** TMDB 网络请求最小间隔:每秒最多 2 次 */
+export const BACKFILL_MIN_INTERVAL_MS = 500;
+const PROGRESS_EVERY_MS = 5_000;
+
+export interface BackfillResult {
+  status: "ok" | "skipped";
+  total: number;
+  refreshed: number;
+  skipped: number;
+  failed: number;
+  requests: number | null;
+  errors: { mediaEntityId: string; title: string; error: string }[];
+  summary: string;
+}
+
+export async function backfillLanguages(
+  rt: SourcesRuntime,
+  opts: { runId?: string; jobId?: string; minIntervalMs?: number; disableWhenDone?: boolean } = {},
+): Promise<BackfillResult> {
+  const { ctx } = rt;
+  const empty = { total: 0, refreshed: 0, skipped: 0, failed: 0, requests: null, errors: [] };
+  const batch = await rt.tmdbForBatch({ minIntervalMs: opts.minIntervalMs ?? BACKFILL_MIN_INTERVAL_MS });
+  if (!batch) return { status: "skipped", ...empty, summary: "TMDB 未配置，未执行（配置后可手动执行）" };
+  const rows = await sql<{ id: string; title: string }>`
+    select e.id, e.title from media_entity e
+     where e.status in ('active', 'provisional')
+       and exists (select 1 from external_identity x where x.media_entity_id = e.id and x.provider = 'tmdb'
+                     and x.status = 'accepted' and x.entity_type in ('movie', 'tv'))
+     order by e.created_at, e.id`.execute(ctx.db);
+  const total = rows.rows.length;
+  const r: BackfillResult = { status: "ok", ...empty, total, summary: "" };
+  let lastProgress = Date.now();
+  const writeProgress = async (done: number) => {
+    if (!opts.runId) return;
+    await ctx.db
+      .updateTable("scheduled_job_run")
+      .set({ summary: `进度 ${done}/${total}：刷新 ${r.refreshed}，跳过 ${r.skipped}，失败 ${r.failed}` })
+      .where("id", "=", opts.runId)
+      .execute()
+      .catch(() => undefined);
+  };
+  for (const [i, e] of rows.rows.entries()) {
+    try {
+      const res = await refreshMetadata(rt, e.id, { full: true, tmdb: batch.client });
+      if (res.status === "refreshed") r.refreshed++;
+      else if (res.status === "skipped") r.skipped++;
+      else {
+        r.failed++;
+        r.errors.push({ mediaEntityId: e.id, title: e.title, error: res.reason ?? "failed" });
+      }
+    } catch (err) {
+      r.failed++;
+      r.errors.push({ mediaEntityId: e.id, title: e.title, error: (err as Error).message.slice(0, 300) });
+      ctx.logger.warn({ entityId: e.id, err: { message: (err as Error).message } }, "元数据回刷失败");
+    }
+    if (Date.now() - lastProgress >= PROGRESS_EVERY_MS) {
+      lastProgress = Date.now();
+      await writeProgress(i + 1);
+    }
+  }
+  r.requests = batch.requests();
+  const failedList = r.errors
+    .slice(0, 5)
+    .map((x) => `「${x.title}」${x.error}`)
+    .join("；");
+  r.summary =
+    `${total} 部作品：刷新 ${r.refreshed}，跳过 ${r.skipped}，失败 ${r.failed}` +
+    (r.requests !== null ? `；TMDB 请求 ${r.requests} 次` : "") +
+    (failedList ? `。失败：${failedList}${r.errors.length > 5 ? " 等" : ""}` : "");
+  if (opts.disableWhenDone && opts.jobId) {
+    await ctx.db
+      .updateTable("scheduled_job")
+      .set({ enabled: false, customized: true })
+      .where("id", "=", opts.jobId)
+      .execute();
+  }
+  return r;
+}

apps/server/src/modules/sources-identity/service/metadata.ts

@@ -2,6 +2,7 @@ import {
   fetchRichDetail,
   fetchRichSeason,
   type MetadataCandidate,
+  type TmdbClient,
   TmdbError,
   type TmdbRichDetail,
   type TmdbRichSeasonDetail,
@@ -129,6 +130,8 @@ export interface RefreshResult {
 export interface RefreshOptions {
   /** true = 拉全部正片季的每集信息(POST /media-entities/{id}/refresh?full=1);默认只拉追踪中 / 库里已有 / 最新的季 */
   full?: boolean;
+  /** 指定 TMDB 客户端(批量回刷用限速客户端);缺省 rt.tmdb() */
+  tmdb?: TmdbClient;
 }
 
 export async function refreshMetadata(
@@ -146,7 +149,7 @@ export async function refreshMetadata(
     .where("provider", "=", "tmdb")
     .where("status", "=", "accepted")
     .executeTakeFirst();
-  const tmdb = await rt.tmdb();
+  const tmdb = opts.tmdb ?? (await rt.tmdb());
   if (!ident || !tmdb || (ident.entity_type !== "movie" && ident.entity_type !== "tv")) {
     return {
       status: "skipped",

apps/server/test/sources-identity/backfill-languages.test.ts

@@ -0,0 +1,125 @@
+import { createMockTmdb, type MockTmdb } from "@umt/testkit";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { registerScheduledJob, scheduleRegistry } from "../../src/modules/schedules/registry.ts";
+import { syncDefinitions } from "../../src/modules/schedules/store.ts";
+import { SourcesRuntime } from "../../src/modules/sources-identity/runtime.ts";
+import {
+  BACKFILL_LANGUAGES_JOB,
+  backfillLanguages,
+} from "../../src/modules/sources-identity/service/backfill-languages.ts";
+import { createEntityFromTmdbCandidate } from "../../src/modules/sources-identity/service/entity-from-candidate.ts";
+import { createHarness, type Harness, pgAvailable } from "./harness.ts";
+
+// FIX-5 存量元数据回刷(metadata.backfill_languages)
+
+const up = await pgAvailable();
+
+describe.skipIf(!up)("元数据语言回刷", () => {
+  let h: Harness;
+  let tmdb: MockTmdb;
+  const ids: string[] = [];
+  const actor = { type: "user" as const, id: "tester" };
+
+  beforeAll(async () => {
+    tmdb = await createMockTmdb();
+    h = await createHarness("backfill", { tmdbUrl: tmdb.url });
+    for (const [externalId, kind] of [
+      ["95396", "tv"],
+      ["872585", "movie"],
+    ] as const) {
+      const r = await createEntityFromTmdbCandidate(h.db, () => h.rt.tmdb(), { externalId, kind }, actor);
+      ids.push(r.mediaEntityId);
+    }
+    // 没有 TMDB 身份的作品不参与
+    await h.db
+      .insertInto("media_entity")
+      .values({ kind: "series", title: "本地短剧", status: "provisional" } as never)
+      .execute();
+  });
+  afterAll(async () => {
+    await h?.close();
+    await tmdb?.close();
+  });
+
+  it("登记为维护组任务,首次登记即到期;跑完按新语言链刷新并自动停用", async () => {
+    registerScheduledJob(h.ctx, {
+      key: BACKFILL_LANGUAGES_JOB,
+      defaults: {
+        name: "元数据语言回刷",
+        group: "maintain",
+        schedule: { type: "interval", everyMs: 86_400_000 },
+        firstRun: "now",
+      },
+      run: async () => undefined,
+    });
+    const now = new Date();
+    await syncDefinitions(h.db, scheduleRegistry(h.ctx).list(), now);
+    const before = await h.db
+      .selectFrom("scheduled_job")
+      .selectAll()
+      .where("id", "=", BACKFILL_LANGUAGES_JOB)
+      .executeTakeFirstOrThrow();
+    expect(before).toMatchObject({ enabled: true, board_group: "maintain" });
+    expect(before.next_run_at!.getTime()).toBeLessThanOrEqual(now.getTime());
+
+    const r = await backfillLanguages(h.rt, { jobId: BACKFILL_LANGUAGES_JOB, disableWhenDone: true, minIntervalMs: 0 });
+    expect(r).toMatchObject({ status: "ok", total: 2, refreshed: 2, failed: 0 });
+    expect(r.summary).toContain("2 部作品：刷新 2");
+    const tv = await h.db.selectFrom("media_entity").selectAll().where("id", "=", ids[0]!).executeTakeFirstOrThrow();
+    expect(tv.title).toBe("人生切割术");
+    const eps = await h.db
+      .selectFrom("content_unit")
+      .select(["episode_number", "title"])
+      .where("media_entity_id", "=", ids[0]!)
+      .where("kind", "=", "episode")
+      .where("season_number", "=", 1)
+      .orderBy("episode_number")
+      .execute();
+    expect(eps.length).toBeGreaterThan(0);
+
+    const after = await h.db
+      .selectFrom("scheduled_job")
+      .selectAll()
+      .where("id", "=", BACKFILL_LANGUAGES_JOB)
+      .executeTakeFirstOrThrow();
+    expect(after).toMatchObject({ enabled: false, customized: true });
+    // 重启后同步定义不会重新打开
+    await syncDefinitions(h.db, scheduleRegistry(h.ctx).list(), new Date());
+    const synced = await h.db
+      .selectFrom("scheduled_job")
+      .select(["enabled"])
+      .where("id", "=", BACKFILL_LANGUAGES_JOB)
+      .executeTakeFirstOrThrow();
+    expect(synced.enabled).toBe(false);
+  });
+
+  it("按设置建的客户端:网络请求限速,走 PG 请求缓存,第二次不再请求 TMDB", async () => {
+    await h.db
+      .insertInto("settings")
+      .values({ key: "metadata", value: JSON.stringify({ tmdbBaseUrl: `${tmdb.url}/3` }) } as never)
+      .onConflict((oc) => oc.column("key").doUpdateSet({ value: JSON.stringify({ tmdbBaseUrl: `${tmdb.url}/3` }) }))
+      .execute();
+    const prev = Deno.env.get("TMDB_API_KEY");
+    Deno.env.set("TMDB_API_KEY", "test-tmdb-key");
+    try {
+      const rt = new SourcesRuntime(h.ctx, {});
+      const t0 = performance.now();
+      const first = await backfillLanguages(rt, { minIntervalMs: 100 });
+      const elapsed = performance.now() - t0;
+      expect(first).toMatchObject({ status: "ok", total: 2, refreshed: 2 });
+      expect(first.requests).toBeGreaterThan(2);
+      expect(elapsed).toBeGreaterThanOrEqual((first.requests! - 1) * 90);
+      const second = await backfillLanguages(rt, { minIntervalMs: 100 });
+      expect(second).toMatchObject({ refreshed: 2, requests: 0 });
+    } finally {
+      if (prev === undefined) Deno.env.delete("TMDB_API_KEY");
+      else Deno.env.set("TMDB_API_KEY", prev);
+    }
+  });
+
+  it("TMDB 未配置:跳过,不停用", async () => {
+    const rt = new SourcesRuntime(h.ctx, { tmdb: () => null });
+    const r = await backfillLanguages(rt, { jobId: "x", disableWhenDone: true });
+    expect(r.status).toBe("skipped");
+  });
+});