diff --git a/apps/server/src/modules/acquisition-organize/__tests__/vertical.test.ts b/apps/server/src/modules/acquisition-organize/__tests__/vertical.test.ts index 7e97533c..5f76aaeb 100644 --- a/apps/server/src/modules/acquisition-organize/__tests__/vertical.test.ts +++ b/apps/server/src/modules/acquisition-organize/__tests__/vertical.test.ts @@ -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 () => { diff --git a/apps/server/src/modules/acquisition-organize/routes/library.ts b/apps/server/src/modules/acquisition-organize/routes/library.ts index 93da034c..721c7b8a 100644 --- a/apps/server/src/modules/acquisition-organize/routes/library.ts +++ b/apps/server/src/modules/acquisition-organize/routes/library.ts @@ -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 = { tmdb: "Tmdb", imdb: "Imdb", tvdb: "Tvdb", douban: "Douban" }; + +type MsInstance = MediaServerInventory["instances"][number]; +type MsInstanceRow = Parameters[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 { + 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 | undefined; + const timeout = new Promise((_, 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>[] = 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), + }); + } + } 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 = { 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), - }); - } - } 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 }, ); diff --git a/apps/server/src/modules/file-manager/identify.ts b/apps/server/src/modules/file-manager/identify.ts index 93451706..b12c2b86 100644 --- a/apps/server/src/modules/file-manager/identify.ts +++ b/apps/server/src/modules/file-manager/identify.ts @@ -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 { + opts: { wait?: boolean } = {}, +): Promise { + 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 { + 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 }> { const db = d.ctx.db; + const pendingPaths = new Set(); 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(); - const remoteCache = new Map(); + const remoteCache = new Map(); const entityKinds = new Map(); + // G8:候选海报(TMDB poster_path;本地作品取投影后的 metadata.posterPath) + const posters = new Map(); const localFor = async (p: ParsedRelease): Promise => { 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"] { diff --git a/apps/server/src/modules/file-manager/index.ts b/apps/server/src/modules/file-manager/index.ts index 45d619ed..98b7171a 100644 --- a/apps/server/src/modules/file-manager/index.ts +++ b/apps/server/src/modules/file-manager/index.ts @@ -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) { diff --git a/apps/server/src/modules/file-manager/routes.ts b/apps/server/src/modules/file-manager/routes.ts index c96d2227..4aef9aa7 100644 --- a/apps/server/src/modules/file-manager/routes.ts +++ b/apps/server/src/modules/file-manager/routes.ts @@ -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)); diff --git a/apps/server/src/modules/sources-identity/capabilities.ts b/apps/server/src/modules/sources-identity/capabilities.ts index 62712cab..40745e88 100644 --- a/apps/server/src/modules/sources-identity/capabilities.ts +++ b/apps/server/src/modules/sources-identity/capabilities.ts @@ -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(p: Promise, signal: AbortSignal | undefined): Promise { + if (!signal) return p; + if (signal.aborted) return Promise.reject(signal.reason); + p.catch(() => {}); + return new Promise((resolve, reject) => { + const onAbort = () => reject(signal.reason); + signal.addEventListener("abort", onAbort, { once: true }); + p.then(resolve, reject).finally(() => signal.removeEventListener("abort", onAbort)); + }); +} diff --git a/apps/server/test/file-manager/identify-organize.test.ts b/apps/server/test/file-manager/identify-organize.test.ts index 7c029144..1ddf5f72 100644 --- a/apps/server/test/file-manager/identify-organize.test.ts +++ b/apps/server/test/file-manager/identify-organize.test.ts @@ -1,6 +1,7 @@ import { appendFile, mkdir, writeFile } from "node:fs/promises"; import { join } from "node:path"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { IDENTIFY_COMPLETED_EVENT, runIdentifyRemote } from "../../src/modules/file-manager/identify.ts"; import { createFmHarness, EXEC_KEY, @@ -69,16 +70,74 @@ describe.skipIf(skip)("files 识别 / 整理 / 探测 / 刮削", () => { expect(pool).toHaveLength(0); }); - it("identify:远端候选(catalog.lookup)", async () => { - const r = await h.api("POST", "/files/identify", { + it("identify:远端候选(catalog.lookup,wait=true 同步模式)", async () => { + const r = await h.api("POST", "/files/identify?wait=true", { items: [{ root: h.dlRoot, path: "Remote.Film.2019.1080p.mkv" }], minSizeBytes: 0, }); expect(r.status).toBe(200); + expect(r.body.data.remoteTaskId).toBeNull(); const c = r.body.data.items[0].candidates; expect(c[0]).toMatchObject({ provider: "tmdb", externalId: "999", mediaEntityId: null, kind: "movie" }); }); + it("identify:缺省入队远端查询,先回本地命中 + remoteTaskId;任务完成发 identify.completed(S-7)", async () => { + const before = h.engine.enqueued.length; + const body = { + items: [ + { root: h.dlRoot, path: "Remote.Film.2019.1080p.mkv" }, + { root: h.dlRoot, path: "Blue.Film.2021.2160p.BluRay.x265-GRP.mkv" }, + ], + minSizeBytes: 0, + }; + const r = await h.api("POST", "/files/identify", body); + expect(r.status).toBe(200); + const [remote, local] = r.body.data.items; + // 本地命中照常返回;远端候选不在请求内查 + expect(remote.candidates).toEqual([]); + expect(remote.warnings.some((w: string) => w.includes("后台"))).toBe(true); + expect(local.candidates[0]).toMatchObject({ mediaEntityId: film }); + expect(r.body.data.remotePending).toBe(2); + const job = h.engine.enqueued.slice(before).find((e) => e.task === "identifyRemote"); + expect(job).toBeTruthy(); + expect(r.body.data.remoteTaskId).toMatch(/^run-/); + + // 模拟 worker 执行任务 + const out = await runIdentifyRemote( + h.module.deps(), + r.body.data.remoteTaskId, + job!.input as Parameters[2], + ); + expect(out.status).toBe("succeeded"); + const ev = h.events.filter((e) => e.type === IDENTIFY_COMPLETED_EVENT).at(-1); + expect(ev?.payload.taskId).toBe(r.body.data.remoteTaskId); + const done = ev!.payload.items.find((i: { path: string }) => i.path === "Remote.Film.2019.1080p.mkv"); + expect(done.candidates[0]).toMatchObject({ provider: "tmdb", externalId: "999" }); + expect(done.warnings.some((w: string) => w.includes("后台"))).toBe(false); + }); + + it("identify:无需远端查询时不入队(remoteTaskId=null)", async () => { + const before = h.engine.enqueued.length; + await mkdir(join(h.dl, "untitled"), { recursive: true }); + await writeFile(join(h.dl, "untitled", "1080p.mkv"), "x"); + const r = await h.api("POST", "/files/identify", { items: [{ root: h.dlRoot, path: "untitled/1080p.mkv" }] }); + expect(r.status).toBe(200); + expect(r.body.data.remoteTaskId).toBeNull(); + expect(h.engine.enqueued.slice(before).some((e) => e.task === "identifyRemote")).toBe(false); + }); + + it("identify:任务里识别失败也发 identify.completed(status=failed)", async () => { + const out = await runIdentifyRemote(h.module.deps(), "run-x", { + body: { items: [{ root: h.dlRoot, path: "no-such-file.mkv" }] } as never, + actorId: "system", + }); + expect(out.status).toBe("failed"); + expect(h.events.at(-1)).toMatchObject({ + type: IDENTIFY_COMPLETED_EVENT, + payload: { taskId: "run-x", status: "failed" }, + }); + }); + it("identify:去重后的远端查询单次上限 20,超出的项只本地召回并给 warning(验收 07 S-6)", async () => { await mkdir(join(h.dl, "many"), { recursive: true }); const names = Array.from({ length: 23 }, (_, i) => `Distinct Title ${String.fromCharCode(65 + i)}.2019.1080p.mkv`); diff --git a/apps/server/test/inventory/media-servers.test.ts b/apps/server/test/inventory/media-servers.test.ts new file mode 100644 index 00000000..3796c8da --- /dev/null +++ b/apps/server/test/inventory/media-servers.test.ts @@ -0,0 +1,104 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + createHarness, + type Harness, + pgAvailable, + seedEntity, +} from "../../src/modules/acquisition-organize/__tests__/harness.ts"; +import { MEDIA_SERVER_LIMITS } from "../../src/modules/acquisition-organize/routes/library.ts"; + +// C1-D:GET /inventory/media-servers 按实例并行(allSettled)+ 实例内并发上限 + 整实例超时;部分失败 partial=true。 + +const hasPg = await pgAvailable(); +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +describe.skipIf(!hasPg)("GET /inventory/media-servers 并行 + 超时 + partial", () => { + let h: Harness; + let entityId: string; + const ids: Record = {}; + const saved = { ...MEDIA_SERVER_LIMITS }; + const inflight: Record = {}; + + beforeAll(async () => { + h = await createHarness("msinv"); + MEDIA_SERVER_LIMITS.timeoutMs = 400; + MEDIA_SERVER_LIMITS.perInstanceConcurrency = 2; + entityId = await seedEntity(h.db, { kind: "movie", title: "并行库存", year: 2020 }); + await h.db + .insertInto("external_identity") + .values( + ["tmdb", "imdb", "tvdb"].map((provider, i) => ({ + media_entity_id: entityId, + provider, + entity_type: "movie", + external_id: `x${i}`, + status: "accepted", + })) as never, + ) + .execute(); + for (const name of ["a-ok", "b-slow", "c-boom"]) { + const r = await h.db + .insertInto("media_server_instance") + .values({ name, base_url: `http://${name}.invalid` }) + .returning("id") + .executeTakeFirstOrThrow(); + ids[name] = r.id; + inflight[name] = { now: 0, max: 0 }; + } + h.d.ports.mediaServer = (async (inst: { name: string }) => { + if (inst.name === "c-boom") throw new Error("connect refused"); + return { + async findByProviderId(provider: string) { + const f = inflight[inst.name]!; + f.now++; + f.max = Math.max(f.max, f.now); + try { + await sleep(inst.name === "b-slow" ? 5_000 : 60); + return provider === "Tmdb" + ? { + status: "present" as const, + items: [{ itemId: "i1", type: "Movie", name: "并行库存", year: 2020, path: "/m", versions: [] }], + } + : { status: "not_found" as const }; + } finally { + f.now--; + } + }, + }; + }) as never; + }, 60_000); + + afterAll(async () => { + Object.assign(MEDIA_SERVER_LIMITS, saved); + await h?.close(); + }); + + it("实例并行、慢实例超时、异常实例 unknown,整体 partial", async () => { + const t0 = Date.now(); + const r = await h.api("GET", `/inventory/media-servers?mediaEntityId=${entityId}`); + const took = Date.now() - t0; + expect(r.status).toBe(200); + type Inst = { name: string; status: string; reason: string | null; items: unknown[] }; + const by = Object.fromEntries((r.body.data.instances as Inst[]).map((x) => [x.name, x])); + expect(by["a-ok"]).toMatchObject({ status: "present", reason: null }); + expect(by["a-ok"]!.items).toHaveLength(1); + expect(by["b-slow"]).toMatchObject({ status: "unknown" }); + expect(by["b-slow"]!.reason).toMatch(/^timeout/); + expect(by["c-boom"]).toMatchObject({ status: "unknown", reason: "connect refused" }); + expect(r.body.data.partial).toBe(true); + // 串行时至少 0.4s(b 超时)+ 3×60ms;并行应接近单个超时 + expect(took).toBeLessThan(2_000); + // 实例内并发上限 2(3 个外部 ID) + expect(inflight["a-ok"]!.max).toBe(2); + }); + + it("全部成功时 partial=false;无外部 ID 的 unknown 不算失败", async () => { + await h.db.updateTable("media_server_instance").set({ enabled: false }).where("name", "!=", "a-ok").execute(); + const r = await h.api("GET", `/inventory/media-servers?mediaEntityId=${entityId}`); + expect(r.body.data.partial).toBe(false); + const bareId = await seedEntity(h.db, { kind: "movie", title: "无外部ID", year: 2021 }); + const r2 = await h.api("GET", `/inventory/media-servers?mediaEntityId=${bareId}`); + expect(r2.body.data.instances[0]).toMatchObject({ status: "unknown", reason: "no_external_id" }); + expect(r2.body.data.partial).toBe(false); + }); +}); diff --git a/docs/runbook/dev.md b/docs/runbook/dev.md index 2fb6cafd..b58f45a2 100644 --- a/docs/runbook/dev.md +++ b/docs/runbook/dev.md @@ -88,6 +88,18 @@ curl localhost:4380/api/v1/health 需要 PG 的测试(db、tasks)默认连 `DATABASE_URL` 或 55433 开发库,不可达时跳过。tasks 包可用 `TASKS_TEST_DATABASE_URL` 指定独立库;每个测试文件使用带 pid+随机后缀的独立 schema,结束时删除,多份测试并发跑也不冲突。 +测试被中断(超时、kill、并发跑挂掉)时 schema 不会删,开发库会越积越多。清理脚本: + +```sh +deno run -A scripts/dev-db-clean.ts # dry-run:只打印清单 +deno run -A scripts/dev-db-clean.ts --verbose # 同上,逐个列出 schema 名 +deno run -A scripts/dev-db-clean.ts --yes # 真正删除 +``` + +- 只在 `DATABASE_URL`(缺省 `postgres://umt:umt@localhost:55433/umt`)指向 localhost:55433 时运行,否则直接拒绝。 +- 删除的 schema:`tasks_test_`、`test_`、`e2e_`、`umt_test_`、`ao_test_`、`fm_test_`、`w2a_test_`、`w2b_`、`w7a_mig_`、`w9g_search_`、`server_test_` 前缀。`public`、`openworkflow` 不动;有其它会话持锁的 schema(测试正在跑)跳过,但最好在没有测试运行时执行。 +- 删除的 public 数据:名字以 `w32-`、`w33-`、`fe4-`、`qa-` 开头的作品(title)、站点(name)、订阅(title_query),以及按 restrict 外键挂在它们下面的行,在一个事务里删。`admin_user`(`w31-admin`)、`pi_upstream`、没有这些前缀的站点(fake 上游)和手工登记的数据不动;要牵连到受保护表时脚本中止。 + ## 排障 - `配置无效`:按提示检查 `.env`,`SECRET_MASTER_KEY` 至少 16 字符,`DATABASE_URL` 必须是 postgres://。 diff --git a/scripts/dev-db-clean.ts b/scripts/dev-db-clean.ts new file mode 100644 index 00000000..d83cf623 --- /dev/null +++ b/scripts/dev-db-clean.ts @@ -0,0 +1,215 @@ +// 开发库清理:只针对 55433 开发库(docker umt-dev-postgres-1),清掉测试残留。 +// +// 用法: +// deno run -A scripts/dev-db-clean.ts # 同 --dry-run:只打印清单,不改库 +// deno run -A scripts/dev-db-clean.ts --yes # 真正删除 +// DATABASE_URL=postgres://umt:umt@localhost:55433/umt deno run -A scripts/dev-db-clean.ts --dry-run +// +// 清理范围: +// 1. 测试 schema:名字匹配 TEST_SCHEMA_PATTERNS(各测试 harness 建的带 pid+随机后缀 schema)。 +// public / openworkflow / 系统 schema 永不删;有其它会话持有锁的 schema(测试正在跑)跳过。 +// 2. public 里名字带测试前缀(TEST_ENTITY_PREFIX:w32- / w33- / fe4- / qa-)的作品、站点、订阅, +// 连同按外键 restrict 挂在它们下面的行(cascade / set null 交给数据库)。 +// 不动:admin_user(w31-admin)、pi_upstream、名字不带测试前缀的站点(fake 上游 观众/HDSky/馒头 等)与用户登记的数据。 + +import pg from "pg"; + +const DEFAULT_URL = "postgres://umt:umt@localhost:55433/umt"; +const ALLOWED_HOSTS = new Set(["localhost", "127.0.0.1", "[::1]", "::1"]); +const DEV_PORT = "55433"; + +/** 测试 schema 前缀(见 apps/server/test 与 packages 下各 harness) */ +export const TEST_SCHEMA_PATTERNS: RegExp[] = [ + /^tasks_test_/, + /^test_/, + /^e2e_/, + /^umt_test_/, + /^ao_test_/, + /^fm_test_/, + /^w2a_test_/, + /^w2b_/, + /^w7a_mig_/, + /^w9g_search_/, + /^server_test_/, +]; +const PROTECTED_SCHEMAS = new Set(["public", "openworkflow", "information_schema"]); + +/** 测试实体名前缀(不区分大小写) */ +export const TEST_ENTITY_PREFIX = /^(w32|w33|fe4|qa)-/i; +const PREFIX_SQL = "^(w32|w33|fe4|qa)-"; + +/** 根:表 → 按哪一列匹配前缀 */ +const ROOTS: { table: string; column: string }[] = [ + { table: "subscription", column: "title_query" }, + { table: "media_entity", column: "title" }, + { table: "site", column: "name" }, +]; +/** 即使被外键牵连也绝不删除的表 */ +const NEVER_TOUCH = new Set(["admin_user", "admin_session", "api_key", "pi_upstream", "settings", "secret"]); + +export function assertDevUrl(url: string): void { + let u: URL; + try { + u = new URL(url); + } catch { + throw new Error(`DATABASE_URL 不是合法 URL`); + } + if (!/^postgres(ql)?:$/.test(u.protocol)) throw new Error("DATABASE_URL 必须是 postgres://"); + if (!ALLOWED_HOSTS.has(u.hostname) || u.port !== DEV_PORT) + throw new Error(`拒绝运行:只清理 localhost:${DEV_PORT} 开发库,当前是 ${u.hostname}:${u.port || "5432"}`); +} + +export const isTestSchema = (name: string) => + !PROTECTED_SCHEMAS.has(name) && !name.startsWith("pg_") && TEST_SCHEMA_PATTERNS.some((p) => p.test(name)); + +interface Fk { + child: string; + column: string; + parent: string; + parentColumn: string; + /** r/a = restrict/no action(需要先删子行);c = cascade;n/d = set null/default */ + rule: string; + childHasId: boolean; +} + +async function loadFks(c: pg.Client): Promise { + const r = await c.query(` + select cl.relname as child, a.attname as column, pc.relname as parent, pa.attname as "parentColumn", + con.confdeltype as rule, + exists (select 1 from pg_attribute i where i.attrelid = con.conrelid and i.attname = 'id' and not i.attisdropped) + as "childHasId" + from pg_constraint con + join pg_class cl on cl.oid = con.conrelid + join pg_class pc on pc.oid = con.confrelid + join pg_attribute a on a.attrelid = con.conrelid and a.attnum = con.conkey[1] + join pg_attribute pa on pa.attrelid = con.confrelid and pa.attnum = con.confkey[1] + where con.contype = 'f' and con.connamespace = 'public'::regnamespace and array_length(con.conkey, 1) = 1`); + return r.rows; +} + +type Plan = Map>; + +/** 从根行出发,沿 restrict/no action 外键收集必须先删的子行(只收集,不删) */ +async function collect(c: pg.Client, fks: Fk[], table: string, ids: string[], plan: Plan): Promise { + const seen = plan.get(table) ?? new Set(); + const fresh = ids.filter((id) => !seen.has(id)); + if (fresh.length === 0) return; + if (NEVER_TOUCH.has(table)) throw new Error(`会牵连受保护表 ${table},中止`); + for (const id of fresh) seen.add(id); + plan.set(table, seen); + for (const fk of fks.filter((f) => f.parent === table && (f.rule === "r" || f.rule === "a"))) { + if (!fk.childHasId) throw new Error(`${fk.child} 没有 id 列且以 restrict 引用 ${table},需手工处理`); + const r = await c.query<{ id: string }>( + `select id::text as id from public."${fk.child}" where "${fk.column}"::text = any($1::text[])`, + [fresh], + ); + if (r.rows.length) + await collect( + c, + fks, + fk.child, + r.rows.map((x) => x.id), + plan, + ); + } +} + +/** 删除顺序:子表先于父表(按 restrict 依赖拓扑;自引用表内一次删完) */ +function deleteOrder(plan: Plan, fks: Fk[]): string[] { + const tables = [...plan.keys()]; + const out: string[] = []; + const visiting = new Set(); + const visit = (t: string) => { + if (out.includes(t) || visiting.has(t)) return; + visiting.add(t); + for (const fk of fks) + if (fk.parent === t && fk.child !== t && plan.has(fk.child) && (fk.rule === "r" || fk.rule === "a")) + visit(fk.child); + visiting.delete(t); + out.push(t); + }; + for (const t of tables) visit(t); + return out; +} + +async function main() { + const args = new Set(process.argv.slice(2)); + const yes = args.has("--yes"); + if (yes && args.has("--dry-run")) throw new Error("--yes 与 --dry-run 不能同时给"); + const url = process.env.DATABASE_URL || DEFAULT_URL; + assertDevUrl(url); + const c = new pg.Client({ connectionString: url, connectionTimeoutMillis: 3000 }); + await c.connect(); + try { + // ── 1. 测试 schema ── + const schemas = ( + await c.query<{ name: string; tables: number; locked: boolean }>(` + select n.nspname as name, + (select count(*)::int from pg_class k where k.relnamespace = n.oid and k.relkind = 'r') as tables, + exists (select 1 from pg_locks l join pg_class k on k.oid = l.relation + where k.relnamespace = n.oid and l.pid <> pg_backend_pid()) as locked + from pg_namespace n order by n.nspname`) + ).rows.filter((s) => isTestSchema(s.name)); + const busy = schemas.filter((s) => s.locked); + const drop = schemas.filter((s) => !s.locked); + const groups = new Map(); + for (const s of drop) { + const g = TEST_SCHEMA_PATTERNS.find((p) => p.test(s.name))!.source.replace(/^\^/, ""); + groups.set(g, (groups.get(g) ?? 0) + 1); + } + console.log(`测试 schema:${drop.length} 个待删${busy.length ? `,${busy.length} 个正被使用(跳过)` : ""}`); + for (const [g, n] of [...groups].sort((a, b) => b[1] - a[1])) console.log(` ${g.padEnd(14)} ${n}`); + if (args.has("--verbose")) for (const s of drop) console.log(` ${s.name} (${s.tables} 表)`); + for (const s of busy) console.log(` 跳过(有锁):${s.name}`); + + // ── 2. public 里的测试实体 ── + const fks = await loadFks(c); + const plan: Plan = new Map(); + const roots: string[] = []; + for (const r of ROOTS) { + const rows = await c.query<{ id: string; label: string }>( + `select id::text as id, "${r.column}" as label from public."${r.table}" where "${r.column}" ~* $1`, + [PREFIX_SQL], + ); + for (const x of rows.rows) roots.push(` ${r.table.padEnd(13)} ${x.label}`); + if (rows.rows.length) + await collect( + c, + fks, + r.table, + rows.rows.map((x) => x.id), + plan, + ); + } + const order = deleteOrder(plan, fks); + const total = [...plan.values()].reduce((a, s) => a + s.size, 0); + console.log(`测试实体(前缀 w32-/w33-/fe4-/qa-):${roots.length} 个根,连同依赖共 ${total} 行`); + for (const line of roots) console.log(line); + for (const t of order) console.log(` 删除 ${t}: ${plan.get(t)!.size} 行`); + + if (!yes) { + console.log("\n(dry-run)未改动。确认后加 --yes 执行。"); + return; + } + for (const s of drop) await c.query(`drop schema if exists "${s.name}" cascade`); + await c.query("begin"); + try { + for (const t of order) + await c.query(`delete from public."${t}" where id::text = any($1::text[])`, [[...plan.get(t)!]]); + await c.query("commit"); + } catch (e) { + await c.query("rollback"); + throw e; + } + console.log(`\n已删除 ${drop.length} 个 schema、${total} 行测试实体。`); + } finally { + await c.end(); + } +} + +if (import.meta.main) { + main().catch((e) => { + console.error(e instanceof Error ? e.message : e); + process.exitCode = 1; + }); +}