perf(server): parallel media-server inventory, queued identify remote lookups, dev DB cleanup

- GET /inventory/media-servers: per-instance Promise.allSettled, per-instance
  concurrency cap (2) and timeout (8s); failed/timed-out instances are unknown,
  response carries partial=true (field not yet in contracts)
- S-7: POST /files/identify returns local matches plus remoteTaskId/remotePending
  and enqueues identifyRemote; the worker runs the TMDB lookups and publishes SSE
  identify.completed {taskId,status,items,truncated,excludedBySize,error};
  ?wait=true keeps the synchronous path, capped at 15s overall
- catalog.lookup / release.search honour the caller's abort signal
- scripts/dev-db-clean.ts: dry-run by default, --yes to drop leftover test
  schemas and w32-/w33-/fe4-/qa- prefixed entities; refuses non localhost:55433
- runbook: dev DB cleanup

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-09-28 19:47:41 +08:00
1 parent 5b8927d1dc
commit fd2f7bc33d
10 files changed
+683 -67

No files matched your search

@@ -463,7 +463,7 @@ describe.skipIf(!hasPg)("acquisition-organize 纵切(真实 PG + mock qB + 临
const inv = await h.api("GET", `/inventory?entityIds=${movieId}`);
expect(inv.body.data.items[0].versions[0]).toMatchObject({ resolution: "1080p", available: true });
const ms = await h.api("GET", `/inventory/media-servers?mediaEntityId=${movieId}`);
expect(ms.body.data).toEqual({ mediaEntityId: movieId, instances: [] });
expect(ms.body.data).toEqual({ mediaEntityId: movieId, instances: [], partial: false });
});
it("自动整理被拦(attention)后手动整理成功:Slot 经 verifying 落 satisfied,整理事项自动关闭(QA U1)", async () => {
@@ -33,6 +33,85 @@ import { mount, paginate } from "./helpers.ts";
const TAG = "library";
const ACTIVE_DL = ["queued", "submitting", "downloading", "stalled", "paused", "completed", "importing"] as const;
/** Emby 库存查询:每实例内部并发上限与整实例超时(毫秒) */
export const MEDIA_SERVER_LIMITS = { perInstanceConcurrency: 2, timeoutMs: 8_000 };
const MS_PROVIDER_KEY: Record<string, string> = { tmdb: "Tmdb", imdb: "Imdb", tvdb: "Tvdb", douban: "Douban" };
type MsInstance = MediaServerInventory["instances"][number];
type MsInstanceRow = Parameters<AoDeps["ports"]["mediaServer"]>[0];
const errText = (e: unknown) => (e instanceof Error ? e.message : String(e));
/**
* 查单个实例:外部 ID 逐个 findByProviderId(实例内并发 ≤ perInstanceConcurrency),整实例超过 timeoutMs 记 unknown。
* 超时后底层请求由 Emby 客户端自身超时收尾,结果丢弃。
*/
export async function queryMediaServer(
d: AoDeps,
inst: MsInstanceRow,
usable: { provider: string; external_id: string }[],
type: "Movie" | "Series",
limits: { perInstanceConcurrency: number; timeoutMs: number },
): Promise<MsInstance> {
const observedAt = new Date().toISOString();
const base = { instanceId: inst.id, name: inst.name, observedAt };
if (usable.length === 0) return { ...base, status: "unknown", reason: "no_external_id", items: [] };
let timer: ReturnType<typeof setTimeout> | undefined;
const timeout = new Promise<never>((_, reject) => {
timer = setTimeout(
() => reject(new Error(`timeout: 媒体服务器 ${limits.timeoutMs / 1000}s 内未响应`)),
limits.timeoutMs,
);
});
const work = (async () => {
const client = await d.ports.mediaServer(inst);
const results: Awaited<ReturnType<typeof client.findByProviderId>>[] = new Array(usable.length);
let next = 0;
const lane = async () => {
while (next < usable.length) {
const k = next++;
const i = usable[k]!;
results[k] = await client.findByProviderId(MS_PROVIDER_KEY[i.provider]!, i.external_id, type);
}
};
await Promise.all(Array.from({ length: Math.min(limits.perInstanceConcurrency, usable.length) }, lane));
return results;
})();
work.catch(() => {});
try {
const results = await Promise.race([work, timeout]);
let status: MsInstance["status"] = "not_found";
let reason: string | null = null;
const items: MsInstance["items"] = [];
for (const r of results) {
if (r.status === "present") {
status = "present";
reason = null;
for (const it of r.items) {
if (items.some((x) => x.itemId === it.itemId)) continue;
items.push({
itemId: it.itemId,
type: it.type,
name: it.name,
year: it.year ?? null,
path: it.path ?? null,
versions: it.versions.map((v) => JSON.parse(JSON.stringify(v)) as Record<string, unknown>),
});
}
} else if (r.status === "unknown" && status !== "present") {
status = "unknown";
reason = r.reason;
}
}
return { ...base, status, reason, items };
} catch (err) {
return { ...base, status: "unknown", reason: errText(err), items: [] };
} finally {
clearTimeout(timer);
}
}
const emptyCounts = (): SlotStatusCounts => ({
empty: 0,
reserved: 0,
@@ -414,54 +493,26 @@ export function registerLibraryRoutes(app: ModuleApp, d: AoDeps) {
.orderBy("name")
.execute();
const type = e!.kind === "movie" ? "Movie" : "Series";
const KEY: Record<string, string> = { tmdb: "Tmdb", imdb: "Imdb", tvdb: "Tvdb", douban: "Douban" };
const out: MediaServerInventory = { mediaEntityId: e!.id, instances: [] };
for (const inst of insts) {
const observedAt = new Date().toISOString();
const usable = idents.filter((i) => KEY[i.provider]);
if (usable.length === 0) {
out.instances.push({
instanceId: inst.id,
name: inst.name,
status: "unknown",
reason: "no_external_id",
items: [],
observedAt,
});
continue;
}
let status: "present" | "not_found" | "unknown" = "not_found";
let reason: string | null = null;
const items: MediaServerInventory["instances"][number]["items"] = [];
try {
const client = await d.ports.mediaServer(inst);
for (const i of usable) {
const r = await client.findByProviderId(KEY[i.provider]!, i.external_id, type);
if (r.status === "present") {
status = "present";
for (const it of r.items) {
if (items.some((x) => x.itemId === it.itemId)) continue;
items.push({
itemId: it.itemId,
type: it.type,
name: it.name,
year: it.year ?? null,
path: it.path ?? null,
versions: it.versions.map((v) => JSON.parse(JSON.stringify(v)) as Record<string, unknown>),
});
}
} else if (r.status === "unknown" && status !== "present") {
status = "unknown";
reason = r.reason;
}
}
} catch (err) {
status = "unknown";
reason = (err as Error).message;
}
out.instances.push({ instanceId: inst.id, name: inst.name, status, reason, items, observedAt });
}
return out;
const usable = idents.filter((i) => MS_PROVIDER_KEY[i.provider]);
// 按实例并行(allSettled):单个 Emby 慢/挂只让自己 unknown,不拖住其它实例;每实例内部限并发 + 总超时
const settled = await Promise.allSettled(
insts.map((inst) => queryMediaServer(d, inst, usable, type, MEDIA_SERVER_LIMITS)),
);
const instances = settled.map((r, i) =>
r.status === "fulfilled"
? r.value
: {
instanceId: insts[i]!.id,
name: insts[i]!.name,
status: "unknown" as const,
reason: errText(r.reason),
items: [],
observedAt: new Date().toISOString(),
},
);
// partial:至少一个实例因查询失败/超时为 unknown(no_external_id 不算失败)。契约未列此字段,前端可按需读取
const partial = instances.some((x) => x.status === "unknown" && x.reason !== "no_external_id");
return { mediaEntityId: e!.id, instances, partial } as MediaServerInventory;
},
{ tag: TAG },
);
@@ -18,6 +18,43 @@ import { kindOfName } from "./list.ts";
// 识别(只读,不落库):parseReleasePath(文件 + 父/祖父目录上下文)→ 本地召回(别名 pg_trgm + 外部身份)
// → 远端候选(catalog.lookup 能力,走 TMDB 缓存;失败只降级为提示)→ rankCandidates,每项前 5。
//
// S-7(C1-D):HTTP 请求内默认只做本地召回;需要远端查询时入队持久任务 identifyRemote,响应带 remoteTaskId(= 引擎 runId)。
// 任务在 worker 里跑完整识别(本地 + 远端),完成后发 SSE `identify.completed`(见 IdentifyCompletedPayload)。
// `?wait=true` 保留同步模式:请求内查远端,整体上限 IDENTIFY_WAIT_MAX_MS。
/** 远端结果推送事件(尚未进 contracts EventType;前端按 taskId 匹配) */
export const IDENTIFY_COMPLETED_EVENT = "identify.completed";
/** 持久任务名 */
export const IDENTIFY_REMOTE_TASK = "identifyRemote";
/** 同步模式(wait=true)整个请求的远端查询上限 */
export const IDENTIFY_WAIT_MAX_MS = 15_000;
export interface IdentifyCompletedPayload {
taskId: string;
status: "succeeded" | "failed";
/** 完整识别结果(本地 + 远端),按 root + path 替换 HTTP 响应里的同名项 */
items: FileIdentifyItem[];
truncated: boolean;
excludedBySize: number;
error: string | null;
}
/** HTTP 响应 = 契约 FilesIdentifyResult + 两个待入契约的字段 */
export type FilesIdentifyResponse = FilesIdentifyResult & {
/** 远端查询任务 id(引擎 runId);null = 无需远端或同步模式已查完 */
remoteTaskId: string | null;
/** 待远端查询的去重标题数(0 = 无) */
remotePending: number;
};
interface CoreOpts {
/** local:只本地召回,统计待查远端数;remote:本地 + 远端 */
mode: "local" | "remote";
actorId: string;
/** remote 模式整体截止(同步 wait 或任务取消) */
signal?: AbortSignal;
}
/** 单次 /files/identify 请求内去重后的远端(TMDB)查询上限 */
export const REMOTE_QUERY_MAX = 20;
@@ -79,12 +116,80 @@ function segments(rel: string) {
return { file, ...(parent ? { parent } : {}), ...(grandParent ? { grandParent } : {}) };
}
/**
* /files/identify 处理:wait=false(缺省)本地召回 + 入队远端任务;wait=true 请求内查远端(≤15s)。
*/
export async function identify(
d: FmDeps,
principal: Principal | undefined,
body: FilesIdentifyRequest,
): Promise<FilesIdentifyResult> {
opts: { wait?: boolean } = {},
): Promise<FilesIdentifyResponse> {
const actorId = actorOfP(principal);
if (opts.wait) {
const r = await identifyCore(d, body, {
mode: "remote",
actorId,
signal: AbortSignal.timeout(IDENTIFY_WAIT_MAX_MS),
});
return { ...r.result, remoteTaskId: null, remotePending: 0 };
}
const r = await identifyCore(d, body, { mode: "local", actorId });
if (r.remotePending === 0) return { ...r.result, remoteTaskId: null, remotePending: 0 };
let taskId: string;
try {
({ runId: taskId } = await d.ctx.engine.enqueue(
IDENTIFY_REMOTE_TASK,
{ body, actorId },
{ idempotencyKey: `files.identify:${randomUUID()}`, traceId: randomUUID() },
));
} catch (e) {
for (const it of r.result.items)
if (r.pendingPaths.has(`${it.root}\u0000${it.path}`))
it.warnings.push(`远端查询入队失败,只查了本地:${errMessage(e).slice(0, 200)}`);
return { ...r.result, remoteTaskId: null, remotePending: 0 };
}
for (const it of r.result.items)
if (r.pendingPaths.has(`${it.root}\u0000${it.path}`)) it.warnings.push("远端候选在后台查询,查完自动补上");
return { ...r.result, remoteTaskId: taskId, remotePending: r.remotePending };
}
/** 持久任务 identifyRemote 的处理体:完整识别后发 identify.completed(失败也发,status=failed) */
export async function runIdentifyRemote(
d: FmDeps,
taskId: string,
input: { body: FilesIdentifyRequest; actorId: string },
signal?: AbortSignal,
): Promise<IdentifyCompletedPayload> {
let payload: IdentifyCompletedPayload;
try {
const r = await identifyCore(d, input.body, {
mode: "remote",
actorId: input.actorId,
...(signal ? { signal } : {}),
});
payload = {
taskId,
status: "succeeded",
items: r.result.items,
truncated: r.result.truncated,
excludedBySize: r.result.excludedBySize ?? 0,
error: null,
};
} catch (e) {
payload = { taskId, status: "failed", items: [], truncated: false, excludedBySize: 0, error: errMessage(e) };
}
await d.ctx.events.publish(IDENTIFY_COMPLETED_EVENT, payload);
return payload;
}
async function identifyCore(
d: FmDeps,
body: FilesIdentifyRequest,
opts: CoreOpts,
): Promise<{ result: FilesIdentifyResult; remotePending: number; pendingPaths: Set<string> }> {
const db = d.ctx.db;
const pendingPaths = new Set<string>();
const { videos, truncated } = await collectVideos(d, body);
// G2:最小体积(请求/向导 > settings.organize.minSizeBytes,缺省 50 MiB;设置读不到时不过滤)
const orgSettings = (await d.ctx.settings.get("organize").catch(() => null)) as { minSizeBytes?: unknown } | null;
@@ -102,8 +207,10 @@ export async function identify(
: { rows: [] };
const recordsBy = new Map(recordRows.rows.map((r) => [r.source_path, r]));
const localCache = new Map<string, IdentityCandidate[]>();
const remoteCache = new Map<string, { list: IdentityCandidate[]; warning: string | null }>();
const remoteCache = new Map<string, { list: IdentityCandidate[]; warning: string | null; pending?: boolean }>();
const entityKinds = new Map<string, string>();
// G8:候选海报(TMDB poster_path;本地作品取投影后的 metadata.posterPath)
const posters = new Map<string, string>();
const localFor = async (p: ParsedRelease): Promise<IdentityCandidate[]> => {
const key = `${queryTitles(p).join("|")}|${body.hints?.tmdbId ?? ""}`;
@@ -148,6 +255,8 @@ export async function identify(
.execute();
for (const e of ents) {
entityKinds.set(e.id, e.kind);
const pp = (e.metadata as { posterPath?: unknown } | null)?.posterPath;
if (typeof pp === "string" && pp) posters.set(`local:${e.id}`, pp);
const c = localIdentityCandidate(
{ id: e.id, kind: e.kind, title: e.title, originalTitle: e.original_title, year: e.year },
aliases.filter((a) => a.media_entity_id === e.id).map((a) => a.alias),
@@ -163,13 +272,15 @@ export async function identify(
};
let remoteQueries = 0;
const remoteFor = async (p: ParsedRelease) => {
const remoteFor = async (
p: ParsedRelease,
): Promise<{ list: IdentityCandidate[]; warning: string | null; pending?: boolean }> => {
const q = p.titleCn ?? p.title ?? p.titleEn;
if (!q) return { list: [], warning: null };
const key = `${q}|${body.hints?.year ?? p.year ?? ""}`;
const hit = remoteCache.get(key);
if (hit) return hit;
let res: { list: IdentityCandidate[]; warning: string | null };
let res: { list: IdentityCandidate[]; warning: string | null; pending?: boolean };
// 单次请求内去重后的远端查询设上限,超出的项只做本地召回(验收 07 S-6)
if (remoteQueries >= REMOTE_QUERY_MAX) {
res = {
@@ -180,6 +291,17 @@ export async function identify(
return res;
}
remoteQueries += 1;
if (opts.mode === "local") {
// 只登记待查,由 identifyRemote 任务查;同键的后续项命中缓存,也标为待查
res = { list: [], warning: null, pending: true };
remoteCache.set(key, res);
return res;
}
if (opts.signal?.aborted) {
res = { list: [], warning: `远端查询超时(${IDENTIFY_WAIT_MAX_MS / 1000}s),此项只查了本地` };
remoteCache.set(key, res);
return res;
}
try {
const input = {
query: q,
@@ -188,17 +310,25 @@ export async function identify(
...(body.hints?.year ? { year: body.hints.year } : {}),
};
// 经进程共享的 TMDB 信号量(并发 4)+ 15s 超时,与能力调用端点同一个限流器
const out = (await sharedInvokeLimiter().run("catalog.lookup", input, undefined, (signal) =>
const out = (await sharedInvokeLimiter().run("catalog.lookup", input, opts.signal, (signal) =>
d.ctx.capabilities.invoke("catalog.lookup", input, {
actorId: actorOfP(principal),
actorId: opts.actorId,
taskId: `files.identify:${randomUUID()}`,
scopeIds: [],
signal: signal ?? AbortSignal.timeout(EXTERNAL_INVOKE_TIMEOUT_MS),
source: "web",
}),
)) as {
remote?: { provider: string; entityType: string; externalId: string; title: string; year: number | null }[];
remote?: {
provider: string;
entityType: string;
externalId: string;
title: string;
year: number | null;
posterPath?: string | null;
}[];
};
for (const m of out.remote ?? []) if (m.posterPath) posters.set(`${m.provider}:${m.externalId}`, m.posterPath);
const list: IdentityCandidate[] = (out.remote ?? [])
.filter((m) => m.entityType === "movie" || m.entityType === "tv")
.map((m) => ({
@@ -254,6 +384,7 @@ export async function identify(
const local = parsed.title || parsed.explicitIds.length || body.hints?.tmdbId ? await localFor(parsed) : [];
const remote = parsed.title ? await remoteFor(parsed) : { list: [], warning: null };
if (remote.warning) warnings.push(remote.warning);
if (remote.pending) pendingPaths.add(`${v.root}\u0000${v.rel}`);
if (!parsed.title) warnings.push("文件名无法解析出标题");
// 已关联到本地作品的远端候选不重复出现
const linked = new Set(local.flatMap((c) => (c.linkedExternalIds ?? []).map((l) => `${l.provider}:${l.id}`)));
@@ -276,6 +407,12 @@ export async function identify(
level: LEVEL[r.level] ?? "low",
score: 5 - i,
reasons: r.evidence,
posterUrl: posterUrlOf(
c.localEntityId
? (posters.get(`local:${c.localEntityId}`) ??
c.linkedExternalIds?.map((l) => posters.get(`${l.provider}:${l.id}`)).find(Boolean))
: posters.get(`${c.provider}:${c.externalId}`),
),
};
});
const best = candidates[0];
@@ -321,7 +458,14 @@ export async function identify(
}
if (videos.length === 0 && body.items.length > 0 && !truncated)
throw new FilesError("bad_request", "所选项中没有视频文件");
return { items, truncated, excludedBySize };
return { result: { items, truncated, excludedBySize }, remotePending: remoteQueries, pendingPaths };
}
/** TMDB 图片 CDN 缩略图(w154);已是完整 URL 的原样返回 */
export function posterUrlOf(path: string | null | undefined): string | null {
if (!path) return null;
if (/^https?:\/\//.test(path)) return path;
return `https://image.tmdb.org/t/p/w154${path.startsWith("/") ? path : `/${path}`}`;
}
function parsedOut(parsed: ParsedRelease): FileIdentifyItem["parsed"] {
@@ -1,5 +1,6 @@
import { randomUUID } from "node:crypto";
import type { TmdbClient } from "@umt/adapters";
import { FilesIdentifyRequest } from "@umt/contracts";
import { defineTask } from "@umt/tasks";
import { z } from "zod";
import type { AoOverrides } from "../acquisition-organize/index.ts";
@@ -8,6 +9,7 @@ import { registerScheduledJob } from "../schedules/registry.ts";
import { SourcesRuntime } from "../sources-identity/runtime.ts";
import type { ServerContext, ServerModule } from "../types.ts";
import { defaultOptions, type FmDeps, type FmOptions } from "./common.ts";
import { IDENTIFY_REMOTE_TASK, runIdentifyRemote } from "./identify.ts";
import { registerFileRoutes } from "./routes.ts";
import { purgeTrash } from "./trash.ts";
@@ -64,6 +66,18 @@ export function createFileManagerModule(overrides: FmOverrides = {}): ServerModu
return { purged: r.purged.length, failed: r.failed.length };
}),
}),
// S-7:/files/identify 的远端(TMDB)查询;只读,失败发 status=failed 的 identify.completed,不重试
defineTask({
name: IDENTIFY_REMOTE_TASK,
version: 1,
inputSchema: z.object({ body: FilesIdentifyRequest, actorId: z.string() }),
retryPolicy: { maximumAttempts: 1 },
handler: async (t) =>
t.step.run("identify", async () => {
const r = await runIdentifyRemote(lazy(ctx), t.runId, t.input, t.abortSignal);
return { status: r.status, items: r.items.length };
}),
}),
);
},
start(ctx) {
@@ -105,7 +105,10 @@ export function registerFileRoutes(app: ModuleApp, d: FmDeps) {
mount(app, F.trashPurge, async (c, { body }) =>
idem(d, c, "trash_purge", body, () => purgeTrash(d, c.get("principal"), body)),
);
mount(app, F.identify, async (c, { body }) => identify(d, c.get("principal"), body));
// S-7:缺省只回本地命中 + remoteTaskId(远端结果经 SSE identify.completed 推送);?wait=true 请求内查远端,≤15s
mount(app, F.identify, async (c, { body }) =>
identify(d, c.get("principal"), body, { wait: /^(1|true)$/i.test(c.req.query("wait") ?? "") }),
);
mount(app, F.scrape, async (c, { body }) => idem(d, c, "scrape", body, () => scrape(d, c.get("principal"), body)));
mount(app, F.probe, async (_c, { body }) => probeFiles(d, body));
mount(app, F.records, async (_c, { query }) => findRecords(d, query));
@@ -16,7 +16,7 @@ import { liveSearch } from "./service/workbench.ts";
export function registerSourcesCapabilities(registry: CapabilityRegistry, rt: () => SourcesRuntime): void {
registerSiteCapabilities(registry, rt);
registry.register(capabilitySpec("catalog.lookup"), async (input) => {
registry.register(capabilitySpec("catalog.lookup"), async (input, ctx) => {
const { db } = rt().ctx;
const local: { entity: unknown; score: number; matchedAlias: string | null }[] = [];
if (input.externalId) {
@@ -57,10 +57,9 @@ export function registerSourcesCapabilities(registry: CapabilityRegistry, rt: ()
}[] = [];
const tmdb = await rt().tmdb();
if (input.includeRemote && input.query && tmdb) {
for (const m of (await tmdb.searchAll(input.query, input.year ? { year: input.year } : {})).slice(
0,
input.limit,
)) {
// 认调用方的 signal(HTTP 15s 超时 / 任务取消):中止后立即返回,不再等 TMDB(底层请求由客户端自身超时收尾)
const found = await abortable(tmdb.searchAll(input.query, input.year ? { year: input.year } : {}), ctx.signal);
for (const m of found.slice(0, input.limit)) {
remote.push({
provider: "tmdb",
entityType: m.entityType,
@@ -73,7 +72,7 @@ export function registerSourcesCapabilities(registry: CapabilityRegistry, rt: ()
return { local: local as never, remote };
});
registry.register(capabilitySpec("release.search"), async (input) => {
registry.register(capabilitySpec("release.search"), async (input, ctx) => {
const r = rt();
const { db } = r.ctx;
let q = db.selectFrom("source_release").selectAll();
@@ -93,7 +92,10 @@ export function registerSourcesCapabilities(registry: CapabilityRegistry, rt: ()
const coverage: { siteId: string; fromSnapshot: boolean; live: boolean; error: string | null }[] = [];
const needLive = input.allowLiveSearch && !!text && rows.length === 0;
for (const s of sites) {
if (needLive) {
if (needLive && ctx.signal.aborted) {
// 已超时/取消:剩余站点不再实时补查
coverage.push({ siteId: s.id, fromSnapshot: true, live: false, error: "aborted" });
} else if (needLive) {
const res = await liveSearch(r, s, text!);
coverage.push({ siteId: s.id, fromSnapshot: false, live: res.error === null, error: res.error });
} else coverage.push({ siteId: s.id, fromSnapshot: true, live: false, error: null });
@@ -188,3 +190,15 @@ async function identities(r: SourcesRuntime, entityId: string) {
.execute();
return toApiList(rows) as never[];
}
/** 调用方 signal 中止时立即 reject(原 Promise 结果丢弃) */
function abortable<T>(p: Promise<T>, signal: AbortSignal | undefined): Promise<T> {
if (!signal) return p;
if (signal.aborted) return Promise.reject(signal.reason);
p.catch(() => {});
return new Promise<T>((resolve, reject) => {
const onAbort = () => reject(signal.reason);
signal.addEventListener("abort", onAbort, { once: true });
p.then(resolve, reject).finally(() => signal.removeEventListener("abort", onAbort));
});
}