import { createHash, randomUUID } from "node:crypto";
import { createReadStream, createWriteStream } from "node:fs";
import { chmod, copyFile, mkdir, readFile, rename, rm, stat, writeFile } from "node:fs/promises";
import { basename, dirname, extname, join, resolve, sep } from "node:path";
import { Readable, Transform } from "node:stream";
import { pipeline } from "node:stream/promises";

export class KeepError extends Error {
  constructor(status, message) {
    super(message);
    this.name = "KeepError";
    this.status = status;
  }
}

function cleanText(value, maximumLength) {
  return typeof value === "string" ? value.trim().slice(0, maximumLength) : "";
}

function finiteSize(value) {
  const size = Number(value);
  return Number.isSafeInteger(size) && size > 0 ? size : 0;
}

function isCookingCategory(category) {
  return category === "recipes" || category === "courses";
}

export function keepCookingCategory(relativePath) {
  const path = String(relativePath || "");
  if (path.startsWith("Recipes/")) return "recipes";
  if (path.startsWith("Courses/")) return "courses";
  return null;
}

export function sanitizeKeepFilename(name, fallback = "kept-media") {
  const base = basename(cleanText(name, 180) || fallback)
    .replace(/[<>:"/\\|?*\u0000-\u001f]/g, "-")
    .replace(/\s+/g, " ")
    .trim()
    .replace(/^\.+/, "")
    || fallback;
  return base.slice(0, 180);
}

export function sanitizeKeepSeriesFolder(name) {
  const folder = sanitizeKeepFilename(name, "");
  return folder && folder !== "kept-media" ? folder : "";
}

export function sanitizeKeepRelativePath(value) {
  const segments = cleanText(value, 800).replaceAll("\\", "/").split("/");
  if (segments.length === 0 || segments.some((segment) => !segment || segment === "." || segment === "..")) {
    throw new KeepError(400, "Keep requires a safe relative destination path.");
  }
  return segments.map((segment) => sanitizeKeepFilename(segment)).join("/");
}

export function validateKeepDownloadPath(value, itemId) {
  const expected = `/api/keep/content/${itemId}/file`;
  const path = cleanText(value, 200) || expected;
  if (path !== expected) throw new KeepError(400, "The Keep source file endpoint is invalid.");
  return path;
}

export function validateKeepSourceURL(value) {
  let url;
  try {
    url = new URL(cleanText(value, 500));
  } catch {
    throw new KeepError(400, "The source peer address is invalid.");
  }
  if ((url.protocol !== "https:" && url.protocol !== "http:") || url.username || url.password || url.hash) {
    throw new KeepError(400, "Keep source peers must use http(s) without embedded credentials.");
  }
  return url.origin;
}

export function publicKeepJob(job) {
  return {
    id: job.id,
    state: job.state,
    mode: job.mode || "peer",
    itemId: job.itemId,
    contentId: job.contentId || null,
    title: job.title,
    seriesName: job.seriesName || null,
    sourceBaseURL: job.sourceBaseURL,
    sourceName: job.sourceName || null,
    category: job.category,
    relativePath: job.relativePath || null,
    removeSource: job.removeSource === true,
    sourceDeleted: job.sourceDeleted === true,
    sourceDeleteError: job.sourceDeleteError || null,
    bytesTotal: job.bytesTotal,
    bytesReceived: job.bytesReceived,
    progress: job.bytesTotal > 0 ? Math.min(1, job.bytesReceived / job.bytesTotal) : null,
    filename: job.filename,
    keepPath: job.keepPath || null,
    error: job.error || null,
    createdAt: job.createdAt,
    updatedAt: job.updatedAt,
    finishedAt: job.finishedAt || null,
  };
}

function isInsideDirectory(filePath, directory) {
  const root = resolve(directory) + sep;
  const target = resolve(filePath);
  return target === resolve(directory) || target.startsWith(root);
}

async function uniqueDestination(directory, filename, jobId) {
  let destination = join(directory, filename);
  try {
    await stat(destination);
    const extension = extname(filename);
    const stem = basename(filename, extension);
    destination = join(directory, `${stem}-${jobId.slice(0, 8)}${extension}`);
  } catch {
    // Destination is free.
  }
  return destination;
}

export function createPeerKeepRegistry({
  keepDirectory,
  categoryDirectories = {},
  partialsDirectory,
  statePath = join(partialsDirectory || "", "jobs.json"),
  prepareLocal,
  onComplete,
  fetchImpl = globalThis.fetch,
  maxConcurrent = 2,
  now = Date.now,
} = {}) {
  if (!keepDirectory || !partialsDirectory) {
    throw new Error("Keep registry requires keepDirectory and partialsDirectory.");
  }

  const jobs = new Map();
  const transferWaiters = [];
  let activeTransfers = 0;
  let persistenceEnabled = false;
  let persistenceChain = Promise.resolve();

  function serializedJob(job) {
    return {
      id: job.id,
      mode: job.mode,
      state: job.state,
      itemId: job.itemId,
      contentId: job.contentId,
      title: job.title,
      kind: job.kind,
      seriesName: job.seriesName,
      sourceBaseURL: job.sourceBaseURL,
      sourceTokenRequired: Boolean(job.sourceToken || job.sourceTokenUnavailable),
      sourceName: job.sourceName,
      sourcePath: job.sourcePath,
      sourceDownloadPath: job.sourceDownloadPath,
      category: job.category,
      relativePath: job.relativePath,
      removeSource: job.removeSource,
      sourceDeleted: job.sourceDeleted,
      sourceDeleteError: job.sourceDeleteError,
      sha256: job.sha256,
      bytesTotal: job.bytesTotal,
      bytesReceived: job.bytesReceived,
      filename: job.filename,
      keepPath: job.keepPath,
      error: job.error,
      createdAt: job.createdAt,
      updatedAt: job.updatedAt,
      finishedAt: job.finishedAt,
    };
  }

  async function writeState() {
    await mkdir(dirname(statePath), { recursive: true });
    const temporaryPath = `${statePath}.tmp`;
    const payload = JSON.stringify({ version: 1, jobs: [...jobs.values()].map(serializedJob) }, null, 2);
    await writeFile(temporaryPath, `${payload}\n`, { encoding: "utf8", mode: 0o600 });
    await chmod(temporaryPath, 0o600);
    await rename(temporaryPath, statePath);
  }

  function queueStateWrite() {
    if (!persistenceEnabled) return persistenceChain;
    persistenceChain = persistenceChain.then(writeState, writeState);
    return persistenceChain;
  }

  async function restoreState() {
    let stored;
    try {
      stored = JSON.parse(await readFile(statePath, "utf8"));
    } catch (error) {
      if (error?.code !== "ENOENT") console.warn(`Could not restore Keep jobs: ${error.message}`);
      persistenceEnabled = true;
      return;
    }

    let restoredActiveJob = false;
    for (const saved of Array.isArray(stored?.jobs) ? stored.jobs : []) {
      const id = cleanText(saved?.id, 20).toLowerCase();
      const itemId = cleanText(saved?.itemId, 20).toLowerCase();
      if (!/^[a-f0-9]{20}$/.test(id) || !/^[a-f0-9]{20}$/.test(itemId)) continue;
      const mode = saved?.mode === "local" ? "local" : "peer";
      const wasActive = saved?.state === "running" || saved?.state === "queued";
      restoredActiveJob ||= wasActive;
      const sourceTokenUnavailable = mode === "peer" && saved?.sourceTokenRequired === true;
      const state = sourceTokenUnavailable ? "failed" : wasActive ? "paused" : saved?.state;
      if (!["completed", "paused", "cancelled", "failed"].includes(state)) continue;
      const createdAt = cleanText(saved?.createdAt, 40) || new Date(now()).toISOString();
      const updatedAt = cleanText(saved?.updatedAt, 40) || createdAt;
      jobs.set(id, {
        ...saved,
        id,
        mode,
        state,
        itemId,
        sourceToken: "",
        sourceTokenUnavailable,
        bytesTotal: finiteSize(saved?.bytesTotal),
        bytesReceived: Math.max(0, Number(saved?.bytesReceived) || 0),
        partialPath: join(partialsDirectory, `${id}.partial`),
        createdAt,
        updatedAt,
        error: sourceTokenUnavailable
          ? "The agent restarted without saving the legacy access token. Clear this transfer, then start it again from the browser."
          : wasActive
            ? "The agent restarted. Resume to continue from the saved partial copy."
          : cleanText(saved?.error, 500) || null,
        controller: null,
        runPromise: null,
      });
    }
    persistenceEnabled = true;
    if (restoredActiveJob) queueStateWrite();
  }

  const readyPromise = restoreState();

  async function acquireTransferSlot() {
    const limit = Math.max(1, Math.min(Number(maxConcurrent) || 2, 6));
    if (activeTransfers < limit) {
      activeTransfers += 1;
      return;
    }
    await new Promise((resolvePromise) => transferWaiters.push(resolvePromise));
  }

  function releaseTransferSlot() {
    const next = transferWaiters.shift();
    if (next) next();
    else activeTransfers = Math.max(0, activeTransfers - 1);
  }

  function touch(job, patch = {}) {
    Object.assign(job, patch, { updatedAt: new Date(now()).toISOString() });
    queueStateWrite();
    return job;
  }

  function getJob(jobId) {
    const job = jobs.get(cleanText(jobId, 80));
    if (!job) throw new KeepError(404, "Keep job not found.");
    return job;
  }

  async function ensureDirs() {
    await mkdir(keepDirectory, { recursive: true });
    for (const directory of Object.values(categoryDirectories)) {
      if (directory) await mkdir(directory, { recursive: true });
    }
    await mkdir(partialsDirectory, { recursive: true });
  }

  async function removeSourceAfterKeep(job) {
    if (!job.removeSource) return { sourceDeleted: false, sourceDeleteError: null };
    try {
      const headers = job.sourceToken ? { "X-Private-Reel-Token": job.sourceToken } : undefined;
      const response = await fetchImpl(`${job.sourceBaseURL}/api/keep/content/${job.itemId}`, {
        method: "DELETE",
        headers,
        redirect: "error",
        cache: "no-store",
        signal: AbortSignal.timeout(12_000),
      });
      if (!response.ok) throw new Error(`Source removal returned ${response.status}.`);
      return { sourceDeleted: true, sourceDeleteError: null };
    } catch (error) {
      return {
        sourceDeleted: false,
        sourceDeleteError: error instanceof Error ? error.message : "The source copy could not be removed.",
      };
    }
  }

  async function partialSize(path) {
    try {
      return (await stat(path)).size;
    } catch {
      return 0;
    }
  }

  async function runJob(job) {
    await acquireTransferSlot();
    if (job.state === "cancelled") {
      releaseTransferSlot();
      return publicKeepJob(job);
    }
    const controller = new AbortController();
    job.controller = controller;
    touch(job, { state: "running", error: null });

    try {
      await ensureDirs();

      if (!isCookingCategory(job.category) && job.sha256 && typeof prepareLocal === "function") {
        const prepared = await prepareLocal({
          sha256: job.sha256,
          size: job.bytesTotal,
          mediaType: job.kind,
        });
        if (prepared?.alreadyAvailable) {
          const sourceRemoval = await removeSourceAfterKeep(job);
          touch(job, {
            state: "completed",
            bytesReceived: job.bytesTotal,
            contentId: prepared.contentId || job.contentId,
            finishedAt: new Date(now()).toISOString(),
            keepPath: null,
            error: null,
            ...sourceRemoval,
          });
          return publicKeepJob(job);
        }
      }

      const received = await partialSize(job.partialPath);
      job.bytesReceived = received;
      if (received > job.bytesTotal) {
        await rm(job.partialPath, { force: true });
        job.bytesReceived = 0;
      }

      const headers = {
        "X-Private-Reel-Token": job.sourceToken,
      };
      if (job.bytesReceived > 0) headers.Range = `bytes=${job.bytesReceived}-`;

      const response = await fetchImpl(`${job.sourceBaseURL}${job.sourceDownloadPath}`, {
        method: "GET",
        headers,
        redirect: "error",
        cache: "no-store",
        signal: controller.signal,
      });

      if (job.bytesReceived > 0 && response.status === 200) {
        await rm(job.partialPath, { force: true });
        job.bytesReceived = 0;
      } else if (job.bytesReceived > 0 && response.status !== 206) {
        throw new KeepError(502, `The source peer could not resume the download (${response.status}).`);
      } else if (job.bytesReceived === 0 && !response.ok) {
        throw new KeepError(502, `The source peer download failed (${response.status}).`);
      }

      if (!response.body) throw new KeepError(502, "The source peer returned an empty body.");

      const file = createWriteStream(job.partialPath, {
        flags: job.bytesReceived > 0 ? "a" : "w",
      });
      let transferred = job.bytesReceived;
      const counting = new Transform({
        transform(chunk, _encoding, callback) {
          transferred += chunk.length;
          job.bytesReceived = transferred;
          job.updatedAt = new Date(now()).toISOString();
          callback(null, chunk);
        },
      });
      await pipeline(Readable.fromWeb(response.body), counting, file);

      if (job.bytesTotal > 0 && job.bytesReceived !== job.bytesTotal) {
        throw new KeepError(502, "The kept file size did not match the source size.");
      }

      if (job.sha256) {
        const hash = createHash("sha256");
        await new Promise((resolvePromise, reject) => {
          const stream = createReadStream(job.partialPath);
          stream.on("data", (chunk) => hash.update(chunk));
          stream.on("error", reject);
          stream.on("end", resolvePromise);
        });
        const digest = hash.digest("hex");
        if (digest !== job.sha256) {
          await rm(job.partialPath, { force: true });
          throw new KeepError(502, "The kept file failed the integrity check.");
        }
      }

      const categoryRoot = isCookingCategory(job.category)
        ? categoryDirectories[job.category] || keepDirectory
        : keepDirectory;
      const destinationDirectory = isCookingCategory(job.category) && job.relativePath
        ? join(categoryRoot, ...job.relativePath.split("/").slice(0, -1))
        : job.seriesName
          ? join(categoryRoot, job.seriesName)
          : categoryRoot;
      await mkdir(destinationDirectory, { recursive: true });
      let destination = isCookingCategory(job.category) && job.relativePath
        ? join(categoryRoot, ...job.relativePath.split("/"))
        : join(destinationDirectory, job.filename);
      if (!isCookingCategory(job.category)) {
        try {
          await stat(destination);
          const extension = extname(job.filename);
          const stem = basename(job.filename, extension);
          destination = join(destinationDirectory, `${stem}-${job.id.slice(0, 8)}${extension}`);
        } catch {
          // Destination is free.
        }
      }

      await rename(job.partialPath, destination);
      const sourceRemoval = await removeSourceAfterKeep(job);
      touch(job, {
        state: "completed",
        keepPath: destination,
        finishedAt: new Date(now()).toISOString(),
        error: null,
        ...sourceRemoval,
      });
      if (typeof onComplete === "function") await onComplete(job);
    } catch (error) {
      if (job.state === "cancelled") return publicKeepJob(job);
      if (error?.name === "AbortError") {
        touch(job, {
          state: "cancelled",
          error: "The keep transfer was cancelled.",
          finishedAt: new Date(now()).toISOString(),
        });
        return publicKeepJob(job);
      }
      const message = error instanceof KeepError
        ? error.message
        : error instanceof Error
          ? error.message
          : "The keep transfer failed.";
      // Soft pause so resume can continue from the partial when the source returns.
      touch(job, { state: "paused", error: message });
    } finally {
      job.controller = null;
      job.runPromise = null;
      releaseTransferSlot();
    }

    return publicKeepJob(job);
  }

  async function runLocalJob(job) {
    touch(job, { state: "running", error: null });
    try {
      await ensureDirs();
      if (!job.sourcePath) throw new KeepError(400, "Local Keep requires a source path.");
      if (isInsideDirectory(job.sourcePath, keepDirectory)) {
        touch(job, {
          state: "completed",
          bytesReceived: job.bytesTotal,
          keepPath: job.sourcePath,
          finishedAt: new Date(now()).toISOString(),
          error: null,
        });
        return publicKeepJob(job);
      }

      const destinationDirectory = job.seriesName
        ? join(keepDirectory, job.seriesName)
        : keepDirectory;
      await mkdir(destinationDirectory, { recursive: true });
      const destination = await uniqueDestination(destinationDirectory, job.filename, job.id);
      await copyFile(job.sourcePath, destination);
      touch(job, {
        state: "completed",
        bytesReceived: job.bytesTotal,
        keepPath: destination,
        finishedAt: new Date(now()).toISOString(),
        error: null,
      });
      if (typeof onComplete === "function") await onComplete(job);
    } catch (error) {
      const message = error instanceof KeepError
        ? error.message
        : error instanceof Error
          ? error.message
          : "The local keep failed.";
      touch(job, { state: "paused", error: message });
    } finally {
      job.runPromise = null;
    }
    return publicKeepJob(job);
  }

  function buildJobShell(input, { mode, sourceBaseURL = null, sourceToken = "", sourcePath = null }) {
    const itemId = cleanText(input?.itemId, 20).toLowerCase();
    if (!/^[a-f0-9]{20}$/.test(itemId)) throw new KeepError(400, "Keep requires a valid media id.");
    const bytesTotal = finiteSize(input?.size);
    if (!bytesTotal) throw new KeepError(400, "Keep requires a positive size.");
    const sha256 = cleanText(input?.sha256, 64).toLowerCase();
    if (sha256 && !/^[a-f0-9]{64}$/.test(sha256)) {
      throw new KeepError(400, "Keep sha256 must be a 64-character hex digest.");
    }
    const title = cleanText(input?.title, 240) || "Kept media";
    const kind = input?.kind === "image" || input?.kind === "file" ? input.kind : "video";
    const category = ["movies", "series", "recipes", "courses"].includes(input?.category)
      ? input.category
      : "movies";
    const relativePath = isCookingCategory(category)
      ? sanitizeKeepRelativePath(input?.relativePath || "")
      : null;
    const seriesName = sanitizeKeepSeriesFolder(input?.seriesName || "");
    const extension = cleanText(input?.extension, 12).toLowerCase()
      || (kind === "image" ? ".jpg" : ".mp4");
    const normalizedExtension = extension.startsWith(".") ? extension : `.${extension}`;
    const filename = sanitizeKeepFilename(
      cleanText(input?.filename, 180) || `${title}${normalizedExtension}`,
      `kept-${itemId}${normalizedExtension}`,
    );
    const id = randomUUID().replace(/-/g, "").slice(0, 20);
    const createdAt = new Date(now()).toISOString();
    return {
      id,
      mode,
      state: "queued",
      itemId,
      contentId: cleanText(input?.contentId, 20).toLowerCase() || null,
      title,
      kind,
      seriesName: seriesName || null,
      sourceBaseURL,
      sourceToken,
      sourceName: cleanText(input?.sourceName, 120) || null,
      sourcePath,
      sourceDownloadPath: mode === "peer"
        ? validateKeepDownloadPath(input?.sourceDownloadPath, itemId)
        : null,
      category,
      relativePath,
      removeSource: mode === "peer" && input?.removeSource === true,
      sourceDeleted: false,
      sourceDeleteError: null,
      sha256: sha256 || null,
      bytesTotal,
      bytesReceived: 0,
      filename,
      partialPath: join(partialsDirectory, `${id}.partial`),
      keepPath: null,
      error: null,
      createdAt,
      updatedAt: createdAt,
      finishedAt: null,
      controller: null,
      runPromise: null,
    };
  }

  return {
    list() {
      return [...jobs.values()]
        .sort((a, b) => String(b.updatedAt).localeCompare(String(a.updatedAt)))
        .map(publicKeepJob);
    },

    get(jobId) {
      return publicKeepJob(getJob(jobId));
    },

    async start(input) {
      await readyPromise;
      const sourceBaseURL = validateKeepSourceURL(input?.sourceBaseURL);
      const job = buildJobShell(input, {
        mode: "peer",
        sourceBaseURL,
        sourceToken: cleanText(input?.sourceToken, 500),
      });
      jobs.set(job.id, job);
      queueStateWrite();
      job.runPromise = runJob(job);
      return publicKeepJob(job);
    },

    async startLocal(input) {
      await readyPromise;
      const sourcePath = cleanText(input?.sourcePath, 1000);
      if (!sourcePath) throw new KeepError(400, "Local Keep requires a source path.");
      const job = buildJobShell(input, {
        mode: "local",
        sourceBaseURL: null,
        sourcePath,
      });
      job.sourceName = cleanText(input?.sourceName, 120) || "This Mac";
      jobs.set(job.id, job);
      queueStateWrite();
      job.runPromise = runLocalJob(job);
      return publicKeepJob(job);
    },

    async resume(jobId) {
      await readyPromise;
      const job = getJob(jobId);
      if (job.state === "completed") return publicKeepJob(job);
      if (job.state === "running" || job.runPromise) return publicKeepJob(job);
      if (job.state === "cancelled") throw new KeepError(409, "Cancelled keep jobs cannot resume.");
      if (job.sourceTokenUnavailable) {
        throw new KeepError(409, "This restored transfer needs its legacy source token. Start it again from the browser.");
      }
      job.runPromise = job.mode === "local" ? runLocalJob(job) : runJob(job);
      return publicKeepJob(job);
    },

    async cancel(jobId) {
      await readyPromise;
      const job = getJob(jobId);
      if (job.state === "completed") throw new KeepError(409, "Completed keep jobs cannot be cancelled.");
      touch(job, { state: "cancelled", error: "Cancelled.", finishedAt: new Date(now()).toISOString() });
      job.controller?.abort();
      await rm(job.partialPath, { force: true });
      return publicKeepJob(job);
    },

    async deleteKept(jobId) {
      await readyPromise;
      const job = getJob(jobId);
      job.controller?.abort();
      await rm(job.partialPath, { force: true });
      // Never delete the original when Keep only marked an existing Private Reel path.
      if (job.keepPath && job.keepPath !== job.sourcePath) {
        await rm(job.keepPath, { force: true });
      }
      const removedCopy = Boolean(job.keepPath && job.keepPath !== job.sourcePath);
      jobs.delete(job.id);
      await queueStateWrite();
      if (typeof onComplete === "function" && removedCopy) await onComplete(job);
      return { deleted: true, id: job.id };
    },

    async ready() {
      await readyPromise;
    },

    async flush() {
      await readyPromise;
      await persistenceChain;
    },
  };
}
