fofos-admin/api/sync.ts

import { mkdir } from 'node:fs/promises';
import { dirname } from 'node:path';
import { deriveOne, derivedDir, listOriginals, originalsDir, targetFor } from './derive';

/**
 * 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.
 *
 * ── Two at a time ──
 *
 * av-club runs two jobs concurrently and queues the rest, so a third caller would only wait in its
 * queue while making this module's progress harder to report. Matching its number keeps both slots
 * busy and keeps "what is happening now" answerable.
 */

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;
  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;
};

const CONCURRENCY = 2;

const idle = (): SyncState => ({
  running: false,
  total: null,
  done: 0,
  built: 0,
  skipped: 0,
  errors: [],
  current: [],
  startedAt: null,
  finishedAt: null,
  fatal: null,
});

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): { 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() };
  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 };
}

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

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

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

        state.current.push(file);
        try {
          const target = targetFor(to, file);
          await mkdir(dirname(target), { recursive: true });
          const { built } = await deriveOne(`${from}/${file}`, target, email);
          if (built) state.built++;
          else state.skipped++;
        } 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;
            return;
          }
          state.errors.push({ 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));
  } catch (error) {
    state.fatal = (error as Error).message;
  } finally {
    state.running = false;
    state.current = [];
    state.finishedAt = new Date().toISOString();
  }
}