// @pdsjs/core/handlers/account-backup - The snapshot engine and its account API routes.
//
// A snapshot is `snapshots/<stamp>/{repo.car, manifest.json, key.enc?}` in the
// backup target, with blobs deduplicated across snapshots at `blobs/<cid>` —
// content-addressed, so a run only uploads what the target does not already
// hold. The manifest is written last: a snapshot without one is incomplete and
// never counts for restore or blob retention.

import {
  BACKUP_CHECKPOINT_MS,
  BACKUP_RUNS_KEPT,
  defaultBackupState,
  isStaleRun,
  keyFingerprint,
  sealKeyBackup,
  snapshotStamp,
} from '../backup.js';
import { base64UrlDecode } from '../crypto.js';

/** Heads a snapshot tries before giving up, should commits keep landing. */
const SNAPSHOT_TRIES = 3;

/**
 * Count the bytes of a stream as they pass, and give the reader a place to
 * checkpoint between them. One blob is more than one request to the target,
 * and the run's heartbeat has to keep beating across them.
 * @param {ReadableStream<Uint8Array>} body
 * @param {(bytes: number) => Promise<void>} onBytes
 * @returns {ReadableStream<Uint8Array>}
 */
function meterStream(body, onBytes) {
  /** @type {TransformStream<Uint8Array, Uint8Array>} */
  const meter = new TransformStream({
    async transform(chunk, controller) {
      controller.enqueue(chunk);
      await onBytes(chunk.byteLength);
    },
  });
  return body.pipeThrough(meter);
}

/**
 * These are structural types. Thus a member does not name the module that owns
 * the function you give it.
 *
 * @typedef {Object} BackupContext
 * @property {import('../ports.js').ActorStoragePort} actorStorage
 * @property {import('../ports.js').BlobPort} blobs
 * @property {import('../ports.js').BackupTargetPort|null} backupTarget - Absent, backups report unavailable and the scheduler never runs
 * @property {string} [hostname] - Recorded in the manifest
 * @property {() => Promise<string|null>} getDid
 * @property {(commitCid: string) => Promise<Uint8Array>} buildFullRepoCar
 * @property {(request: Request, url: URL, handler: (did: string, body: any) => Promise<unknown>, options?: {write?: boolean}) => Promise<Response>} accountApi
 */

/**
 * @param {BackupContext} ctx
 * @returns {{
 *   backupsAvailable: () => boolean,
 *   readBackupState: () => Promise<import('../backup.js').BackupState>,
 *   writeBackupState: (state: import('../backup.js').BackupState) => Promise<void>,
 *   maybeRunScheduledBackup: (now?: number) => Promise<import('../backup.js').BackupRun|null>,
 *   runBackup: (trigger?: 'manual'|'scheduled') => Promise<import('../backup.js').BackupRun>,
 *   pruneBackups: (retain: number) => Promise<void>,
 *   publicBackupView: (state: import('../backup.js').BackupState) => Promise<{available: boolean, settings: {enabled: boolean, intervalHours: number, retain: number, keyBackup: {createdAt: string, stale: boolean}|null}, nextRunAt: number|null, runs: import('../backup.js').BackupRun[]}>,
 *   routes: import('../pds.js').Routes,
 * }}
 */
export function createBackupHandlers(ctx) {
  const {
    actorStorage,
    blobs,
    backupTarget,
    hostname,
    getDid,
    buildFullRepoCar,
    accountApi,
  } = ctx;

  // This flag applies to one isolate. The heartbeat in the stored run is the
  // claim that continues after a restart.
  let backupRunning = false;

  /**
   * Whether this deployment can take backups: a target to write to, and an
   * adapter that can hold the settings document.
   * @returns {boolean}
   */
  function backupsAvailable() {
    return Boolean(
      backupTarget &&
        actorStorage.getBackupState &&
        actorStorage.setBackupState,
    );
  }

  /**
   * The stored backup document, over defaults so a document written before a
   * field existed still reads whole.
   * @returns {Promise<import('../backup.js').BackupState>}
   */
  async function readBackupState() {
    const fallback = defaultBackupState();
    const stored = actorStorage.getBackupState
      ? await actorStorage.getBackupState()
      : null;
    if (!stored) return fallback;
    return {
      config: { ...fallback.config, ...stored.config },
      runs: stored.runs || [],
      nextRunAt: stored.nextRunAt ?? null,
    };
  }

  /**
   * @param {import('../backup.js').BackupState} state
   * @returns {Promise<void>}
   */
  async function writeBackupState(state) {
    if (actorStorage.setBackupState) {
      await actorStorage.setBackupState(state);
    }
  }

  /**
   * Run a backup if one is enabled and due. Called from each platform's
   * hourly tick (interval on Node and Deno, alarm on Cloudflare), so
   * schedules land within an hour of their mark.
   * @param {number} [now] - Epoch ms
   * @returns {Promise<import('../backup.js').BackupRun|null>} The finished
   *   run, or null when nothing was due
   */
  async function maybeRunScheduledBackup(now = Date.now()) {
    if (!backupsAvailable() || backupRunning) return null;
    const state = await readBackupState();
    if (!state.config.enabled || !state.nextRunAt || now < state.nextRunAt) {
      return null;
    }
    // A run in another isolate is still proving itself alive; don't stack.
    const running = state.runs.find((r) => r.status === 'running');
    if (running && !isStaleRun(running, now)) return null;
    return runBackup('scheduled');
  }

  /**
   * Take one snapshot now. Execution failures land in the returned run as
   * status 'error' rather than throwing; only being unable to start —
   * no target, no account, a run already going — throws.
   * @param {'manual'|'scheduled'} [trigger]
   * @returns {Promise<import('../backup.js').BackupRun>}
   */
  async function runBackup(trigger = 'manual') {
    const target = backupTarget;
    if (!target || !backupsAvailable()) {
      throw new Error('No backup target is configured');
    }
    if (backupRunning) throw new Error('A backup is already running');
    backupRunning = true;
    try {
      const did = await getDid();
      if (!did) throw new Error('No account on this server yet');
      let commit = await actorStorage.getLatestCommit();
      if (!commit) throw new Error('No repository to back up');

      const state = await readBackupState();
      const startedAt = new Date();

      // The lock is the persisted heartbeat, not just the flag above: a run
      // whose isolate died mid-copy stops beating and loses its claim, so an
      // interruption can never wedge backups until a restart.
      const running = state.runs.find((r) => r.status === 'running');
      if (running && !isStaleRun(running, startedAt.getTime())) {
        throw new Error('A backup is already running');
      }
      if (running) {
        running.status = 'error';
        running.error = 'Interrupted — the server stopped mid-run';
        running.finishedAt = running.heartbeatAt || running.startedAt;
      }
      // Two runs can share a millisecond; nudge the stamp until the name is
      // free so the second run cannot overwrite the first.
      let stampTime = startedAt.getTime();
      let stamp = snapshotStamp(startedAt);
      while (state.runs.some((r) => r.id === stamp)) {
        stampTime += 1;
        stamp = snapshotStamp(new Date(stampTime));
      }
      const snapshot = `snapshots/${stamp}`;
      /** @type {import('../backup.js').BackupRun} */
      const run = {
        id: stamp,
        trigger,
        status: 'running',
        startedAt: startedAt.toISOString(),
        finishedAt: null,
        heartbeatAt: startedAt.toISOString(),
        blobsCopied: 0,
        blobsTotal: 0,
        blobsFailed: 0,
        carBytes: 0,
        bytesUploaded: 0,
        snapshot,
        keyIncluded: Boolean(state.config.keyBackup),
        keyStale: false,
      };
      // Record the run and advance the schedule before doing the work, so a
      // failed run waits out the interval instead of retrying every tick.
      state.runs = [run, ...state.runs].slice(0, BACKUP_RUNS_KEPT);
      if (state.config.enabled) {
        state.nextRunAt =
          startedAt.getTime() + state.config.intervalHours * 3600_000;
      }
      await writeBackupState(state);

      // Persist the run's counters every few seconds: the page polls them as
      // live progress, and the fresh heartbeat is what keeps this run's claim
      // on the lock. Re-read before writing so a settings save that lands
      // mid-run is never clobbered.
      let lastCheckpoint = Date.now();
      const checkpoint = async (force = false) => {
        if (!force && Date.now() - lastCheckpoint < BACKUP_CHECKPOINT_MS) {
          return;
        }
        lastCheckpoint = Date.now();
        run.heartbeatAt = new Date().toISOString();
        const fresh = await readBackupState();
        fresh.runs = [run, ...fresh.runs.filter((r) => r.id !== run.id)].slice(
          0,
          BACKUP_RUNS_KEPT,
        );
        await writeBackupState(fresh);
      };

      try {
        /** @type {Uint8Array|null} */
        let car = null;
        for (let tries = 1; !car; tries++) {
          try {
            car = await buildFullRepoCar(commit.cid);
          } catch (err) {
            // A commit landing during the walk drops nodes of the tree it has
            // not reached yet: the snapshot is of the new head instead.
            const head = await actorStorage.getLatestCommit();
            if (tries === SNAPSHOT_TRIES || !head || head.cid === commit.cid) {
              throw err;
            }
            commit = head;
          }
        }
        run.carBytes = car.byteLength;
        await checkpoint(true);

        // Blobs the target already holds are skipped, which is what makes
        // every run after the first cheap.
        const existing = new Set(await target.list('blobs/'));
        /** @type {string[]} */
        const blobCids = [];
        /** @type {string[]} */
        const missing = [];
        /** @type {Array<{cid: string, error: string}>} */
        const blobErrors = [];
        /** @type {string|null} */
        let cursor = null;
        do {
          const page = await actorStorage.listBlobs(cursor, 500);
          for (const cid of page.cids) {
            blobCids.push(cid);
            if (existing.has(`blobs/${cid}`)) continue;
            try {
              // A blob can be larger than the memory this process has, and
              // larger than one request to the target. Stream it when both
              // sides can, and read it whole only when one of them cannot.
              if (target.putStream && blobs.getStream) {
                const streamed = await blobs.getStream(did, cid);
                if (!streamed) {
                  missing.push(cid);
                  continue;
                }
                await target.putStream(
                  `blobs/${cid}`,
                  meterStream(streamed.body, async (bytes) => {
                    run.bytesUploaded += bytes;
                    await checkpoint();
                  }),
                  streamed.mimeType,
                );
              } else {
                const blob = await blobs.get(did, cid);
                if (!blob) {
                  missing.push(cid);
                  continue;
                }
                await target.put(`blobs/${cid}`, blob.data, blob.mimeType);
                run.bytesUploaded += blob.data.byteLength;
              }
            } catch (err) {
              // One blob the target refuses costs the snapshot that blob, not
              // the repo and not the other blobs. The next run tries it again:
              // a blob the target does not hold is never skipped. A target
              // that is failing for every blob fails the CAR too, a few lines
              // down, and that does end the run.
              const message = err instanceof Error ? err.message : String(err);
              missing.push(cid);
              blobErrors.push({ cid, error: message });
              run.blobsFailed += 1;
              if (!run.blobError) run.blobError = message;
              await checkpoint();
              continue;
            }
            run.blobsCopied += 1;
            await checkpoint();
          }
          cursor = page.cursor;
        } while (cursor);
        run.blobsTotal = blobCids.length;
        await checkpoint(true);

        const keyBackup = state.config.keyBackup;
        if (keyBackup) {
          const stored = await actorStorage.getPrivateKey();
          run.keyStale =
            !stored || (await keyFingerprint(stored)) !== keyBackup.keyHash;
        }

        await target.put(
          `${snapshot}/repo.car`,
          car,
          'application/vnd.ipld.car',
        );
        run.bytesUploaded += car.byteLength;

        if (keyBackup) {
          const keyBytes = new TextEncoder().encode(
            JSON.stringify(keyBackup, null, 2),
          );
          await target.put(`${snapshot}/key.enc`, keyBytes, 'application/json');
          run.bytesUploaded += keyBytes.byteLength;
        }

        const manifest = {
          version: 1,
          did,
          handle: await actorStorage.getHandle(),
          hostname: hostname ?? null,
          createdAt: startedAt.toISOString(),
          commit: { cid: commit.cid, rev: commit.rev, seq: commit.seq },
          carBytes: car.byteLength,
          blobs: blobCids,
          // Everything this snapshot does not hold, whatever the cause, so a
          // restore reads one list. `blobErrors` says why for the ones the
          // target refused.
          missingBlobs: missing,
          blobErrors,
          keyIncluded: Boolean(keyBackup),
          keyStale: run.keyStale,
        };
        const manifestBytes = new TextEncoder().encode(
          JSON.stringify(manifest, null, 2),
        );
        await target.put(
          `${snapshot}/manifest.json`,
          manifestBytes,
          'application/json',
        );
        run.bytesUploaded += manifestBytes.byteLength;
        await checkpoint(true);

        await pruneBackups(state.config.retain);
        run.status = 'ok';
      } catch (err) {
        run.status = 'error';
        run.error = err instanceof Error ? err.message : String(err);
      }
      run.finishedAt = new Date().toISOString();

      // Re-read before persisting the finished run: settings may have been
      // saved while the copy ran, and this write must not undo them.
      const fresh = await readBackupState();
      fresh.runs = [run, ...fresh.runs.filter((r) => r.id !== run.id)].slice(
        0,
        BACKUP_RUNS_KEPT,
      );
      await writeBackupState(fresh);
      return run;
    } finally {
      backupRunning = false;
    }
  }

  /**
   * Drop snapshots beyond the newest `retain`, then delete blobs no kept
   * manifest references. Blob GC is skipped entirely if no kept manifest is
   * readable — better to hold garbage than to reap blobs still needed.
   * @param {number} retain
   * @returns {Promise<void>}
   */
  async function pruneBackups(retain) {
    const target = backupTarget;
    if (!target) return;

    /** @type {Map<string, string[]>} */
    const byStamp = new Map();
    for (const key of await target.list('snapshots/')) {
      const stamp = key.slice('snapshots/'.length).split('/')[0];
      if (!stamp) continue;
      const keys = byStamp.get(stamp) ?? [];
      keys.push(key);
      byStamp.set(stamp, keys);
    }
    const stamps = [...byStamp.keys()].sort().reverse();
    const keep = stamps.slice(0, Math.max(1, retain));
    for (const stamp of stamps.slice(Math.max(1, retain))) {
      for (const key of byStamp.get(stamp) ?? []) {
        await target.delete(key);
      }
    }

    const referenced = new Set();
    let readableManifests = 0;
    for (const stamp of keep) {
      const bytes = await target.get(`snapshots/${stamp}/manifest.json`);
      if (!bytes) continue;
      try {
        const manifest = JSON.parse(new TextDecoder().decode(bytes));
        readableManifests += 1;
        for (const cid of manifest.blobs || []) referenced.add(`blobs/${cid}`);
      } catch {
        // An unreadable manifest names blobs we cannot know; the guard
        // below keeps GC from running on partial knowledge.
        readableManifests = 0;
        break;
      }
    }
    if (readableManifests === 0) return;
    for (const key of await target.list('blobs/')) {
      if (!referenced.has(key)) await target.delete(key);
    }
  }

  /**
   * The backup document as the account API tells it: everything but the
   * sealed key ciphertext, which never leaves the server this way.
   * @param {import('../backup.js').BackupState} state
   * @returns {Promise<{available: boolean, settings: {enabled: boolean, intervalHours: number, retain: number, keyBackup: {createdAt: string, stale: boolean}|null}, nextRunAt: number|null, runs: import('../backup.js').BackupRun[]}>}
   */
  async function publicBackupView(state) {
    const keyBackup = state.config.keyBackup;
    /** @type {{createdAt: string, stale: boolean}|null} */
    let keyView = null;
    if (keyBackup) {
      const stored = await actorStorage.getPrivateKey();
      keyView = {
        createdAt: keyBackup.createdAt,
        stale: !stored || (await keyFingerprint(stored)) !== keyBackup.keyHash,
      };
    }
    return {
      available: backupsAvailable(),
      settings: {
        enabled: state.config.enabled,
        intervalHours: state.config.intervalHours,
        retain: state.config.retain,
        keyBackup: keyView,
      },
      nextRunAt: state.config.enabled ? state.nextRunAt : null,
      // A run that stopped beating is shown as what it is. The stored row is
      // corrected for real when the next run claims the lock.
      runs: state.runs.map((r) =>
        isStaleRun(r, Date.now())
          ? {
              ...r,
              status: /** @type {const} */ ('error'),
              error: 'Interrupted — the server stopped mid-run',
              finishedAt: r.heartbeatAt || r.startedAt,
            }
          : r,
      ),
    };
  }

  /**
   * GET /account/api/backups - Settings, schedule, and run history
   * @param {Request} request
   * @param {URL} url
   * @returns {Promise<Response>}
   */
  async function handleApiBackups(request, url) {
    return accountApi(request, url, async () =>
      publicBackupView(await readBackupState()),
    );
  }

  /**
   * POST /account/api/backups/settings - Change any subset of the settings;
   * a passphrase here seals the signing key into future snapshots, and
   * removeKeyBackup takes it back out.
   * @param {Request} request
   * @param {URL} url
   * @returns {Promise<Response>}
   */
  async function handleApiBackupsSettings(request, url) {
    return accountApi(
      request,
      url,
      async (_did, body) => {
        if (!backupsAvailable()) {
          throw new Error('No backup target is configured');
        }
        const state = await readBackupState();
        const config = state.config;

        if (typeof body.enabled === 'boolean') config.enabled = body.enabled;
        if (body.intervalHours !== undefined) {
          const hours = Number(body.intervalHours);
          if (!Number.isFinite(hours) || hours < 1 || hours > 720) {
            throw new Error('intervalHours must be between 1 and 720');
          }
          config.intervalHours = hours;
        }
        if (body.retain !== undefined) {
          const retain = Number(body.retain);
          if (!Number.isInteger(retain) || retain < 1 || retain > 50) {
            throw new Error('retain must be between 1 and 50');
          }
          config.retain = retain;
        }
        if (body.removeKeyBackup === true) config.keyBackup = null;
        // The browser derives the AES key from the passphrase (PBKDF2 at a
        // count Workers' WebCrypto refuses to run) and sends key, salt and
        // count. The passphrase itself never reaches the server.
        if (body.keyDerived) {
          const { key, salt, iterations } = body.keyDerived;
          if (typeof key !== 'string' || typeof salt !== 'string') {
            throw new Error('keyDerived needs base64url key and salt');
          }
          if (
            !Number.isInteger(iterations) ||
            iterations < 100_000 ||
            iterations > 10_000_000
          ) {
            throw new Error('keyDerived.iterations out of range');
          }
          const derivedKey = base64UrlDecode(key);
          const saltBytes = base64UrlDecode(salt);
          if (derivedKey.byteLength !== 32 || saltBytes.byteLength < 8) {
            throw new Error('keyDerived key must be 32 bytes, salt at least 8');
          }
          const stored = await actorStorage.getPrivateKey();
          if (!stored) throw new Error('Server has no signing key');
          config.keyBackup = await sealKeyBackup(
            { derivedKey, salt: saltBytes, iterations },
            stored,
          );
        }

        if (config.enabled) {
          // Due one interval after the last run, whatever its outcome — or
          // right away if there has never been one. The hourly tick picks it
          // up from there.
          const lastRun = state.runs[0];
          state.nextRunAt = lastRun
            ? Date.parse(lastRun.startedAt) + config.intervalHours * 3600_000
            : Date.now();
        } else {
          state.nextRunAt = null;
        }

        await writeBackupState(state);
        return publicBackupView(state);
      },
      { write: true },
    );
  }

  /**
   * POST /account/api/backups/run - Take a snapshot now. Answers with the
   * finished run; a run that failed partway is an 'error' run, not an HTTP
   * error.
   * @param {Request} request
   * @param {URL} url
   * @returns {Promise<Response>}
   */
  async function handleApiBackupsRun(request, url) {
    return accountApi(request, url, async () => runBackup('manual'), {
      write: true,
    });
  }

  return {
    backupsAvailable,
    readBackupState,
    writeBackupState,
    maybeRunScheduledBackup,
    runBackup,
    pruneBackups,
    publicBackupView,
    routes: {
      '/account/api/backups': {
        method: 'GET',
        handler: handleApiBackups,
      },
      '/account/api/backups/settings': {
        method: 'POST',
        handler: handleApiBackupsSettings,
      },
      '/account/api/backups/run': {
        method: 'POST',
        handler: handleApiBackupsRun,
      },
    },
  };
}