merge: feat/alist-storage

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei

# Conflicts:
#	packages/adapters/deno.json
This commit is contained in:
root committed 2026-10-07 00:48:53 +08:00
commit 339105e151
108 files changed
+7255 -214

No files matched your search

+2
View File
@@ -17,6 +17,7 @@ import type { EventBusPort } from "./http/sse.ts";
import { assertCapabilityCount, createCapabilityRegistry } from "./modules/agent/capability-registry.ts";
import { createSecretStore, createSettingsReader } from "./modules/auth-settings/provided.ts";
import type { OutboxPort, ServerContext, ServerModule, ServerRole } from "./modules/types.ts";
import { configureStorage } from "./storage/registry.ts";
// 进程装配:ServerContext、任务注册表、outbox 路由。api / worker / all 三种角色共用。
@@ -200,6 +201,7 @@ export async function bootstrap(opts: BootstrapOptions): Promise<Bootstrapped> {
// 授权(automation_grant / approval + 自动化总开关)、审计(audit_event)、幂等键与操作账本在这里注入
capabilities: createCapabilityRegistry({ logger, db, settings }),
};
configureStorage(ctx);
const tasks = buildTaskRegistry(opts.modules, ctx);
for (const m of opts.modules) m.registerCapabilities?.(ctx.capabilities, ctx);
ctx.hooks = buildHookRegistry(opts.modules, ctx);
@@ -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) {
@@ -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<string, unknown>;
const apiKey = (await resolveSecret(inst.secret_ref)) ?? "";
@@ -1,5 +1,4 @@
import { randomUUID } from "node:crypto";
import { mkdir, realpath, rmdir } from "node:fs/promises";
import { basename, dirname, join, relative } from "node:path";
import {
type CopyReceipt,
@@ -13,6 +12,9 @@ import {
import type { OrganizationRecord } from "@umt/contracts";
import { sql } from "@umt/db";
import { withAttemptToken } from "@umt/tasks";
import { mkdir, realpath, rmdir } from "../../../storage/local-fs.ts";
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<void> {
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<FileOutcome> {
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<FileOutcome> {
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) => ({
@@ -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<Execu
for (const f of filesPre) {
if (f.status === "succeeded" && f.role === "main" && f.target_path) {
const abs = join(frozenPre.rootPath, f.target_path);
const r = await d.ports.probe(abs);
let r = await d.ports.probe(abs);
// Alist 目标:直链探测拿不到(驱动不给直链等)时退回探测本地源(复制时还在)
if (r.status === "unknown" && isRemotePath(abs) && !isRemotePath(f.source_path)) {
const local = await d.ports.probe(f.source_path);
if (local.status === "ok") r = local;
}
probes.set(f.id, r);
const names = await readdir(dirname(abs)).catch(() => [] as string[]);
for (const o of findOtherVersions(
@@ -1,6 +1,8 @@
import { realpath } from "node:fs/promises";
import { isAbsolute, resolve, sep } from "node:path";
import type { Principal } from "../../../http/auth.ts";
import { isRemotePath, vrealpath } from "../../../storage/vfs.ts";
import { toVPath } from "../../../storage/vpath.ts";
import { canonicalJson } from "../util.ts";
import { type AoDeps, type Exec, hostPathOf, loadInstances, upsertDecision } from "./common.ts";
import { hnrSatisfied } from "./retention.ts";
@@ -41,8 +43,17 @@ export async function allowedSourceRoots(exec: Exec): Promise<string[]> {
for (const m of t.mappings) roots.add(m.hostPath);
if (t.mappings.length === 0 && t.localSavePath) roots.add(t.localSavePath);
}
// Alist 存储:实例根路径(文件管理 st: 根、整理来源可选)
for (const s of await exec
.selectFrom("storage_instance")
.select(["id", "root_path"])
.where("enabled", "=", true)
.execute())
roots.add(toVPath(s.id, s.root_path));
const out: string[] = [];
for (const r of roots) if (r && isAbsolute(r)) out.push(await real(r.replace(/\/+$/, "") || "/"));
for (const r of roots)
if (r && isAbsolute(r))
out.push(isRemotePath(r) ? r.replace(/\/+$/, "") : await real(r.replace(/\/+$/, "") || "/"));
// 根目录本身("/")不能作为白名单,否则等于不设防
return [...new Set(out)].filter((r) => r !== "/");
}
@@ -62,7 +73,7 @@ export async function sourceGuard(exec: Exec): Promise<SourceGuard> {
if (p.split("/").some((seg) => seg === "..")) throw new PathGuardError(`源路径不允许包含 "..": ${p}`, p);
let r: string;
try {
r = await realpath(p);
r = isRemotePath(p) ? await vrealpath(p) : await realpath(p);
} catch {
throw new PathGuardError(`源路径不存在或不可访问:${p}`, p);
}
@@ -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<PlanRes
return fail("slot_entity_mismatch", { slotId: req.slotId }, entity.id);
}
}
const sentinel = root.sentinel_file ?? DEFAULT_ROOT_SENTINEL;
const sentinel = isRemotePath(root.local_path) ? "" : (root.sentinel_file ?? DEFAULT_ROOT_SENTINEL);
const health = await checkRoot(root.local_path, sentinel);
if (health.kind === "offline") {
const r = await fail(
@@ -881,7 +886,8 @@ export async function planImport(d: AoDeps, req: ImportRequest): Promise<PlanRes
continue;
}
const e: ExistingTarget = { path: f.dst, size: st.facts.size };
if (st.facts.size === f.size) {
// 涉及 Alist 时不算哈希(网盘方向不核验,用户决定):同尺寸也按目标已存在交给策略判定
if (st.facts.size === f.size && !isRemotePath(probe) && !isRemotePath(f.src)) {
// 同尺寸已存在的目标:全量 sha256 判定是否同一文件(子进程,见 adapters sha256Full)
e.sha256 = await cachedSha256(probe, st.facts);
const src = files.find((x) => x.path === f.src);
@@ -1,9 +1,17 @@
import { randomUUID } from "node:crypto";
import { readdir, rmdir } from "node:fs/promises";
import { dirname, join, relative } from "node:path";
import { checkRoot, DEFAULT_ROOT_SENTINEL, statSafe } from "@umt/adapters";
import { DEFAULT_ROOT_SENTINEL } from "@umt/adapters";
import { fileKindOf, type OrganizeFileKind, type OrganizePlan, type PlannedFile } from "@umt/domain";
import { organizationStatusFromFiles, organizePlanToFileResults } from "@umt/mapping";
import { readdir, rmdir } from "../../../storage/local-fs.ts";
import {
isRemotePath,
vstatSafe as statSafe,
vcheckRoot,
vreaddir,
vreaddirStats,
vremove,
} from "../../../storage/vfs.ts";
import { hashJson, jsonb } from "../util.ts";
import type { AoDeps, Exec } from "./common.ts";
import { type ExecuteResult, emitRecord, type FrozenPlan, type ImportRequest, type PlanResult } from "./organize.ts";
@@ -27,7 +35,8 @@ export interface TransferItem {
export interface TransferSpec {
items: TransferItem[];
targetLibraryRootId: string;
/** 目标库根;目标是 Alist 存储根(文件管理 st:)时为 null */
targetLibraryRootId: string | null;
/** 目标库根 realpath(相对路径以它为基准) */
targetRootPath: string;
/** 遍历到的每个文件都经它复核(realpath 在允许的根下) */
@@ -36,7 +45,7 @@ export interface TransferSpec {
export interface FrozenTransfer {
items: (TransferItem & { isDir: boolean; files: number })[];
targetLibraryRootId: string;
targetLibraryRootId: string | null;
}
const KIND_OF = (p: string): OrganizeFileKind => {
@@ -50,6 +59,18 @@ async function walkTree(
out: { files: string[]; dirs: string[]; links: string[] },
limit: number,
): Promise<boolean> {
if (isRemotePath(dir)) {
// Alist:一次 list 拿到子项与类型,没有符号链接
for (const it of (await vreaddirStats(dir)).sort((a, b) => (a.name < b.name ? -1 : 1))) {
if (out.files.length >= limit) return true;
const p = join(dir, it.name);
if (it.stats.isDirectory()) {
out.dirs.push(p);
if (await walkTree(p, out, limit)) return true;
} else out.files.push(p);
}
return false;
}
const names = (await readdir(dir)).sort();
for (const n of names) {
if (out.files.length >= limit) return true;
@@ -71,14 +92,14 @@ export async function planTransfer(d: AoDeps, req: ImportRequest, t: TransferSpe
const dry = req.dryRun === true;
const action = req.action ?? "copy";
const failed = (reason: string): PlanResult => ({ recordId: null, status: "failed", reason });
const root = await db
.selectFrom("library_root")
.selectAll()
.where("id", "=", t.targetLibraryRootId)
.executeTakeFirst();
if (root?.management_mode !== "managed") return failed("target_root_not_managed");
const sentinel = root.sentinel_file ?? DEFAULT_ROOT_SENTINEL;
const health = await checkRoot(t.targetRootPath, sentinel);
const lib = t.targetLibraryRootId
? await db.selectFrom("library_root").selectAll().where("id", "=", t.targetLibraryRootId).executeTakeFirst()
: null;
if (t.targetLibraryRootId && lib?.management_mode !== "managed") return failed("target_root_not_managed");
if (!t.targetLibraryRootId && !isRemotePath(t.targetRootPath)) return failed("target_root_not_managed");
const root = { id: lib?.id ?? null };
const sentinel = isRemotePath(t.targetRootPath) ? "" : (lib?.sentinel_file ?? DEFAULT_ROOT_SENTINEL);
const health = await vcheckRoot(t.targetRootPath, sentinel);
if (health.kind === "offline") return failed(`library_root_offline:${health.reason}`);
const max = d.ports.options.walkMaxFiles;
@@ -148,7 +169,7 @@ export async function planTransfer(d: AoDeps, req: ImportRequest, t: TransferSpe
organize: plan,
entityId: req.mediaEntityId ?? "",
mediaType: "movie",
libraryRootId: root.id,
libraryRootId: root.id ?? "",
rootPath: t.targetRootPath,
sentinel,
targetDirs: [...dirs].sort(),
@@ -162,7 +183,7 @@ export async function planTransfer(d: AoDeps, req: ImportRequest, t: TransferSpe
};
if (dry) return { recordId: null, status: "queued", reason: null, plan: frozen, ...(truncated ? { truncated } : {}) };
const drafts = organizePlanToFileResults(plan, { targetLibraryRootId: root.id });
const drafts = organizePlanToFileResults(plan, { targetLibraryRootId: root.id ?? "" });
const now = d.ports.now();
const rel = (abs: string | null | undefined) => (abs ? relative(t.targetRootPath, abs) : null);
const heldSrc = new Set(files.filter((f) => f.action === "hold").map((f) => f.src));
@@ -321,6 +342,15 @@ export async function relocate(trx: Exec, input: RelocateInput): Promise<Relocat
/** 目录下的子目录,深的在前(删空目录用) */
async function dirsDeepFirst(dir: string): Promise<string[]> {
const out: string[] = [];
if (isRemotePath(dir)) {
const walk = async (p: string) => {
const ents = await vreaddirStats(p).catch(() => []);
for (const e of ents) if (e.stats.isDirectory()) await walk(join(p, e.name));
out.push(p);
};
await walk(dir);
return out;
}
const walk = async (p: string) => {
const ents = await readdir(p, { withFileTypes: true }).catch(() => []);
for (const e of ents) if (e.isDirectory()) await walk(join(p, e.name));
@@ -330,6 +360,16 @@ async function dirsDeepFirst(dir: string): Promise<string[]> {
return out;
}
/** 只删空目录(Alist 的 remove 会递归删,先确认为空) */
async function removeEmptyDir(dir: string): Promise<void> {
if (!isRemotePath(dir)) {
await rmdir(dir).catch(() => undefined);
return;
}
const left = await vreaddir(dir).catch(() => null);
if (left && left.length === 0) await vremove(dir).catch(() => undefined);
}
/** 原样搬运的终态:记录状态 + move 的位置跟随/绑定修订 + 删源侧空目录;行锁 fencing,重复调用幂等 */
export async function finalizeTransfer(d: AoDeps, recordId: string): Promise<ExecuteResult> {
const db = d.ctx.db;
@@ -351,7 +391,7 @@ export async function finalizeTransfer(d: AoDeps, recordId: string): Promise<Exe
// move:整个目录都搬完才删源侧空目录(先于终态提交;失败/中断重跑幂等)
if (pre.action === "move" && (pre.status === "running" || pre.status === "queued"))
for (const it of tr.items.filter((x) => x.isDir && allOk(x)))
for (const dir of await dirsDeepFirst(it.src)) await rmdir(dir).catch(() => undefined);
for (const dir of await dirsDeepFirst(it.src)) await removeEmptyDir(dir);
return db.transaction().execute(async (trx) => {
const rec = await trx
@@ -373,9 +413,9 @@ export async function finalizeTransfer(d: AoDeps, recordId: string): Promise<Exe
if (rec.action === "move") {
const actor = `organize:${recordId}`;
for (const it of t.items) {
if (it.srcRel === null || it.srcLibraryRootId === null) continue;
if (it.srcRel === null || it.srcLibraryRootId === null || t.targetLibraryRootId === null) continue;
const mine = files.filter((f) => under(it, f.source_path));
const base = { fromLib: it.srcLibraryRootId, toLib: t.targetLibraryRootId, actor, opKind: "move" as const };
const base = { fromLib: it.srcLibraryRootId, toLib: t.targetLibraryRootId!, actor, opKind: "move" as const };
if (mine.length > 0 && mine.every((f) => f.status === "succeeded")) {
// 整项搬完:按前缀跟随(目录含路径绑定修订)
const r = await relocate(trx, { ...base, fromRel: it.srcRel, toRel: it.dstRel, isDir: it.isDir });
@@ -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";
@@ -1,13 +1,52 @@
import { randomUUID } from "node:crypto";
import { mkdir, readdir, realpath, rm } from "node:fs/promises";
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 { mkdir, readdir, realpath, rm } from "../../../storage/local-fs.ts";
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<FileFacts>, b: Partial<FileFacts>) =>
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<string | null> {
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
@@ -1,7 +1,6 @@
import { createHash, randomUUID } from "node:crypto";
import { lstat, mkdir, readdir, readFile, realpath, rename, rm, writeFile } from "node:fs/promises";
import { dirname, isAbsolute, join, relative, resolve } from "node:path";
import { checkRoot, DEFAULT_ROOT_SENTINEL } from "@umt/adapters";
import { DEFAULT_ROOT_SENTINEL } from "@umt/adapters";
import {
DEFAULT_SETTINGS,
type OrganizeSettings,
@@ -24,6 +23,16 @@ import {
seasonOfDirName,
workNfo,
} from "@umt/domain";
import { mkdir, rename, rm, writeFile } from "../../../storage/local-fs.ts";
import {
isRemotePath,
vlstat as lstat,
vreadFile as readFile,
vrealpath as realpath,
vcheckRoot,
vreaddirents,
vwriteFileAtomic,
} from "../../../storage/vfs.ts";
import {
createEntityEpisodeMapper,
type EntityEpisodeMapper,
@@ -587,6 +596,8 @@ export async function scrapeAfterImport(
/** 原子写:同目录临时文件 → rename;目录不存在时建(只在登记目录内) */
async function atomicWrite(abs: string, content: Uint8Array | string): Promise<void> {
// Alist 目标:本地生成后上传到对应目录(同 MP 刮削 _save_file → upload_file),已存在时上传旁名后替换
if (isRemotePath(abs)) return vwriteFileAtomic(abs, content);
await mkdir(dirname(abs), { recursive: true });
const tmp = join(dirname(abs), `.${randomUUID()}.umt-scrape.tmp`);
try {
@@ -621,7 +632,7 @@ async function pool<T, R>(items: readonly T[], n: number, fn: (t: T) => Promise<
}
async function listDir(rootReal: string, rel: string) {
const names = await readdir(join(rootReal, rel), { withFileTypes: true }).catch(() => []);
const names = await vreaddirents(join(rootReal, rel)).catch(() => []);
const visible = names.filter((n) => !n.name.startsWith(".")).sort((a, b) => a.name.localeCompare(b.name));
return {
videos: visible
@@ -948,7 +959,7 @@ export async function runScrapeRecord(
.executeTakeFirst();
if (!lr?.enabled) return await fail("scrape_root_offline");
if (lr.management_mode !== "managed") return await fail("scrape_root_read_only");
const health = await checkRoot(lr.local_path, lr.sentinel_file ?? DEFAULT_ROOT_SENTINEL);
const health = await vcheckRoot(lr.local_path, lr.sentinel_file ?? DEFAULT_ROOT_SENTINEL);
if (health.kind !== "online") return await fail("scrape_root_offline");
}
const rootReal = await realpath(rec.root_path).catch(() => null);
@@ -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";
@@ -7,6 +7,7 @@ import { registerEngineRoutes } from "./engine-routes.ts";
import { registerInstanceRoutes } from "./instances-routes.ts";
import { registerKeyRoutes } from "./keys-routes.ts";
import { registerSettingsRoutes } from "./settings-routes.ts";
import { registerStorageRoutes } from "./storage-routes.ts";
/**
* W2-A 认证/设置/实例/引擎面板:管理员登录门禁、settings、api_key、secret 加密服务、
@@ -38,6 +39,7 @@ export function createAuthSettingsModule(opts: AuthSettingsModuleOptions = {}):
registerSettingsRoutes(app, ctx);
registerKeyRoutes(app, ctx);
registerInstanceRoutes(app, ctx);
registerStorageRoutes(app, ctx);
registerEngineRoutes(app, ctx, opts.engineSchema ? { schema: opts.engineSchema } : {});
if (opts.setupTokenOnStart !== false && ctx.db) {
ensureSetupToken(ctx).catch((err) => ctx.logger.error({ err }, "生成 setup token 失败"));
@@ -25,6 +25,8 @@ import {
import { z } from "zod";
import { ApiException } from "../../http/errors.ts";
import { redact, redactString } from "../../logger.ts";
import { isRemotePath, vcheckRoot } from "../../storage/vfs.ts";
import { parseVPath } from "../../storage/vpath.ts";
import type { ModuleApp, ServerContext } from "../types.ts";
import {
afterCursor,
@@ -184,6 +186,17 @@ export function registerInstanceRoutes(app: ModuleApp, ctx: ServerContext) {
// ── library_root ──
const sentinel = async (localPath: string, file: string | null | undefined) => {
// Alist 库根(/@alist/<存储 id>/<路径>):不写哨兵(同 MP),只确认目录存在且存储可连
if (isRemotePath(localPath)) {
if (!parseVPath(localPath)) throw new ApiException("validation_failed", "Alist 路径格式不对");
const h = await vcheckRoot(localPath);
if (h.kind !== "online")
throw new ApiException(
"precondition_failed",
`Alist 目录不可用(${h.reason}),先在 Alist 里建好目录并测试存储连接`,
);
return null;
}
const name = file || DEFAULT_ROOT_SENTINEL;
try {
const st = await stat(localPath);
@@ -229,7 +242,9 @@ export function registerInstanceRoutes(app: ModuleApp, ctx: ServerContext) {
test("/library-roots/{id}/test", "检查媒体库目录在线(目录 + 哨兵文件)", "library-roots"),
async ({ params, actor }) => {
const row = await getRow("library_root", params.id);
const h = await checkRoot(String(row.local_path), (row.sentinel_file as string | null) ?? DEFAULT_ROOT_SENTINEL);
const h = isRemotePath(String(row.local_path))
? await vcheckRoot(String(row.local_path))
: await checkRoot(String(row.local_path), (row.sentinel_file as string | null) ?? DEFAULT_ROOT_SENTINEL);
const ok = h.kind === "online";
await update("library_root", row.id, {
health_status: ok ? "healthy" : "offline",
@@ -0,0 +1,264 @@
import { AlistError, normAlistPath, redactAlistText } from "@umt/adapters";
import {
Api,
StorageInstance,
StorageInstanceCreate,
StorageInstanceUpdate,
type StorageTestResult,
secretInput,
} from "@umt/contracts";
import { ApiException } from "../../http/errors.ts";
import { buildClient, clearStorageCache, configureStorage, resolvePlain } from "../../storage/registry.ts";
import type { ModuleApp, ServerContext } from "../types.ts";
import {
afterCursor,
audit,
crudDefs,
cursorCol,
decodeCursor,
endpoint,
pageOf,
pgConflict,
toApi,
toDb,
} from "./endpoint.ts";
// 存储实例(Alist / OpenList)CRUD + 测试连接(docs/design/storage-alist.md §3、§4)。
// 凭据(令牌 / 密码)与路径密码只写不读:省略或空串 = 不变,null = 清除;明文只在 secret 表里加密保存。
const JSON_COLS = ["options", "health_detail", "version_info"] as const;
const tags = ["storage-instances"];
type Row = Record<string, unknown> & {
id: string;
secret_ref: string | null;
path_password_ref: string | null;
auth_kind: "token" | "password" | "guest";
username: string | null;
};
export function registerStorageRoutes(app: ModuleApp, ctx: ServerContext) {
configureStorage(ctx);
const S = {
...crudDefs("/storage-instances", "存储实例", {
row: StorageInstance,
create: StorageInstanceCreate,
update: StorageInstanceUpdate,
}),
test: Api.storageInstances.test,
remove: Api.storageInstances.remove,
};
const db = () => ctx.db;
const getRow = async (id: string): Promise<Row> => {
const r = await db().selectFrom("storage_instance").selectAll().where("id", "=", id).executeTakeFirst();
if (!r) throw new ApiException("not_found", "存储实例不存在");
return r as unknown as Row;
};
const normalize = (b: Partial<StorageInstanceCreate>) => {
const out: Record<string, unknown> = { ...b };
if (b.baseUrl !== undefined) out.baseUrl = b.baseUrl.trim().replace(/\/+$/, "");
if (b.rootPath !== undefined) {
try {
out.rootPath = normAlistPath(b.rootPath);
} catch {
throw new ApiException("validation_failed", "根路径不合法");
}
}
return out;
};
const checkAuth = (authKind: string, username: string | null | undefined, hasCredential: boolean) => {
if (authKind === "token" && !hasCredential) throw new ApiException("validation_failed", "令牌认证需要填写令牌");
if (authKind === "password" && (!username || !hasCredential))
throw new ApiException("validation_failed", "账号密码认证需要填写用户名和密码");
};
const putSecret = async (oldRef: string | null, plaintext: string) => {
if (oldRef) {
await ctx.secrets.replace(oldRef, plaintext);
return oldRef;
}
return ctx.secrets.create({ purpose: "storage_credential", plaintext });
};
const view = (r: object) => toApi(r);
endpoint(app, { ...S.list, tags }, async ({ query }) => {
const q0 = query as { cursor?: string; limit: number };
const cur = decodeCursor(q0.cursor);
let q = db().selectFrom("storage_instance").selectAll().select(cursorCol("created_at"));
if (cur) q = q.where(afterCursor("created_at", cur));
const rows = await q
.orderBy("created_at", "desc")
.orderBy("id", "desc")
.limit(q0.limit + 1)
.execute();
const page = pageOf(rows as unknown as { cursor_us: number; id: string }[], q0.limit);
return { items: page.items.map(view), nextCursor: page.nextCursor } as never;
});
endpoint(app, { ...S.get, tags }, async ({ params }) => view(await getRow(params.id)) as never);
endpoint(app, { ...S.create, tags }, async ({ body, actor }) => {
const { credential: rawCredential, pathPassword: rawPw, ...rest } = body;
const credential = secretInput(rawCredential) ?? undefined;
const pathPassword = secretInput(rawPw) ?? undefined;
checkAuth(rest.authKind ?? "token", rest.username, !!credential);
const refs: string[] = [];
try {
const secretRef =
credential && (rest.authKind ?? "token") !== "guest"
? await ctx.secrets.create({ purpose: "storage_credential", plaintext: credential })
: null;
if (secretRef) refs.push(secretRef);
const pwRef = pathPassword
? await ctx.secrets.create({ purpose: "storage_credential", plaintext: pathPassword })
: null;
if (pwRef) refs.push(pwRef);
const row = await db()
.insertInto("storage_instance")
.values({
...(toDb(normalize(rest), JSON_COLS) as { name: string; base_url: string }),
secret_ref: secretRef,
path_password_ref: pwRef,
})
.returningAll()
.executeTakeFirstOrThrow()
.catch((e: unknown) => pgConflict(e, "存储实例名称重复"));
await audit(
ctx,
actor,
"storage_instance.create",
{ type: "storage_instance", id: row.id },
{ name: row.name, authKind: row.auth_kind },
);
return view(row) as never;
} catch (e) {
for (const r of refs) await ctx.secrets.remove(r).catch(() => {});
throw e;
}
});
endpoint(app, { ...S.update, tags }, async ({ params, body, actor }) => {
const { credential: rawCredential, pathPassword: rawPw, ...rest } = body;
const credential = secretInput(rawCredential);
const pathPassword = secretInput(rawPw);
const cur = await getRow(params.id);
const authKind = rest.authKind ?? cur.auth_kind;
const username = rest.username !== undefined ? rest.username : cur.username;
const hasCred = credential === null ? false : credential ? true : !!cur.secret_ref;
checkAuth(authKind, username, hasCred);
const values = toDb(normalize(rest), JSON_COLS);
const drop: string[] = [];
if (authKind === "guest" || credential === null) {
if (cur.secret_ref) {
values.secret_ref = null;
drop.push(cur.secret_ref);
}
} else if (credential) values.secret_ref = await putSecret(cur.secret_ref, credential);
if (pathPassword === null) {
if (cur.path_password_ref) {
values.path_password_ref = null;
drop.push(cur.path_password_ref);
}
} else if (pathPassword) values.path_password_ref = await putSecret(cur.path_password_ref, pathPassword);
// 改了凭据也要让注册表重建客户端:updated_at 由触发器刷新;无字段变化时补一次 touch
const row = await db()
.updateTable("storage_instance")
.set({ ...(values as Record<string, never>), updated_at: new Date() })
.where("id", "=", params.id)
.returningAll()
.executeTakeFirst()
.catch((e: unknown) => pgConflict(e, "存储实例名称重复"));
if (!row) throw new ApiException("not_found", "存储实例不存在");
for (const r of drop) await ctx.secrets.remove(r).catch(() => {});
clearStorageCache(row.id);
await audit(
ctx,
actor,
"storage_instance.update",
{ type: "storage_instance", id: row.id },
{
fields: Object.keys(body).map((k) =>
k === "credential" || k === "pathPassword" ? `${k}(${body[k] === null ? "cleared" : "set"})` : k,
),
},
);
return view(row) as never;
});
endpoint(app, { ...S.remove, tags }, async ({ params, actor }) => {
const row = await getRow(params.id);
const prefix = `/@alist/${row.id}`;
const used = await db()
.selectFrom("library_root")
.select("name")
.where((eb) => eb.or([eb("local_path", "=", prefix), eb("local_path", "like", `${prefix}/%`)]))
.execute();
if (used.length)
throw new ApiException(
"conflict",
`存储实例仍被媒体库目录使用:${used.map((u) => u.name).join("、")},先删除或改用其它存储`,
);
await db().deleteFrom("storage_instance").where("id", "=", row.id).execute();
for (const r of [row.secret_ref, row.path_password_ref]) if (r) await ctx.secrets.remove(r).catch(() => {});
clearStorageCache(row.id);
await audit(ctx, actor, "storage_instance.delete", { type: "storage_instance", id: row.id });
return { ok: true as const };
});
endpoint(app, { ...S.test, tags }, async ({ params, actor }) => {
const row = await getRow(params.id);
const result = await testStorage(ctx, row);
await db()
.updateTable("storage_instance")
.set({
health_status: result.healthStatus,
last_health_at: new Date(),
health_detail: JSON.stringify({ message: result.message, rootOk: result.rootOk, usage: result.usage }),
...(result.version ? { version_info: JSON.stringify({ version: result.version, user: result.user }) } : {}),
})
.where("id", "=", row.id)
.execute();
clearStorageCache(row.id);
await audit(ctx, actor, "storage_instance.test", { type: "storage_instance", id: row.id }, { ok: result.ok });
return result;
});
}
/** 测试连接(设置页「测试连接」与健康检查共用) */
export async function testStorage(
ctx: Pick<ServerContext, "secrets" | "logger">,
row: Row & { base_url?: unknown; root_path?: unknown; options?: unknown },
): Promise<StorageTestResult> {
let plain: { credential: string | null; pathPassword: string | null } = { credential: null, pathPassword: null };
const fail = (healthStatus: StorageTestResult["healthStatus"], message: string): StorageTestResult => ({
ok: false,
healthStatus,
version: null,
user: null,
rootOk: false,
usage: null,
message: redactAlistText(message, [plain.credential, plain.pathPassword]).slice(0, 500),
});
try {
plain = await resolvePlain(ctx.secrets, row);
const client = buildClient(row as never, plain);
const version = await client.version();
const user = row.auth_kind === "guest" ? "guest" : (await client.me()).username;
const root = await client.stat(String(row.root_path ?? "/"));
const usage = await client.usage(String(row.root_path ?? "/"));
if (!root?.isDir) return { ...fail("degraded", `根路径不存在或不是目录:${row.root_path}`), version, user, usage };
return { ok: true, healthStatus: "healthy", version, user, rootOk: true, usage, message: null };
} catch (e) {
if (e instanceof AlistError) {
const label =
e.kind === "auth"
? row.auth_kind === "token"
? "令牌失效或无权限"
: "登录失败(用户名或密码不对)"
: e.kind === "unavailable" || e.kind === "timeout"
? "连不上"
: "接口出错";
return fail("offline", `${label}:${e.message}`);
}
return fail("offline", String((e as Error)?.message ?? e));
}
}
+78 -5
View File
@@ -1,10 +1,13 @@
import { createHash } from "node:crypto";
import { lstat, realpath, stat } from "node:fs/promises";
import { basename, dirname, join, sep } from "node:path";
import { checkRoot, DEFAULT_ROOT_SENTINEL, type TmdbClient } from "@umt/adapters";
import { AlistError, checkRoot, DEFAULT_ROOT_SENTINEL, type TmdbClient } from "@umt/adapters";
import type { ApiErrorCode, FileRootId, FilesErrorReason } from "@umt/contracts";
import type { LibraryRootRow } from "@umt/db";
import type { Principal } from "../../http/auth.ts";
import { listStorages } from "../../storage/registry.ts";
import { isRemotePath, remoteError, vcheckRoot, vlstat, vrealpath } from "../../storage/vfs.ts";
import { parseVPath, toVPath } from "../../storage/vpath.ts";
import type { AoDeps, Exec } from "../acquisition-organize/services/common.ts";
import { loadInstances } from "../acquisition-organize/services/common.ts";
import type { SourcesRuntime } from "../sources-identity/runtime.ts";
@@ -122,13 +125,15 @@ export const isOfflineErr = (e: unknown) => OFFLINE_CODES.has(errnoOf(e) ?? "");
* 处理器里漏出的文件系统错误(检查与使用之间目录被删、挂载掉线等竞态)→ 业务错误,而不是 500:
* 离线类 → 503 root_unavailable(失败 ≠ 空目录),ENOENT/ENOTDIR → 404,EACCES/EPERM → 403,其余 errno → 400。
*/
export function fsErrorToFiles(e: unknown): FilesError | null {
if (e instanceof FilesError) return e;
export function fsErrorToFiles(e0: unknown): FilesError | null {
if (e0 instanceof FilesError) return e0;
const e = e0 instanceof AlistError ? remoteError(e0) : e0;
const code = errnoOf(e);
if (typeof code !== "string" || !/^E[A-Z]+$/.test(code)) return null;
const msg = e instanceof Error ? e.message : String(e);
if (OFFLINE_CODES.has(code)) return new FilesError("root_unavailable", `读取失败(${code}),根可能已离线:${msg}`);
if (code === "ENOENT" || code === "ENOTDIR") return new FilesError("not_found", `路径已不存在(${code})`);
if (code === "EEXIST") return new FilesError("exists", `目标已存在:${msg}`);
if (code === "EACCES" || code === "EPERM") return new FilesError("forbidden", `没有访问权限(${code})`);
return new FilesError("bad_request", `文件系统错误(${code}): ${msg}`);
}
@@ -137,7 +142,7 @@ export function fsErrorToFiles(e: unknown): FilesError | null {
export interface RootDef {
id: FileRootId;
kind: "library" | "download";
kind: "library" | "download" | "storage";
name: string;
hostPath: string;
mediaKind: LibraryRootRow["media_kind"] | null;
@@ -146,6 +151,9 @@ export interface RootDef {
libraryRootId: string | null;
instanceId: string | null;
sentinel: string | null;
/** 所在 Alist 存储实例(本机为 null);hostPath 是虚拟路径 /@alist/<id>/… */
storageId: string | null;
storageName: string | null;
}
const norm = (p: string) => p.replace(/\/+$/, "") || "/";
@@ -153,10 +161,14 @@ const norm = (p: string) => p.replace(/\/+$/, "") || "/";
/** 已登记的根:启用的库根 + 下载器路径映射本机前缀(无映射 = 同机,取保存路径);"/" 永不出现 */
export async function loadRootDefs(exec: Exec): Promise<RootDef[]> {
const out: RootDef[] = [];
const storages = await listStorages(exec as never).catch(() => []);
const storageName = new Map(storages.map((s) => [s.id, s.name]));
const libs = await exec.selectFrom("library_root").selectAll().where("enabled", "=", true).orderBy("name").execute();
for (const r of libs) {
const hostPath = norm(r.local_path);
if (hostPath === "/" || !hostPath.startsWith("/")) continue;
const v = parseVPath(hostPath);
if (isRemotePath(hostPath) && !v) continue;
out.push({
id: `lib:${r.id}`,
kind: "library",
@@ -166,7 +178,9 @@ export async function loadRootDefs(exec: Exec): Promise<RootDef[]> {
modifiable: r.management_mode === "managed",
libraryRootId: r.id,
instanceId: null,
sentinel: r.sentinel_file ?? DEFAULT_ROOT_SENTINEL,
sentinel: v ? null : (r.sentinel_file ?? DEFAULT_ROOT_SENTINEL),
storageId: v?.storageId ?? null,
storageName: v ? (storageName.get(v.storageId) ?? null) : null,
});
}
for (const t of await loadInstances(exec, { enabledOnly: false })) {
@@ -184,9 +198,28 @@ export async function loadRootDefs(exec: Exec): Promise<RootDef[]> {
libraryRootId: null,
instanceId: t.inst.id,
sentinel: null,
storageId: null,
storageName: null,
});
});
}
// Alist 存储(对照 MP 存储切换):整个实例从根路径开始可浏览
for (const s of storages) {
if (!s.enabled) continue;
out.push({
id: `st:${s.id}`,
kind: "storage",
name: s.name,
hostPath: toVPath(s.id, s.root_path),
mediaKind: null,
modifiable: true,
libraryRootId: null,
instanceId: null,
sentinel: null,
storageId: s.id,
storageName: s.name,
});
}
return out;
}
@@ -211,6 +244,11 @@ const withTimeout = <T>(p: Promise<T>, ms: number, fallback: T): Promise<T> =>
export async function probeRoot(def: RootDef, timeoutMs: number): Promise<RootHealthView> {
const checkedAt = new Date();
const run = async (): Promise<RootHealthView> => {
if (def.storageId) {
const h = await vcheckRoot(def.hostPath);
if (h.kind === "offline") return { health: "offline", reason: h.reason, checkedAt, real: null };
return { health: "healthy", reason: null, checkedAt, real: def.hostPath };
}
if (def.sentinel) {
const h = await checkRoot(def.hostPath, def.sentinel);
if (h.kind === "offline") return { health: "offline", reason: h.reason, checkedAt, real: null };
@@ -283,6 +321,7 @@ export async function resolveIn(
const def = opts.def ?? (await rootDef(d.ctx.db, rootId));
const rootReal = opts.rootReal ?? (await onlineRoot(d, def));
const rel = checkRel(relIn);
if (def.storageId) return resolveRemote(def, rootReal, rel, opts.mustExist !== false);
if (rel === "") {
const lst = await lstat(rootReal).catch(() => null);
return { def, rootReal, rel, abs: rootReal, lst, isSymlink: false };
@@ -314,8 +353,42 @@ export async function resolveIn(
return { def, rootReal, rel: relOf(rootReal, abs), abs, lst, isSymlink };
}
/** 远端根:没有符号链接,路径即 realpath;lst 是 vfs 的 stat(不存在为 null,连不上抛 root_unavailable) */
async function resolveRemote(def: RootDef, rootReal: string, rel: string, mustExist: boolean): Promise<Resolved> {
const abs = rel === "" ? rootReal : join(rootReal, rel);
let lst: Resolved["lst"] = null;
try {
lst = (await vlstat(abs)) as unknown as Resolved["lst"];
} catch (e) {
if (errnoOf(e) !== "ENOENT") throw remoteFilesError(e, def);
}
if (!lst && mustExist) throw new FilesError("not_found", `不存在:${rel}`);
return { def, rootReal, rel, abs, lst, isSymlink: false };
}
/** 远端错误 → 业务错误(连不上 / 超时 → 503 根不可用;鉴权 → 403;冲突 → 409) */
export function remoteFilesError(e: unknown, def?: RootDef): FilesError {
if (e instanceof FilesError) return e;
const code = errnoOf(e);
const msg = e instanceof Error ? e.message : String(e);
if (code === "ENOENT") return new FilesError("not_found", msg);
if (code === "EEXIST") return new FilesError("exists", msg);
if (code === "EACCES") return new FilesError("forbidden", msg);
return new FilesError("root_unavailable", `${def ? `存储 ${def.storageName ?? def.name} ` : ""}不可用:${msg}`);
}
/** 执行阶段复核(TOCTOU):再次 realpath,必须仍在根内 */
export async function recheck(abs: string, rootReal: string): Promise<string> {
if (isRemotePath(abs)) {
try {
const r = await vrealpath(abs);
if (!within(r, rootReal)) throw new FilesError("path_escape", `执行时路径指向根外:${abs}`);
return r;
} catch (e) {
if (errnoOf(e) === "ENOENT") throw new FilesError("plan_stale", `源已不存在:${abs}`);
throw remoteFilesError(e);
}
}
let r: string;
try {
r = await realpath(abs);
@@ -1,6 +1,7 @@
import { basename, extname } from "node:path";
import type { FilesDownloadQuery } from "@umt/contracts";
import type { Principal } from "../../http/auth.ts";
import { remote, remoteError } from "../../storage/vfs.ts";
import { FilesError, type FmDeps, isAdminSession, recheck, resolveIn } from "./common.ts";
import { audit } from "./ledger.ts";
import { kindOfName } from "./list.ts";
@@ -55,6 +56,7 @@ export async function downloadFile(
if (!ALLOWED.has(kind) && !pk)
throw new FilesError("forbidden", `只能下载 NFO、字幕、图片、文本和 .torrent,此文件类型为 ${kind}`);
const real = await recheck(r.abs, r.rootReal);
if (r.def.storageId) return downloadRemote(d, principal, q, r.rel, real, name, kind, pk);
const file = await Deno.open(real, { read: true });
let size: number;
try {
@@ -86,3 +88,47 @@ export async function downloadFile(
},
});
}
/** Alist 文件:经服务端流式转发(不落本地盘,同一下载上限与类型限制;MP 是先下到临时目录再回传) */
async function downloadRemote(
d: FmDeps,
principal: Principal | undefined,
q: FilesDownloadQuery,
rel: string,
abs: string,
name: string,
kind: string,
pk: ReturnType<typeof previewKindOf>,
): Promise<Response> {
const { v, client } = await remote(abs);
let size: number;
let res: Response;
try {
const o = await client.get(v.remote);
if (o.isDir) throw new FilesError("bad_request", "只能下载文件,不能下载目录");
size = o.size;
if (size > d.options.downloadMaxBytes)
throw new FilesError(
"too_large",
`文件 ${size} 字节,超过下载上限 ${d.options.downloadMaxBytes}(FILES_DOWNLOAD_MAX_BYTES)`,
);
res = await client.openRead(v.remote);
} catch (e) {
if (e instanceof FilesError) throw e;
throw remoteError(e, abs);
}
const ext = extname(name).replace(/^\./, "").toLowerCase();
const inline = q.inline && (kind === "image" || pk === "image") && ext !== "svg";
await audit(d, principal, "download", { root: q.root, path: rel, size, inline });
return new Response(res.body, {
status: 200,
headers: {
"content-type": (ext !== "svg" && MIME[ext]) || "application/octet-stream",
...(size ? { "content-length": String(size) } : {}),
"content-disposition": disposition(inline ? "inline" : "attachment", name),
"x-content-type-options": "nosniff",
"content-security-policy": "default-src 'none'; sandbox",
"cache-control": "private, no-store",
},
});
}
@@ -1,5 +1,4 @@
import { randomUUID } from "node:crypto";
import { readdir, stat } from "node:fs/promises";
import { join } from "node:path";
import {
EventType,
@@ -21,6 +20,7 @@ import {
} from "@umt/domain";
import { localIdentityCandidate } from "@umt/mapping";
import type { Principal } from "../../http/auth.ts";
import { vlstat as stat, vreaddirents } from "../../storage/vfs.ts";
import { successfulOrganizeCounts } from "../acquisition-organize/services/organize-history.ts";
import { EXTERNAL_INVOKE_TIMEOUT_MS, sharedInvokeLimiter } from "../agent/invoke-limits.ts";
import { createEntityEpisodeMapper, type EntityEpisodeMapper } from "../sources-identity/service/episode-resolution.ts";
@@ -90,7 +90,7 @@ async function collectVideos(d: FmDeps, body: FilesIdentifyRequest): Promise<{ v
const stack = [r.rel];
while (stack.length > 0) {
const rel = stack.shift()!;
const names = (await readdir(join(r.rootReal, rel), { withFileTypes: true }).catch(() => [])).sort((a, b) =>
const names = (await vreaddirents(join(r.rootReal, rel)).catch(() => [])).sort((a, b) =>
a.name.localeCompare(b.name),
);
for (const n of names) {
@@ -3,6 +3,7 @@ import type { TmdbClient } from "@umt/adapters";
import { FilesIdentifyRequest } from "@umt/contracts";
import { defineTask } from "@umt/tasks";
import { z } from "zod";
import { configureStorage } from "../../storage/registry.ts";
import type { AoOverrides } from "../acquisition-organize/index.ts";
import { createPorts } from "../acquisition-organize/ports.ts";
import { registerScheduledJob } from "../schedules/registry.ts";
@@ -57,10 +58,12 @@ export function createFileManagerModule(overrides: FmOverrides = {}): ServerModu
return deps;
},
registerRoutes(app, ctx) {
configureStorage(ctx);
registerFileRoutes(app, lazy(ctx));
registerOrganizeSourceRoutes(app, lazy(ctx));
},
registerTasks(registry, ctx) {
configureStorage(ctx);
registry.register(
defineTask({
name: "scanOrganizeSource",
+99 -12
View File
@@ -17,6 +17,8 @@ import type {
} from "@umt/contracts";
import { sql } from "@umt/db";
import { fileKindOf } from "@umt/domain";
import { storageEntry } from "../../storage/registry.ts";
import { isRemotePath, vlstat, vreaddir, vreaddirStats, vrealpath } from "../../storage/vfs.ts";
import { hostPathOf, loadInstances } from "../acquisition-organize/services/common.ts";
import { hnrSatisfied } from "../acquisition-organize/services/retention.ts";
import {
@@ -29,6 +31,7 @@ import {
probeRoot,
type RootDef,
relOf,
remoteFilesError,
resolveIn,
within,
} from "./common.ts";
@@ -43,7 +46,11 @@ export async function listRoots(d: FmDeps): Promise<{ items: FileRoot[] }> {
const h = await probeRoot(def, d.options.healthTimeoutMs);
let writable = false;
let usage: FileRoot["usage"] = null;
if (h.real) {
if (h.real && def.storageId) {
writable = await isWritableTarget(def, h.real);
const u = await storageUsage(def.storageId, def.hostPath);
if (u) usage = { totalBytes: u.total, freeBytes: u.free };
} else if (h.real) {
writable = await isWritableTarget(def, h.real);
const du = await d.ao.ports.diskUsage(h.real).catch(() => null);
if (du?.status === "ok") usage = { totalBytes: du.totalBytes, freeBytes: du.availableBytes };
@@ -59,6 +66,13 @@ export async function listRoots(d: FmDeps): Promise<{ items: FileRoot[] }> {
checkedAt: h.checkedAt.toISOString(),
healthReason: h.reason,
usage,
storage: def.storageId
? {
id: def.storageId,
name: def.storageName ?? def.name,
path: def.hostPath.replace(/^\/@alist\/[0-9a-f-]{36}/i, "") || "/",
}
: null,
...(def.kind === "download" && h.real && h.health !== "offline"
? await pendingOf(d, def.id, h.real)
: { pendingCount: null, pendingPartial: false }),
@@ -68,6 +82,23 @@ export async function listRoots(d: FmDeps): Promise<{ items: FileRoot[] }> {
return { items };
}
/** Alist 容量(拿不到为 null,不报错;缓存 5 分钟,避免每次列根都打 admin 接口) */
const usageCache = new Map<string, { at: number; v: { total: number; free: number } | null }>();
async function storageUsage(storageId: string, hostPath: string) {
const hit = usageCache.get(hostPath);
if (hit && Date.now() - hit.at < 5 * 60_000) return hit.v;
let v: { total: number; free: number } | null = null;
try {
const e = await storageEntry(storageId);
const remote = hostPath.replace(/^\/@alist\/[0-9a-f-]{36}/i, "") || "/";
v = await e.client.usage(remote);
} catch {
v = null;
}
usageCache.set(hostPath, { at: Date.now(), v });
return v;
}
// ── G6:下载目录待识别计数(缓存 5 分钟;文件变更事件使对应根失效) ──
const PENDING_TTL_MS = 5 * 60_000;
@@ -121,6 +152,7 @@ async function canWrite(p: string): Promise<boolean> {
}
export async function isWritableTarget(def: RootDef, rootReal: string): Promise<boolean> {
if (def.storageId) return (def.kind === "library" || def.kind === "storage") && def.modifiable;
return def.kind === "library" && def.modifiable && (await canWrite(rootReal));
}
@@ -144,7 +176,7 @@ export function kindOfName(name: string): FileKindTag {
async function rawOf(abs: string, rel: string, rootReal: string): Promise<RawEntry | null> {
let st: Awaited<ReturnType<typeof lstat>>;
try {
st = await lstat(abs);
st = (isRemotePath(abs) ? await vlstat(abs) : await lstat(abs)) as Awaited<ReturnType<typeof lstat>>;
} catch (e) {
if (isOfflineErr(e)) throw new FilesError("root_unavailable", `读取失败(${errnoOf(e)})`);
return null; // 读取期间被删除
@@ -436,6 +468,7 @@ const matchFilter = (e: FileEntry, f: FileFilter) => {
export async function listDir(d: FmDeps, q: FilesListQuery): Promise<FilesListResponse> {
const r = await resolveIn(d, q.root, q.path);
if (r.def.storageId) return listRemoteDir(d, q, r);
if (r.isSymlink || !r.lst?.isDirectory()) {
const st = await stat(r.abs).catch(() => null);
if (!st?.isDirectory()) throw new FilesError("bad_request", `不是目录:${r.rel || "/"}`);
@@ -463,13 +496,7 @@ export async function listDir(d: FmDeps, q: FilesListQuery): Promise<FilesListRe
throw new FilesError("bad_request", `读取目录失败:${errnoOf(e) ?? String(e)}`);
}
let pool = names;
if (q.q) {
const needle = q.q;
const re = /[*?]/.test(needle) ? globToRegex(needle) : null;
const lower = needle.toLowerCase();
pool = names.filter((n) => (re ? re.test(n) : n.toLowerCase().includes(lower)));
}
const pool = filterNames(names, q.q);
const raws: RawEntry[] = [];
for (let i = 0; i < pool.length; i += 64) {
const batch = await Promise.all(
@@ -477,6 +504,66 @@ export async function listDir(d: FmDeps, q: FilesListQuery): Promise<FilesListRe
);
for (const b of batch) if (b) raws.push(b);
}
return finishList(d, q, r, raws, dirMtime, cursor, truncated);
}
function filterNames(names: string[], needle: string | undefined): string[] {
if (!needle) return names;
const re = /[*?]/.test(needle) ? globToRegex(needle) : null;
const lower = needle.toLowerCase();
return names.filter((n) => (re ? re.test(n) : n.toLowerCase().includes(lower)));
}
/** Alist 目录:一次 list 拿全(自动翻页),refresh 只在用户点刷新时带(§9-8);不做递归目录摘要(避免逐目录打网盘接口) */
async function listRemoteDir(
d: FmDeps,
q: FilesListQuery,
r: Awaited<ReturnType<typeof resolveIn>>,
): Promise<FilesListResponse> {
if (!r.lst?.isDirectory()) throw new FilesError("bad_request", `不是目录:${r.rel || "/"}`);
const cursor = decodeCursor(q.cursor);
const dirMtime = Number(r.lst.mtimeMs) || 0;
let items: Awaited<ReturnType<typeof vreaddirStats>>;
try {
items = await vreaddirStats(r.abs, { refresh: q.refresh === true });
} catch (e) {
throw remoteFilesError(e, r.def);
}
const visible = items.filter((it) => q.hidden || !isHidden(it.name, r.def));
const truncated = visible.length > d.options.listMaxEntries;
const keep = new Set(
filterNames(
visible.slice(0, d.options.listMaxEntries).map((it) => it.name),
q.q,
),
);
const raws: RawEntry[] = visible
.filter((it) => keep.has(it.name))
.map((it) => {
const abs = join(r.abs, it.name);
const dir = it.stats.isDirectory();
return {
name: it.name,
rel: relOf(r.rootReal, abs),
abs,
type: dir ? "dir" : "file",
size: dir ? 0 : Number(it.stats.size),
mtimeMs: Number(it.stats.mtimeMs) || null,
escape: false,
} satisfies RawEntry;
});
return finishList(d, q, r, raws, dirMtime, cursor, truncated);
}
async function finishList(
d: FmDeps,
q: FilesListQuery,
r: Awaited<ReturnType<typeof resolveIn>>,
raws: RawEntry[],
dirMtime: number,
cursor: { o: number; m: number } | null,
truncated: boolean,
): Promise<FilesListResponse> {
const downloads = await managedDownloads(d);
let entries = await annotate(d, r.def, raws, downloads);
if (q.filter?.length) entries = entries.filter((e) => q.filter!.some((f) => matchFilter(e, f)));
@@ -512,7 +599,7 @@ export async function listDir(d: FmDeps, q: FilesListQuery): Promise<FilesListRe
/** 当前页目录的懒摘要:递归数视频(上限 dirSummaryMax 项)+ 一次查询数已登记 */
async function fillDirSummaries(d: FmDeps, def: RootDef, rootReal: string, page: FileEntry[]) {
const dirs = page.filter((e) => e.type === "dir");
if (dirs.length === 0) return;
if (dirs.length === 0 || def.storageId) return;
const reg = new Map<string, number>();
if (def.kind === "library") {
const likes = dirs.map((e) => `${e.path.replace(/[\\%_]/g, (m) => `\\${m}`)}/%`);
@@ -558,7 +645,7 @@ export async function sidecarsOf(abs: string): Promise<string[]> {
const name = basename(abs);
const stem = name.slice(0, name.length - extname(name).length);
if (!stem) return [];
const names = await readdir(dirname(abs)).catch(() => [] as string[]);
const names = await vreaddir(dirname(abs)).catch(() => [] as string[]);
return names.filter((n) => {
if (n === name) return false;
if (!(n.startsWith(`${stem}.`) || n.startsWith(`${stem}-`))) return false;
@@ -602,7 +689,7 @@ export async function statEntry(d: FmDeps, root: string, path: string): Promise<
const [entry] = await annotate(d, r.def, [raw]);
if (entry!.type === "dir") await fillDirSummaries(d, r.def, r.rootReal, [entry!]);
const st = r.lst!;
const real = await realpath(r.abs).catch(() => r.abs);
const real = await vrealpath(r.abs).catch(() => r.abs);
const nlink = Number(st.nlink);
return {
entry: entry!,
+16 -9
View File
@@ -1,9 +1,9 @@
import { lstat, mkdir } from "node:fs/promises";
import { basename, dirname, extname, join } from "node:path";
import { moveSafe } from "@umt/adapters";
import type { FilesMkdirRequest, FilesMkdirResult, FilesRenameRequest, FilesRenameResult } from "@umt/contracts";
import { sanitizeSegment } from "@umt/domain";
import type { Principal } from "../../http/auth.ts";
import { lstat, mkdir } from "../../storage/local-fs.ts";
import { isRemotePath, vexists, vlstat, vmkdirp, vrenameNoClobber } from "../../storage/vfs.ts";
import { actorOfP, errMessage, errnoOf, FilesError, type FmDeps, recheck, relOf, resolveIn } from "./common.ts";
import { audit, filesChanged } from "./ledger.ts";
import { downloadsFor, isWritableTarget, kindOfName, managedDownloads, sidecarsOf } from "./list.ts";
@@ -31,7 +31,10 @@ export async function mkdirIn(
await recheck(r.abs, r.rootReal);
const abs = join(r.abs, body.name);
try {
await mkdir(abs);
if (r.def.storageId) {
if (await vexists(abs)) throw Object.assign(new Error("exists"), { code: "EEXIST" });
await vmkdirp(abs);
} else await mkdir(abs);
} catch (e) {
if (errnoOf(e) === "EEXIST") throw new FilesError("exists", `已存在:${body.name}`);
throw new FilesError("bad_request", `创建目录失败:${errMessage(e)}`);
@@ -88,7 +91,9 @@ export async function renameEntry(
// 不覆盖:任一目标已存在 → 409,原文件不动
const taken: string[] = [];
for (const p of pairs) {
const st = await lstat(p.toAbs).catch((e) => (errnoOf(e) === "ENOENT" ? null : e));
const st = await (isRemotePath(p.toAbs) ? vlstat(p.toAbs) : lstat(p.toAbs)).catch((e) =>
errnoOf(e) === "ENOENT" ? null : e,
);
if (st) taken.push(p.to);
}
if (taken.length > 0) throw new FilesError("exists", `目标已存在,不覆盖:${taken.join(", ")}`);
@@ -116,17 +121,19 @@ export async function renameEntry(
// 执行:逐个 no-clobber 改名;任一失败回滚已改的
const done: Pair[] = [];
const rollback = async () => {
for (const p of done.reverse()) await moveSafe(p.toAbs, p.fromAbs).catch(() => undefined);
for (const p of done.reverse()) await vrenameNoClobber(p.toAbs, p.fromAbs).catch(() => undefined);
};
for (const p of pairs) {
try {
await recheck(p.fromAbs, r.rootReal);
const before = await lstat(p.fromAbs);
const res = await moveSafe(p.fromAbs, p.toAbs);
const remote = isRemotePath(p.fromAbs);
const before = remote ? null : await lstat(p.fromAbs);
const res = await vrenameNoClobber(p.fromAbs, p.toAbs);
if (res.status === "exists") throw new FilesError("exists", `目标已存在,不覆盖:${p.to}`);
if (res.status !== "moved") throw new FilesError("bad_request", `重命名失败(${res.status}): ${p.from}`);
const after = await lstat(p.toAbs);
if (after.ino !== before.ino || after.dev !== before.dev)
// 远端没有 inode:Alist 改名接口成功即算成功(同 MP)
const after = remote ? null : await lstat(p.toAbs);
if (before && after && (after.ino !== before.ino || after.dev !== before.dev))
throw new FilesError("bad_request", `重命名后 inode 不一致,已中止:${p.to}`);
done.push(p);
} catch (e) {
@@ -74,14 +74,19 @@ export async function saveOrganizeSource(d: FmDeps, id: string, input: Omit<Orga
const target = await resolveIn(d, input.target.root, input.target.path);
if (
target.isSymlink ||
target.def.kind !== "library" ||
!target.lst?.isDirectory() ||
!(await isWritableTarget(target.def, target.rootReal)) ||
!(await access(target.abs, constants.W_OK).then(
() => true,
() => false,
))
(!target.def.storageId &&
!(await access(target.abs, constants.W_OK).then(
() => true,
() => false,
)))
)
throw new FilesError("not_writable", "目标必须为可写的纳管媒体库目录");
// Alist 来源(对照 MP 远程目录监控):只能手动或目录快照触发,qB 标签对网盘没有意义
if (source.def.storageId && input.trigger === "qb_tag")
throw new FilesError("bad_request", "Alist 来源只能用手动或目录快照触发");
if (input.trigger === "qb_tag") {
if (!input.downloaderInstanceId || !input.tag?.trim())
throw new FilesError("bad_request", "标签触发需要下载器和标签");
@@ -1,9 +1,9 @@
import { randomUUID } from "node:crypto";
import { stat } from "node:fs/promises";
import { join, relative } from "node:path";
import type { FilesOrganizeGroup, FilesOrganizeRequest, FilesOrganizeResult } from "@umt/contracts";
import type { Principal } from "../../http/auth.ts";
import { ApiException } from "../../http/errors.ts";
import { vlstat as stat } from "../../storage/vfs.ts";
import { ensureTaskRow, markTask } from "../acquisition-organize/services/common.ts";
import { type FileOverrideInput, type FrozenPlan, planImport } from "../acquisition-organize/services/organize.ts";
import {
@@ -4,6 +4,7 @@ import type { FilePreview, FilesPreviewQuery } from "@umt/contracts";
import type { Principal } from "../../http/auth.ts";
import { runSandbox, SandboxError } from "../../sandbox/client.ts";
import { b64encode } from "../../sandbox/jobs.ts";
import { isRemotePath, remote, remoteError } from "../../storage/vfs.ts";
import { FilesError, type FmDeps, isAdminSession, recheck, resolveIn } from "./common.ts";
import { BAD_TORRENT } from "./preview-build.ts";
import { extOf, previewKindOf } from "./preview-parse.ts";
@@ -16,6 +17,20 @@ const IMAGE_HEAD = 512 * 1024;
const TORRENT_MAX = 4 * 1024 * 1024;
async function readHead(abs: string, max: number): Promise<{ buf: Uint8Array; size: number }> {
if (isRemotePath(abs)) {
// Alist:按 Range 只取头部
const { v, client } = await remote(abs);
try {
const o = await client.get(v.remote);
if (o.isDir) throw new FilesError("bad_request", "不是普通文件");
const res = await client.openRead(v.remote, { range: `bytes=0-${Math.max(0, max - 1)}` });
const buf = new Uint8Array(await res.arrayBuffer()).subarray(0, max);
return { buf, size: o.size };
} catch (e) {
if (e instanceof FilesError) throw e;
throw remoteError(e, abs);
}
}
const fh = await open(abs, "r");
try {
const st = await fh.stat();
@@ -1,5 +1,4 @@
import { createHash, randomUUID } from "node:crypto";
import { mkdir, readFile, stat, writeFile } from "node:fs/promises";
import { join, resolve } from "node:path";
import type { ProbeResult } from "@umt/adapters";
import type {
@@ -11,6 +10,8 @@ import type {
} from "@umt/contracts";
import { sql } from "@umt/db";
import type { Principal } from "../../http/auth.ts";
import { mkdir, readFile, writeFile } from "../../storage/local-fs.ts";
import { vlstat } from "../../storage/vfs.ts";
import { FilesError, type FmDeps, jsonb, resolveIn } from "./common.ts";
import { audit } from "./ledger.ts";
import { kindOfName, mediaSummary } from "./list.ts";
@@ -54,7 +55,7 @@ export async function probeFiles(d: FmDeps, body: FilesProbeRequest): Promise<Fi
});
continue;
}
const st = await stat(r.abs);
const st = await vlstat(r.abs);
const key = createHash("sha1").update(`${r.abs}|${st.size}|${st.mtimeMs}`).digest("hex");
const cacheFile = join(cacheDir, `${key}.json`);
let result: ProbeResult | null = null;
@@ -1,5 +1,4 @@
import { randomUUID } from "node:crypto";
import { lstat } from "node:fs/promises";
import { basename, isAbsolute, join } from "node:path";
import type {
FileRef,
@@ -9,6 +8,7 @@ import type {
OrganizationDeleteResult,
} from "@umt/contracts";
import type { Principal } from "../../http/auth.ts";
import { vlstat as lstat } from "../../storage/vfs.ts";
import { FilesError, type FmDeps, isAdmin, loadRootDefs, type RootDef, within } from "./common.ts";
import { audit } from "./ledger.ts";
import { deleteFiles } from "./trash.ts";
@@ -23,6 +23,7 @@ import { suggestName } from "./suggest-name.ts";
import { transfer } from "./transfer.ts";
import { deleteFiles, listTrash, purgeTrash, restoreTrash } from "./trash.ts";
import { recordSources, treeOf } from "./tree.ts";
import { uploadFile } from "./upload.ts";
// /files 路由:scope 由 contracts 端点定义(读 organize.read / 写 organize.execute),处理器内再按授权矩阵细分。
// FilesError → 信封 + 对应状态(根离线 503,message 以 `<reason>: ` 开头)。
@@ -120,6 +121,13 @@ export function registerFileRoutes(app: ModuleApp, d: FmDeps) {
mount(app, F.probe, async (_c, { body }) => probeFiles(d, body));
mount(app, F.records, async (_c, { query }) => findRecords(d, query));
mount(app, F.download, async (c, { query }) => downloadFile(d, c.get("principal"), query), true);
// 上传:请求体是文件原始字节(不走 JSON 校验),query 照常校验;幂等键防重放
mount(app, F.upload, async (c, { query }) => {
const { idempotencyKey, ...rest } = query as { idempotencyKey: string } & Record<string, unknown>;
return idempotent(d, c.get("principal"), "upload", idempotencyKey, rest, () =>
uploadFile(d, c.get("principal"), query, c.req.raw.body, c.req.raw.signal),
);
});
mount(app, F.preview, async (c, { query }) => previewFile(d, c.get("principal"), query));
// W9-H 文件管理 v2
mount(app, F.entityFromCandidate, async (c, { body }) => entityFromCandidate(d, c.get("principal"), body));
@@ -3,6 +3,8 @@ import { basename, join } from "node:path";
import { isDownloadComplete } from "@umt/adapters";
import type { FileRef, OrganizeSourceConfig } from "@umt/contracts";
import { sql } from "@umt/db";
import { storageEntry } from "../../storage/registry.ts";
import { vreaddirStats } from "../../storage/vfs.ts";
import { hostPathOf, loadInstances, upsertDecision } from "../acquisition-organize/services/common.ts";
import { planImport } from "../acquisition-organize/services/organize.ts";
import { downloadTasksUnder } from "../acquisition-organize/services/takeover.ts";
@@ -15,6 +17,7 @@ import {
jsonb,
recheck,
relOf,
remoteFilesError,
resolveIn,
rootDef,
within,
@@ -46,11 +49,17 @@ export function advanceSnapshot(previous: Snapshot | null, observed: Record<stri
return { snapshot: { files }, eligible };
}
export async function walkSource(d: FmDeps, config: OrganizeSourceConfig, path = config.source.path) {
export async function walkSource(
d: FmDeps,
config: OrganizeSourceConfig,
path = config.source.path,
prev?: { files: Record<string, Omit<SnapshotFile, "accepted">>; since: number | null },
) {
if (!within(path, config.source.path) && config.source.path !== "")
throw new FilesError("path_escape", "扫描路径必须在来源目录内");
const start = await resolveIn(d, config.source.root, path);
if (start.isSymlink || !start.lst?.isDirectory()) throw new FilesError("bad_request", "来源必须为目录");
if (start.def.storageId) return walkRemoteSource(start, prev);
const files: Record<string, Omit<SnapshotFile, "accepted">> = {};
const stack = [start.abs];
let seen = 0;
@@ -78,6 +87,59 @@ export async function walkSource(d: FmDeps, config: OrganizeSourceConfig, path =
return { files, truncated, rootReal: start.rootReal };
}
/**
* Alist 来源快照(对照 MP StorageBase.snapshot,docs/design/storage-alist.md §8):BFS 列目录,深度 ≤ snapshotMaxDepth;
* 开了「按目录修改时间跳过」时,子目录修改时间不晚于上次扫描就沿用上次快照里该子树的文件(根目录每轮都列,用于发现删除);
* 子目录列表失败保留旧快照(MP「查询异常时保留基线」);根目录列表失败抛错(记 last_error,保留旧快照)。
* 请求经实例限速器(QPS / 并发),不会打爆网盘。
*/
async function walkRemoteSource(
start: Awaited<ReturnType<typeof resolveIn>>,
prev?: { files: Record<string, Omit<SnapshotFile, "accepted">>; since: number | null },
) {
const opts = (await storageEntry(start.def.storageId!)).options;
const files: Record<string, Omit<SnapshotFile, "accepted">> = {};
const carry = (relDir: string) => {
for (const [k, v] of Object.entries(prev?.files ?? {}))
if (k === relDir || k.startsWith(`${relDir}/`)) files[k] = { size: v.size, mtimeMs: v.mtimeMs };
};
let seen = 0;
let truncated = false;
const queue: { abs: string; depth: number }[] = [{ abs: start.abs, depth: 0 }];
while (queue.length && !truncated) {
const { abs, depth } = queue.shift()!;
let items: Awaited<ReturnType<typeof vreaddirStats>>;
try {
items = await vreaddirStats(abs);
} catch (e) {
if (depth === 0) throw remoteFilesError(e, start.def);
carry(relOf(start.rootReal, abs));
continue;
}
for (const it of items) {
if (++seen > SOURCE_SCAN_LIMIT) {
truncated = true;
break;
}
if (it.name.startsWith(".") || isTempName(it.name)) continue;
const child = join(abs, it.name);
const rel = relOf(start.rootReal, child);
if (it.stats.isDirectory()) {
if (depth + 1 >= opts.snapshotMaxDepth) continue;
const unchanged =
opts.snapshotCheckFolderModtime &&
prev?.since != null &&
it.stats.mtimeMs > 0 &&
it.stats.mtimeMs <= prev.since;
if (unchanged) carry(rel);
else queue.push({ abs: child, depth: depth + 1 });
} else if (kindOfName(it.name) === "video" && Number(it.stats.size) > 0)
files[rel] = { size: Number(it.stats.size), mtimeMs: Number(it.stats.mtimeMs) };
}
}
return { files, truncated, rootReal: start.rootReal };
}
/** Query unfiltered QB state once: an untagged download may never have a DB task row. */
async function activeDownloaderPaths(d: FmDeps, config: OrganizeSourceConfig): Promise<string[]> {
const source = await rootDef(d.ctx.db, config.source.root);
@@ -220,7 +282,10 @@ export async function scanOrganizeSource(d: FmDeps, input: { sourceId: string })
const now = new Date();
try {
const liveActive = await activeDownloaderPaths(d, config);
const scan = await walkSource(d, config);
const scan = await walkSource(d, config, config.source.path, {
files: row?.baseline_at ? ((row.snapshot as unknown as Snapshot).files ?? {}) : {},
since: row?.baseline_at && row.last_scan_at ? new Date(row.last_scan_at).getTime() : null,
});
if (scan.truncated) throw new FilesError("too_large", "来源超过单次扫描上限,请选择更小的目录");
const baseline = !row?.baseline_at;
const advance = advanceSnapshot(baseline ? null : (row!.snapshot as unknown as Snapshot), scan.files);
@@ -1,7 +1,7 @@
import { lstat, readdir, stat } from "node:fs/promises";
import { basename, dirname, join } from "node:path";
import type { FilePlanPair, FilesMoveRequest, FilesTransferResult, OrganizationStatus } from "@umt/contracts";
import type { Principal } from "../../http/auth.ts";
import { isRemotePath, vlstat as lstat, vreaddir as readdir } from "../../storage/vfs.ts";
import { ensureTaskRow, markTask } from "../acquisition-organize/services/common.ts";
import { planImport } from "../acquisition-organize/services/organize.ts";
import { authorizeMoveEntities, PathGuardError, sourceGuard } from "../acquisition-organize/services/organize-guard.ts";
@@ -94,8 +94,10 @@ export async function transfer(
const mode = body.mode;
const t = await resolveIn(d, body.target.root, body.target.path);
if (!t.lst?.isDirectory() || t.isSymlink) throw new FilesError("bad_request", "目标必须是已存在的目录");
if (!t.def.libraryRootId || !(await isWritableTarget(t.def, t.rootReal)))
throw new FilesError("not_writable", `目标根 ${t.def.name} 不可写,只能选 可写的媒体库目录`);
// 目标:可写的媒体库目录(本地或 Alist),或 Alist 存储根(文件管理的 st: 根,对照 MP 存储内移动 / 复制)
const storageTarget = t.def.kind === "storage";
if ((!t.def.libraryRootId && !storageTarget) || !(await isWritableTarget(t.def, t.rootReal)))
throw new FilesError("not_writable", `目标根 ${t.def.name} 不可写,只能选 可写的媒体库目录或 Alist 存储`);
const sources: Resolved[] = [];
for (const it of body.items) {
@@ -118,7 +120,7 @@ export async function transfer(
for (const h of downloadsFor(downloads, s.abs))
hnrBlocks.push({ taskId: h.taskId, path: s.rel, hnr: h.hnr, status: h.status });
const targetDev = Number((await stat(t.abs)).dev);
const targetDev = Number((await lstat(t.abs)).dev);
const spec: TransferSpec = {
items: sources.map((s, i) => {
const dst = join(t.abs, names[i]!);
@@ -164,6 +166,11 @@ export async function transfer(
? (await relocatePreview(d.ctx.db, s.def.libraryRootId, s.rel, snaps[i]!.isDir)).locations > 0
: false;
const crossDevice = Number((await lstat(s.abs)).dev) !== targetDev;
if (storageTarget && mode === "move" && registered)
throw new FilesError(
"needs_retire",
`${s.rel} 含已整理的文件,只能移动到媒体库目录;要放到 Alist 存储请用复制或先从媒体库移除`,
);
pairs.push({
source: { root: body.items[i]!.root, path: s.rel },
target: { root: body.target.root, path: relOf(t.rootReal, dst) },
@@ -175,7 +182,7 @@ export async function transfer(
});
if (!(mode === "move" && !crossDevice)) needBytes += snaps[i]!.size;
}
const du = await d.ao.ports.diskUsage(t.abs).catch(() => null);
const du = isRemotePath(t.abs) ? null : await d.ao.ports.diskUsage(t.abs).catch(() => null);
const freeBytes = du?.status === "ok" ? du.availableBytes : null;
const spaceOk = freeBytes === null ? true : needBytes <= freeBytes;
const truncated = snaps.some((s) => s.truncated) || planned.truncated === true;
@@ -184,10 +191,14 @@ export async function transfer(
const request = {
sourcePaths: sources.map((s) => s.abs),
action: mode,
targetLibraryRootId: t.def.libraryRootId,
...(t.def.libraryRootId ? { targetLibraryRootId: t.def.libraryRootId } : {}),
};
const allowCross = (await d.ctx.settings.get("organize")).allowCrossVolumeMove;
const crossBlocked = mode === "move" && !allowCross && pairs.some((p) => p.crossDevice);
// 跨卷开关只管本地 ↔ 本地;涉及 Alist 的移动一律「传输成功 → 删源」(同 MP)
const crossBlocked =
mode === "move" &&
!allowCross &&
pairs.some((p, i) => p.crossDevice && !isRemotePath(sources[i]!.abs) && !isRemotePath(t.abs));
const warnings: string[] = [];
if (crossBlocked) warnings.push("未开启跨卷移动,执行会失败。改用复制,或在 设置 → 整理 中开启");
if (conflictSrc.size > 0) warnings.push(`${conflictSrc.size} 个文件在目标已存在,不覆盖,执行时记为失败`);
+91 -8
View File
@@ -1,5 +1,4 @@
import { randomUUID } from "node:crypto";
import { lstat, mkdir, readdir, realpath, rm } from "node:fs/promises";
import { basename, dirname, join, resolve } from "node:path";
import { moveSafe } from "@umt/adapters";
import type {
@@ -16,6 +15,9 @@ import type {
} from "@umt/contracts";
import { sql, type TrashItemRow } from "@umt/db";
import type { Principal } from "../../http/auth.ts";
import { lstat, mkdir, readdir, realpath, rm } from "../../storage/local-fs.ts";
import { isRemotePath, vlstat, vmkdirp, vreaddirStats, vremove, vrenameNoClobber } from "../../storage/vfs.ts";
import { parseVPath, toVPath } from "../../storage/vpath.ts";
import { upsertDecision } from "../acquisition-organize/services/common.ts";
import { fsRuntime } from "../acquisition-organize/services/organize.ts";
import {
@@ -47,6 +49,7 @@ const likePrefix = (rel: string) => `${rel.replace(/[\\%_]/g, (m) => `\\${m}`)}/
const day = (t: Date) => t.toISOString().slice(0, 10).replace(/-/g, "");
async function sizeOf(abs: string, cap = 200_000): Promise<{ size: number; files: string[] }> {
if (isRemotePath(abs)) return remoteSizeOf(abs, cap);
const st = await lstat(abs);
if (!st.isDirectory()) return { size: Number(st.size), files: [""] };
let size = 0;
@@ -68,6 +71,39 @@ async function sizeOf(abs: string, cap = 200_000): Promise<{ size: number; files
return { size, files };
}
/** 远端:按目录 list(每个子目录一次请求) */
async function remoteSizeOf(abs: string, cap: number): Promise<{ size: number; files: string[] }> {
const st = await vlstat(abs);
if (!st.isDirectory()) return { size: Number(st.size), files: [""] };
let size = 0;
const files: string[] = [];
const walk = async (dir: string, prefix: string) => {
for (const it of await vreaddirStats(dir)) {
if (files.length >= cap) return;
const rel = prefix ? `${prefix}/${it.name}` : it.name;
if (it.stats.isDirectory()) await walk(join(dir, it.name), rel);
else {
size += Number(it.stats.size);
files.push(rel);
}
}
};
await walk(abs, "");
return { size, files };
}
/**
* 回收站所在目录:本地 = <根>/.umt-trash;远端同布局,用 Alist 同驱动 move(立即完成)。
* 远端根是 Alist 的 "/"(各挂载的上一级,不能建目录)时放进该项所在挂载的 .umt-trash。
*/
export function trashBase(rootReal: string, abs: string): string {
const v = parseVPath(rootReal);
if (v?.remote !== "/") return rootReal;
const a = parseVPath(abs);
const mount = a?.remote.split("/").filter(Boolean)[0];
return mount ? toVPath(v.storageId, `/${mount}`) : rootReal;
}
interface CheckCtx {
abs: string | null;
rootReal: string | null;
@@ -234,7 +270,8 @@ export async function deleteFiles(
await recheck(abs, ctx.rootReal!);
let where: string | null = null;
if (body.mode === "permanent") {
await rm(abs, { recursive: true, force: false });
if (isRemotePath(abs)) await vremove(abs);
else await rm(abs, { recursive: true, force: false });
deleted.push(check.path);
} else {
const row = await moveToTrash(d, principal, check, ctx, expiresAt);
@@ -379,6 +416,16 @@ async function moveToTrash(
const abs = ctx.abs!;
const id = randomUUID();
const stamp = day(d.ao.ports.now());
if (isRemotePath(abs)) {
// Alist:同存储 move 进 <根>/.umt-trash(MP 无回收站;Alist 驱动自己的回收站不依赖)
const base = trashBase(ctx.rootReal!, abs);
const rel = base === ctx.rootReal ? check.path : relOfRoot(base, abs);
const trashPath = join(base, TRASH_DIR, stamp, id, rel);
await vmkdirp(dirname(trashPath));
const r = await vrenameNoClobber(abs, trashPath);
if (r.status !== "moved") throw new FilesError("exists", `移入回收站失败(${r.status}):${trashPath}`);
return insertTrashRow(d, principal, check, ctx, { id, rootHost: base, trashPath, expiresAt });
}
let trashPath = join(ctx.rootReal!, TRASH_DIR, stamp, id, check.path);
let moved = false;
try {
@@ -404,13 +451,24 @@ async function moveToTrash(
await copyTree(d, abs, trashPath, files);
await rm(abs, { recursive: true });
}
return insertTrashRow(d, principal, check, ctx, { id, rootHost: ctx.rootReal!, trashPath, expiresAt });
}
function insertTrashRow(
d: FmDeps,
principal: Principal | undefined,
check: FileDeleteCheck,
ctx: CheckCtx,
x: { id: string; rootHost: string; trashPath: string; expiresAt: Date },
): Promise<TrashItemRow> {
const { id, trashPath, expiresAt } = x;
return d.ctx.db
.insertInto("trash_item")
.values({
id,
root_id: check.root,
library_root_id: ctx.libraryRootId,
root_host_path: ctx.rootReal!,
root_host_path: x.rootHost,
original_path: check.path,
trash_path: trashPath,
is_dir: check.isDir,
@@ -477,6 +535,11 @@ function trashPathSafe(d: FmDeps, row: TrashItemRow): boolean {
*/
export async function mkdirWithin(rootReal: string, rel: string): Promise<void> {
if (rel === "" || rel === ".") return;
if (isRemotePath(rootReal)) {
if (rel.split("/").includes("..")) throw new FilesError("path_escape", `路径含 ..: ${rel}`);
await vmkdirp(join(rootReal, rel));
return;
}
let cur = rootReal;
for (const seg of rel.split("/")) {
if (seg === "" || seg === ".") continue;
@@ -494,6 +557,13 @@ export async function mkdirWithin(rootReal: string, rel: string): Promise<void>
}
}
const relOfRoot = (root: string, p: string) => (p === root ? "" : p.slice(root.length + 1));
const sameStorageOf = (a: string, b: string) => parseVPath(a)?.storageId === parseVPath(b)?.storageId;
/** <…>/.umt-trash/<yyyymmdd>/<uuid>:回收站路径里 uuid 那一级 */
function trashUuidDir(row: TrashItemRow): string | null {
const segs = row.trash_path.split("/");
const i = segs.indexOf(row.id);
return i > 0 ? segs.slice(0, i + 1).join("/") : null;
}
export async function restoreTrash(
d: FmDeps,
@@ -537,8 +607,14 @@ export async function restoreTrash(
);
dst = parent.abs;
}
if (await lstat(dst).catch(() => null)) throw new FilesError("exists", "原位置已被占用,可选择「还原到…」");
const r = await moveSafe(row.trash_path, dst);
if (await (isRemotePath(dst) ? vlstat(dst) : lstat(dst)).catch(() => null))
throw new FilesError("exists", "原位置已被占用,可选择「还原到…」");
if (
isRemotePath(dst) !== isRemotePath(row.trash_path) ||
(isRemotePath(dst) && !sameStorageOf(dst, row.trash_path))
)
throw new FilesError("bad_request", "回收站里的项只能还原到它原来所在的存储");
const r = await vrenameNoClobber(row.trash_path, dst);
if (r.status === "exists") throw new FilesError("exists", "原位置已被占用");
if (r.status === "exdev") {
// 跨卷还原:与跨卷进回收站同一上限(FILES_TRASH_CROSS_MAX_BYTES),复制经 fsproxy,核验后删回收站副本
@@ -601,10 +677,17 @@ export async function purgeTrash(
for (const row of rows) {
try {
if (!trashPathSafe(d, row)) throw new FilesError("path_escape", "回收站路径异常,拒绝清除");
await rm(row.trash_path, { recursive: true, force: true });
// 顺带清掉空的 <uuid>/<yyyymmdd> 目录
const uuidDir = row.trash_path.slice(0, row.trash_path.length - row.original_path.length - 1);
await rm(uuidDir, { recursive: true, force: true }).catch(() => undefined);
const uuidDir = trashUuidDir(row);
if (isRemotePath(row.trash_path)) {
await vremove(row.trash_path).catch((e) => {
if (errnoOf(e) !== "ENOENT") throw e;
});
if (uuidDir) await vremove(uuidDir).catch(() => undefined);
} else {
await rm(row.trash_path, { recursive: true, force: true });
if (uuidDir) await rm(uuidDir, { recursive: true, force: true }).catch(() => undefined);
}
const gone = await d.ctx.db.deleteFrom("trash_item").where("id", "=", row.id).returning("id").executeTakeFirst();
if (gone) purged.push(row.id);
} catch (e) {
+13 -5
View File
@@ -1,4 +1,3 @@
import { lstat, readdir } from "node:fs/promises";
import { join } from "node:path";
import type {
FileRootId,
@@ -8,6 +7,7 @@ import type {
OrganizationStatus,
} from "@umt/contracts";
import { sql } from "@umt/db";
import { isRemotePath, vlstat as lstat, vreaddirents as readdirents } from "../../storage/vfs.ts";
import { FilesError, type FmDeps, isTempName, loadRootDefs, resolveIn, within } from "./common.ts";
import { kindOfName } from "./list.ts";
@@ -20,7 +20,8 @@ const hiddenName = (n: string) => n.startsWith(".") || isTempName(n);
export async function treeOf(d: FmDeps, q: FilesTreeQuery): Promise<FilesTreeResult> {
const r = await resolveIn(d, q.root, q.path);
if (!r.lst?.isDirectory() || r.isSymlink) throw new FilesError("bad_request", `不是目录:${r.rel || "/"}`);
const names = await readdir(r.abs, { withFileTypes: true });
const remote = isRemotePath(r.abs);
const names = await readdirents(r.abs);
const subdirs = names
.filter((n) => n.isDirectory() && !hiddenName(n.name))
.map((n) => n.name)
@@ -30,7 +31,9 @@ export async function treeOf(d: FmDeps, q: FilesTreeQuery): Promise<FilesTreeRes
const dirs = await Promise.all(
subdirs.slice(0, max).map(async (name) => {
const path = r.rel ? `${r.rel}/${name}` : name;
const kids = await readdir(join(r.abs, name), { withFileTypes: true }).catch(() => []);
// Alist:不逐个子目录再列一遍(避免 N 次网盘请求),按「可能有下级」展示,展开时再列
if (remote) return { name, path, hasChildren: true };
const kids = await readdirents(join(r.abs, name)).catch(() => []);
return { name, path, hasChildren: kids.some((k) => k.isDirectory() && !hiddenName(k.name)) };
}),
);
@@ -39,10 +42,15 @@ export async function treeOf(d: FmDeps, q: FilesTreeQuery): Promise<FilesTreeRes
let seen = 0;
let videoCount = 0;
let partial = false;
const stack = [r.abs];
// Alist:只数当前目录这一层(不递归打网盘接口),结果标为不完整
if (remote) {
videoCount = names.filter((n) => n.isFile() && kindOfName(n.name) === "video").length;
partial = names.some((n) => n.isDirectory());
}
const stack = remote ? [] : [r.abs];
while (stack.length > 0 && !partial) {
const dir = stack.pop()!;
for (const n of await readdir(dir, { withFileTypes: true }).catch(() => [])) {
for (const n of await readdirents(dir).catch(() => [])) {
if (++seen > d.options.dirSummaryMax) {
partial = true;
break;
@@ -0,0 +1,83 @@
import { randomUUID } from "node:crypto";
import { join } from "node:path";
import { commitNoClobber, partialPathFor } from "@umt/adapters";
import type { FilesUploadQuery, FilesUploadResult } from "@umt/contracts";
import { sanitizeSegment } from "@umt/domain";
import type { Principal } from "../../http/auth.ts";
import { mkdir, open, rm } from "../../storage/local-fs.ts";
import { relayDir, remote, remoteError, vexists } from "../../storage/vfs.ts";
import { errnoOf, FilesError, type FmDeps, recheck, relOf, resolveIn } from "./common.ts";
import { audit, filesChanged } from "./ledger.ts";
import { isWritableTarget } from "./list.ts";
// PUT /files/upload(docs/design/storage-alist.md §4):请求体是文件字节,放进 {root, path} 目录,同名不覆盖。
// Alist 根:先流式写本机临时文件,再按 §2.6 上传(临时名 → 改名,成功判定同 MP);本地根:同目录 partial → 不覆盖改名。
async function spool(body: ReadableStream<Uint8Array>, file: string, signal?: AbortSignal): Promise<number> {
const fh = await open(file, "w");
let n = 0;
try {
const reader = body.getReader();
for (;;) {
signal?.throwIfAborted();
const { value, done } = await reader.read();
if (done) break;
await fh.write(value);
n += value.length;
}
} finally {
await fh.close();
}
return n;
}
export async function uploadFile(
d: FmDeps,
principal: Principal | undefined,
q: FilesUploadQuery,
body: ReadableStream<Uint8Array> | null,
signal?: AbortSignal,
): Promise<FilesUploadResult> {
if (!body) throw new FilesError("bad_request", "请求体为空");
const r = await resolveIn(d, q.root, q.path);
if (!r.lst?.isDirectory() || r.isSymlink) throw new FilesError("bad_request", `不是目录:${r.rel || "/"}`);
if (!(await isWritableTarget(r.def, r.rootReal))) throw new FilesError("not_writable", `根 ${r.def.name} 不可写`);
if (sanitizeSegment(q.name) !== q.name)
throw new FilesError("bad_request", `文件名含非法字符或首尾有空白/点,可改为:${sanitizeSegment(q.name)}`);
await recheck(r.abs, r.rootReal);
const dst = join(r.abs, q.name);
const rel = relOf(r.rootReal, dst);
let size: number;
if (r.def.storageId) {
if (await vexists(dst).catch((e) => Promise.reject(remoteError(e, dst))))
throw new FilesError("exists", `已存在,不覆盖:${q.name}`);
const dir = relayDir();
await mkdir(dir, { recursive: true });
const tmp = join(dir, `${randomUUID()}-upload`);
try {
size = await spool(body, tmp, signal);
const { v, client } = await remote(dst);
await client.upload(tmp, v.remote, signal ? { signal } : {});
} catch (e) {
if (e instanceof FilesError) throw e;
const err = remoteError(e, dst);
if (errnoOf(err) === "EEXIST") throw new FilesError("exists", `已存在,不覆盖:${q.name}`);
throw new FilesError("upstream", `上传到 Alist 失败:${err.message}`);
} finally {
await rm(tmp, { force: true }).catch(() => undefined);
}
} else {
const partial = partialPathFor(dst);
try {
size = await spool(body, partial, signal);
await commitNoClobber(partial, dst);
} catch (e) {
await rm(partial, { force: true }).catch(() => undefined);
if (errnoOf(e) === "EEXIST") throw new FilesError("exists", `已存在,不覆盖:${q.name}`);
throw e;
}
}
await audit(d, principal, "upload", { root: q.root, path: rel, size });
filesChanged(d, q.root, [r.rel]);
return { path: rel, size };
}
+9 -1
View File
@@ -1,4 +1,3 @@
import { copyFile, mkdir, readdir, rename, rm, stat, writeFile } from "node:fs/promises";
import { dirname, join } from "node:path";
import {
lifecycleKey,
@@ -26,6 +25,8 @@ import type { Selectable } from "kysely";
import { dataPaths } from "../../config.ts";
import { ApiException } from "../../http/errors.ts";
import { runSandbox, SandboxError } from "../../sandbox/client.ts";
import { copyFile, mkdir, readdir, rename, rm, stat, writeFile } from "../../storage/local-fs.ts";
import { isRemotePath } from "../../storage/vpath.ts";
import type { ServerContext } from "../types.ts";
// 补字幕(docs/plan/subtitles.md;对照 MP 插件「海拉鲁字幕大师」upload / matching / subtitle_writer):
@@ -199,6 +200,11 @@ export async function applySubtitles(
failed.push({ name: it.name, reason: "目标视频不在这部作品的媒体库里" });
continue;
}
// Alist 上的媒体库:补字幕暂不支持(MP 的 Alist 能力里也没有),明确失败,不往本机写
if (isRemotePath(t.rootPath)) {
failed.push({ name: it.name, reason: "目标视频在 Alist 存储上,暂不支持补字幕" });
continue;
}
const name = subtitleFileName(t.relativePath, it.language, it.ext as SubtitleExt, defaultLang);
const rel = join(dirname(t.relativePath), name);
if (planned.has(rel)) {
@@ -322,6 +328,8 @@ export async function undoSubtitleWrite(ctx: ServerContext, id: string): Promise
.select("local_path")
.where("id", "=", row.library_root_id)
.executeTakeFirstOrThrow();
if (isRemotePath(root.local_path))
throw new ApiException("precondition_failed", "字幕在 Alist 存储上,不能在这里撤销");
const abs = join(root.local_path, row.relative_path);
if (row.backup_path && (await exists(row.backup_path))) {
await copyFile(row.backup_path, `${abs}.umt-restore.tmp`);
+43 -4
View File
@@ -10,6 +10,9 @@ import {
SystemWorkerLane,
} from "@umt/contracts";
import { type DB, type Kysely, type Selectable, sql } from "@umt/db";
import { storageEntry } from "../../storage/registry.ts";
import { isRemotePath, vcheckRoot } from "../../storage/vfs.ts";
import { toVPath } from "../../storage/vpath.ts";
import { CURRENT_HOST_WINDOW_MS, HEARTBEAT_ROW_RETENTION_MS, HEARTBEAT_STALE_MS } from "./heartbeat.ts";
// GET /system/health 各段的取数(W9-I)。每段独立 try/catch:一段失败只把该段标 unknown,不让整页 500。
@@ -214,7 +217,14 @@ export async function loadPathTargets(db: Kysely<DB>) {
.where("path_mapping.view_kind", "=", "downloader")
.orderBy("path_mapping.local_prefix")
.execute();
return { roots, maps };
const storages = await db
.selectFrom("storage_instance")
.select(["id", "name", "root_path"])
.where("enabled", "=", true)
.orderBy("name")
.execute()
.catch(() => [] as { id: string; name: string; root_path: string }[]);
return { roots, maps, storages };
}
export function diskLevel(ratio: number, t: HealthDeps["thresholds"]): HealthLevel {
@@ -226,11 +236,14 @@ export async function loadDisksAndMounts(d: HealthDeps): Promise<{ disks: System
const usage = d.diskUsage ?? diskUsage;
const check = d.checkRoot ?? checkRoot;
const ms = d.ioTimeoutMs ?? 3000;
const { roots, maps } = await loadPathTargets(d.db);
const { roots, maps, storages } = await loadPathTargets(d.db);
const targets: PathTarget[] = [
{ scope: "data_dir", id: null, name: "DATA_DIR", path: d.dataDir },
...roots.map((r) => ({ scope: "library_root" as const, id: r.id, name: r.name, path: r.local_path })),
// Alist 库根没有本机磁盘可统计(容量在文件管理按 mount_details 显示)
...roots
.filter((r) => !isRemotePath(r.local_path))
.map((r) => ({ scope: "library_root" as const, id: r.id, name: r.name, path: r.local_path })),
...maps.map((m) => ({
scope: "download_dir" as const,
id: m.id,
@@ -280,7 +293,12 @@ export async function loadDisksAndMounts(d: HealthDeps): Promise<{ disks: System
...roots.map(async (r): Promise<SystemMount> => {
const base = { kind: "library_root" as const, id: r.id, name: r.name, path: r.local_path, sentinelChecked: true };
try {
const h = await withTimeout(check(r.local_path, r.sentinel_file ?? DEFAULT_ROOT_SENTINEL), ms);
const h = await withTimeout(
isRemotePath(r.local_path)
? vcheckRoot(r.local_path)
: check(r.local_path, r.sentinel_file ?? DEFAULT_ROOT_SENTINEL),
isRemotePath(r.local_path) ? Math.max(ms, 15_000) : ms,
);
return h.kind === "online"
? { ...base, status: "online", reason: null }
: { ...base, status: "offline", reason: h.reason };
@@ -316,6 +334,27 @@ export async function loadDisksAndMounts(d: HealthDeps): Promise<{ disks: System
return { ...base, status: "offline", reason };
}
}),
// Alist / OpenList 实例:连通 + 根路径可访问(令牌失效、连不上都在这里显示)
...storages.map(async (s): Promise<SystemMount> => {
const path = toVPath(s.id, s.root_path);
const base = {
kind: "storage_instance" as const,
id: s.id,
name: s.name,
path: s.root_path,
sentinelChecked: false,
};
try {
await storageEntry(s.id);
const h = await withTimeout(vcheckRoot(path), Math.max(ms, 15_000));
if (h.kind === "online") return { ...base, status: "online", reason: null };
return { ...base, status: h.reason === "timeout" ? "timeout" : "offline", reason: h.reason };
} catch (err) {
return err instanceof Timeout
? { ...base, status: "timeout", reason: "timeout" }
: { ...base, status: "offline", reason: "error" };
}
}),
]);
return { disks, mounts };
}
+4 -1
View File
@@ -85,7 +85,10 @@ export async function loadMountedPaths(db: Kysely<DB>): Promise<MountedPath[]> {
.where("view_kind", "=", "downloader")
.execute();
return [
...roots.map((r) => ({ kind: "library_root" as const, id: r.id, name: r.name, path: r.local_path })),
// Alist 库根(/@alist/…)不是本机挂载,由健康检查按实例连通性判断
...roots
.filter((r) => !r.local_path.startsWith("/@alist/"))
.map((r) => ({ kind: "library_root" as const, id: r.id, name: r.name, path: r.local_path })),
...maps.map((m) => ({ kind: "path_mapping" as const, id: m.id, name: m.remote_prefix, path: m.local_prefix })),
];
}
+61
View File
@@ -0,0 +1,61 @@
import * as fsp from "node:fs/promises";
import { assertLocalPaths } from "@umt/adapters";
// 本地写入口的远端路径守卫(server 侧):整理 / 刮削 / 字幕 / 回收站 / 上传里直接调 node:fs 写文件的地方都从这里取函数。
// 参数是 Alist 虚拟路径(/@alist/…)就抛 FsError("REMOTE_PATH_ON_LOCAL_FS"),不在本机建出 /@alist 目录或写进文件。
// 读类函数(stat / readdir / readFile …)原样转出:对虚拟路径只会 ENOENT。
export * from "node:fs/promises";
type P = Parameters<typeof fsp.mkdir>[0];
const s = (p: unknown) => (typeof p === "string" ? p : p instanceof URL ? p.pathname : undefined);
export const mkdir: typeof fsp.mkdir = (async (p: P, o?: Parameters<typeof fsp.mkdir>[1]) => {
assertLocalPaths("mkdir", s(p));
return fsp.mkdir(p, o);
}) as typeof fsp.mkdir;
export const writeFile: typeof fsp.writeFile = async (p, data, o) => {
assertLocalPaths("writeFile", s(p));
return fsp.writeFile(p, data, o);
};
export const rename: typeof fsp.rename = async (a, b) => {
assertLocalPaths("rename", s(a), s(b));
return fsp.rename(a, b);
};
export const copyFile: typeof fsp.copyFile = async (a, b, mode) => {
assertLocalPaths("copyFile", s(a), s(b));
return fsp.copyFile(a, b, mode);
};
export const rm: typeof fsp.rm = async (p, o) => {
assertLocalPaths("rm", s(p));
return fsp.rm(p, o);
};
export const rmdir: typeof fsp.rmdir = async (p, o) => {
assertLocalPaths("rmdir", s(p));
return fsp.rmdir(p, o);
};
export const unlink: typeof fsp.unlink = async (p) => {
assertLocalPaths("unlink", s(p));
return fsp.unlink(p);
};
export const link: typeof fsp.link = async (a, b) => {
assertLocalPaths("link", s(a), s(b));
return fsp.link(a, b);
};
/** open:写 / 追加 / 新建模式才拦;只读打开放行(虚拟路径只会 ENOENT) */
export const open: typeof fsp.open = (async (
p: Parameters<typeof fsp.open>[0],
flags?: string | number,
mode?: number,
) => {
if (flags !== undefined && flags !== "r") assertLocalPaths("open", s(p));
return fsp.open(p, flags, mode);
}) as typeof fsp.open;
+23
View File
@@ -0,0 +1,23 @@
import { ffprobe, type ProbeResult, redactAlistText } from "@umt/adapters";
import { vrawUrl } from "./vfs.ts";
import { parseVPath } from "./vpath.ts";
// 远端文件的 ffprobe(docs/design/storage-alist.md §7):不下载,直接对签名直链探测,ffprobe 只读容器头与少量包(HTTP Range)。
// 拿不到直链或探测失败返回 unknown(不阻塞,同本地 ffprobe 失败的口径)。直链进错误信息前脱敏。
export async function probeAny(
path: string,
local: (p: string) => Promise<ProbeResult> = (p) => ffprobe(p),
): Promise<ProbeResult> {
if (!parseVPath(path)) return local(path);
let url: string;
try {
url = await vrawUrl(path);
} catch (e) {
return { status: "unknown", reason: "probe_failed", message: redactAlistText(String((e as Error)?.message ?? e)) };
}
const r = await ffprobe(url, { inputArgs: ["-rw_timeout", "30000000"], timeoutMs: 60_000 });
if (r.status === "unknown" && r.message)
return { ...r, message: redactAlistText(r.message.split(url).join("<直链>")) };
return r;
}
+170
View File
@@ -0,0 +1,170 @@
import { type AlistAuth, AlistClient, AlistError } from "@umt/adapters";
import { STORAGE_OPTION_DEFAULTS, type StorageOptions } from "@umt/contracts";
import type { DB, Kysely, StorageInstanceRow } from "@umt/db";
import type { ServerContext } from "../modules/types.ts";
// 存储实例注册表(docs/design/storage-alist.md §3):按 id 取实例 + 解密凭据 → AlistClient。
// 进程级:模块装配时 configureStorage(ctx);实例行按 updated_at 缓存,配置一改客户端重建(令牌缓存随之作废,同 MP init_storage)。
export interface StorageEntry {
id: string;
name: string;
enabled: boolean;
/** 本系统可用的 Alist 根路径 */
rootPath: string;
options: Required<StorageOptions>;
client: AlistClient;
/** 日志脱敏要遮的秘密值 */
secrets: string[];
}
interface Source {
dataDir?: string | undefined;
db: Kysely<DB>;
secrets: ServerContext["secrets"];
logger?: ServerContext["logger"] | undefined;
}
let source: Source | null = null;
const cache = new Map<string, { stamp: string; entry: StorageEntry; checkedAt: number }>();
const overrides = new Map<string, StorageEntry>();
const RECHECK_MS = 5_000;
/** 模块装配时调用(幂等;最后一次为准) */
export function configureStorage(
ctx: Pick<ServerContext, "db" | "secrets"> & {
logger?: ServerContext["logger"];
config?: Partial<Pick<ServerContext["config"], "DATA_DIR">>;
},
) {
if (source?.db !== ctx.db) cache.clear();
source = { db: ctx.db, secrets: ctx.secrets, logger: ctx.logger, dataDir: ctx.config?.DATA_DIR };
}
/** 数据目录(跨存储中转、上传暂存放在 <DATA_DIR>/cache/storage-relay) */
export const storageDataDir = (): string | undefined => source?.dataDir;
/** 测试用:直接给某个存储 id 指定客户端(mock Alist) */
export function overrideStorage(
id: string,
e: { client: AlistClient; name?: string; rootPath?: string; options?: StorageOptions } | null,
): void {
if (!e) {
overrides.delete(id);
return;
}
overrides.set(id, {
id,
name: e.name ?? "Alist",
enabled: true,
rootPath: e.rootPath ?? "/",
options: { ...STORAGE_OPTION_DEFAULTS, ...(e.options ?? {}) },
client: e.client,
secrets: [],
});
}
export function clearStorageCache(id?: string): void {
if (id) cache.delete(id);
else cache.clear();
}
export const storageOptions = (o: unknown): Required<StorageOptions> => ({
...STORAGE_OPTION_DEFAULTS,
...((o && typeof o === "object" ? o : {}) as StorageOptions),
});
/** 由实例行 + 明文凭据建客户端(测试连接、注册表共用) */
export function buildClient(
row: Pick<StorageInstanceRow, "base_url" | "auth_kind" | "username" | "options">,
plain: { credential: string | null; pathPassword: string | null },
logger?: ServerContext["logger"],
): AlistClient {
const o = storageOptions(row.options);
let auth: AlistAuth;
if (row.auth_kind === "token") {
if (!plain.credential) throw new AlistError("auth", "没有配置令牌");
auth = { kind: "token", token: plain.credential };
} else if (row.auth_kind === "password") {
if (!row.username || !plain.credential) throw new AlistError("auth", "账号密码认证需要用户名和密码");
auth = { kind: "password", username: row.username, password: plain.credential };
} else auth = { kind: "guest" };
return new AlistClient({
baseUrl: row.base_url,
auth,
pathPassword: plain.pathPassword,
qps: o.qps,
concurrency: o.concurrency,
timeoutMs: o.timeoutSec * 1000,
idleTimeoutMs: o.idleTimeoutSec * 1000,
taskTimeoutMs: o.taskTimeoutMin * 60_000,
retries: o.retries,
uploadAsTask: o.uploadAsTask,
...(logger
? {
logger: {
debug: (obj, msg) => logger.debug(obj, msg),
warn: (obj, msg) => logger.warn(obj, msg),
},
}
: {}),
});
}
export async function resolvePlain(
secrets: Source["secrets"],
row: Pick<StorageInstanceRow, "secret_ref" | "path_password_ref">,
) {
return {
credential: row.secret_ref ? await secrets.resolve(row.secret_ref) : null,
pathPassword: row.path_password_ref ? await secrets.resolve(row.path_password_ref) : null,
};
}
/** 取存储实例;不存在 / 停用 / 未装配抛 AlistError(unavailable,按「存储离线」处理,不当成文件不存在) */
export async function storageEntry(id: string, opts: { allowDisabled?: boolean } = {}): Promise<StorageEntry> {
const o = overrides.get(id);
if (o) return o;
if (!source) throw new AlistError("unavailable", "存储注册表未初始化");
const hit = cache.get(id);
if (hit && Date.now() - hit.checkedAt < RECHECK_MS) return check(hit.entry, opts);
const row = await source.db.selectFrom("storage_instance").selectAll().where("id", "=", id).executeTakeFirst();
if (!row) throw new AlistError("unavailable", `存储实例不存在:${id}`);
const stamp = new Date(row.updated_at).toISOString();
if (hit && hit.stamp === stamp) {
hit.checkedAt = Date.now();
return check(hit.entry, opts);
}
const plain = await resolvePlain(source.secrets, row);
const entry: StorageEntry = {
id: row.id,
name: row.name,
enabled: row.enabled,
rootPath: row.root_path,
options: storageOptions(row.options),
client: buildClient(row, plain, source.logger),
secrets: [plain.credential, plain.pathPassword].filter((s): s is string => !!s),
};
cache.set(id, { stamp, entry, checkedAt: Date.now() });
return check(entry, opts);
}
function check(e: StorageEntry, opts: { allowDisabled?: boolean }): StorageEntry {
if (!e.enabled && !opts.allowDisabled) throw new AlistError("unavailable", `存储实例已停用:${e.name}`);
return e;
}
/** 启用的实例(文件管理根、健康检查) */
export async function listStorages(db?: Kysely<DB>): Promise<StorageInstanceRow[]> {
const d = db ?? source?.db;
if (!d) return [];
return d.selectFrom("storage_instance").selectAll().orderBy("name").execute();
}
/** 存储名表(展示「存储名:/路径」) */
export async function storageNames(db?: Kysely<DB>): Promise<Map<string, string>> {
const m = new Map<string, string>();
for (const [id, e] of overrides) m.set(id, e.name);
for (const r of await listStorages(db)) m.set(r.id, r.name);
return m;
}
+278
View File
@@ -0,0 +1,278 @@
import { randomUUID } from "node:crypto";
import { basename, dirname, join, posix } from "node:path";
import { AlistError, type ProgressCallback } from "@umt/adapters";
import { mkdir, rename, rm, stat, unlink } from "./local-fs.ts";
import { invalidateRemote, relayDir, remote, remoteError } from "./vfs.ts";
import { parseVPath, sameStorage } from "./vpath.ts";
// 整理矩阵中涉及 Alist 的方向(docs/design/storage-alist.md §5):
// 本地→Alist 上传、Alist→本地 下载、同实例 Alist→Alist API copy / move、跨实例 下载→上传。
// 成功判定与 MP 一致(用户决定):接口成功 + 目标存在,不比对大小 / 哈希;移动在判定通过后删源,删源失败 = attention。
// 本地 ↔ 本地不经这里(仍走 fsproxy 与现有核验)。
export interface RemoteTransferInput {
kind: "copy" | "move";
src: string;
dst: string;
/** 目标已存在:conflict 失败;replace 先把旧文件移到 replaceTrashDir(同存储),成功后旧文件留在那里,失败移回 */
onExisting: "conflict" | "replace";
replaceTrashDir?: string;
/** 本次是重试(上次可能已完成外部动作) */
resumed?: boolean;
signal?: AbortSignal;
onProgress?: ProgressCallback;
}
export interface RemoteTransferResult {
status: "succeeded" | "failed" | "attention";
bytes?: number;
/** 顺带得到的本地文件 sha256(上传的源 / 下载的结果),只是事实记录 */
checksum?: string;
resumed?: boolean;
error?: { code: string; message: string };
replaced?: { disposition: "trash"; path: string; size: number; mtimeMs: number };
steps?: { step: string; status: string; at: string; detail?: string; error?: { code: string; message: string } }[];
via: "upload" | "download" | "alist_copy" | "alist_move" | "relay";
}
/** Alist / 本地错误 → 整理账本错误码(可重试的 IO/TIMEOUT 走任务重试) */
export function transferErrorCode(e: unknown, role: "源" | "目标" | null = null): { code: string; message: string } {
if (e instanceof AlistError) {
const code =
e.kind === "not_found"
? role === "源"
? "SOURCE_MISSING"
: "IO"
: e.kind === "conflict"
? "EEXIST"
: e.kind === "auth"
? "EACCES"
: e.kind === "timeout" || e.kind === "rate_limited"
? "TIMEOUT"
: e.kind === "aborted"
? "ABORTED"
: "IO";
return { code, message: `Alist:${e.message}` };
}
const c = (e as { code?: string } | null)?.code;
const message = (e as Error)?.message ?? String(e);
if (c === "ENOENT") return { code: role === "源" ? "SOURCE_MISSING" : "IO", message };
if (c === "EEXIST") return { code: "EEXIST", message };
if (c === "EACCES" || c === "EPERM") return { code: "EACCES", message };
if (c === "ENOSPC") return { code: "ENOSPC", message };
if (c === "ETIMEDOUT") return { code: "TIMEOUT", message };
if (c === "ABORT_ERR") return { code: "ABORTED", message };
return { code: "IO", message };
}
const now = () => new Date().toISOString();
async function localExists(p: string): Promise<{ size: number; mtimeMs: number } | null> {
try {
const s = await stat(p);
return { size: s.size, mtimeMs: s.mtimeMs };
} catch (e) {
if ((e as { code?: string }).code === "ENOENT") return null;
throw e;
}
}
/** 任一端是远端路径? */
export const involvesRemote = (src: string, dst: string) => !!parseVPath(src) || !!parseVPath(dst);
/** 执行一次涉及远端的复制 / 移动 */
export async function remoteTransfer(x: RemoteTransferInput): Promise<RemoteTransferResult> {
const rs = parseVPath(x.src);
const rd = parseVPath(x.dst);
if (rs) invalidateRemote(x.src);
if (rd) invalidateRemote(x.dst);
const via: RemoteTransferResult["via"] =
rs && rd
? sameStorage(x.src, x.dst)
? x.kind === "copy"
? "alist_copy"
: "alist_move"
: "relay"
: rd
? "upload"
: "download";
const fail = (e: unknown, role: "源" | "目标" | null = null): RemoteTransferResult => ({
status: "failed",
error: transferErrorCode(e, role),
via,
});
// ── 源 / 目标状态 ──
let srcSize = 0;
let srcExists = true;
try {
if (rs) {
const { client } = await remote(x.src);
const o = await client.stat(rs.remote);
srcExists = !!o;
srcSize = o?.size ?? 0;
if (o?.isDir) return fail(new AlistError("error", `源是目录:${x.src}`));
} else {
const s = await localExists(x.src);
srcExists = !!s;
srcSize = s?.size ?? 0;
}
} catch (e) {
return fail(e, "源");
}
let dstFacts: { size: number; mtimeMs: number } | null;
try {
if (rd) {
const { client } = await remote(x.dst);
const o = await client.stat(rd.remote);
dstFacts = o ? { size: o.size, mtimeMs: o.modified } : null;
} else dstFacts = await localExists(x.dst);
} catch (e) {
return fail(e, "目标");
}
// 对账:上次尝试已完成外部动作但没留下结果(目标都是「临时名 → 改名」落地的,目标在 = 上次已写完)
if (dstFacts && x.resumed && x.onExisting !== "replace") {
if (x.kind === "move" && srcExists) {
const del = await deleteSource(x.src).then(
() => null,
(e) => transferErrorCode(e),
);
if (del)
return {
status: "attention",
bytes: dstFacts.size,
resumed: true,
via,
steps: [{ step: "delete_source", status: "failed", at: now(), error: del }],
};
}
return { status: "succeeded", bytes: dstFacts.size, resumed: true, via };
}
if (!srcExists) {
if (x.kind === "move" && dstFacts) return { status: "succeeded", bytes: dstFacts.size, resumed: true, via };
return { status: "failed", error: { code: "SOURCE_MISSING", message: `源文件不存在:${x.src}` }, via };
}
if (dstFacts && x.onExisting !== "replace")
return { status: "failed", error: { code: "EEXIST", message: `目标已存在:${x.dst}` }, via };
// ── 替换:旧目标先移走 ──
let parked: { path: string; size: number; mtimeMs: number } | null = null;
if (dstFacts && x.onExisting === "replace") {
const trashDir = x.replaceTrashDir ?? join(dirname(x.dst), ".umt-trash");
const to = join(trashDir, `replaced-${Date.now()}-${randomUUID().slice(0, 6)}`, basename(x.dst));
try {
await parkOld(x.dst, to);
parked = { path: to, size: dstFacts.size, mtimeMs: dstFacts.mtimeMs };
} catch (e) {
return fail(e, "目标");
}
}
const restoreParked = async () => {
if (parked) await parkOld(parked.path, x.dst).catch(() => undefined);
};
// ── 传输 ──
let bytes = srcSize;
let checksum: string | undefined;
try {
if (rd && !rs) {
const { client } = await remote(x.dst);
const r = await client.upload(x.src, rd.remote, {
...(x.signal ? { signal: x.signal } : {}),
...(x.onProgress ? { onProgress: x.onProgress } : {}),
});
bytes = r.bytes;
checksum = `sha256:${r.hashes.sha256}`;
} else if (rs && !rd) {
await mkdir(dirname(x.dst), { recursive: true });
const { client } = await remote(x.src);
const r = await client.download(rs.remote, x.dst, {
...(x.signal ? { signal: x.signal } : {}),
...(x.onProgress ? { onProgress: x.onProgress } : {}),
});
bytes = r.bytes;
checksum = `sha256:${r.sha256}`;
} else if (rs && rd && sameStorage(x.src, x.dst)) {
const { client } = await remote(x.src);
const opts = x.signal ? { signal: x.signal } : {};
const o =
x.kind === "copy"
? await client.copyFile(rs.remote, rd.remote, opts)
: await client.moveFile(rs.remote, rd.remote, opts);
bytes = o.size || srcSize;
} else if (rs && rd) {
// 跨实例:下载到本机中转 → 上传(MP 不支持,我们补上)
const tmpDir = join(relayDir(), randomUUID());
await mkdir(tmpDir, { recursive: true });
const tmp = join(tmpDir, posix.basename(rs.remote));
try {
const a = await remote(x.src);
await a.client.download(rs.remote, tmp, {
...(x.signal ? { signal: x.signal } : {}),
...(x.onProgress ? { onProgress: (d, t) => x.onProgress?.(d / 2, t) } : {}),
});
const b = await remote(x.dst);
const r = await b.client.upload(tmp, rd.remote, {
...(x.signal ? { signal: x.signal } : {}),
...(x.onProgress ? { onProgress: (d, t) => x.onProgress?.(t / 2 + d / 2, t) } : {}),
});
bytes = r.bytes;
checksum = `sha256:${r.hashes.sha256}`;
} finally {
await rm(tmpDir, { recursive: true, force: true }).catch(() => undefined);
}
}
} catch (e) {
await restoreParked();
return fail(e);
}
const out: RemoteTransferResult = {
status: "succeeded",
bytes,
...(checksum ? { checksum } : {}),
via,
...(parked ? { replaced: { disposition: "trash" as const, ...parked } } : {}),
};
// 同实例 API move 自己删了源;其它移动在成功后删源
if (x.kind === "move" && via !== "alist_move") {
const del = await deleteSource(x.src).then(
() => null,
(e) => transferErrorCode(e),
);
if (del)
return {
...out,
status: "attention",
steps: [{ step: "delete_source", status: "failed", at: now(), error: del }],
};
}
return out;
}
async function deleteSource(src: string): Promise<void> {
const v = parseVPath(src);
if (!v) {
await unlink(src);
return;
}
const { client } = await remote(src);
try {
await client.removePath(v.remote);
} catch (e) {
throw remoteError(e, src);
}
}
/** 同存储内把 from 移到 to(替换时暂放旧文件 / 失败移回) */
async function parkOld(from: string, to: string): Promise<void> {
const a = parseVPath(from);
const b = parseVPath(to);
if (!a && !b) {
await mkdir(dirname(to), { recursive: true });
await rename(from, to);
return;
}
if (!a || !b || a.storageId !== b.storageId) throw new AlistError("error", "替换暂存必须与目标在同一存储");
const { client } = await remote(from);
await client.moveFile(a.remote, b.remote);
}
+356
View File
@@ -0,0 +1,356 @@
import { createHash, randomUUID } from "node:crypto";
import { tmpdir } from "node:os";
import { basename, dirname, join, posix } from "node:path";
import {
type AlistClient,
AlistError,
checkRoot,
DEFAULT_ROOT_SENTINEL,
type FileFacts,
type RootHealth,
type StatResult,
statSafe,
} from "@umt/adapters";
import { lstat, mkdir, readdir, readFile, realpath, rename, rm, writeFile } from "./local-fs.ts";
import { storageDataDir, storageEntry } from "./registry.ts";
import { parseVPath, toVPath, type VPath } from "./vpath.ts";
// 本地 / Alist 统一的文件操作(docs/design/storage-alist.md §3)。本地分支原样调用 node:fs / adapters,
// 远端分支经注册表取 AlistClient。远端错误映射成 errno 风格的 code:不存在 ENOENT、冲突 EEXIST、鉴权 EACCES,
// 连接 / 服务端失败 EHOSTDOWN / ETIMEDOUT(文件管理按「根不可用」503 处理,整理按可重试 IO 处理,绝不当成不存在)。
export { isRemotePath, parseVPath, toVPath } from "./vpath.ts";
export interface VStats {
size: number;
mtimeMs: number;
ino: number;
dev: number;
nlink: number;
isFile(): boolean;
isDirectory(): boolean;
isSymbolicLink(): boolean;
}
/** Alist 错误 → 带 errno code 的 Error */
export function remoteError(e: unknown, path?: string): Error & { code: string } {
if (e instanceof AlistError) {
const code =
e.kind === "not_found"
? "ENOENT"
: e.kind === "conflict"
? "EEXIST"
: e.kind === "auth"
? "EACCES"
: e.kind === "timeout" || e.kind === "rate_limited"
? "ETIMEDOUT"
: e.kind === "aborted"
? "ABORT_ERR"
: "EHOSTDOWN";
return Object.assign(new Error(`Alist:${e.message}${path ? `(${path})` : ""}`), { code, alist: e.kind });
}
if (e && typeof e === "object" && "code" in e) return e as Error & { code: string };
return Object.assign(new Error(String((e as Error)?.message ?? e)), { code: "EIO" });
}
/** 远端对象的伪设备号:同一存储实例相同,与本机设备号不冲突 */
export function remoteDev(storageId: string): number {
return 2 ** 40 + Number.parseInt(createHash("sha1").update(storageId).digest("hex").slice(0, 8), 16);
}
export async function remote(p: string): Promise<{ v: VPath; client: AlistClient }> {
const v = parseVPath(p);
if (!v) throw Object.assign(new Error(`不是远端路径:${p}`), { code: "EINVAL" });
try {
const e = await storageEntry(v.storageId);
return { v, client: e.client };
} catch (err) {
throw remoteError(err, p);
}
}
const statsOf = (storageId: string, o: { isDir: boolean; size: number; modified: number }): VStats => ({
size: o.isDir ? 0 : o.size,
mtimeMs: o.modified,
ino: 0,
dev: remoteDev(storageId),
nlink: 1,
isFile: () => !o.isDir,
isDirectory: () => o.isDir,
isSymbolicLink: () => false,
});
export const factsOf = (s: VStats): FileFacts => ({
size: Number(s.size),
mtimeMs: Number(s.mtimeMs),
ino: Number(s.ino),
dev: Number(s.dev),
nlink: Number(s.nlink),
isFile: s.isFile(),
isDirectory: s.isDirectory(),
isSymbolicLink: s.isSymbolicLink(),
});
// 远端 stat 短缓存:列目录拿到的子项 stat 在几秒内复用,避免遍历时逐个 get(§9-8 限制请求量);写操作按目录失效
const STAT_TTL_MS = 3000;
const statCache = new Map<string, { at: number; stats: VStats }>();
function cacheStat(p: string, stats: VStats) {
statCache.set(p, { at: Date.now(), stats });
if (statCache.size > 20_000) statCache.delete(statCache.keys().next().value!);
}
/** 写后失效:该路径、其子树与父目录 */
export function invalidateRemote(p: string): void {
const parent = posix.dirname(p);
for (const k of statCache.keys()) if (k === p || k.startsWith(`${p}/`) || k === parent) statCache.delete(k);
}
export async function vlstat(p: string, opts: { refresh?: boolean } = {}): Promise<VStats> {
if (!parseVPath(p)) return (await lstat(p)) as unknown as VStats;
if (!opts.refresh) {
const hit = statCache.get(p);
if (hit && Date.now() - hit.at < STAT_TTL_MS) return hit.stats;
}
const { v, client } = await remote(p);
try {
const st = statsOf(v.storageId, await client.get(v.remote, opts));
cacheStat(p, st);
return st;
} catch (e) {
throw remoteError(e, p);
}
}
/** 远端没有符号链接:realpath = 规范化后的路径(存在才返回) */
export async function vrealpath(p: string): Promise<string> {
const v = parseVPath(p);
if (!v) return realpath(p);
await vlstat(p);
return toVPath(v.storageId, v.remote);
}
export interface VDirent {
name: string;
stats: VStats;
}
/** 列目录(带每项的 stat;远端一次 list 拿全) */
export async function vreaddirStats(p: string, opts: { refresh?: boolean } = {}): Promise<VDirent[]> {
const v = parseVPath(p);
if (!v) {
const names = await readdir(p);
const out: VDirent[] = [];
for (const name of names) {
const st = await lstat(join(p, name)).catch(() => null);
if (st) out.push({ name, stats: st as unknown as VStats });
}
return out;
}
const { client } = await remote(p);
try {
return (await client.list(v.remote, opts)).map((o) => {
const stats = statsOf(v.storageId, o);
cacheStat(join(p, o.name), stats);
return { name: o.name, stats };
});
} catch (e) {
throw remoteError(e, p);
}
}
export interface VDirentLite {
name: string;
isDirectory(): boolean;
isFile(): boolean;
isSymbolicLink(): boolean;
}
/** readdir(withFileTypes) 的统一版本;本地照旧 */
export async function vreaddirents(p: string): Promise<VDirentLite[]> {
if (!parseVPath(p)) return readdir(p, { withFileTypes: true });
return (await vreaddirStats(p)).map((d) => ({
name: d.name,
isDirectory: () => d.stats.isDirectory(),
isFile: () => d.stats.isFile(),
isSymbolicLink: () => false,
}));
}
export async function vreaddir(p: string): Promise<string[]> {
if (!parseVPath(p)) return readdir(p);
return (await vreaddirStats(p)).map((d) => d.name).sort();
}
/** 同 adapters statSafe 的口径;远端失败(连不上 / 超时 / 鉴权)返回 error,只有明确不存在才是 missing */
export async function vstatSafe(p: string, opts: { root?: string; sentinel?: string } = {}): Promise<StatResult> {
if (!parseVPath(p)) return statSafe(p, opts);
if (opts.root) {
const h = await vcheckRoot(opts.root, opts.sentinel);
if (h.kind === "offline") return { kind: "root_offline", root: opts.root, reason: h.reason };
}
try {
return { kind: "ok", facts: factsOf(await vlstat(p)) };
} catch (e) {
const err = remoteError(e, p);
if (err.code === "ENOENT") return { kind: "missing" };
if (err.code === "EACCES") return { kind: "permission_denied" };
return { kind: "error", code: err.code === "ETIMEDOUT" ? "TIMEOUT" : "IO", message: err.message };
}
}
/** 根在线:本地按哨兵;远端 = 根路径存在且是目录(Alist 不用哨兵,同 MP) */
export async function vcheckRoot(root: string, sentinel: string = DEFAULT_ROOT_SENTINEL): Promise<RootHealth> {
const v = parseVPath(root);
if (!v) return checkRoot(root, sentinel);
try {
const st = await vlstat(root);
if (!st.isDirectory()) return { kind: "offline", reason: "root_missing" };
return { kind: "online", dev: remoteDev(v.storageId) };
} catch (e) {
const err = remoteError(e, root);
if (err.code === "ENOENT") return { kind: "offline", reason: "root_missing" };
if (err.code === "EACCES") return { kind: "offline", reason: "permission_denied", code: "EACCES" };
if (err.code === "ETIMEDOUT") return { kind: "offline", reason: "timeout" };
return { kind: "offline", reason: "error", code: err.code };
}
}
export async function vmkdirp(p: string): Promise<void> {
const v = parseVPath(p);
if (!v) {
await mkdir(p, { recursive: true });
return;
}
const { client } = await remote(p);
invalidateRemote(p);
await client.mkdir(v.remote).catch((e) => {
throw remoteError(e, p);
});
}
export async function vexists(p: string): Promise<boolean> {
try {
await vlstat(p);
return true;
} catch (e) {
if ((e as { code?: string }).code === "ENOENT") return false;
throw e;
}
}
export async function vreadFile(p: string, maxBytes = 4 * 1024 * 1024): Promise<Uint8Array> {
const v = parseVPath(p);
if (!v) return new Uint8Array(await readFile(p));
const { client } = await remote(p);
try {
return await client.readFile(v.remote, maxBytes);
} catch (e) {
throw remoteError(e, p);
}
}
/** 本机临时目录(远端写入的本地中转) */
export function relayDir(): string {
return join(storageDataDir() ?? Deno.env.get("DATA_DIR") ?? tmpdir(), "cache", "storage-relay");
}
/**
* 原子写:本地同目录临时文件 → rename;远端本地生成后上传(同 MP 刮削 `_save_file`):
* 上传到旁边的临时名,成功后删旧文件(若有)再改名。
*/
export async function vwriteFileAtomic(p: string, content: Uint8Array | string): Promise<void> {
const v = parseVPath(p);
if (!v) {
await mkdir(dirname(p), { recursive: true });
const tmp = `${p}.${randomUUID().slice(0, 8)}.tmp`;
try {
await writeFile(tmp, content);
await rename(tmp, p);
} catch (e) {
await rm(tmp, { force: true }).catch(() => {});
throw e;
}
return;
}
const { client } = await remote(p);
invalidateRemote(p);
const dir = relayDir();
await mkdir(dir, { recursive: true });
const local = join(dir, `${randomUUID()}-${basename(p)}`);
const stage = `${v.remote}.umt-new`;
try {
await writeFile(local, content);
if (await client.stat(stage)) await client.removePath(stage);
await client.upload(local, stage);
if (await client.stat(v.remote)) await client.removePath(v.remote);
await client.rename(stage, posix.basename(v.remote));
} catch (e) {
await client.removePath(stage).catch(() => undefined);
throw remoteError(e, p);
} finally {
await rm(local, { force: true }).catch(() => {});
}
}
/** 删除文件或目录(远端 remove;本地 rm -r) */
export async function vremove(p: string): Promise<void> {
const v = parseVPath(p);
if (!v) {
await rm(p, { recursive: true, force: true });
return;
}
const { client } = await remote(p);
invalidateRemote(p);
await client.removePath(v.remote).catch((e) => {
throw remoteError(e, p);
});
}
/** 同存储内移动 / 改名(远端用 API move,目标已存在 EEXIST;本地 rename 不覆盖由调用方保证) */
export async function vmove(src: string, dst: string): Promise<void> {
const a = parseVPath(src);
const b = parseVPath(dst);
if (!a && !b) {
await mkdir(dirname(dst), { recursive: true });
await rename(src, dst);
return;
}
if (!a || !b || a.storageId !== b.storageId)
throw Object.assign(new Error(`跨存储不能直接移动:${src} → ${dst}`), { code: "EXDEV" });
const { client } = await remote(src);
invalidateRemote(src);
invalidateRemote(dst);
await client.moveFile(a.remote, b.remote).catch((e) => {
throw remoteError(e, src);
});
}
/** 远端直链(ffprobe 用;日志前要脱敏) */
export async function vrawUrl(p: string): Promise<string> {
const { v, client } = await remote(p);
return client.rawUrl(v.remote).catch((e) => {
throw remoteError(e, p);
});
}
/** 同目录不覆盖改名(远端 Alist rename 本身拒绝同名;本地走 adapters moveSafe) */
export async function vrenameNoClobber(from: string, to: string): Promise<{ status: "moved" | "exists" | "exdev" }> {
const a = parseVPath(from);
if (!a) {
const { moveSafe } = await import("@umt/adapters");
return moveSafe(from, to);
}
const b = parseVPath(to);
if (!b || b.storageId !== a.storageId) return { status: "exdev" };
const { client } = await remote(from);
invalidateRemote(from);
invalidateRemote(to);
try {
if (await client.stat(b.remote)) return { status: "exists" };
if (posix.dirname(a.remote) === posix.dirname(b.remote)) await client.rename(a.remote, posix.basename(b.remote));
else await client.moveFile(a.remote, b.remote);
return { status: "moved" };
} catch (e) {
if (e instanceof AlistError && e.kind === "conflict") return { status: "exists" };
throw remoteError(e, from);
}
}
+45
View File
@@ -0,0 +1,45 @@
import { posix } from "node:path";
// 远端路径的虚拟前缀(docs/design/storage-alist.md §0、§3):/@alist/<存储 id>/<Alist 内路径>。
// 它是合法的 POSIX 绝对路径,join / dirname / relative 照常可用;本机没有 /@alist,
// 不认识前缀的旧代码对它做本地 IO 只会得到 ENOENT(失败安全)。
export const VROOT = "/@alist";
const RE = /^\/@alist\/([0-9a-f-]{36})(\/.*)?$/i;
export interface VPath {
storageId: string;
/** Alist 内绝对路径(以 / 开头,无尾斜杠) */
remote: string;
}
export const isRemotePath = (p: string | null | undefined): boolean =>
!!p && (p === VROOT || p.startsWith(`${VROOT}/`));
export function parseVPath(p: string): VPath | null {
if (!isRemotePath(p)) return null;
const m = RE.exec(p.replace(/\/+$/, ""));
if (!m) return null;
const remote = posix.normalize(m[2] ?? "/");
if (remote.split("/").includes("..")) return null;
return { storageId: m[1]!.toLowerCase(), remote: remote === "/" ? "/" : remote.replace(/\/+$/, "") };
}
export function toVPath(storageId: string, remote: string): string {
const r = posix.normalize(`/${remote}`).replace(/\/+$/, "");
return `${VROOT}/${storageId}${r === "" || r === "/" ? "" : r}`;
}
/** 同一存储实例? */
export function sameStorage(a: string, b: string): boolean {
const pa = parseVPath(a);
const pb = parseVPath(b);
return !!pa && !!pb && pa.storageId === pb.storageId;
}
/** 界面 / 日志展示:「存储名:/路径」;本地原样 */
export function displayPath(p: string, names: ReadonlyMap<string, string> = new Map()): string {
const v = parseVPath(p);
if (!v) return p;
return `${names.get(v.storageId) ?? "Alist"}:${v.remote}`;
}