UHD_Media_Tool/packages/adapters/src/fs/copy.ts

// 迁移来源:MoviePilot app/modules/filemanager/storages/local.py _write_atomically/_partial_path/_cleanup_stale_partials/copy(e6820a8);
// 调整:fsync + size/sha256/sample 校验、no-clobber 提交(link 原子失败),见 docs/porting/moviepilot-transfer.md §2.6。
import { createHash, randomBytes } from "node:crypto";
import { createReadStream, createWriteStream } from "node:fs";
import { link, lstat, mkdir, open, readdir, rename, stat, unlink, utimes } from "node:fs/promises";
import { basename, dirname, join } from "node:path";
import { Transform } from "node:stream";
import { pipeline } from "node:stream/promises";
import { errnoOf, FsError, toFsError } from "./errors.ts";

export type VerifyMethod = "size" | "sha256" | "sample";

/** 字节级进度回调(复制 / 校验回读 / 哈希共用) */
export type ProgressFn = (bytesDone: number, totalBytes: number) => void;

/**
 * 目标已存在时的执行端策略(对应域层 ExistingPolicy,由计划冻结后传入):
 * - conflict(缺省):抛 EEXIST,绝不覆盖
 * - skip:不写,回执 outcome=skipped_existing
 * - replace:新文件先写临时名 → 校验 → 旧文件硬链接留底 → rename 原子替换 → 留底按 replaceOld 处置;不裸覆盖
 */
export type OnExisting = "conflict" | "skip" | "replace";

/** replace 时被替换旧文件的去向:trash = 移到 trashDir(同卷 rename);delete = 删除 */
export interface ReplaceOldPolicy {
  mode: "trash" | "delete";
  /** mode=trash 时的回收目录(应与目标同卷);缺省 <目标目录>/.umt-trash */
  trashDir?: string;
}

/** 写前顺带清理本目录残留 partial 的缺省年龄(对照 MP _cleanup_stale_partials:24h) */
export const DEFAULT_PARTIAL_MAX_AGE_MS = 24 * 3600_000;

export interface CopyOptions {
  /** 临时文件后缀,默认 .uhd-partial;临时文件与目标同目录、以 . 开头 */
  tmpSuffix?: string;
  /** 校验方式,默认 size(总是比对大小;sha256/sample 额外比对内容) */
  hash?: VerifyMethod;
  /** sample 模式的抽样块数与块大小 */
  sampleChunks?: number;
  sampleChunkBytes?: number;
  /** 是否创建目标父目录;默认 false(防止库根离线时在挂载点下写入) */
  createParents?: boolean;
  signal?: AbortSignal;
  onProgress?: ProgressFn;
  /** 校验回读(sha256/sample)的进度;给出时 sha256 走进程内流式以便上报心跳 */
  onVerifyProgress?: ProgressFn;
  /** 目标已存在时的策略,缺省 conflict */
  onExisting?: OnExisting;
  /** onExisting=replace 时旧文件去向,缺省 trash */
  replaceOld?: ReplaceOldPolicy;
  /** 写入前清理目标目录中超过该年龄的残留临时文件;缺省 24h,null 关闭。失败吞掉 */
  sweepPartialsOlderThanMs?: number | null;
}

export interface ReplacedOld {
  /** 旧文件留底后去向:trash=已移到回收目录;deleted=已删除;left=处置失败,留底文件保留在原目录 */
  disposition: "trash" | "deleted" | "left";
  /** 旧文件的大小与 mtime(证据) */
  size: number;
  mtimeMs: number;
  /** trash / left 时旧文件当前所在路径 */
  path?: string;
  error?: string;
}

export interface CopyReceipt {
  src: string;
  dst: string;
  bytes: number;
  verifiedBy: VerifyMethod;
  sha256?: string;
  /** sha256 校验时临时文件的全量哈希方式 */
  sha256Via?: Sha256Via;
  sampleDigest?: string;
  partialPath: string;
  durationMs: number;
  /** committed = 已提交到目标;skipped_existing = onExisting=skip 且目标已存在,未写 */
  outcome: "committed" | "skipped_existing";
  /** onExisting=replace 且确实替换了旧文件 */
  replaced?: ReplacedOld;
  /** 本次写前清走的残留 partial */
  sweptPartials?: string[];
}

async function fsyncPath(p: string, flags = "r"): Promise<void> {
  let fh: Awaited<ReturnType<typeof open>> | undefined;
  try {
    fh = await open(p, flags);
    await fh.sync();
  } catch (e) {
    // 部分文件系统不支持对目录 fsync
    if (!["EISDIR", "EINVAL", "EBADF", "EPERM", "ENOTSUP"].includes(errnoOf(e) ?? "")) throw e;
  } finally {
    await fh?.close();
  }
}

/** 复制/哈希的流块大小:1 MiB(缺省 64 KiB 在大文件上每 GB 要多上万次回调) */
export const IO_CHUNK_BYTES = 1024 * 1024;

/** 进程内流式 sha256(每块之间让出事件循环);onProgress 每块回调一次(fsproxy 心跳用) */
export async function sha256File(p: string, signal?: AbortSignal, onProgress?: ProgressFn): Promise<string> {
  const h = createHash("sha256");
  let done = 0;
  const total = onProgress ? Number((await stat(p)).size) : 0;
  await pipeline(createReadStream(p, { highWaterMark: IO_CHUNK_BYTES }), async (src: AsyncIterable<Buffer>) => {
    for await (const c of src) {
      if (signal?.aborted) throw new FsError("ABORTED", "aborted", { path: p });
      h.update(c);
      done += c.length;
      onProgress?.(done, total);
    }
  });
  return h.digest("hex");
}

/** 全量 sha256 用的方式:外部 sha256sum 子进程,或进程内流式(回退) */
export type Sha256Via = "sha256sum" | "stream";

export interface Sha256Result {
  digest: string;
  via: Sha256Via;
}

const hashCounters: Record<Sha256Via, number> = { sha256sum: 0, stream: 0 };
/** 子进程不可用(未安装 / 没有 --allow-run)后不再尝试,直到进程重启 */
let sha256sumUnavailable: string | null = null;

/** 本进程全量哈希方式计数与子进程不可用原因(诊断 / 指标用) */
export function sha256Stats(): { counts: Record<Sha256Via, number>; subprocessUnavailable: string | null } {
  return { counts: { ...hashCounters }, subprocessUnavailable: sha256sumUnavailable };
}

type DenoLike = {
  Command: new (
    cmd: string,
    opts: { args: string[]; stdout: "piped"; stderr: "piped"; signal?: AbortSignal },
  ) => { output(): Promise<{ success: boolean; code: number; stdout: Uint8Array; stderr: Uint8Array }> };
  errors: { NotFound: new () => Error; NotCapable?: new () => Error; PermissionDenied: new () => Error };
};

async function sha256Subprocess(p: string, signal?: AbortSignal): Promise<string | null> {
  const D = (globalThis as { Deno?: DenoLike }).Deno;
  if (!D?.Command) {
    sha256sumUnavailable = "no_deno_command";
    return null;
  }
  let out: Awaited<ReturnType<InstanceType<DenoLike["Command"]>["output"]>>;
  try {
    out = await new D.Command("sha256sum", {
      args: ["-b", "--", p],
      stdout: "piped",
      stderr: "piped",
      ...(signal ? { signal } : {}),
    }).output();
  } catch (e) {
    if (signal?.aborted) throw new FsError("ABORTED", "aborted", { path: p, cause: e });
    const notCapable = D.errors.NotCapable && e instanceof D.errors.NotCapable;
    if (e instanceof D.errors.NotFound || e instanceof D.errors.PermissionDenied || notCapable) {
      sha256sumUnavailable = e instanceof D.errors.NotFound ? "not_installed" : "not_permitted";
      return null;
    }
    throw toFsError(e, p);
  }
  if (!out.success) {
    const msg = new TextDecoder().decode(out.stderr).trim().slice(0, 300);
    throw new FsError("IO", `sha256sum 失败(${out.code}): ${msg}`, { path: p });
  }
  const digest = new TextDecoder().decode(out.stdout).trim().split(/\s+/)[0] ?? "";
  if (!/^[0-9a-f]{64}$/.test(digest)) throw new FsError("IO", `sha256sum 输出无法解析`, { path: p });
  return digest;
}

/**
 * 全量 sha256:优先外部 `sha256sum` 子进程(读盘 + 哈希完全不占本进程事件循环),
 * 不可用时回退进程内流式哈希。返回所用方式,调用方记进操作结果。
 */
export async function sha256Full(p: string, signal?: AbortSignal, onProgress?: ProgressFn): Promise<Sha256Result> {
  // 需要进度(fsproxy 子进程内按心跳判挂死)时走流式:sha256sum 子进程看不到进度;
  // 在执行子进程里哈希本就不占主进程事件循环,实测两者吞吐相当(见 docs/porting/moviepilot-rust-transfer.md 基准)
  if (!sha256sumUnavailable && !onProgress) {
    const digest = await sha256Subprocess(p, signal);
    if (digest) {
      hashCounters.sha256sum += 1;
      return { digest, via: "sha256sum" };
    }
  }
  const digest = await sha256File(p, signal, onProgress);
  hashCounters.stream += 1;
  return { digest, via: "stream" };
}

/** 测试用:清掉"子进程不可用"缓存 */
export function resetSha256Probe(): void {
  sha256sumUnavailable = null;
}

/** 抽样摘要:首/尾/等距 N 块 + 文件大小。抽样一致不等于完整一致性证明(REQ §8.2)。 */
export async function sampleDigest(
  p: string,
  chunks = 8,
  chunkBytes = 1024 * 1024,
  onProgress?: ProgressFn,
): Promise<string> {
  const fh = await open(p, "r");
  try {
    const { size } = await fh.stat();
    const h = createHash("sha256");
    h.update(String(size));
    const n = Math.max(2, chunks);
    const buf = Buffer.alloc(chunkBytes);
    const offsets = new Set<number>();
    for (let i = 0; i < n; i++) offsets.add(Math.max(0, Math.floor(((size - chunkBytes) * i) / (n - 1))));
    for (const off of [...offsets].sort((a, b) => a - b)) {
      const { bytesRead } = await fh.read(buf, 0, chunkBytes, off);
      h.update(buf.subarray(0, bytesRead));
      onProgress?.(off + bytesRead, size);
    }
    return h.digest("hex");
  } finally {
    await fh.close();
  }
}

export function partialPathFor(dst: string, suffix = ".uhd-partial"): string {
  return join(dirname(dst), `.${basename(dst)}.${process.pid}.${randomBytes(3).toString("hex")}${suffix}`);
}

const LINK_UNSUPPORTED = ["EPERM", "ENOTSUP", "EOPNOTSUPP", "ENOSYS", "EMLINK"];

/** no-clobber 提交:link(tmp,dst) 在目标存在时原子失败;文件系统不支持硬链接时退回 lstat+rename。 */
export async function commitNoClobber(tmp: string, dst: string): Promise<void> {
  try {
    await link(tmp, dst);
    await unlink(tmp);
    return;
  } catch (e) {
    const code = errnoOf(e);
    if (code === "EEXIST") throw new FsError("EEXIST", `target exists: ${dst}`, { path: dst, cause: e });
    if (!LINK_UNSUPPORTED.includes(code ?? "")) throw toFsError(e, dst);
  }
  try {
    await lstat(dst);
    throw new FsError("EEXIST", `target exists: ${dst}`, { path: dst });
  } catch (e) {
    if (e instanceof FsError) throw e;
    if (errnoOf(e) !== "ENOENT") throw toFsError(e, dst);
  }
  await rename(tmp, dst);
}

async function lstatOrNull(p: string): Promise<Awaited<ReturnType<typeof lstat>> | null> {
  try {
    return await lstat(p);
  } catch (e) {
    if (errnoOf(e) === "ENOENT") return null;
    // 失败 ≠ 不存在(REQ §2.1 原则 4):非 ENOENT 一律按错误上抛
    throw toFsError(e, p);
  }
}

/**
 * 替换提交:旧目标先硬链接留底(同目录隐藏名,原子)→ 确认留底的就是计划时看到的旧文件 → rename(tmp,dst) 原子替换
 * (读者看不到空窗)→ fsync 目录 → 留底按策略移到回收目录或删除。文件系统不支持硬链接时退回
 * rename(dst,留底) + no-clobber 提交,提交失败把旧文件改回。处置失败不删旧文件,标 left 交给调用方记待处理。
 */
async function replaceCommit(
  tmp: string,
  dst: string,
  old: { ino: number | bigint; dev: number | bigint; size: number | bigint; mtimeMs: number | bigint },
  policy: ReplaceOldPolicy = { mode: "trash" },
): Promise<ReplacedOld> {
  const keep = join(dirname(dst), `.${basename(dst)}.${process.pid}.${randomBytes(3).toString("hex")}.uhd-replaced`);
  const sameInode = (st: { ino: number | bigint; dev: number | bigint }) =>
    Number(st.ino) === Number(old.ino) && Number(st.dev) === Number(old.dev);
  let linked = true;
  try {
    await link(dst, keep);
  } catch (e) {
    if (!LINK_UNSUPPORTED.includes(errnoOf(e) ?? "")) throw toFsError(e, dst);
    linked = false;
  }
  if (linked) {
    try {
      if (!sameInode(await lstat(keep)))
        throw new FsError("EEXIST", `target changed before replace: ${dst}`, { path: dst });
      await rename(tmp, dst);
    } catch (e) {
      await unlink(keep).catch(() => undefined);
      throw toFsError(e, dst);
    }
  } else {
    const cur = await lstatOrNull(dst);
    if (!cur || !sameInode(cur)) throw new FsError("EEXIST", `target changed before replace: ${dst}`, { path: dst });
    await rename(dst, keep).catch((e) => {
      throw toFsError(e, dst);
    });
    try {
      await commitNoClobber(tmp, dst);
    } catch (e) {
      await rename(keep, dst).catch(() => undefined);
      throw e;
    }
  }
  await fsyncPath(dirname(dst));

  const facts = { size: Number(old.size), mtimeMs: Number(old.mtimeMs) };
  try {
    if (policy.mode === "delete") {
      await unlink(keep);
      return { disposition: "deleted", ...facts };
    }
    const bin = join(
      policy.trashDir ?? join(dirname(dst), ".umt-trash"),
      `replaced-${new Date().toISOString().replace(/[:.]/g, "-")}-${randomBytes(3).toString("hex")}`,
    );
    await mkdir(bin, { recursive: true });
    const to = join(bin, basename(dst));
    await rename(keep, to);
    return { disposition: "trash", path: to, ...facts };
  } catch (e) {
    return { disposition: "left", path: keep, error: (e as Error)?.message ?? String(e), ...facts };
  }
}

/**
 * 复制并校验:写目标同目录临时文件 → fsync → 校验 → 提交 → fsync 目录。
 * 目标已存在按 onExisting:conflict(缺省)抛 EEXIST 绝不覆盖;skip 不写;replace 走 replaceCommit(不裸覆盖)。
 * 写前顺带清理本目录超过 24h 的残留 partial(对照 MP,只清本目录、不全库扫,失败吞掉)。
 * 任何失败都会清理临时文件并抛出 FsError。
 *
 * 故意只保留 atime/mtime,**不复制权限与属主**:目标目录的默认权限/继承 ACL 是媒体库的访问策略,
 * 用源文件权限覆盖会清掉继承 ACL(与 MP _copy_with_target_permissions 同理)。不要补 chmod/chown。
 */
export async function copyWithVerify(src: string, dst: string, opts: CopyOptions = {}): Promise<CopyReceipt> {
  const started = Date.now();
  const method = opts.hash ?? "size";
  const onExisting = opts.onExisting ?? "conflict";
  let srcStat: Awaited<ReturnType<typeof stat>>;
  try {
    srcStat = await stat(src);
  } catch (e) {
    throw toFsError(e, src);
  }
  if (!srcStat.isFile()) throw new FsError("IO", `source is not a regular file: ${src}`, { path: src });
  const existing = await lstatOrNull(dst);
  if (existing) {
    if (onExisting === "skip")
      return {
        src,
        dst,
        bytes: 0,
        verifiedBy: method,
        partialPath: "",
        durationMs: Date.now() - started,
        outcome: "skipped_existing",
      };
    // 只替换普通文件;目录/软链(含失效软链)一律按冲突
    if (onExisting !== "replace" || !existing.isFile())
      throw new FsError("EEXIST", `target exists: ${dst}`, { path: dst });
  }
  if (opts.createParents) await mkdir(dirname(dst), { recursive: true });
  const sweptPartials =
    opts.sweepPartialsOlderThanMs === null
      ? []
      : await sweepStalePartials(
          dirname(dst),
          opts.sweepPartialsOlderThanMs ?? DEFAULT_PARTIAL_MAX_AGE_MS,
          opts.tmpSuffix,
        ).catch(() => [] as string[]);

  const tmp = partialPathFor(dst, opts.tmpSuffix);
  const hasher = method === "sha256" ? createHash("sha256") : null;
  let copied = 0;
  try {
    const meter = new Transform({
      transform(chunk: Buffer, _enc, cb) {
        copied += chunk.length;
        hasher?.update(chunk);
        opts.onProgress?.(copied, srcStat.size);
        cb(null, chunk);
      },
    });
    await pipeline(
      createReadStream(src, { highWaterMark: IO_CHUNK_BYTES }),
      meter,
      createWriteStream(tmp, { flags: "wx", flush: true, highWaterMark: IO_CHUNK_BYTES }),
      opts.signal ? { signal: opts.signal } : {},
    );
    await fsyncPath(tmp, "r+");

    const after = await stat(src);
    if (after.size !== srcStat.size || after.mtimeMs !== srcStat.mtimeMs) {
      throw new FsError("SOURCE_CHANGED", `source changed during copy: ${src}`, { path: src });
    }
    const tmpStat = await stat(tmp);
    if (tmpStat.size !== srcStat.size || copied !== srcStat.size) {
      throw new FsError("VERIFY_FAILED", `size mismatch ${tmpStat.size} != ${srcStat.size}`, { path: dst });
    }
    const receipt: CopyReceipt = {
      src,
      dst,
      bytes: srcStat.size,
      verifiedBy: method,
      partialPath: tmp,
      durationMs: 0,
      outcome: "committed",
      ...(sweptPartials.length ? { sweptPartials } : {}),
    };
    if (hasher) {
      const expected = hasher.digest("hex");
      // 回读临时文件做全量比对:主进程里走 sha256sum 子进程;fsproxy 子进程里走流式并上报心跳
      const actual = await sha256Full(tmp, opts.signal, opts.onVerifyProgress);
      if (actual.digest !== expected) throw new FsError("VERIFY_FAILED", "sha256 mismatch", { path: dst });
      receipt.sha256 = actual.digest;
      receipt.sha256Via = actual.via;
    } else if (method === "sample") {
      const a = await sampleDigest(src, opts.sampleChunks, opts.sampleChunkBytes, opts.onVerifyProgress);
      const b = await sampleDigest(tmp, opts.sampleChunks, opts.sampleChunkBytes, opts.onVerifyProgress);
      if (a !== b) throw new FsError("VERIFY_FAILED", "sample digest mismatch", { path: dst });
      receipt.sampleDigest = b;
    }
    opts.signal?.throwIfAborted();
    await utimes(tmp, srcStat.atime, srcStat.mtime);
    if (existing) receipt.replaced = await replaceCommit(tmp, dst, existing, opts.replaceOld);
    else await commitNoClobber(tmp, dst);
    await fsyncPath(dirname(dst));
    receipt.durationMs = Date.now() - started;
    return receipt;
  } catch (e) {
    await unlink(tmp).catch(() => undefined);
    if (opts.signal?.aborted) throw new FsError("ABORTED", "copy aborted", { path: dst, cause: e });
    throw toFsError(e, dst);
  }
}

/** 比对/哈希的 IO 选项:中止信号与进度(两文件的累计字节) */
export interface MatchIo {
  signal?: AbortSignal;
  onProgress?: ProgressFn;
}

const asIo = (x: AbortSignal | MatchIo | undefined): MatchIo =>
  x === undefined ? {} : x instanceof AbortSignal ? { signal: x } : x;

/**
 * 两个文件全量 sha256 比对(大小先比);两边并行,各走 sha256Full。返回是否一致与所用方式。
 * 读失败/超时**抛 FsError**(IO/TIMEOUT/ENOENT…),调用方不得把异常当成"内容不同"(EEXIST 误报)。
 */
export async function filesMatchSha256(
  a: string,
  b: string,
  io?: AbortSignal | MatchIo,
): Promise<{ match: boolean; via: Sha256Via | null }> {
  const { signal, onProgress } = asIo(io);
  let sa: Awaited<ReturnType<typeof stat>>;
  let sb: Awaited<ReturnType<typeof stat>>;
  try {
    [sa, sb] = await Promise.all([stat(a), stat(b)]);
  } catch (e) {
    throw toFsError(e);
  }
  if (sa.size !== sb.size) return { match: false, via: null };
  const total = Number(sa.size) * 2;
  const done = [0, 0];
  const prog = (i: 0 | 1) =>
    onProgress
      ? (n: number) => {
          done[i] = n;
          onProgress(done[0]! + done[1]!, total);
        }
      : undefined;
  try {
    const [ha, hb] = await Promise.all([sha256Full(a, signal, prog(0)), sha256Full(b, signal, prog(1))]);
    return { match: ha.digest === hb.digest, via: ha.via === hb.via ? ha.via : "stream" };
  } catch (e) {
    throw toFsError(e);
  }
}

/** 按指定方式比对两个文件是否一致(size 恒比对)。读失败抛 FsError。 */
export async function filesMatch(
  a: string,
  b: string,
  method: VerifyMethod = "size",
  io: MatchIo = {},
): Promise<boolean> {
  if (method === "sha256") return (await filesMatchSha256(a, b, io)).match;
  try {
    const [sa, sb] = await Promise.all([stat(a), stat(b)]);
    if (sa.size !== sb.size) return false;
    if (method === "sample")
      return (
        (await sampleDigest(a, undefined, undefined, io.onProgress)) ===
        (await sampleDigest(b, undefined, undefined, io.onProgress))
      );
    return true;
  } catch (e) {
    throw toFsError(e);
  }
}

/** 清理目录下超过 olderThanMs 的残留临时文件,返回被删除的路径。 */
export async function sweepStalePartials(dir: string, olderThanMs: number, suffix = ".uhd-partial"): Promise<string[]> {
  const removed: string[] = [];
  const now = Date.now();
  for (const name of await readdir(dir)) {
    if (!name.startsWith(".") || !name.endsWith(suffix)) continue;
    const p = join(dir, name);
    const st = await lstat(p).catch(() => null);
    if (st?.isFile() && now - st.mtimeMs > olderThanMs) {
      if (
        await unlink(p).then(
          () => true,
          () => false,
        )
      )
        removed.push(p);
    }
  }
  return removed;
}