fofos-admin/api/sync.ts

import { mkdir, stat } from 'node:fs/promises';
import { dirname } from 'node:path';
import {
  deriveOne,
  derivedDir,
  indexDerived,
  listOriginals,
  originalsDir,
  pruneDerived,
  relocateDerived,
  safeScope,
  targetFor,
} from './derive';
import { appendRun, type RunTrigger } from './ledger';

/**
 * The sync: bring `derived/` into line with `originals/`.
 *
 * ── One job, in memory, per user ──
 *
 * Not a `pipeline_jobs` row. That table belongs to the task runner, which runs a script in a temp
 * directory with a cwd and a log — a good fit for what a user wrote and a poor one for a plugin
 * calling a service in-process. So the state lives here, and the cost is stated plainly: **a restart
 * loses the job.**
 *
 * That cost is small because the work is idempotent. Every derived directory carries a manifest naming
 * the size and mtime it was built from, so a sync that died halfway is finished by running it again —
 * it skips everything already correct and picks up where the interruption left it. There is nothing to
 * reconcile and nothing to clean up, which is the property that made an in-memory job acceptable
 * rather than merely convenient.
 *
 * ── The same code path as a single file ──
 *
 * `deriveOne` is what the watcher will call for one upload. This walks and calls it; it decides
 * nothing about how a file is built. A full sync and a single arrival must never be two implementations
 * that can disagree about what a derived directory should contain.
 *
 * ── One at a time ──
 *
 * It was two, to match av-club's own concurrency. That is the right number on a machine with room and
 * the wrong one here: converting a 12-megapixel HEIC holds a full-resolution bitmap, and fofos has
 * 1.87 GiB of RAM. Two at once left about 520 MiB free with 648 MB already paged out — CPU pinned at
 * 99%, and an out-of-memory kill one large photograph away.
 *
 * Nothing is lost by going single-file. The work is CPU-bound on a box that was already saturated, so
 * the second worker was not buying throughput; it was buying a second resident bitmap. It also makes
 * "what is happening now" one filename instead of two, which is the more useful thing to read.
 */

export type SyncError = { file: string; message: string };

export type SyncState = {
  running: boolean;
  /** How many originals the walk found. Null until the walk finishes. */
  total: number | null;
  /** Attempted — built, skipped and failed together. What a progress bar divides by `total`. */
  done: number;
  built: number;
  skipped: number;
  /** Ladders that were RENAMED into place because their original moved, rather than rebuilt. */
  moved: number;
  /** Derived folders removed because their original is gone. Only on a whole-tree sync. */
  pruned: number;
  errors: SyncError[];
  /** The files being worked on right now, for somebody watching. Empty when idle. */
  current: string[];
  startedAt: string | null;
  finishedAt: string | null;
  /** Set when the run stopped for a reason that was not about any single file — av-club being down. */
  fatal: string | null;
  /**
   * The subtree this run covered, relative to `originals/`. Empty means all of it.
   *
   * Recorded rather than implied, because "3 built" means something different depending on whether it
   * was the whole library or one folder somebody had just dropped photographs into.
   */
  scope: string;
  trigger: RunTrigger;
  /**
   * A running transcript — what happened, to which file, when.
   *
   * The counters say how MUCH; this says what. They answer different questions and a run that stalls
   * on one photograph is invisible in the counters: `done` simply stops moving, and there is nothing
   * to tell "av-club is chewing on a 40-megapixel HEIC" apart from "the process is wedged".
   *
   * Kept on the state rather than written to a file, because the run already lives in memory and a log
   * file would be a second thing to find, rotate and clean up for output whose whole value is that
   * somebody is watching it right now. The ledger is what survives; this is what is happening.
   */
  log: LogLine[];
};

/** One line of the transcript. */
export type LogLine = {
  /**
   * Monotonic, and the reason the client can poll cheaply.
   *
   * A watcher asks for everything after the last number it holds, so a modal open for a long run
   * transfers each line once instead of re-sending the whole transcript every second.
   */
  n: number;
  at: string;
  level: 'info' | 'step' | 'done' | 'warn' | 'error';
  /** The file this is about, when it is about one. Blank for the run's own lines. */
  file: string;
  text: string;
};

/**
 * How many lines to keep.
 *
 * A photograph produces about thirty of them, so this is a few hundred files of detail. Past it the
 * OLDEST go: the interesting end of a long run is the end, and a transcript that stopped recording at
 * the top would drop the failure somebody opened it to read.
 */
const LOG_LIMIT = 4000;

/** One. See the note above — this is a memory limit, not a politeness. */
const CONCURRENCY = 1;

const idle = (): SyncState => ({
  running: false,
  total: null,
  done: 0,
  built: 0,
  skipped: 0,
  moved: 0,
  pruned: 0,
  errors: [],
  current: [],
  startedAt: null,
  finishedAt: null,
  fatal: null,
  scope: '',
  trigger: 'manual',
  log: [],
});

const states = new Map<string, SyncState>();

export const readSync = (email: string): SyncState => states.get(email) ?? idle();

/**
 * Start one, unless one is already running.
 *
 * Returns the state either way, so the caller does not have to decide what "already running" means to
 * a person: the panel shows the run that is in progress, which is the honest answer to "sync now".
 */
export function startSync(
  email: string,
  options: { scope?: string; trigger?: RunTrigger } = {},
): { started: boolean; state: SyncState } {
  const existing = states.get(email);
  if (existing?.running) return { started: false, state: existing };

  const state: SyncState = {
    ...idle(),
    running: true,
    startedAt: new Date().toISOString(),
    scope: safeScope(options.scope ?? ''),
    trigger: options.trigger ?? 'manual',
  };
  states.set(email, state);

  // Deliberately not awaited. The route answers immediately with the job's opening state and the panel
  // polls; an await here would be the request timing out on a folder of any size.
  void run(email, state);

  return { started: true, state };
}

/**
 * Add a line, and keep the buffer bounded.
 *
 * `seq` is per-run and starts where the last one left off only in the sense that a new run resets it —
 * a watcher that reconnects across runs sees the numbers go backwards, which is why the client drops
 * what it holds when `startedAt` changes rather than trusting the numbers alone.
 */
const say = (state: SyncState, level: LogLine['level'], file: string, text: string): void => {
  const n = (state.log.at(-1)?.n ?? 0) + 1;
  state.log.push({ n, at: new Date().toISOString(), level, file, text });
  // Oldest first. See LOG_LIMIT — the end of a long run is the part worth keeping.
  if (state.log.length > LOG_LIMIT) state.log.splice(0, state.log.length - LOG_LIMIT);
};

/** `1.4s`, `320ms`. Short enough to sit at the end of a line without becoming the line. */
const since = (from: number): string => {
  const ms = Date.now() - from;
  return ms < 1000 ? `${ms}ms` : `${(ms / 1000).toFixed(1)}s`;
};

async function run(email: string, state: SyncState): Promise<void> {
  const began = Date.now();
  try {
    const from = await originalsDir(email);
    const to = await derivedDir(email);

    say(state, 'info', '', `Reading ${state.scope || 'originals/'}`);

    const files = await listOriginals(from, state.scope);
    state.total = files.length;

    say(state, 'info', '', `${files.length} file${files.length === 1 ? '' : 's'} to check`);

    /*
      What is already built, indexed by what it was built FROM.

      Built once, before the walk, so a file that has simply MOVED can have its ladder renamed into
      place instead of converted again — minutes of work to produce bytes that already exist a
      directory away. Size and mtime survive a move and are already in both the manifest and the stat,
      so the index costs one pass over derived/ and nothing else.
    */
    const derivedIndex = await indexDerived(to);

    // Worth a line: on a big library this is the pause before anything appears to happen, and without
    // it the transcript opens with a gap somebody reads as a hang.
    say(state, 'info', '', `Indexed ${derivedIndex.size} existing ladder${derivedIndex.size === 1 ? '' : 's'}`);

    let next = 0;
    const worker = async (): Promise<void> => {
      for (;;) {
        const index = next++;
        const file = files[index];
        if (file === undefined) return;

        state.current.push(file);
        const startedFile = Date.now();
        say(state, 'step', file, 'checking');
        try {
          const target = targetFor(to, file);
          await mkdir(dirname(target), { recursive: true });

          /*
            A move, before a build.

            `deriveOne` decides by looking for a manifest at the target, so a file that changed folder
            has none there and would convert from scratch. This asks the index whether the same bytes
            were already built somewhere else and, if exactly one folder matches, renames it across.
            An ambiguous match is refused inside `relocateDerived` — rebuilding is slow, attaching the
            wrong photographs to a name is wrong.
          */
          const info = await stat(`${from}/${file}`).catch(() => null);
          const relocated = info
            ? await relocateDerived(derivedIndex, to, file, { bytes: info.size, mtimeMs: Math.round(info.mtimeMs) })
            : false;

          if (relocated) say(state, 'info', file, 'moved — renaming its ladder instead of rebuilding');

          /*
            Every rung, as it lands. This is the "what is it doing right now" the counters cannot say:
            one photograph is thirty encodes and a video is three of them plus a poster ladder, and a
            run that looks stopped is usually most of the way through one of those.
          */
          const { built } = await deriveOne(`${from}/${file}`, target, email, {
            onStep: (message) => say(state, 'step', file, message),
          });

          if (relocated && !built) {
            state.moved++;
            say(state, 'done', file, `moved · ${since(startedFile)}`);
          } else if (built) {
            state.built++;
            say(state, 'done', file, `built · ${since(startedFile)}`);
          } else {
            state.skipped++;
            // `current` and not `skipped`: nothing was skipped over, the ladder was already right.
            say(state, 'done', file, `already current · ${since(startedFile)}`);
          }
        } catch (error) {
          const message = (error as Error).message;
          /*
            av-club being unreachable is not a fact about this file, and recording it once per file
            would bury the one thing worth reading under four hundred copies of it. Stop the run and
            say so once.
          */
          if (message.includes('av-club is not running')) {
            state.fatal = message;
            say(state, 'error', '', message);
            return;
          }
          state.errors.push({ file, message });
          say(state, 'error', file, message);
        } finally {
          state.current = state.current.filter((name) => name !== file);
          state.done++;
        }

        if (state.fatal) return;
      }
    };

    await Promise.all(Array.from({ length: CONCURRENCY }, worker));

    /*
      Then the orphans — derived folders whose original no longer exists.

      Only on a WHOLE-TREE sync. A scoped one knows about a single folder, so everything outside it
      looks abandoned, and deleting on that basis would take out the library the first time somebody
      re-synced one subfolder. Safe otherwise because derived/ is reproducible by definition, which is
      what the name has meant since these two folders stopped being called `user` and `treated`.
    */
    if (!state.scope && !state.fatal) {
      const orphans = await pruneDerived(from, to);
      state.pruned = orphans.length;
      // Named, not counted. A removal nobody can see the target of is the one thing here somebody
      // might want to argue with afterwards.
      for (const orphan of orphans) say(state, 'warn', orphan, 'removed — its original is gone');
    }
  } catch (error) {
    state.fatal = (error as Error).message;
    say(state, 'error', '', (error as Error).message);
  } finally {
    state.running = false;
    state.current = [];
    state.finishedAt = new Date().toISOString();

    // The last line, and the one still on screen when the modal stops moving.
    say(
      state,
      state.fatal || state.errors.length ? 'warn' : 'done',
      '',
      [
        `${state.built} built`,
        state.skipped ? `${state.skipped} already current` : '',
        state.moved ? `${state.moved} moved` : '',
        state.pruned ? `${state.pruned} removed` : '',
        state.errors.length ? `${state.errors.length} failed` : '',
        `in ${since(began)}`,
      ]
        .filter(Boolean)
        .join(' · '),
    );

    /*
      The one durable record this leaves.

      In `finally`, so a run that died still writes a line — a failure nobody was watching is the whole
      reason the ledger exists, and the runs that go unrecorded would be exactly the ones worth having.
    */
    await appendRun(email, {
      startedAt: state.startedAt ?? state.finishedAt,
      finishedAt: state.finishedAt,
      trigger: state.trigger,
      scope: state.scope,
      total: state.total ?? state.done,
      built: state.built,
      skipped: state.skipped,
      moved: state.moved,
      pruned: state.pruned,
      errors: state.errors,
      fatal: state.fatal,
    });
  }
}