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";

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

export interface CopyOptions {
  /** 临时文件后缀,默认 .uhd-partial;临时文件与目标同目录、以 . 开头 */
  tmpSuffix?: string;
  /** 校验方式,默认 size(总是比对大小;sha256/sample 额外比对内容) */
  hash?: VerifyMethod;
  /** sample 模式的抽样块数与块大小 */
  sampleChunks?: number;
  sampleChunkBytes?: number;
  /** 是否创建目标父目录;默认 false(防止库根离线时在挂载点下写入) */
  createParents?: boolean;
  signal?: AbortSignal;
  onProgress?: (bytesCopied: number, totalBytes: number) => void;
}

export interface CopyReceipt {
  src: string;
  dst: string;
  bytes: number;
  verifiedBy: VerifyMethod;
  sha256?: string;
  sampleDigest?: string;
  partialPath: string;
  durationMs: number;
}

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();
  }
}

export async function sha256File(p: string, signal?: AbortSignal): Promise<string> {
  const h = createHash("sha256");
  await pipeline(createReadStream(p), async (src: AsyncIterable<Buffer>) => {
    for await (const c of src) {
      if (signal?.aborted) throw new FsError("ABORTED", "aborted", { path: p });
      h.update(c);
    }
  });
  return h.digest("hex");
}

/** 抽样摘要:首/尾/等距 N 块 + 文件大小。抽样一致不等于完整一致性证明(REQ §8.2)。 */
export async function sampleDigest(p: string, chunks = 8, chunkBytes = 1024 * 1024): 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));
    }
    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}`);
}

/** 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 (!["EPERM", "ENOTSUP", "EOPNOTSUPP", "ENOSYS", "EMLINK"].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);
}

/**
 * 复制并校验:写目标同目录临时文件 → fsync → 校验 → no-clobber 原子提交 → fsync 目录。
 * 任何失败都会清理临时文件并抛出 FsError;目标已存在抛 EEXIST,绝不覆盖。
 */
export async function copyWithVerify(src: string, dst: string, opts: CopyOptions = {}): Promise<CopyReceipt> {
  const started = Date.now();
  const method = opts.hash ?? "size";
  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 });
  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);
  }
  if (opts.createParents) await mkdir(dirname(dst), { recursive: true });

  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),
      meter,
      createWriteStream(tmp, { flags: "wx", flush: true }),
      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,
    };
    if (hasher) {
      const expected = hasher.digest("hex");
      const actual = await sha256File(tmp, opts.signal);
      if (actual !== expected) throw new FsError("VERIFY_FAILED", "sha256 mismatch", { path: dst });
      receipt.sha256 = actual;
    } else if (method === "sample") {
      const a = await sampleDigest(src, opts.sampleChunks, opts.sampleChunkBytes);
      const b = await sampleDigest(tmp, opts.sampleChunks, opts.sampleChunkBytes);
      if (a !== b) throw new FsError("VERIFY_FAILED", "sample digest mismatch", { path: dst });
      receipt.sampleDigest = b;
    }
    await utimes(tmp, srcStat.atime, srcStat.mtime);
    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);
  }
}

/** 按指定方式比对两个文件是否一致(size 恒比对)。 */
export async function filesMatch(a: string, b: string, method: VerifyMethod = "size"): Promise<boolean> {
  const [sa, sb] = await Promise.all([stat(a), stat(b)]);
  if (sa.size !== sb.size) return false;
  if (method === "sha256") return (await sha256File(a)) === (await sha256File(b));
  if (method === "sample") return (await sampleDigest(a)) === (await sampleDigest(b));
  return true;
}

/** 清理目录下超过 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) {
      await unlink(p).catch(() => undefined);
      removed.push(p);
    }
  }
  return removed;
}