import { test } from "node:test";
import assert from "node:assert/strict";
import {
  chmod,
  lstat,
  mkdir,
  mkdtemp,
  readdir,
  readFile,
  readlink,
  rename,
  rm,
  stat,
  symlink,
  utimes,
  writeFile,
} from "node:fs/promises";
import { writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import type { Paths } from "../lib/paths";
import type { SiteSettings } from "../lib/settings";
import {
  inspectChannelMedia,
  readRelocationMarker,
} from "../lib/channelMedia";
import {
  relocatedMediaDir,
  tierChannelMedia,
  tierLinkTarget,
} from "../lib/mediaTier-server";
import type { ChannelWriter } from "./channelWriters";
import type { JobRecord } from "../jobs/registry";
import {
  assertRelocationRootPresent,
  channelMediaWriters,
  previewRelocation,
  relocateChannelMedia,
  relocationRootPresenceProblem,
} from "./relocateChannelMedia";
import type {
  StorageLocation,
  StorageSettings,
} from "../lib/storageLocations";
import {
  resetStorageProbeMemo,
  type VolumeBins,
} from "../lib/storageVolumes";
import { readChannelConfig } from "./channels";

// Run with:
//   pnpm --filter yt-dlp-transcript-common exec tsx --test controller/relocateChannelMedia.test.ts
//
// These exercise the REAL rsync binary (paths.rsyncBin -> "rsync"), like
// controller/backupSavedVideos.test.ts. Everything happens inside one mkdtemp.
//
// RELEASE 17: the unit of a move is `channels/<slug>/media` (the media tier),
// never `data/`. A seeded channel is CLASSIC — its big files real in
// `data/<id>/` — and the move's preflight tiers it; the crash cases build the
// tiered state first (`tierInPlace`) and then the exact on-disk state of the
// crash.

const MTIME = new Date("2021-03-04T05:06:07.000Z");

// THE SETTINGS SEAM, AND WHY EVERY CALL BELOW PASSES IT.
//
// The mover's space check demands `bytes + resumeMarginGB` free on the
// destination, and the shipped default is 2 GB. These fixtures live under
// `os.tmpdir()`, which on Linux is a TMPFS sized from RAM — with under 2 GB
// free on it, eight of these tests failed with "2.9 KB to move plus a 2 GB
// resume margin", on a machine and in a suite that had nothing to do with disk
// space. A test that fails because of how much RAM is left is not testing the
// mover.
//
// `minFreeDiskGB: 0` is the honest way to say it: the margin exists to clear
// the disk gate's floor, and with the gate off there is no floor to clear —
// which is production's own rule (`minFreeDiskGB > 0 ? resumeMarginGB : 0`),
// not a special case invented here. The margin ITSELF still has its real value,
// so nothing about the arithmetic is faked.
const TEST_IO = {
  read: () =>
    ({
      minFreeDiskGB: 0,
      resumeMarginGB: 2,
      // No configured locations: the root-presence guard is then stat-only,
      // which is what a hand-typed tmpdir root is in production too.
      storage: { locations: [], defaultLocationId: "" },
    }) as unknown as SiteSettings,
};

async function withTmp(
  fn: (paths: Paths, root: string, dir: string) => Promise<void>,
): Promise<void> {
  const dir = await mkdtemp(path.join(tmpdir(), "ttb-relocate-"));
  // THE CORPUS AND THE PLATTER ARE SIBLINGS under the tmpdir, never nested. A
  // root inside the corpus is refused by design (relocationRootProblem: it
  // would copy the channel onto itself and then reclaim the only copy), so a
  // harness that nested them would be exercising a shape the controller
  // rejects — and would have hidden exactly the bug that check exists for.
  const transcriptsDir = path.join(dir, "corpus");
  const paths = {
    transcriptsDir,
    channelsDir: path.join(transcriptsDir, "channels"),
    rsyncBin: "rsync",
  } as Paths;
  const root = path.join(dir, "platter");
  await mkdir(paths.channelsDir, { recursive: true });
  await mkdir(root, { recursive: true });
  try {
    await fn(paths, root, dir);
  } finally {
    await rm(dir, { recursive: true, force: true });
  }
}

async function seed(
  paths: Paths,
  slug: string,
  videos: Record<string, Record<string, string>>,
  config: Record<string, unknown> = {},
): Promise<string> {
  const channelDir = path.join(paths.channelsDir, slug);
  await mkdir(channelDir, { recursive: true });
  await writeFile(
    path.join(channelDir, "config.json"),
    JSON.stringify(
      { handling: "transcribe", url: "https://example.com/c", ...config },
      null,
      2,
    ) + "\n",
  );
  // These live in the CHANNEL dir, not data/ — they must not move.
  await writeFile(path.join(channelDir, "archive"), "youtube v1\n");
  await writeFile(path.join(channelDir, "playlist"), "https://x/v1\n");
  for (const [id, files] of Object.entries(videos)) {
    const videoDir = path.join(channelDir, "data", id);
    await mkdir(videoDir, { recursive: true });
    for (const [name, content] of Object.entries(files)) {
      const file = path.join(videoDir, name);
      await writeFile(file, content);
      // A known mtime, so "rsync -a preserved it" is an assertion and not a
      // coincidence.
      await utimes(file, MTIME, MTIME);
    }
  }
  return channelDir;
}

// A classic channel tiered in place: `media/` a real directory on the corpus
// disk, every big file in it and a relative link left in `data/<id>/` — what a
// move's preflight does first, and the state every crash case starts from.
async function tierInPlace(paths: Paths, slug: string): Promise<string> {
  await tierChannelMedia(paths, slug, { createMediaDir: true });
  return path.join(paths.channelsDir, slug, "media");
}

test("out: copies, links, records the target, keeps mtimes and reclaims the source", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: {
        "audio.m4a": "one".repeat(500),
        "transcript.json": "{}",
        "transcript.live_chat.json": "[]",
      },
      v2: { "audio.m4a": "two".repeat(500), "source-media.mp4": "container" },
    });
    const target = relocatedMediaDir(root, "alpha");
    const mediaLink = path.join(channelDir, "media");

    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    assert.equal(res.direction, "out");
    assert.equal(res.target, target);
    // The two audio files and the raw live chat — the media tier. The text
    // and the persisted container are not in it.
    assert.equal(res.files, 3);
    assert.equal(res.tiered, 3, "the classic channel was tiered first");
    assert.equal(res.resumed, false);

    // `media` is ONE absolute symlink at the target...
    assert.ok((await lstat(mediaLink)).isSymbolicLink());
    assert.equal(await readlink(mediaLink), target);
    // ...`data/` stays a real directory on the corpus disk...
    assert.ok((await lstat(path.join(channelDir, "data"))).isDirectory());
    assert.ok(!(await lstat(path.join(channelDir, "data"))).isSymbolicLink());
    // ...each big file is a RELATIVE link into `media/` that resolves through it.
    const audio = path.join(channelDir, "data", "v1", "audio.m4a");
    assert.ok((await lstat(audio)).isSymbolicLink());
    assert.equal(await readlink(audio), tierLinkTarget("v1", "audio.m4a"));
    assert.equal(await readFile(audio, "utf8"), "one".repeat(500));
    assert.ok(
      (await lstat(
        path.join(channelDir, "data", "v1", "transcript.live_chat.json"),
      )).isSymbolicLink(),
    );
    // ...and the text, and what is not tierable, never left.
    for (const real of [["v1", "transcript.json"], ["v2", "source-media.mp4"]]) {
      const st = await lstat(path.join(channelDir, "data", ...real));
      assert.ok(st.isFile() && !st.isSymbolicLink(), real.join("/"));
    }
    await assert.rejects(() => stat(path.join(target, "v1", "transcript.json")));
    assert.equal(
      (await stat(path.join(target, "v1", "audio.m4a"))).mtime.getTime(),
      MTIME.getTime(),
    );

    // The record is in config.json, written only now, on success.
    const config = await readChannelConfig(paths, "alpha");
    assert.equal(config?.mediaDir, target);
    assert.equal(config?.dataDir, undefined);
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "ok");

    // The parked copy is gone and the marker with it — this is the step that
    // actually frees the source volume.
    const left = await readdir(channelDir);
    assert.deepEqual(
      left.filter((n) => n.startsWith("media.") || n.startsWith("data.")),
      [],
    );
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
    // Channel-level files never moved.
    assert.ok(left.includes("archive"));
    assert.ok(left.includes("playlist"));
  });
});

test("a channel already tiered in place moves without tiering anything", async () => {
  await withTmp(async (paths, root) => {
    await seed(paths, "alpha", { v1: { "audio.m4a": "one" } });
    await tierInPlace(paths, "alpha");
    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    assert.equal(res.tiered, 0);
    assert.equal(res.files, 1);
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "ok");
  });
});

test("abort from an onLog hook leaves the source intact, and the rerun completes", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "x".repeat(20000), "transcript.json": "{}" },
    });
    const target = relocatedMediaDir(root, "alpha");

    // Aborting from the log hook is how a Cancel button reaches this code: the
    // job's onLog and its AbortSignal belong to the same run. Firing on the
    // first line is deterministic — what it pins is the contract (the source is
    // untouched, the marker is kept, a rerun resumes), not rsync's own
    // --partial mechanics, which are rsync's to test.
    const controller = new AbortController();
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "out",
          root,
          signal: controller.signal,
          onLog: () => controller.abort(),
        }),
      /Cancelled/,
    );

    // The source is a REAL `media/` still, with its file, reached through the
    // link in `data/`.
    assert.ok((await lstat(path.join(channelDir, "media"))).isDirectory());
    assert.ok(!(await lstat(path.join(channelDir, "media"))).isSymbolicLink());
    assert.equal(
      (await readFile(path.join(channelDir, "data", "v1", "audio.m4a"), "utf8"))
        .length,
      20000,
    );
    // Nothing was recorded, and the marker says where it was going — and that
    // it moves the media only.
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, undefined);
    const marker = await readRelocationMarker(paths, "alpha");
    assert.equal(marker?.target, target);
    assert.equal(marker?.direction, "out");
    assert.equal(marker?.phase, "copy");
    assert.equal(marker?.scope, "media");
    // Every media guard now reads the channel as in transition; its text stays
    // readable.
    const during = await inspectChannelMedia(paths, "alpha", undefined, {
      fresh: true,
    });
    assert.equal(during.status, "in-transition");
    assert.equal(during.text.readable, true);

    // The rerun finishes it. It is the ONE caller allowed to look past its own
    // marker.
    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    assert.equal(res.resumed, true);
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "ok");
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, target);
  });
});

// A STRAY FILE ON THE DESTINATION IS MIRRORED AWAY, NOT REFUSED (release 16
// slice RM). Until this slice the copy never deleted, so a stray at the target
// passed the itemized dry run and failed the counts — forever: no rerun could
// settle it. The mirror pass removes it from the COPY; the source is never the
// target of --delete.
test("a stray file on the destination is removed by the mirror pass", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one" },
    });
    const target = relocatedMediaDir(root, "alpha");
    await mkdir(path.join(target, "v9"), { recursive: true });
    await writeFile(path.join(target, "v9", "stray.m4a"), "not ours");

    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    assert.equal(res.files, 1);
    await assert.rejects(() => stat(path.join(target, "v9")));
    assert.equal(
      await readFile(path.join(target, "v1", "audio.m4a"), "utf8"),
      "one",
    );
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, target);
    assert.deepEqual(await leftoverCopies(channelDir), []);
  });
});

// WHAT STILL REFUSES: a source that keeps changing. A writer that lands a file
// before every check is seen by the first verify (one more mirror pass) and
// again by the second, and the move refuses naming the paths by kind — with the
// source untouched, the config unwritten, and the marker left for a Reconcile
// and resume once the writer has stopped.
test("a verify failure keeps the source and does not write the config", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one" },
    });
    let n = 0;
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "out",
          root,
          onLog: (m) => {
            if (m.startsWith("$ ") && m.includes("--dry-run")) {
              n++;
              writeFileSync(
                path.join(channelDir, "media", "v1", `chunk-${n}.json`),
                "{}",
              );
            }
          },
        }),
      /Verification failed: after a second mirror pass .*1 missing on the destination \(v1\/chunk-2\.json\)/,
    );

    assert.ok((await lstat(path.join(channelDir, "media"))).isDirectory());
    assert.ok(!(await lstat(path.join(channelDir, "media"))).isSymbolicLink());
    assert.equal(
      await readFile(path.join(channelDir, "data", "v1", "audio.m4a"), "utf8"),
      "one",
    );
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, undefined);
    assert.equal((await readRelocationMarker(paths, "alpha"))?.phase, "copy");
  });
});

test("back: restores a real directory, clears the config and reclaims the target", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one", "transcript.json": "{}" },
    });
    const target = relocatedMediaDir(root, "alpha");
    const mediaLink = path.join(channelDir, "media");
    await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });

    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "back",
      onLog: () => {},
    });
    assert.equal(res.direction, "back");
    assert.equal(res.files, 1);
    assert.equal(res.tiered, 0);

    // `media/` is a REAL directory on the corpus disk again, and the per-file
    // link was never touched: it resolves there now (no "untier").
    const back = await lstat(mediaLink);
    assert.ok(back.isDirectory());
    assert.ok(!back.isSymbolicLink());
    const audio = path.join(channelDir, "data", "v1", "audio.m4a");
    assert.ok((await lstat(audio)).isSymbolicLink());
    assert.equal(await readlink(audio), tierLinkTarget("v1", "audio.m4a"));
    assert.equal(await readFile(audio, "utf8"), "one");
    assert.equal((await stat(audio)).mtime.getTime(), MTIME.getTime());
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, undefined);
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "in-place");
    // The target is reclaimed, and so is <root>/<slug> once it is empty.
    assert.equal(await pathThere(target), false);
    assert.equal(await pathThere(path.dirname(target)), false);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
    assert.equal(
      (await readdir(channelDir)).includes("media.incoming"),
      false,
    );

    // And the next move out has nothing to tier.
    const again = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    assert.equal(again.tiered, 0);
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "ok");
  });
});

test("moving back a channel that was never moved is refused", async () => {
  await withTmp(async (paths) => {
    await seed(paths, "alpha", { v1: { "audio.m4a": "one" } });
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "back",
          onLog: () => {},
        }),
      /already in place/,
    );
  });
});

test("a relocated channel is refused a second move without a move back", async () => {
  await withTmp(async (paths, root, dir) => {
    await seed(paths, "alpha", { v1: { "audio.m4a": "one" } });
    await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    const other = path.join(dir, "platter2");
    await mkdir(other, { recursive: true });
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "out",
          root: other,
          onLog: () => {},
        }),
      /already relocated/,
    );
  });
});

test("an unmounted root is refused before anything is written", async () => {
  await withTmp(async (paths, _root, dir) => {
    const channelDir = await seed(paths, "alpha", { v1: { "audio.m4a": "one" } });
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "out",
          root: path.join(dir, "not-mounted"),
          onLog: () => {},
        }),
      /does not exist or is not a directory/,
    );
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
    assert.ok((await lstat(path.join(channelDir, "data"))).isDirectory());
    // Not even tiered: the root is refused before the preflight runs.
    assert.equal(await pathIsThere(path.join(channelDir, "media")), false);
    assert.ok(
      (await lstat(path.join(channelDir, "data", "v1", "audio.m4a"))).isFile(),
    );
  });
});

test("a social channel has no media to relocate", async () => {
  await withTmp(async (paths, root) => {
    await seed(paths, "poster", {}, { sourceKind: "social" });
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "poster",
          direction: "out",
          root,
          onLog: () => {},
        }),
      /social channel/,
    );
  });
});

test("preview tiers a classic channel first, measures media/ and both volumes, and moves nothing", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "abcde", "transcript.json": "{}" },
      v2: { "audio.m4a": "fgh" },
    });
    const preview = await previewRelocation({ paths, slug: "alpha", root });
    assert.equal(preview.target, relocatedMediaDir(root, "alpha"));
    // The media tier only: two audio files, not the transcript.
    assert.equal(preview.files, 2);
    assert.equal(preview.bytesToMove, 5 + 3);
    assert.equal(preview.tieredFirst, 2);
    assert.equal(preview.existingPartial, false);
    assert.ok(preview.freeOnRoot > 0);
    assert.ok(preview.freeOnSource > 0);
    // Both dirs are under one tmpdir, so this really is the same volume.
    assert.equal(preview.sameDevice, true);
    // Tiered in place — `media/` a real directory, the files linked — and
    // nothing moved off the corpus disk.
    assert.ok((await lstat(path.join(channelDir, "data"))).isDirectory());
    assert.ok(!(await lstat(path.join(channelDir, "media"))).isSymbolicLink());
    assert.ok(
      (await lstat(path.join(channelDir, "data", "v1", "audio.m4a"))).isSymbolicLink(),
    );
    assert.equal(await pathThere(path.join(root, "alpha")), false);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
    assert.equal(
      (await inspectChannelMedia(paths, "alpha", undefined, { fresh: true })).status,
      "in-place",
    );

    // Idempotent: a second preview tiers nothing and says the same figures.
    const again = await previewRelocation({ paths, slug: "alpha", root });
    assert.equal(again.tieredFirst, 0);
    assert.equal(again.files, 2);
    assert.equal(again.bytesToMove, 8);
  });
});

// THE RETIRED LAYOUT (release 17): a channel whose whole `data/` was moved by
// the old mover is `legacy`, and the mover refuses it — out, back and the
// preview — with the sentence that names the migration. Nothing is touched.
test("a legacy channel is refused by the job, the move back and the preview, naming migrate-tier", async () => {
  await withTmp(async (paths, root, dir) => {
    const channelDir = await seed(paths, "alpha", { v1: { "audio.m4a": "one" } });
    const oldTarget = path.join(root, "alpha", "data");
    await mkdir(path.dirname(oldTarget), { recursive: true });
    await rename(path.join(channelDir, "data"), oldTarget);
    await symlink(oldTarget, path.join(channelDir, "data"));
    await setConfigField(paths, "alpha", "dataDir", oldTarget);
    assert.equal(
      (await inspectChannelMedia(paths, "alpha", undefined, { fresh: true })).status,
      "legacy",
    );
    const other = path.join(dir, "platter2");
    await mkdir(other, { recursive: true });

    const refused = /cannot be moved: its media layout is the retired whole-directory one — run archilyzer storage migrate-tier alpha/;
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "out",
          root: other,
          onLog: () => {},
        }),
      refused,
    );
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "back",
          onLog: () => {},
        }),
      refused,
    );
    await assert.rejects(
      () => previewRelocation({ paths, slug: "alpha", root: other }),
      refused,
    );
    // Untouched: no marker, no `media`, the old link and its bytes as they were.
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
    assert.equal(await pathIsThere(path.join(channelDir, "media")), false);
    assert.equal(await readlink(path.join(channelDir, "data")), oldTarget);
    assert.equal(
      await readFile(path.join(oldTarget, "v1", "audio.m4a"), "utf8"),
      "one",
    );
    assert.equal(await pathThere(path.join(other, "alpha")), false);
  });
});

test("a tier migration's marker is never resumed or replaced by the mover", async () => {
  await withTmp(async (paths, root) => {
    await seed(paths, "alpha", { v1: { "audio.m4a": "one" } });
    const target = relocatedMediaDir(root, "alpha");
    await seedMarker(paths, "alpha", {
      target,
      direction: "out",
      phase: "copy",
      scope: "tier-migration",
    });
    for (const direction of ["out", "back"] as const) {
      await assert.rejects(
        () =>
          relocateChannelMedia({
            io: TEST_IO,
            paths,
            slug: "alpha",
            direction,
            root,
            onLog: () => {},
          }),
        /media-tier migration in flight or interrupted .* archilyzer storage migrate-tier alpha/,
      );
    }
    // The preview gives the same sentence, and tiers nothing (review N9).
    await assert.rejects(
      () => previewRelocation({ paths, slug: "alpha", root }),
      /media-tier migration in flight or interrupted .* archilyzer storage migrate-tier alpha/,
    );
    assert.equal(
      await pathIsThere(path.join(paths.channelsDir, "alpha", "media")),
      false,
    );
    assert.equal((await readRelocationMarker(paths, "alpha"))?.scope, "tier-migration");
  });
});

async function pathThere(p: string): Promise<boolean> {
  try {
    await stat(p);
    return true;
  } catch {
    return false;
  }
}

// ---------------------------------------------------------------------------
// CRASH RECOVERY: one case per phase per direction.
//
// The plan's contract is "a rerun with a matching marker re-verifies and
// continues from swap/reclaim", and every one of these states is reachable from
// a kill -9 between two syscalls. What they pin is that a rerun OBSERVES the
// disk rather than assuming the previous run finished the step the marker names:
// the marker says how far the last attempt got, it cannot say how far it got
// THROUGH a phase, and the disk is the only evidence.
//
// Each seeds the exact on-disk state of a crash at that point and asserts the
// rerun reaches a clean `ok` / `in-place` — with the config right, the marker
// gone and NO `media.relocated-*` or `media.incoming` left holding a second
// copy on the volume the move exists to free.

async function seedMarker(
  paths: Paths,
  slug: string,
  marker: {
    target: string;
    direction: "out" | "back";
    phase: "copy" | "swap" | "reclaim";
    scope?: "media" | "tier-migration";
  },
): Promise<void> {
  await writeFile(
    path.join(paths.channelsDir, slug, ".relocating.json"),
    JSON.stringify({ ...marker, startedAt: new Date().toISOString() }) + "\n",
  );
}

async function setConfigField(
  paths: Paths,
  slug: string,
  key: "mediaDir" | "dataDir",
  value: string | null,
): Promise<void> {
  const file = path.join(paths.channelsDir, slug, "config.json");
  const config = JSON.parse(await readFile(file, "utf8")) as Record<
    string,
    unknown
  >;
  if (value === null) delete config[key];
  else config[key] = value;
  await writeFile(file, JSON.stringify(config, null, 2) + "\n");
}

async function setConfigMediaDir(
  paths: Paths,
  slug: string,
  value: string | null,
): Promise<void> {
  await setConfigField(paths, slug, "mediaDir", value);
}

async function leftoverCopies(channelDir: string): Promise<string[]> {
  return (await readdir(channelDir))
    .filter((n) => n.startsWith("media.relocated-") || n === "media.incoming")
    .sort();
}

// rsync -a of the tree, so the seeded "already copied" target is byte-for-byte
// what a completed copy phase would have left — including mtimes, which the
// re-verify compares. Over `media/` only: it holds no links.
async function copyTree(src: string, dest: string): Promise<void> {
  await mkdir(dest, { recursive: true });
  for (const entry of await readdir(src, { withFileTypes: true })) {
    const from = path.join(src, entry.name);
    const to = path.join(dest, entry.name);
    if (entry.isDirectory()) {
      await copyTree(from, to);
    } else {
      await writeFile(to, await readFile(from));
      await utimes(to, MTIME, MTIME);
    }
  }
}

// A COMPLETED MOVE OUT, built by hand: tiered, `media/` copied to the target
// and replaced by the absolute link, `mediaDir` recorded.
async function seedRelocated(
  paths: Paths,
  slug: string,
  root: string,
): Promise<{ mediaLink: string; target: string }> {
  const mediaLink = await tierInPlace(paths, slug);
  const target = relocatedMediaDir(root, slug);
  await copyTree(mediaLink, target);
  await rm(mediaLink, { recursive: true, force: true });
  await symlink(target, mediaLink);
  await setConfigMediaDir(paths, slug, target);
  return { mediaLink, target };
}

test("out @ swap: crash before the rename — the rerun re-verifies and completes", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one".repeat(500) },
    });
    const mediaLink = await tierInPlace(paths, "alpha");
    const target = relocatedMediaDir(root, "alpha");
    // The copy finished; the process died before `media/` was parked.
    await copyTree(mediaLink, target);
    await seedMarker(paths, "alpha", { target, direction: "out", phase: "swap" });

    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    assert.equal(res.resumed, true);
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "ok");
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, target);
    assert.deepEqual(await leftoverCopies(channelDir), []);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
  });
});

test("out @ swap: crash after the config write — the first run's parked copy is still reclaimed", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one".repeat(500) },
    });
    const mediaLink = await tierInPlace(paths, "alpha");
    const target = relocatedMediaDir(root, "alpha");
    await copyTree(mediaLink, target);
    // Everything through writeChannelConfig ran; the reclaim marker never
    // landed. A full second copy of the media is parked on the source volume —
    // the disk the whole move exists to free — and the marker still says
    // "swap", so the old code minted a SECOND parked name and reclaimed only
    // that one, orphaning this forever.
    const parked = path.join(channelDir, "media.relocated-1700000000000");
    await rename(mediaLink, parked);
    await symlink(target, mediaLink);
    await setConfigMediaDir(paths, "alpha", target);
    await seedMarker(paths, "alpha", { target, direction: "out", phase: "swap" });

    await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "ok");
    assert.deepEqual(await leftoverCopies(channelDir), []);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
    // The media itself is untouched by the sweep, and reached from `data/`.
    assert.equal(
      await readFile(path.join(channelDir, "data", "v1", "audio.m4a"), "utf8"),
      "one".repeat(500),
    );
  });
});

test("out @ reclaim: the rerun sweeps every parked copy and clears the marker", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one".repeat(500) },
    });
    const mediaLink = await tierInPlace(paths, "alpha");
    const target = relocatedMediaDir(root, "alpha");
    await copyTree(mediaLink, target);
    await rename(mediaLink, path.join(channelDir, "media.relocated-1700000000000"));
    await symlink(target, mediaLink);
    await setConfigMediaDir(paths, "alpha", target);
    await seedMarker(paths, "alpha", {
      target,
      direction: "out",
      phase: "reclaim",
    });

    await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "ok");
    assert.deepEqual(await leftoverCopies(channelDir), []);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
  });
});

// THE MISSING CELL of the resume matrix: out @ {swap, reclaim} and back @
// {swap(before), swap(after), reclaim} all had cases, and back @ copy had none —
// the only one where the rerun has to finish an rsync rather than a rename, and
// the only one where the space check is asked to credit a PARTIAL copy.
test("back @ copy: a half-copied media.incoming is resumed, not restarted", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one".repeat(500), "transcript.json": "{}" },
      v2: { "audio.m4a": "two".repeat(500) },
    });
    const { mediaLink, target } = await seedRelocated(paths, "alpha", root);
    // The copy got one video in and died. The marker says `copy`, so the rerun
    // resumes the rsync — and the surviving file keeps its mtime, which is what
    // says rsync skipped it rather than re-sending it.
    const incoming = path.join(channelDir, "media.incoming");
    await copyTree(path.join(target, "v1"), path.join(incoming, "v1"));
    await seedMarker(paths, "alpha", { target, direction: "back", phase: "copy" });

    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "back",
      onLog: () => {},
    });
    assert.equal(res.resumed, true);
    assert.equal(res.files, 2);
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "in-place");
    assert.ok((await lstat(mediaLink)).isDirectory());
    assert.ok(!(await lstat(mediaLink)).isSymbolicLink());
    assert.equal(
      await readFile(path.join(channelDir, "data", "v2", "audio.m4a"), "utf8"),
      "two".repeat(500),
    );
    assert.equal(
      (await stat(path.join(mediaLink, "v1", "audio.m4a"))).mtime.getTime(),
      MTIME.getTime(),
    );
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, undefined);
    assert.equal(await pathThere(target), false);
    assert.deepEqual(await leftoverCopies(channelDir), []);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
  });
});

test("back @ swap: crash before the rename — the rerun finishes the swap", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one".repeat(500) },
    });
    // A completed relocation, then a move-back whose copy finished.
    const { mediaLink, target } = await seedRelocated(paths, "alpha", root);
    await copyTree(target, path.join(channelDir, "media.incoming"));
    await seedMarker(paths, "alpha", { target, direction: "back", phase: "swap" });

    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "back",
      onLog: () => {},
    });
    assert.equal(res.resumed, true);
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "in-place");
    assert.ok((await lstat(mediaLink)).isDirectory());
    assert.ok(!(await lstat(mediaLink)).isSymbolicLink());
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, undefined);
    assert.deepEqual(await leftoverCopies(channelDir), []);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
  });
});

test("back @ swap: crash AFTER the rename — the rerun does not ENOENT forever", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one".repeat(500) },
    });
    const mediaLink = await tierInPlace(paths, "alpha");
    const target = relocatedMediaDir(root, "alpha");
    await copyTree(mediaLink, target);
    // rename(incoming, media) committed; the process died before the config
    // was cleared. `media/` is a real directory and `incoming` is gone — so the
    // old sequence unlinked nothing (swallowed), then renamed a path that no
    // longer exists, and failed identically on every rerun.
    await setConfigMediaDir(paths, "alpha", target);
    await seedMarker(paths, "alpha", { target, direction: "back", phase: "swap" });

    await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "back",
      onLog: () => {},
    });
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "in-place");
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, undefined);
    assert.equal(await pathIsThere(target), false);
    assert.deepEqual(await leftoverCopies(channelDir), []);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
  });
});

test("back @ reclaim: the config is already clear, and the rerun still finishes", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one".repeat(500) },
    });
    const mediaLink = await tierInPlace(paths, "alpha");
    const target = relocatedMediaDir(root, "alpha");
    // The swap completed and cleared config.mediaDir — which is why this case
    // was UNREACHABLE: with no mediaDir the entry point threw "is not
    // relocated" and the marker named the only place that knew where the media
    // had been.
    await copyTree(mediaLink, target);
    await seedMarker(paths, "alpha", {
      target,
      direction: "back",
      phase: "reclaim",
    });

    await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "back",
      onLog: () => {},
    });
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "in-place");
    assert.equal(await pathIsThere(target), false);
    assert.deepEqual(await leftoverCopies(channelDir), []);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
  });
});

test("a marker for the other direction is never resumed into", async () => {
  await withTmp(async (paths, root) => {
    await seed(paths, "alpha", { v1: { "audio.m4a": "one" } });
    const target = relocatedMediaDir(root, "alpha");
    await setConfigMediaDir(paths, "alpha", target);
    await seedMarker(paths, "alpha", { target, direction: "out", phase: "swap" });
    await assert.rejects(
      relocateChannelMedia({
        io: TEST_IO,
        paths,
        slug: "alpha",
        direction: "back",
        onLog: () => {},
      }),
      /move-out .* in flight|interrupted/,
    );
  });
});

async function pathIsThere(p: string): Promise<boolean> {
  try {
    await lstat(p);
    return true;
  } catch {
    return false;
  }
}

// ---------------------------------------------------------------------------
// TWO WAYS THE MOVE USED TO EAT THE MEDIA. Both were reachable from the shipped
// UI and both ended with rm -r on the only copy, so both get their own cases.
// ---------------------------------------------------------------------------

// `inconsistent` — config.json records a target while `media/` is a REAL
// directory — is what `rsync --copy-links` of a channel produces, which
// WORKTREES.md documents as the way to carry media into a shard. inspect()
// reports relocated:true for it, so the panel offered "Move back in place", and
// the run copied the target to the incoming dir, verified it, found the live
// dir already real, logged "the swap had completed", cleared the config, rm -r'd
// the target and then swept the incoming copy. Three copies in, zero out.
test("back: an inconsistent channel is refused, and the target keeps its bytes", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one" },
    });
    const target = relocatedMediaDir(root, "alpha");
    const mediaLink = path.join(channelDir, "media");
    await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    // Turn the link back into a real directory WITHOUT clearing
    // config.mediaDir — the --copy-links shape, reproduced.
    await rm(mediaLink);
    await copyTree(target, mediaLink);
    assert.equal(
      (await inspectChannelMedia(paths, "alpha")).status,
      "inconsistent",
    );

    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "back",
          onLog: () => {},
        }),
      /not in a movable state/,
    );

    // The relocated copy is still there, the local one is still there, and
    // nothing was parked or marked.
    assert.equal(
      await readFile(path.join(target, "v1", "audio.m4a"), "utf8"),
      "one",
    );
    assert.equal(
      await readFile(path.join(channelDir, "data", "v1", "audio.m4a"), "utf8"),
      "one",
    );
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, target);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
    assert.equal((await readdir(channelDir)).includes("media.incoming"), false);
  });
});

// An `unreachable` channel (the drive is not mounted) is refused for the same
// reason: nothing can vouch for what the target holds, and the run ends by
// deleting it.
test("back: an unreachable channel is refused", async () => {
  await withTmp(async (paths, root) => {
    await seed(paths, "alpha", { v1: { "audio.m4a": "one" } });
    const target = relocatedMediaDir(root, "alpha");
    await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    // The platter goes away. The link dangles; config still names the target.
    await rm(path.dirname(target), { recursive: true, force: true });
    const away = await inspectChannelMedia(paths, "alpha");
    assert.equal(away.status, "unreachable");
    // The text does not care.
    assert.equal(away.text.readable, true);
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "back",
          onLog: () => {},
        }),
      /not in a movable state|not reachable/,
    );
  });
});

// THE SELF-RELOCATION. root = <transcriptsDir>/channels makes the target the
// source: rsync src/ src/ succeeds, verifyCopy compares the tree with itself,
// the swap parks `media/` (which IS the verified "target") and links to a path
// that no longer exists, and the reclaim sweep deletes the only copy. The
// panel's own help text describes transcripts/channels almost word for word.
test("a root inside the corpus is refused by the job and by the preview", async () => {
  await withTmp(async (paths) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one" },
    });
    const roots = [
      paths.channelsDir,
      paths.transcriptsDir,
      channelDir,
      path.join(channelDir, "data"),
      path.join(paths.channelsDir, "beta"),
    ];
    for (const bad of roots) {
      await assert.rejects(
        () =>
          relocateChannelMedia({
            io: TEST_IO,
            paths,
            slug: "alpha",
            direction: "out",
            root: bad,
            onLog: () => {},
          }),
        /inside the corpus|inside the channel directory/,
        `job accepted ${bad}`,
      );
      await assert.rejects(
        () => previewRelocation({ paths, slug: "alpha", root: bad }),
        /inside the corpus|inside the channel directory/,
        `preview accepted ${bad}`,
      );
    }
    // Untouched: still a real file in a real directory (not even tiered), no
    // config, no marker, no parked copy.
    assert.ok((await lstat(path.join(channelDir, "data"))).isDirectory());
    assert.ok(
      (await lstat(path.join(channelDir, "data", "v1", "audio.m4a"))).isFile(),
    );
    assert.equal(
      await readFile(path.join(channelDir, "data", "v1", "audio.m4a"), "utf8"),
      "one",
    );
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, undefined);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
    assert.equal(
      (await readdir(channelDir)).filter(
        (n) => n.startsWith("media") || n.startsWith("data."),
      ).length,
      0,
    );
  });
});

// A root OUTSIDE the corpus whose `<slug>` level is a symlink back into it. The
// root's own realpath cannot see this — the link is one level down — which is
// why the target is resolved separately.
test("a root that links back into the channel dir is refused", async () => {
  await withTmp(async (paths, _root, dir) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one" },
    });
    const sneaky = path.join(dir, "sneaky");
    await mkdir(sneaky, { recursive: true });
    await symlink(channelDir, path.join(sneaky, "alpha"));
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "out",
          root: sneaky,
          onLog: () => {},
        }),
      /inside the channel directory|inside the corpus/,
    );
    assert.ok((await lstat(path.join(channelDir, "data"))).isDirectory());
  });
});

test("a relative root is refused by the preview, not only by the job", async () => {
  await withTmp(async (paths) => {
    await seed(paths, "alpha", { v1: { "audio.m4a": "one" } });
    await assert.rejects(
      () => previewRelocation({ paths, slug: "alpha", root: "platter/media" }),
      /must be an absolute path/,
    );
  });
});

// THE OMNIMIRROR REFUSAL (2026-09-13), in a tmpdir.
//
// A 131 GB copy landed byte-complete and the verify refused it. The whole of the
// drift was one directory timestamp: a sidecar written into a video dir while
// rsync was already past that directory bumped the SOURCE directory's mtime and
// left the target's behind. `.d..t` is rsync's itemization for exactly that — a
// directory, and only its time differs — and it means nothing about the bytes.
//
// Both cases run at phase `swap`, where the re-verify runs on its own with no
// `rsync -a` ahead of it. That is the shape that can see the difference: in the
// copy phase the transfer itself would have set the timestamps.

test("out @ swap: a directory timestamp is settled by one more pass, not refused", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one".repeat(500), "transcript.json": "{}" },
      v2: { "audio.m4a": "two".repeat(500) },
    });
    const mediaLink = await tierInPlace(paths, "alpha");
    const target = relocatedMediaDir(root, "alpha");
    await copyTree(mediaLink, target);
    // The echo of the sidecar: the target's copy of v1/ carries a different
    // mtime from the source's. Every file underneath is identical.
    await utimes(path.join(target, "v1"), MTIME, MTIME);
    await seedMarker(paths, "alpha", { target, direction: "out", phase: "swap" });

    const lines: string[] = [];
    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: (m) => lines.push(m),
    });

    assert.equal(res.retried, true);
    assert.equal(res.resumed, true);
    assert.ok(
      lines.some((l) => l.includes("directory timestamp")),
      "the extra pass is announced in the job log",
    );
    // And the move completed: link, config, no leftovers, no marker.
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "ok");
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, target);
    assert.deepEqual(await leftoverCopies(channelDir), []);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
    assert.equal(
      await readFile(path.join(channelDir, "data", "v1", "audio.m4a"), "utf8"),
      "one".repeat(500),
    );
    assert.equal(
      await readFile(path.join(channelDir, "data", "v1", "transcript.json"), "utf8"),
      "{}",
    );
  });
});

// CONTENT drift at the swap's re-verify, arriving the way the timestamp did —
// a file written into the source after the copy. Until release 16 slice RM
// any file line refused; now the re-verify runs in mirror mode while `media/`
// is still the live directory, and one change gets one more mirror pass,
// exactly as in the copy phase. A second change is still a refusal ("a verify
// failure keeps the source …" above).
test("out @ swap: a file the target is missing is mirrored by one more pass", async () => {
  await withTmp(async (paths, root) => {
    await seed(paths, "alpha", {
      v1: { "audio.m4a": "one".repeat(500) },
    });
    const mediaLink = await tierInPlace(paths, "alpha");
    const target = relocatedMediaDir(root, "alpha");
    await copyTree(mediaLink, target);
    await writeFile(path.join(mediaLink, "v1", "transcript.live_chat.json"), "[]");
    await seedMarker(paths, "alpha", { target, direction: "out", phase: "swap" });

    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    assert.equal(res.retried, true);
    assert.equal(
      await readFile(path.join(target, "v1", "transcript.live_chat.json"), "utf8"),
      "[]",
    );
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "ok");
    assert.equal((await readChannelConfig(paths, "alpha"))?.mediaDir, target);
  });
});

// A PARKED COPY IS NEVER MIRRORED FROM. Once the rename has committed, the
// link is what readers follow and `media.relocated-*` is not live media any
// more: the re-verify against it is the strict one, as it always was — a
// difference refuses, and nothing on the target is deleted to match a stale
// copy.
test("out @ swap: after the rename, the re-verify against the parked copy is strict", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one".repeat(500) },
    });
    const mediaLink = await tierInPlace(paths, "alpha");
    const target = relocatedMediaDir(root, "alpha");
    await copyTree(mediaLink, target);
    // A file on the target the parked copy lacks.
    await writeFile(path.join(target, "v1", "newer.json"), "{}");
    await rename(mediaLink, path.join(channelDir, "media.relocated-1"));
    await seedMarker(paths, "alpha", { target, direction: "out", phase: "swap" });

    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "out",
          root,
          onLog: () => {},
        }),
      /Verification failed/,
    );
    assert.equal(
      await readFile(path.join(target, "v1", "newer.json"), "utf8"),
      "{}",
      "nothing on the target was deleted to match the parked copy",
    );
  });
});

// ---------------------------------------------------------------------------
// Release 16 slice RM — a move holds the writers and mirrors its copy
// ---------------------------------------------------------------------------

const TRANSCRIBING: ChannelWriter = {
  source: "job",
  jobId: "J1",
  kind: "whisper-all",
  label: "Transcribe all",
  status: "running",
  videoId: "v50t5yt",
  taskKind: "transcribe",
};

// THE REFUSAL OVER A RUNNING JOB, the job's first step. The registry is
// injected: what is pinned is that the move asks, refuses naming the writer,
// and has touched nothing — no marker, no target directory, no tiering, the
// source as it was.
test("a move refuses to start over a running job, naming it, and touches nothing", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one" },
    });
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "out",
          root,
          onLog: () => {},
          writers: () => [TRANSCRIBING],
        }),
      /Cannot move the media of "alpha" now: a transcription of v50t5yt is running \(Transcribe all, job J1\) — wait for it or cancel it\. Nothing has been touched\./,
    );
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
    await assert.rejects(() => stat(path.join(root, "alpha")));
    assert.ok((await lstat(path.join(channelDir, "data"))).isDirectory());
    assert.equal(await pathIsThere(path.join(channelDir, "media")), false);
    assert.ok(
      (await lstat(path.join(channelDir, "data", "v1", "audio.m4a"))).isFile(),
    );
  });
});

test("the preview gives the same refusal before the operator commits", async () => {
  await withTmp(async (paths, root) => {
    await seed(paths, "alpha", { v1: { "audio.m4a": "one" } });
    await assert.rejects(
      () =>
        previewRelocation({
          paths,
          slug: "alpha",
          root,
          writers: () => [TRANSCRIBING],
        }),
      /a transcription of v50t5yt is running/,
    );
    // And with nothing writing, the preview answers.
    const ok = await previewRelocation({
      paths,
      slug: "alpha",
      root,
      writers: () => [],
    });
    assert.equal(ok.files, 1);
  });
});

// A WRITER THAT STARTED AFTER THE FIRST STEP, before the marker landed. The
// third ask, right after the marker is written, sees it; the refusal removes the
// marker this run created, so the channel is not left held by a move that never
// copied a byte.
test("a writer seen once the marker is written refuses, and a fresh move's marker is removed", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one" },
    });
    let asked = 0;
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "out",
          root,
          onLog: () => {},
          writers: () => (++asked === 1 ? [] : [TRANSCRIBING]),
        }),
      /a transcription of v50t5yt is running/,
    );
    assert.equal(asked, 2);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
    // Tiered in place (the preflight ran) and nothing else: an in-place channel.
    assert.equal(
      (await inspectChannelMedia(paths, "alpha", undefined, { fresh: true })).status,
      "in-place",
    );
    assert.deepEqual(
      await readdir(path.join(relocatedMediaDir(root, "alpha"))),
      [],
      "nothing was copied",
    );
    assert.ok((await lstat(path.join(channelDir, "data"))).isDirectory());
    assert.equal(
      await readFile(path.join(channelDir, "data", "v1", "audio.m4a"), "utf8"),
      "one",
    );
  });
});

// A MOVE HOLDS ONLY THE MEDIA WRITERS (release 17 ruling): a digest — a job or
// the digest lane's unit — reads and writes the text, which never moves, so it
// is not a writer for this question. A running move of the channel IS one,
// though its kind is not `needsMedia`: the preview and the panel must not
// offer a second move over it.
test("channelMediaWriters: media jobs, a move and the media lanes — never a digest", () => {
  const job = (id: string, kind: string): JobRecord =>
    ({
      id,
      kind,
      queueKey: "q",
      status: "running",
      queuedAt: 1,
      logPath: "/dev/null",
      channelSlug: "alpha",
    }) as JobRecord;
  const jobs = [
    job("J1", "digest-channel-local"),
    job("J2", "whisper-all"),
    job("J3", "relocate-channel-media"),
  ];
  const unit = (lane: "digest" | "transcription", videoId: string) => ({
    lane,
    unit: { videoId, leafId: "leaf", channelSlug: "alpha", startedAt: 1 },
  });
  const source = {
    jobs: () => jobs,
    units: () => [unit("digest", "d1"), unit("transcription", "t1")],
  };
  const names = (ws: ChannelWriter[]) =>
    ws.map((w) => (w.source === "job" ? w.jobId : `${w.lane}:${w.videoId}`));
  assert.deepEqual(names(channelMediaWriters("alpha", { source })), [
    "J2",
    "J3",
    "transcription:t1",
  ]);
  // The move's own first step leaves itself out.
  assert.deepEqual(
    names(
      channelMediaWriters("alpha", {
        source,
        ignoreKinds: ["relocate-channel-media"],
      }),
    ),
    ["J2", "transcription:t1"],
  );
});

// THE 2026-10-01 CASE. A move killed mid-copy left a transcriber's scratch dir
// on the destination that has since gone from the source. The resume used to
// copy everything else and refuse on the counts (1755 against 1750), and no
// rerun could settle it; the mirror pass now does.
test("a resume with a stale extra dir on the destination completes", async () => {
  await withTmp(async (paths, root) => {
    await seed(paths, "alpha", {
      v50t5yt: { "audio.mp3": "a".repeat(64), "transcript.json": "{}" },
      v51fpcd: { "audio.mp3": "b".repeat(64) },
    });
    const mediaLink = await tierInPlace(paths, "alpha");
    const target = relocatedMediaDir(root, "alpha");
    await copyTree(mediaLink, target);
    const scratch = path.join(target, "v50t5yt", ".audio.mp3.parakeet");
    await mkdir(scratch, { recursive: true });
    for (const f of ["meta", "win-0000", "win-0001", "win-0002", "win-0003"]) {
      await writeFile(path.join(scratch, `${f}.json`), "{}");
    }
    await seedMarker(paths, "alpha", { target, direction: "out", phase: "copy" });

    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    assert.equal(res.resumed, true);
    assert.equal(res.files, 2);
    await assert.rejects(() => stat(scratch));
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "ok");
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
  });
});

test("back: a stale extra on the copy coming home is removed on resume", async () => {
  await withTmp(async (paths, root) => {
    const channelDir = await seed(paths, "alpha", {
      v1: { "audio.m4a": "one".repeat(100) },
    });
    await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      onLog: () => {},
    });
    const target = relocatedMediaDir(root, "alpha");
    // An interrupted move back: `media.incoming` holds the copy, plus a file
    // the target (the source of this direction) no longer has.
    const incoming = path.join(channelDir, "media.incoming");
    await copyTree(target, incoming);
    await writeFile(path.join(incoming, "v1", "gone.json"), "{}");
    await seedMarker(paths, "alpha", { target, direction: "back", phase: "copy" });

    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "back",
      onLog: () => {},
    });
    assert.equal(res.resumed, true);
    const mediaLink = path.join(channelDir, "media");
    assert.ok((await lstat(mediaLink)).isDirectory());
    assert.ok(!(await lstat(mediaLink)).isSymbolicLink());
    assert.deepEqual(await readdir(path.join(mediaLink, "v1")), ["audio.m4a"]);
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
  });
});

// RECONCILE AND RESUME — the remediation (the ruling's last bullet). An extra
// file and a changed one on the destination: the job says what it found, by
// kind, makes the copy match the source and finishes the move.
test("reconcile: an extra and a changed file on the destination are settled, and the move completes", async () => {
  await withTmp(async (paths, root) => {
    await seed(paths, "alpha", {
      v1: {
        "audio.m4a": "one".repeat(100),
        "transcript.live_chat.json": '{"v":2}',
      },
    });
    const mediaLink = await tierInPlace(paths, "alpha");
    const target = relocatedMediaDir(root, "alpha");
    await copyTree(mediaLink, target);
    await writeFile(path.join(target, "v1", "stale.json"), "{}");
    await writeFile(path.join(target, "v1", "transcript.live_chat.json"), '{"v":1}');
    await seedMarker(paths, "alpha", { target, direction: "out", phase: "copy" });

    const lines: string[] = [];
    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      reconcile: true,
      onLog: (m) => lines.push(m),
    });
    assert.equal(res.resumed, true);
    assert.equal(res.reconciled, true);
    const said = lines.find((l) => l.startsWith("Reconciling:")) ?? "";
    assert.match(said, /1 extra on the destination \(v1\/stale\.json\)/);
    assert.match(said, /1 changed \(v1\/transcript\.live_chat\.json\)/);
    assert.deepEqual(
      (await readdir(path.join(target, "v1"))).sort(),
      ["audio.m4a", "transcript.live_chat.json"],
    );
    assert.equal(
      await readFile(path.join(target, "v1", "transcript.live_chat.json"), "utf8"),
      '{"v":2}',
    );
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "ok");
    assert.equal(await readRelocationMarker(paths, "alpha"), null);
  });
});

// A marker past the copy phase has nothing to reconcile: the run is a plain
// resume, and says so (the review's L3).
test("reconcile: a marker past the copy phase resumes, and does not claim a reconcile", async () => {
  await withTmp(async (paths, root) => {
    await seed(paths, "alpha", {
      v1: { "audio.m4a": "one".repeat(100) },
    });
    const mediaLink = await tierInPlace(paths, "alpha");
    const target = relocatedMediaDir(root, "alpha");
    await copyTree(mediaLink, target);
    await seedMarker(paths, "alpha", { target, direction: "out", phase: "swap" });
    const lines: string[] = [];
    const res = await relocateChannelMedia({
      io: TEST_IO,
      paths,
      slug: "alpha",
      direction: "out",
      root,
      reconcile: true,
      onLog: (m) => lines.push(m),
    });
    assert.equal(res.resumed, true);
    assert.equal(res.reconciled, false);
    assert.equal(lines.some((l) => l.startsWith("Reconciling:")), false);
    assert.equal((await inspectChannelMedia(paths, "alpha")).status, "ok");
  });
});

test("reconcile: with no marker there is nothing to reconcile", async () => {
  await withTmp(async (paths, root) => {
    await seed(paths, "alpha", { v1: { "audio.m4a": "one" } });
    await assert.rejects(
      () =>
        relocateChannelMedia({
          io: TEST_IO,
          paths,
          slug: "alpha",
          direction: "out",
          root,
          reconcile: true,
          onLog: () => {},
        }),
      /no relocation marker — there is no interrupted move to reconcile/,
    );
  });
});

// ---------------------------------------------------------------------------
// assertRelocationRootPresent — the move must not MATERIALISE the mountpoint
// ---------------------------------------------------------------------------
//
// Every absolute mkdir in the two movers is `{recursive: true}`, so a move
// aimed at an unmounted platter used to build the whole path on the root
// filesystem and fill it. The fake findmnt is the same one
// `lib/storageVolumes.test.ts` uses, for the same reason: a real one needs a
// real disk.

const FAKE_FINDMNT = `#!/usr/bin/env node
import { readFileSync } from "node:fs";
import path from "node:path";

const control = JSON.parse(
  readFileSync(path.join(import.meta.dirname, "control.json"), "utf8"),
);
const argv = process.argv.slice(2);
if (argv.includes("--fstab")) {
  if (!control.fstab) process.exit(1);
  process.stdout.write(control.fstab + "\\n");
  process.exit(0);
}
if (argv.includes("-S")) {
  if (!control.uuidTarget) process.exit(1);
  process.stdout.write(control.uuidTarget + "\\n");
  process.exit(0);
}
if (control.identity === null) process.exit(1);
process.stdout.write(
  JSON.stringify({ filesystems: [control.identity] }) + "\\n",
);
`;

const PLATTER_UUID = "11111111-2222-3333-4444-555555555555";
const OTHER_UUID = "99999999-8888-7777-6666-555555555555";

async function withProbe(
  fn: (
    ctx: {
      dir: string;
      bins: VolumeBins;
      byUuidDir: string;
      control: (c: Record<string, unknown>) => Promise<void>;
    },
  ) => Promise<void>,
): Promise<void> {
  const dir = await mkdtemp(path.join(tmpdir(), "ttb-rootpresent-"));
  const bin = path.join(dir, "fake-findmnt.mjs");
  await writeFile(bin, FAKE_FINDMNT);
  await chmod(bin, 0o755);
  const byUuidDir = path.join(dir, "by-uuid");
  await mkdir(byUuidDir, { recursive: true });
  try {
    await fn({
      dir,
      byUuidDir,
      bins: { findmntBin: bin, udisksctlBin: path.join(dir, "no-udisksctl") },
      control: (c) =>
        writeFile(path.join(dir, "control.json"), JSON.stringify(c)),
    });
  } finally {
    await rm(dir, { recursive: true, force: true });
  }
}

function storageWith(locations: StorageLocation[]): StorageSettings {
  return { locations, defaultLocationId: "" };
}

function platterLocation(root: string, uuid?: string): StorageLocation {
  return {
    id: "platter",
    label: "Platter",
    root,
    autoRepoint: false,
    ...(uuid
      ? {
          volume: {
            uuid,
            fstype: "ext4",
            mountpoint: path.dirname(root),
            relPath: path.basename(root),
          },
        }
      : {}),
  };
}

test("root present: a missing root is refused before anything is created", async () => {
  await withProbe(async (h) => {
    resetStorageProbeMemo();
    const root = path.join(h.dir, "mnt", "platter", "media");
    await h.control({ identity: null });
    const problem = await relocationRootPresenceProblem(
      root,
      storageWith([platterLocation(root, PLATTER_UUID)]),
      h.bins,
      { byUuidDir: h.byUuidDir, findmntTimeoutMs: 1_000 },
    );
    assert.match(String(problem), /does not exist or is not a directory/);
    assert.match(String(problem), /location "platter"/);
    // And nothing was created on the way to finding out — that IS the bug.
    await assert.rejects(() => stat(path.join(h.dir, "mnt")));
  });
});

test("root present: a directory on the WRONG volume is refused", async () => {
  await withProbe(async (h) => {
    resetStorageProbeMemo();
    // Exactly the shape the guard exists for: the mountpoint directory is
    // there, empty, on the root filesystem — the platter is unplugged.
    const root = path.join(h.dir, "mnt", "platter", "media");
    await mkdir(root, { recursive: true });
    await h.control({
      identity: {
        target: "/",
        fstype: "ext4",
        label: "ROOTFS",
        uuid: OTHER_UUID,
      },
      fstab: `UUID=${OTHER_UUID}`,
    });
    const problem = await relocationRootPresenceProblem(
      root,
      storageWith([platterLocation(root, PLATTER_UUID)]),
      h.bins,
      { byUuidDir: h.byUuidDir, findmntTimeoutMs: 1_000 },
    );
    assert.match(String(problem), new RegExp(OTHER_UUID));
    assert.match(String(problem), new RegExp(PLATTER_UUID));
  });
});

test("root present: the right volume passes", async () => {
  await withProbe(async (h) => {
    resetStorageProbeMemo();
    const root = path.join(h.dir, "mnt", "platter", "media");
    await mkdir(root, { recursive: true });
    await h.control({
      identity: {
        target: path.join(h.dir, "mnt", "platter"),
        fstype: "ext4",
        label: "PLATTER",
        uuid: PLATTER_UUID,
      },
      fstab: `UUID=${PLATTER_UUID}`,
    });
    assert.equal(
      await relocationRootPresenceProblem(
        root,
        storageWith([platterLocation(root, PLATTER_UUID)]),
        h.bins,
        { byUuidDir: h.byUuidDir, findmntTimeoutMs: 1_000 },
      ),
      null,
    );
  });
});

test("root present: a root nobody named is stat-only", async () => {
  await withProbe(async (h) => {
    resetStorageProbeMemo();
    const root = path.join(h.dir, "hand-typed");
    await mkdir(root, { recursive: true });
    // No location covers it, so there is no recorded identity to compare
    // against — and a hand-typed root the operator just made must not be
    // refused for failing a comparison that cannot be made. The findmnt here
    // would answer a MISMATCH if it were asked; it must not be asked.
    await h.control({
      identity: { target: "/", fstype: "ext4", uuid: OTHER_UUID },
    });
    assert.equal(
      await relocationRootPresenceProblem(
        root,
        storageWith([platterLocation(path.join(h.dir, "elsewhere"), PLATTER_UUID)]),
        h.bins,
        { byUuidDir: h.byUuidDir, findmntTimeoutMs: 1_000 },
      ),
      null,
    );
  });
});

test("root present: unknown identity FAILS OPEN — a container has no block devices", async () => {
  await withProbe(async (h) => {
    resetStorageProbeMemo();
    const root = path.join(h.dir, "mnt", "platter", "media");
    await mkdir(root, { recursive: true });
    // findmnt exits 1: no identity at all. Refusing here would break
    // relocation on exactly the deployments that cannot answer the question.
    await h.control({ identity: null });
    assert.equal(
      await relocationRootPresenceProblem(
        root,
        storageWith([platterLocation(root, PLATTER_UUID)]),
        h.bins,
        { byUuidDir: h.byUuidDir, findmntTimeoutMs: 1_000 },
      ),
      null,
    );
  });
});

test("assertRelocationRootPresent throws the problem it finds", async () => {
  await withProbe(async (h) => {
    resetStorageProbeMemo();
    await assert.rejects(
      () =>
        assertRelocationRootPresent(
          path.join(h.dir, "nope"),
          storageWith([]),
          h.bins,
          { byUuidDir: h.byUuidDir, findmntTimeoutMs: 1_000 },
        ),
      /does not exist or is not a directory/,
    );
  });
});