UHD_Media_Tool/apps/server/src/storage/transfer.ts

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