UHD_Media_Tool/apps/server/src/modules/file-manager/transfer.ts

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 { ensureTaskRow, markTask } from "../acquisition-organize/services/common.ts";
import { planImport } from "../acquisition-organize/services/organize.ts";
import { authorizeMove, PathGuardError, sourceGuard } from "../acquisition-organize/services/organize-guard.ts";
import type { TransferSpec } from "../acquisition-organize/services/organize-transfer.ts";
import {
  actorOfP,
  FilesError,
  type FmDeps,
  hashJson,
  isAdmin,
  type Resolved,
  relOf,
  resolveIn,
  within,
} from "./common.ts";
import { audit, filesChanged, loadPlan, retirePlan, savePlan } from "./ledger.ts";
import { downloadsFor, isWritableTarget, managedDownloads } from "./list.ts";
import { relocatePreview } from "./records.ts";

// 移动/复制到目录(W5-A2):与整理共用 planImport(原样搬运变体 transfer)、冻结计划存储与 plan_stale 校验。
// dryRun:冻结源事实(文件 size/mtime/inode;目录递归文件清单指纹)+ planImport 逐文件前后路径 → planId;
// 执行:校验 planId(未过期、请求一致)、源事实与重算的逐文件路径未变,否则 409 plan_stale;
// 然后写 organization_record 并入队 verifyAndImport(executeRecord:逐文件账本可恢复、fsproxy 挂死隔离、
// 跨卷 Copy→核验→删源、EXDEV 不降级),请求内最多等 FILES_TRANSFER_WAIT_SEC,超时返回 pending。
// move 完成后已登记位置跟随、整目录搬完时路径绑定写修订、源侧空目录删除(finalizeTransfer)。

interface SourceSnap {
  root: string;
  rel: string;
  abs: string;
  isDir: boolean;
  size: number;
  fingerprint: string;
  files: { rel: string; size: number }[];
  truncated: boolean;
}

async function snapshot(d: FmDeps, r: Resolved, rootId: string): Promise<SourceSnap> {
  const st = await lstat(r.abs);
  if (!st.isDirectory()) {
    const f = { size: Number(st.size), mtimeMs: Number(st.mtimeMs), ino: Number(st.ino) };
    return {
      root: rootId,
      rel: r.rel,
      abs: r.abs,
      isDir: false,
      size: f.size,
      fingerprint: hashJson(f),
      files: [{ rel: "", size: f.size }],
      truncated: false,
    };
  }
  const max = d.ao.ports.options.walkMaxFiles;
  const files: { rel: string; size: number; mtimeMs: number }[] = [];
  let truncated = false;
  const walk = async (dir: string, prefix: string): Promise<void> => {
    const names = (await readdir(dir)).sort();
    for (const n of names) {
      if (files.length >= max) {
        truncated = true;
        return;
      }
      const p = join(dir, n);
      const s = await lstat(p);
      const rel = prefix ? `${prefix}/${n}` : n;
      if (s.isDirectory()) await walk(p, rel);
      else if (s.isFile()) files.push({ rel, size: Number(s.size), mtimeMs: Number(s.mtimeMs) });
    }
  };
  await walk(r.abs, "");
  return {
    root: rootId,
    rel: r.rel,
    abs: r.abs,
    isDir: true,
    size: files.reduce((a, f) => a + f.size, 0),
    fingerprint: hashJson(files),
    files: files.map(({ rel, size }) => ({ rel, size })),
    truncated,
  };
}

const TERMINAL = new Set<OrganizationStatus>(["succeeded", "partial", "failed", "cancelled"]);

export async function transfer(
  d: FmDeps,
  principal: Principal | undefined,
  body: FilesMoveRequest,
): Promise<FilesTransferResult> {
  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} 不可写(仅 managed 库根可作为目标)`);

  const sources: Resolved[] = [];
  for (const it of body.items) {
    const s = await resolveIn(d, it.root, it.path);
    if (s.rel === "") throw new FilesError("bad_request", "不能移动/复制根目录本身");
    if (mode === "move" && !s.def.modifiable)
      throw new FilesError("not_writable", `源根 ${s.def.name} 为只读登记,不能移动(可复制)`);
    if (within(t.abs, s.abs)) throw new FilesError("bad_request", `目标在源目录之内: ${s.rel}`);
    if (dirname(s.abs) === t.abs) throw new FilesError("bad_request", `源已在目标目录中: ${s.rel}`);
    if (s.isSymlink) throw new FilesError("bad_request", `不移动符号链接本身: ${s.rel}`);
    sources.push(s);
  }
  const names = sources.map((s) => basename(s.abs));
  if (new Set(names).size !== names.length) throw new FilesError("bad_request", "多个源同名,会在目标目录冲突");

  const snaps = await Promise.all(sources.map((s, i) => snapshot(d, s, body.items[i]!.root)));
  const downloads = await managedDownloads(d);
  const hnrBlocks: FilesTransferResult["hnrBlocks"] = [];
  if (mode === "move")
    for (const s of sources)
      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 spec: TransferSpec = {
    items: sources.map((s, i) => {
      const dst = join(t.abs, names[i]!);
      return {
        src: s.abs,
        dst,
        srcLibraryRootId: s.def.libraryRootId,
        srcRel: s.def.libraryRootId ? s.rel : null,
        dstRel: relOf(t.rootReal, dst),
      };
    }),
    targetLibraryRootId: t.def.libraryRootId,
    targetRootPath: t.rootReal,
  };
  const actorType = principal?.kind === "api_key" ? ("api_key" as const) : ("user" as const);
  const importReq = (dryRun: boolean, extra: { taskId?: string; guard?: (p: string) => Promise<string> } = {}) => ({
    transfer: { ...spec, ...(extra.guard ? { guard: extra.guard } : {}) },
    action: mode,
    trigger: "manual" as const,
    actorType,
    actorId: actorOfP(principal),
    dryRun,
    ...(extra.taskId ? { taskId: extra.taskId } : {}),
  });
  const planned = await planImport(d.ao, importReq(true));
  if (!planned.plan) {
    const reason = planned.reason ?? "plan_failed";
    throw new FilesError(
      reason.startsWith("library_root_offline") ? "root_unavailable" : "bad_request",
      `无法生成搬运计划:${reason}`,
    );
  }
  const planFiles = planned.plan.organize.files;
  const filePairs = planFiles.map((f) => ({ src: f.src, dst: f.dst }));
  const conflictSrc = new Set(planFiles.filter((f) => f.action === "hold").map((f) => f.src));
  const under = (item: string, p: string) => p === item || p.startsWith(`${item}/`);

  const pairs: FilePlanPair[] = [];
  let needBytes = 0;
  for (const [i, s] of sources.entries()) {
    const dst = spec.items[i]!.dst;
    const registered = s.def.libraryRootId
      ? (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;
    pairs.push({
      source: { root: body.items[i]!.root, path: s.rel },
      target: { root: body.target.root, path: relOf(t.rootReal, dst) },
      size: snaps[i]!.size,
      conflict: !!(await lstat(dst).catch(() => null)) || [...conflictSrc].some((p) => under(s.abs, p)),
      registered,
      files: snaps[i]!.files.length,
      crossDevice,
    });
    if (!(mode === "move" && !crossDevice)) needBytes += snaps[i]!.size;
  }
  const du = 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;
  const totalBytes = snaps.reduce((a, s) => a + s.size, 0);
  const requestHash = hashJson({ items: body.items, target: body.target, mode });
  const request = {
    sourcePaths: sources.map((s) => s.abs),
    action: mode,
    targetLibraryRootId: t.def.libraryRootId,
  };
  const allowCross = (await d.ctx.settings.get("organize")).allowCrossVolumeMove;
  const crossBlocked = mode === "move" && !allowCross && pairs.some((p) => p.crossDevice);
  const warnings: string[] = [];
  if (crossBlocked) warnings.push("跨卷移动未开启(设置 → 整理 → 允许跨卷 Move),执行会被拒绝;可改用复制");
  if (conflictSrc.size > 0) warnings.push(`${conflictSrc.size} 个文件在目标已存在,不覆盖,执行时记为失败`);
  if (truncated) warnings.push("源文件数超过上限,请分批操作");
  const base = { mode, pairs, totalBytes, freeBytes, spaceOk, hnrBlocks, truncated, warnings };

  if (body.dryRun) {
    const plan = await savePlan(d, "transfer", requestHash, principal, {
      request,
      sources: snaps.map((s) => ({ abs: s.abs, fingerprint: s.fingerprint })),
      files: filePairs,
      pairs,
    });
    return { dryRun: true, planId: plan.id, planExpiresAt: plan.expiresAt.toISOString(), ...base, executed: null };
  }

  // ── 执行 ──
  if (!body.planId) throw new FilesError("bad_request", "执行需要 dryRun 返回的 planId");
  const plan = await loadPlan<{ sources: { abs: string; fingerprint: string }[]; files?: typeof filePairs }>(
    d,
    body.planId,
    "transfer",
    requestHash,
  );
  for (const [i, s] of snaps.entries()) {
    const frozen = plan.content.sources[i];
    if (!frozen || frozen.abs !== s.abs || frozen.fingerprint !== s.fingerprint)
      throw new FilesError("plan_stale", `源在预览后发生变化(大小/修改时间/文件清单): ${s.rel}`);
  }
  if (plan.content.files && hashJson(plan.content.files) !== hashJson(filePairs))
    throw new FilesError("plan_stale", "重新计算的前后路径与预览不一致(目标已变化),请重新预览");
  if (truncated) throw new FilesError("bad_request", "源目录文件数超过上限,请分批操作");
  if (!spaceOk) throw new FilesError("bad_request", `目标卷空间不足:需要 ${needBytes},可用 ${freeBytes}`);
  if (crossBlocked)
    throw new FilesError(
      "bad_request",
      "跨卷移动未开启(settings.organize.allowCrossVolumeMove),请改用复制或开启后重试",
    );
  if (mode === "move") {
    if (hnrBlocks.length > 0)
      throw new FilesError(
        "download_active",
        `源属于纳管下载任务(${[...new Set(hnrBlocks.map((h) => h.taskId))].join(", ")}),裸移动会让下载器丢失文件;请用整理入库(Move 会按 H&R 判定并移交)`,
      );
    if (!isAdmin(principal)) {
      const entity = await d.ctx.db
        .selectFrom("asset_location as l")
        .innerJoin("media_asset as a", "a.id", "l.media_asset_id")
        .select(["a.media_entity_id"])
        .where("l.library_root_id", "=", sources[0]!.def.libraryRootId ?? "00000000-0000-0000-0000-000000000000")
        .where("l.relative_path", "=", sources[0]!.rel)
        .executeTakeFirst();
      const auth = await authorizeMove(d.ao, principal, request, entity?.media_entity_id ?? null);
      if (!auth.allowed) {
        await audit(d, principal, "move_denied", { reason: auth.reason, planId: plan.id });
        throw new FilesError("approval_required", auth.reason);
      }
    }
  }
  await retirePlan(d, plan.id);

  const db = d.ctx.db;
  const taskId = await ensureTaskRow(db, {
    kind: "verifyAndImport",
    idempotencyKey: `files:${plan.id}`,
    input: { planId: plan.id, op: "transfer", mode, target: body.target },
    subjectType: "organize_request",
    triggeredBy: actorOfP(principal),
  });
  const guard = await sourceGuard(db);
  let rec: Awaited<ReturnType<typeof planImport>>;
  try {
    rec = await planImport(d.ao, importReq(false, { taskId, guard: guard.check }));
  } catch (e) {
    await markTask(db, taskId, "failed", { error: e instanceof Error ? e.message : String(e) });
    if (e instanceof PathGuardError) throw new FilesError("path_escape", e.message);
    throw e;
  }
  if (!rec.recordId || rec.status === "failed") {
    await markTask(db, taskId, "failed", { error: rec.reason ?? rec.status });
    throw new FilesError("bad_request", `搬运计划不可执行:${rec.reason ?? rec.status}`);
  }
  const recordId = rec.recordId;
  await db.updateTable("organization_record").set({ task_id: taskId }).where("id", "=", recordId).execute();
  const { runId } = await d.ctx.engine.enqueue(
    "verifyAndImport",
    { recordId, taskId },
    { idempotencyKey: `verifyAndImport:files:${plan.id}`, traceId: taskId },
  );
  await db.updateTable("task").set({ engine_run_id: runId }).where("id", "=", taskId).execute();

  // 请求内等待执行完成(小操作即时出结果);超时返回 pending,后台继续,可按 recordId 追踪
  const deadline = Date.now() + d.options.transferWaitMs;
  let status: OrganizationStatus = "running";
  for (;;) {
    const r = await db
      .selectFrom("organization_record")
      .select("status")
      .where("id", "=", recordId)
      .executeTakeFirstOrThrow();
    status = r.status;
    if (TERMINAL.has(status) || Date.now() >= deadline) break;
    await new Promise((res) => setTimeout(res, 150));
  }
  const pending = !TERMINAL.has(status);
  const succeeded: string[] = [];
  const failed: { path: string; code: string; message: string }[] = [];
  if (!pending) {
    const results = await db
      .selectFrom("organization_file_result")
      .select(["source_path", "status", "error"])
      .where("organization_record_id", "=", recordId)
      .execute();
    for (const s of sources) {
      const mine = results.filter((f) => under(s.abs, f.source_path) && f.status !== "skipped");
      const bad = mine.find((f) => f.status !== "succeeded");
      if (!bad) {
        succeeded.push(s.rel);
        continue;
      }
      const err = (bad.error ?? {}) as { code?: string; message?: string };
      failed.push({
        path: s.rel,
        code: err.code === "EEXIST" ? "exists" : (err.code ?? bad.status),
        message: `${err.message ?? bad.status}${mine.length > 1 ? `(${mine.filter((f) => f.status === "succeeded").length}/${mine.length} 个文件已完成)` : ""}`,
      });
    }
  }
  await audit(d, principal, mode, {
    planId: plan.id,
    succeeded,
    failed,
    recordId,
    taskId,
    pending,
    target: body.target,
  });
  for (const it of body.items) filesChanged(d, it.root, [it.path.includes("/") ? dirname(it.path) : ""]);
  filesChanged(d, body.target.root, [t.rel]);
  return {
    dryRun: false,
    planId: plan.id,
    planExpiresAt: plan.expiresAt.toISOString(),
    ...base,
    executed: { succeeded, failed, recordId, taskId, status, pending },
  };
}