"use strict";

// Transfer engine tests against a simulated SFTP server backed by a real temporary directory.
// These are local unit/engine tests (no SSH); real servers are covered by filesIntegration.test.js.
const test = require("node:test");
const assert = require("node:assert/strict");
const crypto = require("node:crypto");
const fs = require("node:fs");
const os = require("node:os");
const path = require("node:path");
const { EventEmitter } = require("node:events");
const { TransferManager, summarize } = require("../src/main/files/transferManager");

const sha = (data) => crypto.createHash("sha256").update(data).digest("hex");
const tick = (ms = 0) => new Promise((resolve) => setTimeout(resolve, ms));
async function until(predicate, timeout = 10000) {
  const started = Date.now();
  while (!predicate()) { if (Date.now() - started > timeout) throw new Error("Timed out waiting for condition"); await tick(5); }
}
function sftpError(error) {
  const code = error.code === "ENOENT" ? 2 : error.code === "EACCES" ? 3 : 4;
  return Object.assign(new Error(error.message), { code });
}

class FakeServer {
  constructor(root) {
    this.root = root; this.writes = []; this.writeDelay = 0; this.failAfterWrites = Infinity; this.channels = 0; this.helperCalls = [];
    this.connected = true; this.readOnly = new Set();
  }
  local(remote) { return path.join(this.root, remote); }
  attrs(stats) {
    return { size: stats.size, mtime: Math.floor(stats.mtimeMs / 1000), mode: stats.mode, isDirectory: () => stats.isDirectory(), isFile: () => stats.isFile(), isSymbolicLink: () => stats.isSymbolicLink() };
  }
  sftp() {
    const server = this; const handles = new Map(); let ended = false; this.channels += 1;
    const guard = (cb) => { if (ended || !server.connected) { cb(Object.assign(new Error("No response from server"), { code: 7 })); return false; } return true; };
    return {
      open(p, flags, attrs, cb) {
        if (typeof attrs === "function") { cb = attrs; attrs = {}; }
        if (!guard(cb)) return;
        if (flags !== "r" && server.readOnly.has(path.posix.dirname(p))) return cb(Object.assign(new Error("Permission denied"), { code: 3 }));
        fs.open(server.local(p), flags, attrs?.mode ?? 0o644, (error, fd) => {
          if (error) return cb(sftpError(error));
          const handle = Buffer.from(crypto.randomBytes(4)); handles.set(handle.toString("hex"), fd); cb(null, handle);
        });
      },
      read(handle, buf, off, len, pos, cb) { if (!guard(cb)) return; fs.read(handles.get(handle.toString("hex")), buf, off, len, pos, (error, n) => error ? cb(sftpError(error)) : cb(undefined, n, buf, pos)); },
      write(handle, buf, off, len, pos, cb) {
        if (!guard(cb)) return;
        server.writes.push(pos);
        if (server.writes.length > server.failAfterWrites) { server.connected = false; return cb(Object.assign(new Error("No response from server"), { code: 7 })); }
        setTimeout(() => {
          if (ended || !server.connected) return cb(Object.assign(new Error("No response from server"), { code: 7 }));
          fs.write(handles.get(handle.toString("hex")), buf, off, len, pos, (error) => error ? cb(sftpError(error)) : cb(undefined, off + len));
        }, server.writeDelay);
      },
      fstat(handle, cb) { if (!guard(cb)) return; fs.fstat(handles.get(handle.toString("hex")), (error, stats) => error ? cb(sftpError(error)) : cb(null, server.attrs(stats))); },
      fsetstat(handle, attrs, cb) { fs.ftruncate(handles.get(handle.toString("hex")), attrs.size, (error) => cb(error ? sftpError(error) : undefined)); },
      close(handle, cb) { const fd = handles.get(handle.toString("hex")); handles.delete(handle.toString("hex")); fs.close(fd, () => cb()); },
      stat(p, cb) { if (!guard(cb)) return; fs.stat(server.local(p), (error, stats) => error ? cb(sftpError(error)) : cb(null, server.attrs(stats))); },
      lstat(p, cb) { if (!guard(cb)) return; fs.lstat(server.local(p), (error, stats) => error ? cb(sftpError(error)) : cb(null, server.attrs(stats))); },
      unlink(p, cb) { if (!guard(cb)) return; fs.unlink(server.local(p), (error) => cb(error ? sftpError(error) : undefined)); },
      rename(from, to, cb) {
        if (!guard(cb)) return;
        if (fs.existsSync(server.local(to))) return cb(Object.assign(new Error("Failure"), { code: 4 }));
        fs.rename(server.local(from), server.local(to), (error) => cb(error ? sftpError(error) : undefined));
      },
      mkdir(p, attrs, cb) { if (typeof attrs === "function") cb = attrs; fs.mkdir(server.local(p), (error) => cb(error ? sftpError(error) : undefined)); },
      readdir(p, cb) {
        fs.readdir(server.local(p), (error, names) => error ? cb(sftpError(error)) : cb(null, names.map((filename) => ({ filename, attrs: server.attrs(fs.lstatSync(path.join(server.local(p), filename))) }))));
      },
      end() { ended = true; for (const fd of handles.values()) try { fs.closeSync(fd); } catch {} handles.clear(); }
    };
  }
  // Simulated fixed remote helper (the real Python helper is exercised on Linux by remoteEditorLinux/filesIntegration).
  helper(request, auth) {
    this.helperCalls.push({ operation: request.operation, sudo: auth.sudo === true, password: auth.sudoPassword, retry: request.retry });
    if (!this.connected) throw Object.assign(new Error("lost"), { code: "SESSION_NOT_CONNECTED" });
    const lite = (stats) => ({ dev: stats.dev, ino: stats.ino, size: stats.size, mtimeNs: String(stats.mtimeNs ?? Math.trunc(stats.mtimeMs * 1e6)), ctimeNs: String(Math.trunc(stats.ctimeMs * 1e6)), uid: stats.uid, gid: stats.gid, mode: stats.mode & 0o7777, nlink: stats.nlink });
    const workspace = "/home/user/ch-j-sm/transfers";
    if (request.operation === "transfer-workspace") { fs.mkdirSync(this.local(workspace), { recursive: true, mode: 0o700 }); return { path: workspace, uid: 1000 }; }
    if (request.operation === "stat") {
      const stats = fs.existsSync(this.local(request.path)) ? fs.lstatSync(this.local(request.path)) : null;
      return stats ? { exists: true, type: stats.isDirectory() ? "directory" : "file", resolvedPath: request.path, baseline: lite(stats) } : { exists: false, resolvedPath: request.path };
    }
    if (request.operation === "hash") { const data = fs.readFileSync(this.local(request.path)); return { hash: sha(data), size: data.length }; }
    if (request.operation === "transfer-remove") { const file = this.local(`${workspace}/${request.id}.part`); const existed = fs.existsSync(file); fs.rmSync(file, { force: true }); return { removed: existed }; }
    if (request.operation === "transfer-list") { const dir = this.local(workspace); return { items: fs.existsSync(dir) ? fs.readdirSync(dir).map((name) => ({ id: name.slice(0, 32), size: 1, mtime: 0 })) : [] }; }
    if (request.operation === "mkdir") { fs.mkdirSync(this.local(request.path), { recursive: true }); return { path: request.path }; }
    if (request.operation === "stage-out") {
      if (!auth.sudo) throw Object.assign(new Error("denied"), { code: "FILE_PERMISSION_DENIED" });
      const data = fs.readFileSync(this.local(request.path)); fs.mkdirSync(this.local(workspace), { recursive: true });
      fs.writeFileSync(this.local(`${workspace}/${request.id}.part`), data, { mode: 0o600 });
      return { path: `${workspace}/${request.id}.part`, size: data.length, hash: sha(data), mtimeNs: "1" };
    }
    if (request.operation === "publish") {
      const target = this.local(request.path);
      if (this.publishFailure) { const error = this.publishFailure; this.publishFailure = null; throw error; }
      const current = fs.existsSync(target) ? lite(fs.lstatSync(target)) : null;
      if (request.retry && current && request.baseline && JSON.stringify(current) !== JSON.stringify(request.baseline) && sha(fs.readFileSync(target)) === request.hash) return { status: "published", alreadyPublished: true, hash: request.hash, resolvedPath: request.path };
      if (current && !request.baseline && !request.overwrite) throw Object.assign(new Error("exists"), { code: "FILE_EXISTS" });
      if (request.baseline && JSON.stringify(current) !== JSON.stringify(request.baseline)) throw Object.assign(new Error("conflict"), { code: "FILE_CONFLICT" });
      const data = fs.readFileSync(this.local(request.source));
      if (sha(data) !== request.hash || data.length !== request.size) throw Object.assign(new Error("verify"), { code: "FILE_VERIFY_FAILED" });
      const mode = current ? current.mode : request.mode;
      fs.mkdirSync(path.dirname(target), { recursive: true });
      if (request.strategy === "rename") { fs.chmodSync(this.local(request.source), mode); fs.renameSync(this.local(request.source), target); }
      else { fs.writeFileSync(target + ".tmp", data, { mode }); fs.renameSync(target + ".tmp", target); fs.rmSync(this.local(request.source)); }
      return { status: "published", hash: request.hash, resolvedPath: request.path, created: !current, metadataPreserved: Boolean(current), sourceRemoved: true };
    }
    throw new Error(`Unexpected helper operation ${request.operation}`);
  }
}

function setup(t, { storageRoot, server: existing, selected = [], downloadDirectory, downloadPath, retryDelaysMs, stopGraceMs } = {}) {
  const root = existing?.root || fs.mkdtempSync(path.join(os.tmpdir(), "chj-transfer-remote-"));
  const local = fs.mkdtempSync(path.join(os.tmpdir(), "chj-transfer-local-"));
  const storage = storageRoot || fs.mkdtempSync(path.join(os.tmpdir(), "chj-transfer-state-"));
  t.after(() => { for (const dir of [local, storage, root]) fs.rmSync(dir, { recursive: true, force: true }); });
  const server = existing || new FakeServer(root);
  for (const dir of ["/srv", "/etc/nginx"]) fs.mkdirSync(server.local(dir), { recursive: true });
  const sessions = new EventEmitter();
  sessions.list = () => server.connected ? [{ sessionId: "s1", profileId: "p", host: "h", port: 22, username: "user", state: "connected" }] : [];
  sessions.openSftp = async () => { if (!server.connected) throw Object.assign(new Error("not connected"), { code: "SESSION_NOT_CONNECTED" }); return server.sftp(); };
  const files = {
    helper: async (_sessionId, request, options = {}) => server.helper(request, options),
    _withSftp: async (_sessionId, action) => { const sftp = await sessions.openSftp(); try { return await action(sftp); } finally { sftp.end(); } },
    privileges: async () => ({ mode: "sudo-password", username: "user", sudoInstalled: true })
  };
  const manager = new TransferManager({ sessionManager: sessions, remoteFileService: files, storageRoot: storage, logger: null, retryDelaysMs, stopGraceMs,
    selectUploadPath: async () => ({ canceled: false, paths: selected.map((name) => path.join(local, name)) }),
    selectDownloadDirectory: async () => ({ canceled: false, path: downloadDirectory || local }),
    selectDownloadPath: async (name) => ({ canceled: false, path: downloadPath || path.join(local, name) }) });
  t.after(() => manager.shutdown());
  const updates = [];
  manager.on("update", (update) => updates.push(update));
  return { server, manager, local, storage, sessions, updates, setSelected: (names) => { selected = names; } };
}
const job = (manager) => manager.list("fm").jobs[0];

test("summary is byte based, keeps unknown sizes indeterminate and never shows a false 100 %", () => {
  const base = { speed: 0, eta: null };
  const summary = summarize([
    { ...base, state: "completed", size: 100, transferred: 100 },
    { ...base, state: "transferring", size: 900, transferred: 0, speed: 50, eta: 18 },
    { ...base, state: "failed", size: 0, transferred: 0 },
    { ...base, state: "cancelled", size: 1000, transferred: 10 }
  ]);
  assert.equal(summary.percent, 10, "1 of 2 files done but only 10 % of the bytes");
  assert.equal(summary.total, 4); assert.equal(summary.completed, 1); assert.equal(summary.active, 1); assert.equal(summary.failed, 1); assert.equal(summary.cancelled, 1);
  assert.equal(summary.totalBytes, 1000); assert.equal(summary.eta, 18);
  assert.equal(summarize([{ ...base, state: "transferring", size: null, transferred: 5 }]).percent, null);
  assert.equal(summarize([{ ...base, state: "verifying", size: 10, transferred: 10 }]).percent, 99.9);
  assert.equal(summarize([{ ...base, state: "completed", size: 0, transferred: 0 }]).percent, 100);
  assert.equal(summarize([{ ...base, state: "transferring", size: 10, transferred: 1, speed: 0, eta: null }]).eta, null, "no ETA without a reliable rate");
});

test("upload of a regular file streams to a sibling partial file and publishes it", async (t) => {
  const f = setup(t, { selected: ["a.bin"] });
  const data = crypto.randomBytes(3 * 1024 * 1024 + 123);
  fs.writeFileSync(path.join(f.local, "a.bin"), data);
  const plan = await f.manager.planUpload("fm", "s1", "/srv");
  assert.equal(plan.files, 1); assert.equal(plan.mode, "direct"); assert.equal(plan.bytes, data.length);
  await f.manager.start("fm", plan.planId);
  await until(() => job(f.manager)?.state === "completed");
  assert.deepEqual(fs.readFileSync(f.server.local("/srv/a.bin")), data);
  assert.deepEqual(fs.readdirSync(f.server.local("/srv")), ["a.bin"], "no partial files left");
  const done = job(f.manager);
  assert.equal(done.percent, 100); assert.equal(done.transferred, data.length);
  assert.equal(done.verification.method, "sha256");
  assert.equal(f.server.helperCalls.find((call) => call.operation === "publish").sudo, false, "no sudo for a writable target");
  const seen = f.updates.flatMap((update) => update.jobs).map((entry) => entry.transferred);
  assert.deepEqual(seen, [...seen].sort((a, b) => a - b), "progress never goes backwards");
});

test("pause stops the data flow; resume continues from the confirmed offset", async (t) => {
  const f = setup(t, { selected: ["big.bin"] });
  const data = crypto.randomBytes(8 * 1024 * 1024);
  fs.writeFileSync(path.join(f.local, "big.bin"), data);
  f.server.writeDelay = 15;
  const plan = await f.manager.planUpload("fm", "s1", "/srv");
  await f.manager.start("fm", plan.planId);
  await until(() => job(f.manager)?.transferred >= 2 * 1024 * 1024);
  f.manager.pause("fm", "all");
  await until(() => job(f.manager).state === "paused");
  const paused = job(f.manager);
  const writesAtPause = f.server.writes.length;
  await tick(150);
  assert.equal(f.server.writes.length, writesAtPause, "no SFTP writes while paused");
  assert.ok(paused.transferred > 0 && paused.transferred < data.length);
  assert.equal(paused.transferred % (256 * 1024), 0, "offset is a confirmed chunk boundary");
  f.server.writeDelay = 0;
  f.manager.resume("fm", "all");
  await until(() => job(f.manager).state === "completed");
  assert.equal(Math.min(...f.server.writes.slice(writesAtPause)), paused.transferred, "the resumed run starts at the confirmed offset");
  assert.ok(f.server.writes.slice(writesAtPause).every((position) => position >= paused.transferred), "nothing before the offset is sent again");
  assert.deepEqual(fs.readFileSync(f.server.local("/srv/big.bin")), data);
});

test("a quick resume is not stopped later by the grace timer of the earlier pause", async (t) => {
  const f = setup(t, { selected: ["slow.bin"], stopGraceMs: 80 });
  fs.writeFileSync(path.join(f.local, "slow.bin"), crypto.randomBytes(4 * 1024 * 1024));
  f.server.writeDelay = 15;
  const plan = await f.manager.planUpload("fm", "s1", "/srv");
  await f.manager.start("fm", plan.planId);
  await until(() => job(f.manager)?.transferred > 0);
  f.manager.pause("fm", "all");
  await until(() => job(f.manager).state === "paused");
  const states = [];
  const set = f.manager._set.bind(f.manager);
  f.manager._set = (entry, state, error) => { states.push(state); return set(entry, state, error); };
  f.manager.resume("fm", "all");
  await tick(250); // well past the old pause's grace period while the new run is active
  await until(() => job(f.manager).state === "completed");
  assert.equal(states.includes("paused"), false, "the resumed run was never stopped by the old timer");
});

test("SSH loss interrupts with a safe offset; reconnect resumes and completes", async (t) => {
  const f = setup(t, { selected: ["c.bin"] });
  const data = crypto.randomBytes(4 * 1024 * 1024);
  fs.writeFileSync(path.join(f.local, "c.bin"), data);
  f.manager.updateSettings("fm", { autoRetry: true });
  const plan = await f.manager.planUpload("fm", "s1", "/srv");
  f.server.failAfterWrites = 6;
  await f.manager.start("fm", plan.planId);
  await until(() => ["interrupted", "reconnecting"].includes(job(f.manager).state));
  const interrupted = job(f.manager);
  assert.equal(interrupted.error.code, "SESSION_NOT_CONNECTED");
  assert.ok(interrupted.transferred <= 6 * 256 * 1024);
  assert.deepEqual(fs.readdirSync(f.server.local("/srv")).filter((name) => name === "c.bin"), [], "target not published from partial data");
  f.server.failAfterWrites = Infinity; f.server.connected = true;
  const offset = interrupted.transferred; const before = f.server.writes.length;
  f.sessions.emit("state", { state: "connected", profileId: "p", host: "h", port: 22, username: "user" });
  await until(() => job(f.manager).state === "completed");
  assert.equal(Math.min(...f.server.writes.slice(before)), offset);
  assert.deepEqual(fs.readFileSync(f.server.local("/srv/c.bin")), data);
});

test("automatic retries are bounded and end in a manual interrupted state", async (t) => {
  const f = setup(t, { selected: ["d.bin"], retryDelaysMs: [5, 10, 20] });
  fs.writeFileSync(path.join(f.local, "d.bin"), crypto.randomBytes(1024 * 1024));
  const plan = await f.manager.planUpload("fm", "s1", "/srv");
  const states = [];
  const set = f.manager._set.bind(f.manager);
  f.manager._set = (entry, state, error) => { states.push(state); return set(entry, state, error); };
  f.server.connected = false; // The server stays unreachable: retries must stop on their own.
  await f.manager.start("fm", plan.planId);
  await until(() => job(f.manager).state === "interrupted" && job(f.manager).attempts === 3);
  await tick(200);
  assert.equal(job(f.manager).state, "interrupted");
  assert.equal(job(f.manager).attempts, 3, "exactly the configured number of attempts");
  assert.equal(states.filter((state) => state === "reconnecting").length, 3);
  assert.equal(job(f.manager).actions.resume, true, "manual resume stays available");
});

test("application restart offers resume and continues the persisted job", async (t) => {
  const storage = fs.mkdtempSync(path.join(os.tmpdir(), "chj-transfer-state-"));
  const first = setup(t, { storageRoot: storage, selected: ["e.bin"] });
  const data = crypto.randomBytes(6 * 1024 * 1024);
  fs.writeFileSync(path.join(first.local, "e.bin"), data);
  first.server.writeDelay = 10;
  const plan = await first.manager.planUpload("fm", "s1", "/srv");
  await first.manager.start("fm", plan.planId);
  await until(() => job(first.manager).transferred >= 1024 * 1024);
  first.manager.shutdown();
  await until(() => !first.manager.runtime.get(job(first.manager).id)?.running);
  const persisted = JSON.parse(fs.readFileSync(path.join(storage, "transfers.json"), "utf8"));
  assert.equal(JSON.stringify(persisted).includes("password"), false, "no secrets in persisted state");
  const offset = persisted.jobs[0].transferred;
  assert.ok(offset > 0);
  first.server.writeDelay = 0;
  const second = setup(t, { storageRoot: storage, server: first.server });
  const restored = job(second.manager);
  assert.equal(restored.state, "interrupted"); assert.equal(restored.error.code, "APP_RESTARTED"); assert.equal(restored.actions.resume, true);
  const before = first.server.writes.length;
  second.manager.resume("fm", [restored.id]);
  await until(() => job(second.manager).state === "completed");
  assert.equal(Math.min(...first.server.writes.slice(before)), offset);
  assert.deepEqual(fs.readFileSync(first.server.local("/srv/e.bin")), data);
});

test("cancel stops the transfer, removes only its partial file and keeps the existing target", async (t) => {
  const f = setup(t, { selected: ["keep.conf"] });
  fs.writeFileSync(path.join(f.local, "keep.conf"), crypto.randomBytes(6 * 1024 * 1024));
  fs.writeFileSync(f.server.local("/srv/keep.conf"), "original");
  f.server.writeDelay = 15;
  const plan = await f.manager.planUpload("fm", "s1", "/srv");
  assert.equal(plan.conflictCount, 1);
  await assert.rejects(() => f.manager.start("fm", plan.planId), { code: "FILE_EXISTS" });
  await f.manager.start("fm", plan.planId, { conflict: "overwrite" });
  await until(() => job(f.manager).transferred > 0);
  await f.manager.cancel("fm", "all");
  await until(() => job(f.manager).state === "cancelled");
  assert.equal(fs.readFileSync(f.server.local("/srv/keep.conf"), "utf8"), "original");
  assert.deepEqual(fs.readdirSync(f.server.local("/srv")), ["keep.conf"]);
});

test("several files run concurrently within the limit with correct individual and total progress", async (t) => {
  const f = setup(t, { selected: ["dir"] });
  fs.mkdirSync(path.join(f.local, "dir/sub"), { recursive: true });
  const sizes = { "dir/one.bin": 1024 * 1024, "dir/two.bin": 2 * 1024 * 1024, "dir/sub/three.bin": 512 * 1024, "dir/empty.txt": 0 };
  for (const [name, size] of Object.entries(sizes)) fs.writeFileSync(path.join(f.local, name), crypto.randomBytes(size));
  f.manager.updateSettings("fm", { maxConcurrent: 2 });
  f.server.writeDelay = 5;
  const plan = await f.manager.planUpload("fm", "s1", "/srv");
  assert.equal(plan.files, 4); assert.equal(plan.directories, 2);
  let peak = 0;
  f.manager.on("update", (update) => { peak = Math.max(peak, update.summary.active); });
  await f.manager.start("fm", plan.planId);
  await until(() => f.manager.list("fm").jobs.every((entry) => entry.state === "completed"));
  assert.ok(peak <= 2, `at most 2 active, saw ${peak}`);
  const summary = f.manager.list("fm").summary;
  assert.equal(summary.percent, 100); assert.equal(summary.totalBytes, 3.5 * 1024 * 1024); assert.equal(summary.completed, 4);
  for (const [name] of Object.entries(sizes)) assert.deepEqual(fs.readFileSync(f.server.local(`/srv/${name}`)), fs.readFileSync(path.join(f.local, name)));
});

test("a source changed during an interruption is detected before resuming", async (t) => {
  const f = setup(t, { selected: ["s.bin"] });
  const file = path.join(f.local, "s.bin");
  fs.writeFileSync(file, crypto.randomBytes(4 * 1024 * 1024));
  f.server.writeDelay = 10;
  const plan = await f.manager.planUpload("fm", "s1", "/srv");
  await f.manager.start("fm", plan.planId);
  await until(() => job(f.manager).transferred >= 512 * 1024);
  f.manager.pause("fm", "all");
  await until(() => job(f.manager).state === "paused");
  fs.appendFileSync(file, "changed");
  f.manager.resume("fm", "all");
  await until(() => job(f.manager).state === "failed");
  assert.equal(job(f.manager).error.code, "FILE_SOURCE_CHANGED");
  assert.equal(job(f.manager).actions.retry, true);
  await f.manager.retry("fm", "all");
  f.server.writeDelay = 0;
  await until(() => job(f.manager).state === "completed");
  assert.deepEqual(fs.readFileSync(f.server.local("/srv/s.bin")), fs.readFileSync(file));
});

test("partial data that no longer matches the source is never continued", async (t) => {
  const f = setup(t, { selected: ["m.bin"] });
  fs.writeFileSync(path.join(f.local, "m.bin"), crypto.randomBytes(4 * 1024 * 1024));
  f.server.writeDelay = 10;
  const plan = await f.manager.planUpload("fm", "s1", "/srv");
  await f.manager.start("fm", plan.planId);
  await until(() => job(f.manager).transferred >= 1024 * 1024);
  f.manager.pause("fm", "all");
  await until(() => job(f.manager).state === "paused");
  const part = fs.readdirSync(f.server.local("/srv")).find((name) => name.includes(".chj-part-"));
  const corrupt = (position) => { const fd = fs.openSync(f.server.local(`/srv/${part}`), "r+"); fs.writeSync(fd, Buffer.from("corrupt"), 0, 7, position); fs.closeSync(fd); };
  corrupt(job(f.manager).transferred - 100); // inside the verified tail window
  f.manager.resume("fm", "all");
  await until(() => job(f.manager).state === "failed");
  assert.equal(job(f.manager).error.code, "FILE_RESUME_MISMATCH");
  assert.equal(fs.existsSync(f.server.local("/srv/m.bin")), false);
  // Damage before the tail window is caught by the end-to-end SHA-256 before publication.
  await f.manager.retry("fm", "all");
  await until(() => job(f.manager).transferred >= 2 * 1024 * 1024);
  f.manager.pause("fm", "all");
  await until(() => job(f.manager).state === "paused");
  corrupt(1000);
  f.server.writeDelay = 0;
  f.manager.resume("fm", "all");
  await until(() => job(f.manager).state === "failed");
  assert.equal(job(f.manager).error.code, "FILE_VERIFY_FAILED");
  assert.equal(fs.existsSync(f.server.local("/srv/m.bin")), false, "corrupted data is never published");
});

test("upload into a protected directory stages privately and publishes through sudo", async (t) => {
  const f = setup(t, { selected: ["nginx.conf"] });
  const data = Buffer.from("worker_processes auto;\n".repeat(5000));
  fs.writeFileSync(path.join(f.local, "nginx.conf"), data);
  fs.writeFileSync(f.server.local("/etc/nginx/nginx.conf"), "old");
  fs.chmodSync(f.server.local("/etc/nginx/nginx.conf"), 0o640);
  f.server.readOnly.add("/etc/nginx");
  const plan = await f.manager.planUpload("fm", "s1", "/etc/nginx");
  assert.equal(plan.mode, "privileged"); assert.equal(plan.requiresSudo, true); assert.equal(plan.sudoPasswordRequired, true);
  await f.manager.start("fm", plan.planId, { conflict: "overwrite", sudoPassword: "secret" });
  await until(() => job(f.manager).state === "completed");
  assert.deepEqual(fs.readFileSync(f.server.local("/etc/nginx/nginx.conf")), data);
  // Windows does not expose POSIX permission bits for this local fake server.
  if (process.platform !== "win32") assert.equal(fs.statSync(f.server.local("/etc/nginx/nginx.conf")).mode & 0o777, 0o640, "original mode kept");
  const publish = f.server.helperCalls.find((call) => call.operation === "publish");
  assert.equal(publish.sudo, true); assert.equal(publish.password, "secret");
  assert.deepEqual(fs.readdirSync(f.server.local("/home/user/ch-j-sm/transfers")), [], "staged copy removed");
  assert.equal(job(f.manager).mode, "privileged"); assert.equal(job(f.manager).finalize.state, "done");
});

test("lost confirmation of the privileged replacement is retried idempotently", async (t) => {
  const f = setup(t, { selected: ["site.conf"] });
  fs.writeFileSync(path.join(f.local, "site.conf"), "server {}\n");
  f.server.readOnly.add("/etc/nginx");
  const plan = await f.manager.planUpload("fm", "s1", "/etc/nginx");
  f.manager.updateSettings("fm", { autoRetry: false });
  f.server.publishFailure = Object.assign(new Error("lost"), { code: "FILE_RESULT_UNKNOWN" });
  await f.manager.start("fm", plan.planId, { sudoPassword: "secret" });
  await until(() => job(f.manager).state === "interrupted");
  assert.equal(job(f.manager).finalize.state, "unknown");
  f.manager.resume("fm", "all");
  await until(() => job(f.manager).state === "completed");
  assert.equal(f.server.helperCalls.filter((call) => call.operation === "publish").at(-1).retry, true);
  assert.equal(fs.readFileSync(f.server.local("/etc/nginx/site.conf"), "utf8"), "server {}\n");
});

test("download of a regular file, pause/resume and checksum verification", async (t) => {
  const f = setup(t);
  const data = crypto.randomBytes(5 * 1024 * 1024 + 7);
  fs.writeFileSync(f.server.local("/srv/remote.bin"), data);
  const plan = await f.manager.planDownload("fm", "s1", ["/srv/remote.bin"]);
  assert.equal(plan.files, 1);
  const server = f.server; const sftp = server.sftp.bind(server);
  let reads = 0;
  server.sftp = () => { const channel = sftp(); const read = channel.read; channel.read = (...args) => { reads += 1; setTimeout(() => read(...args), 8); }; return channel; };
  await f.manager.start("fm", plan.planId);
  await until(() => job(f.manager).transferred >= 1024 * 1024);
  f.manager.pause("fm", "all");
  await until(() => job(f.manager).state === "paused");
  const atPause = reads; await tick(100); assert.equal(reads, atPause, "no reads while paused");
  f.manager.resume("fm", "all");
  await until(() => job(f.manager).state === "completed");
  assert.deepEqual(fs.readFileSync(path.join(f.local, "remote.bin")), data);
  assert.equal(job(f.manager).verification.method, "sha256");
  assert.equal(job(f.manager).verification.matched, true);
  assert.deepEqual(fs.readdirSync(f.local), ["remote.bin"]);
});

test("privileged download stages the protected file and removes the staged copy", async (t) => {
  const f = setup(t);
  fs.writeFileSync(f.server.local("/etc/nginx/secret.conf"), "ssl_certificate_key /x;\n");
  const server = f.server; const sftp = server.sftp.bind(server);
  server.sftp = () => { const channel = sftp(); const open = channel.open; channel.open = (p, flags, attrs, cb) => p.startsWith("/etc/") ? (cb || attrs)(Object.assign(new Error("Permission denied"), { code: 3 })) : open(p, flags, attrs, cb); return channel; };
  const plan = await f.manager.planDownload("fm", "s1", ["/etc/nginx/secret.conf"]);
  await f.manager.start("fm", plan.planId);
  await until(() => job(f.manager).state === "interrupted");
  assert.equal(job(f.manager).authRequired, true);
  f.manager.resume("fm", "all", { sudo: true, sudoPassword: "secret" });
  await until(() => job(f.manager).state === "completed");
  assert.equal(fs.readFileSync(path.join(f.local, "secret.conf"), "utf8"), "ssl_certificate_key /x;\n");
  // POSIX modes are verified on macOS/Linux; Windows uses filesystem ACLs.
  if (process.platform !== "win32") assert.equal(fs.statSync(path.join(f.local, "secret.conf")).mode & 0o777, 0o600, "protected download stays private");
  assert.deepEqual(fs.readdirSync(f.server.local("/home/user/ch-j-sm/transfers")), []);
});

test("a large transfer keeps memory bounded by the pipeline window", async (t) => {
  const f = setup(t, { selected: ["large.bin"] });
  const file = path.join(f.local, "large.bin");
  const fd = fs.openSync(file, "w"); const block = crypto.randomBytes(4 * 1024 * 1024);
  for (let index = 0; index < 48; index += 1) fs.writeSync(fd, block);
  fs.closeSync(fd);
  const plan = await f.manager.planUpload("fm", "s1", "/srv");
  global.gc?.();
  const baseline = process.memoryUsage().rss; let peak = baseline;
  const sampler = setInterval(() => { peak = Math.max(peak, process.memoryUsage().rss); }, 20);
  await f.manager.start("fm", plan.planId);
  await until(() => job(f.manager).state === "completed", 60000);
  clearInterval(sampler);
  t.diagnostic(`192 MiB upload: rss growth ${Math.round((peak - baseline) / 1048576)} MiB`);
  assert.ok(peak - baseline < 128 * 1024 * 1024, "the file is never buffered in memory as a whole");
  assert.equal(fs.statSync(f.server.local("/srv/large.bin")).size, 192 * 1024 * 1024);
});