diff --git a/apps/server/src/modules/acquisition-organize/index.ts b/apps/server/src/modules/acquisition-organize/index.ts index ce9d726a..00d54e7f 100644 --- a/apps/server/src/modules/acquisition-organize/index.ts +++ b/apps/server/src/modules/acquisition-organize/index.ts @@ -1,5 +1,6 @@ import { type EventPayload, EventType } from "@umt/contracts"; import { sql } from "@umt/db"; +import { configureStorage } from "../../storage/registry.ts"; import { registerDecisionHook } from "../decision-hooks.ts"; import { registerScheduledJob } from "../schedules/registry.ts"; import type { ServerContext, ServerModule } from "../types.ts"; @@ -69,6 +70,7 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se return deps; }, registerRoutes(app, ctx) { + configureStorage(ctx); const d = lazy(ctx); // 待处理"处理":接管识别失败 / 目标已存在 两类整理事项在关闭前由本模块执行所选处理 registerDecisionHook(ctx, organizeDecisionHook(d)); @@ -82,6 +84,7 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se registerClassificationRoutes(app, d); }, registerTasks(registry, ctx) { + configureStorage(ctx); registerAoTasks(registry, lazy(ctx)); }, registerCapabilities(registry, ctx) { diff --git a/apps/server/src/modules/acquisition-organize/ports.ts b/apps/server/src/modules/acquisition-organize/ports.ts index 21c33885..7ab27266 100644 --- a/apps/server/src/modules/acquisition-organize/ports.ts +++ b/apps/server/src/modules/acquisition-organize/ports.ts @@ -9,7 +9,6 @@ import { fetchEpisodeGroup, fetchRichDetail, fetchRichSeason, - ffprobe, getSiteAdapter, getSiteDefinition, groupSeasonAsRichSeason, @@ -27,6 +26,7 @@ import type { DownloaderInstanceRow, MediaServerInstanceRow, SiteRow, SourceRele import type { ParsedRelease } from "@umt/domain"; import { dataPaths } from "../../config.ts"; import { externalFetch } from "../../egress.ts"; +import { probeAny } from "../../storage/probe.ts"; import { ImageCache } from "../cache/images.ts"; import { SourcesRuntime } from "../sources-identity/runtime.ts"; import { @@ -250,7 +250,7 @@ export function createPorts( return fetchRichSeason(client, tvId, season, await metadataLanguages(ctx), await metadataTextOptions(ctx)); }, diskUsage: (p) => diskUsage(p), - probe: (p) => ffprobe(p), + probe: (p) => probeAny(p), async mediaServer(inst) { const settings = (inst.settings ?? {}) as Record; const apiKey = (await resolveSecret(inst.secret_ref)) ?? ""; diff --git a/apps/server/src/modules/acquisition-organize/services/organize-exec.ts b/apps/server/src/modules/acquisition-organize/services/organize-exec.ts index b9f4f5a1..c84ff802 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-exec.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-exec.ts @@ -13,6 +13,8 @@ import { import type { OrganizationRecord } from "@umt/contracts"; import { sql } from "@umt/db"; import { withAttemptToken } from "@umt/tasks"; +import { involvesRemote, remoteTransfer } from "../../../storage/transfer.ts"; +import { isRemotePath, vcheckRoot, vmkdirp, vrenameNoClobber, vstatSafe } from "../../../storage/vfs.ts"; import { closeDecisions } from "../../sources-identity/service/common.ts"; import { errMessage, jsonb } from "../util.ts"; import { type AoDeps, upsertDecision } from "./common.ts"; @@ -118,7 +120,9 @@ export async function executeRecord( // 根检查也走执行子进程:NFS/FUSE 挂死 → offline(timeout),按根离线重试,绝不当成文件缺失 // 离线 / 建目录失败不抛错(抛错会耗尽任务重试、记录卡在 running):记录保持 running 并写明原因、开待处理, // 由整理巡检(recoverStalledRecords)在库根恢复后重新入队 - const health = await fsr.proxy.checkRoot(frozen.rootPath, frozen.sentinel); + const health = isRemotePath(frozen.rootPath) + ? await vcheckRoot(frozen.rootPath) + : await fsr.proxy.checkRoot(frozen.rootPath, frozen.sentinel); if (health.kind === "offline") { return holdRecord(d, rec, `${ROOT_OFFLINE}:${health.reason}`, { kind: "other", @@ -133,7 +137,7 @@ export async function executeRecord( } if (rec.action !== "in_place") { try { - for (const dir of frozen.targetDirs) await mkdir(dir, { recursive: true }); + for (const dir of frozen.targetDirs) await vmkdirp(dir); } catch (err) { const code = (err as { code?: string } | null)?.code ?? "EIO"; return holdRecord(d, rec, `${TARGET_DIR_UNAVAILABLE}:${code}`, { @@ -164,6 +168,7 @@ export async function executeRecord( // 写前顺带清本目录残留 partial(对照 MP:只清本目录、>24h;失败吞掉)。copy 路径在 copyWithVerify 内还会再清一次 if (rec.action !== "in_place" && fsr.partialMaxAgeMs > 0) { for (const dir of frozen.targetDirs) { + if (isRemotePath(dir)) continue; await fsr.proxy.sweepStalePartials(dir, fsr.partialMaxAgeMs).catch(() => [] as string[]); } } @@ -250,7 +255,7 @@ export async function executeRecord( : undefined; const stagedOld = selectedOld ? await stagedReplacementBackup(d, selectedOld) : null; if (selectedOld && !stagedOld) { - const current = await fsr.proxy.statSafe(pf.dst, { + const current = await (isRemotePath(pf.dst) ? vstatSafe : fsr.proxy.statSafe.bind(fsr.proxy))(pf.dst, { root: selectedOld.rootPath, sentinel: selectedOld.sentinel, }); @@ -489,18 +494,27 @@ export async function trashReplacedOld( ref: { recordId: string; opId: string }, old: { path?: string; size: number }, ): Promise { - const rootReal = await realpath(frozen.rootPath); - const rel = relative(rootReal, await realpath(dirname(dst)).then((p) => join(p, basename(dst)))); + const remote = isRemotePath(frozen.rootPath); + const rootReal = remote ? frozen.rootPath : await realpath(frozen.rootPath); + const rel = remote + ? relative(rootReal, dst) + : relative(rootReal, await realpath(dirname(dst)).then((p) => join(p, basename(dst)))); if (!rel || rel.startsWith("..") || !old.path) return; const id = randomUUID(); const now = d.ports.now(); const to = join(rootReal, ".umt-trash", now.toISOString().slice(0, 10).replace(/-/g, ""), id, rel); - await mkdir(dirname(to), { recursive: true }); - const m = await fsr.proxy.moveSafe(old.path, to); - if (m.status !== "moved") { - throw new Error(`旧文件移入回收站失败(${m.status}): ${old.path}`); + if (remote) { + await vmkdirp(dirname(to)); + const m = await vrenameNoClobber(old.path, to); + if (m.status !== "moved") throw new Error(`旧文件移入回收站失败(${m.status}): ${old.path}`); + } else { + await mkdir(dirname(to), { recursive: true }); + const m = await fsr.proxy.moveSafe(old.path, to); + if (m.status !== "moved") { + throw new Error(`旧文件移入回收站失败(${m.status}): ${old.path}`); + } + await rmdir(dirname(old.path)).catch(() => undefined); } - await rmdir(dirname(old.path)).catch(() => undefined); const days = Number(Deno.env.get("FILES_TRASH_RETENTION_DAYS") ?? 7) || 7; await d.ctx.db .insertInto("trash_item") @@ -602,7 +616,9 @@ export async function verifyInPlace( path: string, size: number, ): Promise { - const observed = await fsr.proxy.statSafe(path, { root: frozen.rootPath, sentinel: frozen.sentinel }); + const observed = isRemotePath(path) + ? await vstatSafe(path, { root: frozen.rootPath }) + : await fsr.proxy.statSafe(path, { root: frozen.rootPath, sentinel: frozen.sentinel }); if (observed.kind !== "ok" || !observed.facts.isFile || observed.facts.isSymbolicLink) { return { status: "failed", error: { code: "SOURCE_MISSING", message: "原地整理的文件不存在或不可访问" } }; } @@ -645,6 +661,28 @@ export async function transferFile( }, ): Promise { const { kind, src, dst, size, crossDevice, resumed } = x; + // 涉及 Alist 的方向(上传 / 下载 / API copy·move / 跨实例中转):成功判定同 MP(接口成功 + 目标存在),不走本地核验 + if (involvesRemote(src, dst)) { + const r = await remoteTransfer({ + kind, + src, + dst, + onExisting: x.onExisting, + ...(x.replaceOld.trashDir ? { replaceTrashDir: x.replaceOld.trashDir } : {}), + resumed, + ...(x.signal ? { signal: x.signal } : {}), + }); + return { + status: r.status, + bytes: r.bytes ?? size, + ...(r.checksum ? { checksum: r.checksum } : {}), + ...(r.resumed ? { resumed: true } : {}), + ...(r.error ? { error: r.error } : {}), + ...(r.steps ? { steps: r.steps } : {}), + ...(r.replaced ? { replaced: r.replaced } : {}), + hashVia: `alist:${r.via}`, + }; + } const px = fsr.proxy; const call = x.signal ? { signal: x.signal } : {}; const fsErr = (e: unknown) => ({ diff --git a/apps/server/src/modules/acquisition-organize/services/organize-finalize.ts b/apps/server/src/modules/acquisition-organize/services/organize-finalize.ts index a1ea1b65..8da57986 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-finalize.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-finalize.ts @@ -1,10 +1,9 @@ -import { readdir } from "node:fs/promises"; import { dirname, join, relative } from "node:path"; -import { statSafe } from "@umt/adapters"; import type { DownloadTaskRow, OrganizationRecordRow } from "@umt/db"; import { sql } from "@umt/db"; import { findOtherVersions } from "@umt/domain"; import { isUnresolvedExclusion, organizationStatusFromFiles } from "@umt/mapping"; +import { isRemotePath, vreaddir as readdir, vstatSafe as statSafe } from "../../../storage/vfs.ts"; import { errMessage, jsonb } from "../util.ts"; import { type AoDeps, type Exec, loadInstances, upsertDecision } from "./common.ts"; import { applySlotEvent } from "./downloads.ts"; @@ -51,7 +50,12 @@ export async function finalizeRecord(d: AoDeps, recordId: string): Promise [] as string[]); for (const o of findOtherVersions( diff --git a/apps/server/src/modules/acquisition-organize/services/organize-plan.ts b/apps/server/src/modules/acquisition-organize/services/organize-plan.ts index 3710dbe2..11752b6b 100644 --- a/apps/server/src/modules/acquisition-organize/services/organize-plan.ts +++ b/apps/server/src/modules/acquisition-organize/services/organize-plan.ts @@ -1,6 +1,5 @@ -import { readdir } from "node:fs/promises"; import { basename, dirname, join, relative } from "node:path"; -import { checkRoot, DEFAULT_ROOT_SENTINEL, FsError, sha256Full, statSafe } from "@umt/adapters"; +import { DEFAULT_ROOT_SENTINEL, FsError, sha256Full } from "@umt/adapters"; import type { ActorType, OrganizationTrigger, PathBinding } from "@umt/contracts"; import { reasonText } from "@umt/contracts"; import type { DownloadTaskRow, LibraryRootRow, MediaEntityRow, SourceReleaseRow } from "@umt/db"; @@ -27,6 +26,12 @@ import { titleMismatches, } from "@umt/domain"; import { organizePlanToFileResults, toDomainPathBinding } from "@umt/mapping"; +import { + vcheckRoot as checkRoot, + isRemotePath, + vreaddir as readdir, + vstatSafe as statSafe, +} from "../../../storage/vfs.ts"; import { createEntityEpisodeMapper, createStructureEpisodeMapper, @@ -588,7 +593,7 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise x.path === f.src); diff --git a/apps/server/src/modules/acquisition-organize/services/reconcile.ts b/apps/server/src/modules/acquisition-organize/services/reconcile.ts index 52117373..2bd69e30 100644 --- a/apps/server/src/modules/acquisition-organize/services/reconcile.ts +++ b/apps/server/src/modules/acquisition-organize/services/reconcile.ts @@ -1,5 +1,6 @@ import { join } from "node:path"; -import { checkRoot, DEFAULT_ROOT_SENTINEL, statSafe } from "@umt/adapters"; +import { DEFAULT_ROOT_SENTINEL } from "@umt/adapters"; +import { vcheckRoot as checkRoot, vstatSafe as statSafe } from "../../../storage/vfs.ts"; import { errMessage, jsonb } from "../util.ts"; import type { AoDeps } from "./common.ts"; import { applySlotEvent } from "./downloads.ts"; diff --git a/apps/server/src/modules/acquisition-organize/services/replacement.ts b/apps/server/src/modules/acquisition-organize/services/replacement.ts index 05a111e8..2afc5fa1 100644 --- a/apps/server/src/modules/acquisition-organize/services/replacement.ts +++ b/apps/server/src/modules/acquisition-organize/services/replacement.ts @@ -4,10 +4,49 @@ import { basename, dirname, join, relative } from "node:path"; import { DEFAULT_ROOT_SENTINEL, type FileFacts } from "@umt/adapters"; import { sql } from "@umt/db"; import { type ReplacementScope, replacementCovers, replacementScope } from "@umt/domain"; +import { + isRemotePath, + vcheckRoot, + vmkdirp, + vreaddir, + vremove, + vrenameNoClobber, + vstatSafe, +} from "../../../storage/vfs.ts"; import { jsonb } from "../util.ts"; import { type AoDeps, upsertDecision } from "./common.ts"; import { type FrozenPlan, fsRuntime } from "./organize.ts"; +// 本地 / Alist 分派(docs/design/storage-alist.md §6.3):Alist 上的旧文件用同存储 move 暂放 / 进回收站 / 移回。 +// 远端没有 inode 与可靠的修改时间,旧文件「没变」只按大小判断;失败目标不做哈希核对(网盘方向不核验,用户决定)。 +const fsx = (d: AoDeps) => { + const fs = fsRuntime(d); + return { + replaceOld: fs.replaceOld, + checkRoot: (root: string, sentinel: string) => + isRemotePath(root) ? vcheckRoot(root) : fs.proxy.checkRoot(root, sentinel), + statSafe: (p: string, opts: { root?: string; sentinel?: string } = {}) => + isRemotePath(p) ? vstatSafe(p, opts.root ? { root: opts.root } : {}) : fs.proxy.statSafe(p, opts), + realpath: (p: string) => (isRemotePath(p) ? Promise.resolve(p) : realpath(p)), + async move(a: string, b: string): Promise<{ status: string }> { + if (isRemotePath(a) || isRemotePath(b)) { + await vmkdirp(dirname(b)); + return vrenameNoClobber(a, b); + } + await mkdir(dirname(b), { recursive: true }); + return fs.proxy.moveSafe(a, b); + }, + rm: (p: string) => (isRemotePath(p) ? vremove(p) : rm(p, { force: true })), + readdir: (p: string) => (isRemotePath(p) ? vreaddir(p) : readdir(p)), + sha256: (p: string) => fs.proxy.sha256(p), + }; +}; +/** 旧文件事实是否没变:本地比 size/mtime/ino/dev,远端只比 size */ +const sameFacts = (path: string, a: Partial, b: Partial) => + isRemotePath(path) + ? a.size === b.size + : (["size", "mtimeMs", "ino", "dev"] as const).every((k) => a[k] === undefined || a[k] === b[k]); + export interface ReplacementFile { assetId: string; locationId: string; @@ -106,7 +145,7 @@ export async function freezeReplacement( ); for (const file of [loc, ...privateChildren]) { const path = join(loc.local_path, file.relative_path); - const st = await fsRuntime(d).proxy.statSafe(path, { + const st = await fsx(d).statSafe(path, { root: loc.local_path, sentinel: loc.sentinel_file ?? DEFAULT_ROOT_SENTINEL, }); @@ -118,7 +157,7 @@ export async function freezeReplacement( locationId: file.id, rootId: loc.root_id, rootPath: loc.local_path, - sentinel: loc.sentinel_file ?? DEFAULT_ROOT_SENTINEL, + sentinel: isRemotePath(loc.local_path) ? "" : (loc.sentinel_file ?? DEFAULT_ROOT_SENTINEL), path: file.relative_path, facts: { size: st.facts.size, @@ -231,31 +270,26 @@ export async function cleanupReplacement(d: AoDeps, frozen: FrozenPlan, recordId if (location.status === "removed") continue; if (successful.some((f) => frozen.libraryRootId === file.rootId && f.target_path === file.path)) continue; try { - const fs = fsRuntime(d); - if ((await fs.proxy.checkRoot(file.rootPath, file.sentinel)).kind !== "online") { + const fs = fsx(d); + if ((await fs.checkRoot(file.rootPath, file.sentinel)).kind !== "online") { throw new Error("旧版本库根离线"); } - const rootReal = await realpath(file.rootPath); + const rootReal = await fs.realpath(file.rootPath); const abs = join(rootReal, file.path); - const parent = await realpath(dirname(abs)); + const parent = await fs.realpath(dirname(abs)); const rel = relative(rootReal, parent); if (rel.startsWith("..") || file.path.split("/").some((s) => s === "..")) throw new Error("旧文件位置越界"); const trash = join(rootReal, ".umt-trash", file.trashDay, file.trashId, file.path); - const st = await fs.proxy.statSafe(abs, { + const st = await fs.statSafe(abs, { root: rootReal, sentinel: file.sentinel, }); - const moved = await fs.proxy.statSafe(trash); + const moved = await fs.statSafe(trash); const resumed = existing?.status === "started" && st.kind === "missing" && (fs.replaceOld === "delete" || moved.kind === "ok"); - if ( - !resumed && - (st.kind !== "ok" || - st.facts.isSymbolicLink || - Object.entries(file.facts).some(([k, v]) => st.facts[k as keyof typeof file.facts] !== v)) - ) { + if (!resumed && (st.kind !== "ok" || st.facts.isSymbolicLink || !sameFacts(abs, file.facts, st.facts))) { throw new Error("旧文件已变化,保留待核对"); } await db @@ -273,10 +307,9 @@ export async function cleanupReplacement(d: AoDeps, frozen: FrozenPlan, recordId .onConflict((oc) => oc.column("operation_id").doUpdateSet({ status: "started" })) .execute(); if (!resumed) { - if (fs.replaceOld === "delete") await rm(abs); + if (fs.replaceOld === "delete") await fs.rm(abs); else { - await mkdir(dirname(trash), { recursive: true }); - const result = await fs.proxy.moveSafe(abs, trash); + const result = await fs.move(abs, trash); if (result.status !== "moved") { throw new Error(`旧文件回收失败:${result.status}`); } @@ -381,10 +414,10 @@ export async function settleSamePathReplacement(d: AoDeps, frozen: FrozenPlan, r continue; } try { - const fs = fsRuntime(d); + const fs = fsx(d); if (main?.status === "succeeded" && main.media_asset_id) { if (fs.replaceOld === "delete") { - await rm(backup.path, { force: true }); + await fs.rm(backup.path); if (backup.trashItemId) { await d.ctx.db.deleteFrom("trash_item").where("id", "=", backup.trashItemId).execute(); } @@ -402,38 +435,40 @@ export async function settleSamePathReplacement(d: AoDeps, frozen: FrozenPlan, r continue; } if (main?.status !== "failed") continue; - if ((await fs.proxy.checkRoot(old.rootPath, old.sentinel)).kind !== "online") { + if ((await fs.checkRoot(old.rootPath, old.sentinel)).kind !== "online") { throw new Error("旧版本库根离线,保留回收副本"); } + const remote = isRemotePath(pf.dst); const rejected = join(old.rootPath, ".umt-trash", old.trashDay, old.trashId, old.path); - const oldBackup = await fs.proxy.statSafe(backup.path); - const target = await fs.proxy.statSafe(pf.dst); + const oldBackup = await fs.statSafe(backup.path); + const target = await fs.statSafe(pf.dst); const alreadyRestored = - oldBackup.kind === "missing" && - target.kind === "ok" && - Object.entries(old.facts).every(([key, value]) => target.facts[key as keyof typeof old.facts] === value); + oldBackup.kind === "missing" && target.kind === "ok" && sameFacts(pf.dst, old.facts, target.facts); if (!alreadyRestored) { if ( oldBackup.kind !== "ok" || oldBackup.facts.size !== old.facts.size || - oldBackup.facts.ino !== old.facts.ino + (!remote && oldBackup.facts.ino !== old.facts.ino) ) { throw new Error("旧文件回收副本已变化"); } - if (!file.checksum?.startsWith("sha256:")) { + // Alist 目标不做哈希核对(网盘方向不核验,用户决定) + if (!remote && !file.checksum?.startsWith("sha256:")) { throw new Error("失败目标缺少全量校验,旧文件留在回收站"); } if (target.kind === "ok") { - if ((await fs.proxy.sha256(pf.dst)).digest !== file.checksum.slice(7)) { + if (!remote && (await fs.sha256(pf.dst)).digest !== file.checksum!.slice(7)) { throw new Error("失败目标已变化,旧文件留在回收站"); } - await mkdir(dirname(rejected), { recursive: true }); - const parked = await fs.proxy.moveSafe(pf.dst, rejected); + const parked = await fs.move(pf.dst, rejected); if (parked.status !== "moved") throw new Error("失败新文件无法暂存"); - } else if (target.kind !== "missing" || (await fs.proxy.sha256(rejected)).digest !== file.checksum.slice(7)) { + } else if ( + target.kind !== "missing" || + (!remote && (await fs.sha256(rejected)).digest !== file.checksum!.slice(7)) + ) { throw new Error("失败新文件暂存无法核验"); } - const restored = await fs.proxy.moveSafe(backup.path, pf.dst); + const restored = await fs.move(backup.path, pf.dst); if (restored.status !== "moved") { throw new Error("旧文件恢复未完成,副本保留"); } @@ -488,15 +523,11 @@ export async function settleSamePathReplacement(d: AoDeps, frozen: FrozenPlan, r export async function stagedReplacementBackup(d: AoDeps, file: ReplacementFile): Promise { const stage = join(file.rootPath, ".umt-trash", "replacement-staging", file.trashId); - for (const dir of await readdir(stage).catch(() => [] as string[])) { + const fs = fsx(d); + for (const dir of await fs.readdir(stage).catch(() => [] as string[])) { const candidate = join(stage, dir, basename(file.path)); - const st = await fsRuntime(d).proxy.statSafe(candidate); - if ( - st.kind === "ok" && - !st.facts.isSymbolicLink && - Object.entries(file.facts).every(([key, value]) => st.facts[key as keyof typeof file.facts] === value) - ) - return candidate; + const st = await fs.statSafe(candidate); + if (st.kind === "ok" && !st.facts.isSymbolicLink && sameFacts(candidate, file.facts, st.facts)) return candidate; } return null; } @@ -535,7 +566,7 @@ export async function quarantineNotVideo(d: AoDeps, frozen: FrozenPlan, recordId f.skip_reason === "main_failed" && rejectedMains.has(planned.get(f.source_path)?.mainSrc ?? ""))), ); - const fsr = fsRuntime(d); + const fsr = fsx(d); const moved: { path: string; trash: string }[] = []; const left: { path: string; error: string }[] = []; for (const f of targets) { @@ -543,10 +574,10 @@ export async function quarantineNotVideo(d: AoDeps, frozen: FrozenPlan, recordId if (frozen.replacement && pf?.action === "replace") continue; const rel = f.target_path!; try { - if ((await fsr.proxy.checkRoot(frozen.rootPath, frozen.sentinel)).kind !== "online") throw new Error("库根离线"); - const rootReal = await realpath(frozen.rootPath); + if ((await fsr.checkRoot(frozen.rootPath, frozen.sentinel)).kind !== "online") throw new Error("库根离线"); + const rootReal = await fsr.realpath(frozen.rootPath); const abs = join(rootReal, rel); - const st = await fsr.proxy.statSafe(abs, { root: rootReal, sentinel: frozen.sentinel }); + const st = await fsr.statSafe(abs, { root: rootReal, sentinel: frozen.sentinel }); if (st.kind === "missing") continue; if (st.kind !== "ok" || !st.facts.isFile || st.facts.isSymbolicLink) throw new Error("目标不是普通文件"); const registered = await db @@ -562,8 +593,7 @@ export async function quarantineNotVideo(d: AoDeps, frozen: FrozenPlan, recordId const id = randomUUID(); const now = d.ports.now(); const trash = join(rootReal, ".umt-trash", now.toISOString().slice(0, 10).replace(/-/g, ""), id, rel); - await mkdir(dirname(trash), { recursive: true }); - const m = await fsr.proxy.moveSafe(abs, trash); + const m = await fsr.move(abs, trash); if (m.status !== "moved") throw new Error(`移入回收站失败:${m.status}`); const days = Number(Deno.env.get("FILES_TRASH_RETENTION_DAYS") ?? 7) || 7; await db diff --git a/apps/server/src/modules/acquisition-organize/services/takeover.ts b/apps/server/src/modules/acquisition-organize/services/takeover.ts index 64e894d9..dde12628 100644 --- a/apps/server/src/modules/acquisition-organize/services/takeover.ts +++ b/apps/server/src/modules/acquisition-organize/services/takeover.ts @@ -1,6 +1,5 @@ import { randomUUID } from "node:crypto"; import { basename, dirname, extname, join } from "node:path"; -import { statSafe } from "@umt/adapters"; import type { DownloadTaskRow } from "@umt/db"; import { fileKindOf, @@ -10,6 +9,7 @@ import { parseRelease, parseReleasePath, } from "@umt/domain"; +import { vstatSafe as statSafe } from "../../../storage/vfs.ts"; import type { DecisionHook, DecisionHookActor, DecisionResolveBody } from "../../decision-hooks.ts"; import { addAliasIfAbsent } from "../../sources-identity/service/entities.ts"; import type { NameIdentifyResult } from "../ports.ts"; diff --git a/apps/server/test/storage/alist-organize-matrix.test.ts b/apps/server/test/storage/alist-organize-matrix.test.ts new file mode 100644 index 00000000..22e56c2a --- /dev/null +++ b/apps/server/test/storage/alist-organize-matrix.test.ts @@ -0,0 +1,276 @@ +import { access, readFile, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { executeRecord } from "../../src/modules/acquisition-organize/services/organize.ts"; +import { createFmHarness, type FmHarness, key, pgAvailable, seedEntity } from "../file-manager/harness.ts"; +import { addAlist, addAlistLibrary, type TestAlist, text } from "./alist.ts"; + +// 整理矩阵(docs/design/storage-alist.md §5):本地→Alist(上传)、Alist→本地(下载)、同实例 Alist→Alist(API copy/move)、 +// 跨实例(下载 → 上传),复制 / 移动各一条;成功判定同 MP(接口成功 + 目标存在),移动成功后删源; +// 失败路径:上传中断 / 上传后目标取不到 / 下载中断都不删源,连不上按可重试处理而不是「源不存在」。 + +const skip = !(await pgAvailable()); +const exists = (p: string) => + access(p).then( + () => true, + () => false, + ); + +describe.skipIf(skip)("整理矩阵:本地 ↔ Alist ↔ Alist", () => { + let h: FmHarness; + let a: TestAlist; + let b: TestAlist; + let alib: { libraryRootId: string; root: string }; + let blib: { libraryRootId: string; root: string }; + let n = 0; + beforeAll(async () => { + h = await createFmHarness("alist_matrix"); + a = await addAlist(h.db, { name: "A", rootPath: "/s1" }); + b = await addAlist(h.db, { name: "B", rootPath: "/" }); + alib = await addAlistLibrary(h.db, a, "/s1/library", "A 媒体库"); + blib = await addAlistLibrary(h.db, b, "/m/library", "B 媒体库"); + a.mock.mkdirp("/s1/incoming"); + }); + afterAll(async () => h?.close()); + + /** 新建一部电影作品 + 源文件,返回作品 id 与源路径(相对源根) */ + async function film(where: "local" | TestAlist) { + n++; + const title = `Matrix Film ${n}`; + const id = await seedEntity(h.db, { kind: "movie", title, year: 2000 + n }); + const name = `Matrix.Film.${n}.${2000 + n}.1080p.WEB-DL.mkv`; + const body = `video-${n}-${"x".repeat(32)}`; + if (where === "local") { + await writeFile(join(h.dl, name), body); + return { id, root: h.dlRoot, path: name, body, title, year: 2000 + n }; + } + where.mock.put(`${where === a ? "/s1/incoming" : "/m/incoming"}/${name}`, body); + return { + id, + root: where.root, + path: where === a ? `incoming/${name}` : `m/incoming/${name}`, + body, + title, + year: 2000 + n, + }; + } + + /** dryRun → 执行 → 跑 executeRecord(模拟 worker);返回记录与逐文件结果 */ + async function organize( + src: { id: string; root: string; path: string }, + action: "copy" | "move", + targetLibraryRootId: string, + ) { + const items = [{ root: src.root, path: src.path, mediaEntityId: src.id }]; + const dry = await h.api("POST", "/files/organize", { + idempotencyKey: key(), + items, + action, + targetLibraryRootId, + dryRun: true, + }); + expect(dry.status, JSON.stringify(dry.body)).toBe(200); + expect(dry.body.data.groups[0].error).toBeNull(); + const exec = await h.api("POST", "/files/organize", { + idempotencyKey: key(), + items, + action, + targetLibraryRootId, + dryRun: false, + planIds: [dry.body.data.groups[0].planId], + }); + expect(exec.status, JSON.stringify(exec.body)).toBe(200); + const recordId = exec.body.data.recordIds[0] as string; + let thrown: unknown = null; + await executeRecord(h.module.deps().ao, recordId).catch((e) => { + thrown = e; + }); + const rec = await h.db + .selectFrom("organization_record") + .selectAll() + .where("id", "=", recordId) + .executeTakeFirstOrThrow(); + const files = await h.db + .selectFrom("organization_file_result") + .selectAll() + .where("organization_record_id", "=", recordId) + .execute(); + return { rec, files, thrown, main: files.find((f) => f.role === "main")! }; + } + + const assetAt = (libraryRootId: string, rel: string) => + h.db + .selectFrom("asset_location") + .select(["id", "status"]) + .where("library_root_id", "=", libraryRootId) + .where("relative_path", "=", rel) + .executeTakeFirst(); + + it("本地 → Alist 复制:上传到媒体库目录,登记资产,源保留,记下本地 sha256", async () => { + const f = await film("local"); + const r = await organize(f, "copy", alib.libraryRootId); + expect(r.rec.status).toBe("succeeded"); + expect(r.main.target_path).toContain(`${f.title} (${f.year})`); + expect(text(a.mock.read(`/s1/library/${r.main.target_path}`))).toBe(f.body); + expect(await exists(join(h.dl, f.path))).toBe(true); + expect(r.main.checksum).toMatch(/^sha256:[0-9a-f]{64}$/); + expect((await assetAt(alib.libraryRootId, r.main.target_path!))?.status).toBe("available"); + expect(a.mock.requests.some((x) => x.path === "/api/fs/put")).toBe(true); + }); + + it("本地 → Alist 移动:上传成功后删本地源", async () => { + const f = await film("local"); + const r = await organize(f, "move", alib.libraryRootId); + expect(r.rec.status).toBe("succeeded"); + expect(text(a.mock.read(`/s1/library/${r.main.target_path}`))).toBe(f.body); + expect(await exists(join(h.dl, f.path))).toBe(false); + }); + + it("Alist → 本地 复制 / 移动:下载到本地媒体库;移动后删远端源", async () => { + const c = await film(a); + const rc = await organize(c, "copy", h.libraryRootId); + expect(rc.rec.status).toBe("succeeded"); + expect(await readFile(join(h.lib, rc.main.target_path!), "utf8")).toBe(c.body); + expect(a.mock.exists(`/s1/${c.path}`)).toBe(true); + expect(rc.main.checksum).toMatch(/^sha256:/); + + const m = await film(a); + const rm = await organize(m, "move", h.libraryRootId); + expect(rm.rec.status).toBe("succeeded"); + expect(await readFile(join(h.lib, rm.main.target_path!), "utf8")).toBe(m.body); + expect(a.mock.exists(`/s1/${m.path}`)).toBe(false); + }); + + it("Alist → Alist 同实例:复制 / 移动走 Alist API(等任务完成)", async () => { + const c = await film(a); + a.mock.requests.length = 0; + const rc = await organize(c, "copy", alib.libraryRootId); + expect(rc.rec.status).toBe("succeeded"); + expect(text(a.mock.read(`/s1/library/${rc.main.target_path}`))).toBe(c.body); + expect(a.mock.exists(`/s1/${c.path}`)).toBe(true); + expect(a.mock.requests.some((x) => x.path === "/api/fs/copy")).toBe(true); + expect(a.mock.requests.some((x) => x.path.startsWith("/api/task/copy/info"))).toBe(true); + expect(a.mock.requests.some((x) => x.path === "/api/fs/put")).toBe(false); + + const m = await film(a); + const rm = await organize(m, "move", alib.libraryRootId); + expect(rm.rec.status).toBe("succeeded"); + expect(text(a.mock.read(`/s1/library/${rm.main.target_path}`))).toBe(m.body); + expect(a.mock.exists(`/s1/${m.path}`)).toBe(false); + }); + + it("Alist → Alist 跨实例:下载到本机中转再上传;移动后删源", async () => { + const c = await film(a); + const rc = await organize(c, "copy", blib.libraryRootId); + expect(rc.rec.status).toBe("succeeded"); + expect(text(b.mock.read(`/m/library/${rc.main.target_path}`))).toBe(c.body); + expect(a.mock.exists(`/s1/${c.path}`)).toBe(true); + + const m = await film(a); + const rm = await organize(m, "move", blib.libraryRootId); + expect(rm.rec.status).toBe("succeeded"); + expect(text(b.mock.read(`/m/library/${rm.main.target_path}`))).toBe(m.body); + expect(a.mock.exists(`/s1/${m.path}`)).toBe(false); + }); + + it("失败:上传中断 → 不删源、远端不留临时文件,按可重试等待", async () => { + const f = await film("local"); + a.mock.breakUploadAfter(8); + const r = await organize(f, "move", alib.libraryRootId); + expect(String(r.thrown)).toContain("可重试失败"); + expect(await exists(join(h.dl, f.path))).toBe(true); + expect(r.main.status).toBe("running"); + const op = await h.db + .selectFrom("asset_operation") + .select("error") + .where("organization_record_id", "=", r.rec.id) + .executeTakeFirstOrThrow(); + expect((op.error as { code: string }).code).toBe("IO"); + const dir = `/s1/library/${f.title} (${f.year})`; + expect((a.mock.listNames(dir) ?? []).filter((x) => x.endsWith(".uhd-partial"))).toEqual([]); + // 重试(网络恢复)后完成并删源 + await executeRecord(h.module.deps().ao, r.rec.id); + const done = await h.db + .selectFrom("organization_record") + .select("status") + .where("id", "=", r.rec.id) + .executeTakeFirstOrThrow(); + expect(done.status).toBe("succeeded"); + expect(await exists(join(h.dl, f.path))).toBe(false); + }); + + it("失败:上传接口成功但目标取不到 → 判失败,不删源", async () => { + const f = await film("local"); + a.mock.dropNextUpload(); + const r = await organize(f, "move", alib.libraryRootId); + expect(await exists(join(h.dl, f.path))).toBe(true); + const op = await h.db + .selectFrom("asset_operation") + .select(["error", "status"]) + .where("organization_record_id", "=", r.rec.id) + .executeTakeFirstOrThrow(); + expect(op.error).toBeTruthy(); + }); + + it("失败:下载中断 → 不生成目标、不删远端源", async () => { + const f = await film(a); + a.mock.breakDownloadAfter(4); + const r = await organize(f, "move", h.libraryRootId); + expect(String(r.thrown)).toContain("可重试失败"); + expect(a.mock.exists(`/s1/${f.path}`)).toBe(true); + const target = join(h.lib, `${f.title} (${f.year})`); + const names = await import("node:fs/promises").then((m) => m.readdir(target).catch(() => [] as string[])); + expect(names.filter((x) => x.endsWith(".mkv"))).toEqual([]); + }); + + it("连不上 Alist:按可重试 IO 处理,不当成「源不存在」", async () => { + const f = await film(a); + const items = [{ root: f.root, path: f.path, mediaEntityId: f.id }]; + const dry = await h.api("POST", "/files/organize", { + idempotencyKey: key(), + items, + action: "copy", + targetLibraryRootId: h.libraryRootId, + dryRun: true, + }); + const exec = await h.api("POST", "/files/organize", { + idempotencyKey: key(), + items, + action: "copy", + targetLibraryRootId: h.libraryRootId, + dryRun: false, + planIds: [dry.body.data.groups[0].planId], + }); + const recordId = exec.body.data.recordIds[0]; + a.mock.failNext(50); + await expect(executeRecord(h.module.deps().ao, recordId)).rejects.toThrow(/可重试失败/); + const op = await h.db + .selectFrom("asset_operation") + .select("error") + .where("organization_record_id", "=", recordId) + .executeTakeFirstOrThrow(); + expect((op.error as { code: string }).code).not.toBe("SOURCE_MISSING"); + a.mock.failNext(0); + }); + + it("撤销:删除整理记录连带媒体库文件 → Alist 上移进回收站(可还原)", async () => { + const f = await film("local"); + const r = await organize(f, "copy", alib.libraryRootId); + expect(r.rec.status).toBe("succeeded"); + const del = await h.api("POST", "/organization-records/delete", { + ids: [r.rec.id], + deleteSource: false, + deleteTarget: true, + dryRun: false, + }); + expect(del.status, JSON.stringify(del.body)).toBe(200); + expect(a.mock.exists(`/s1/library/${r.main.target_path}`)).toBe(false); + const trash = await h.db + .selectFrom("trash_item") + .select(["trash_path", "library_root_id"]) + .where("library_root_id", "=", alib.libraryRootId) + .execute(); + expect(trash.length).toBeGreaterThan(0); + expect(trash[0]!.trash_path.startsWith(a.v("/s1/library/.umt-trash/"))).toBe(true); + }); +});