spb/drive Public

1/**2 * ─────────────────────────────────────────────3 *  SPB Drive — Personal Cloud Drive4 * ─────────────────────────────────────────────5 *  Author  : Simon-Pierre Boucher6 *  Contact : contact@spboucher.ai7 *  File    : src/preview/queue.mjs8 *  Purpose : Tiny in-process job queue (concurrency 2) for transcodes/conversions9 *  License : MIT © Simon-Pierre Boucher10 * ─────────────────────────────────────────────11 */1213const CONCURRENCY = 2;1415/** @type {Map<string, {promise: Promise<any>, status: string}>} */16const jobs = new Map();17const waiting = [];18let running = 0;1920function pump() {21  while (running < CONCURRENCY && waiting.length > 0) {22    const next = waiting.shift();23    running += 1;24    next.entry.status = 'running';25    Promise.resolve()26      .then(next.fn)27      .then(28        (val) => { next.entry.status = 'done'; next.resolve(val); },29        (err) => { next.entry.status = 'error'; next.entry.error = String(err?.message ?? err); next.reject(err); },30      )31      .finally(() => {32        running -= 1;33        setTimeout(() => jobs.delete(next.key), 60_000);34        pump();35      });36  }37}3839/**40 * Enqueue work keyed by a cache key — duplicate keys share one promise,41 * so a conversion runs at most once per unique blob.42 * @param {string} key e.g. `office:<sha>`43 * @param {() => Promise<any>} fn44 * @returns {Promise<any>}45 */46export function enqueue(key, fn) {47  const existing = jobs.get(key);48  if (existing) return existing.promise;49  const entry = { status: 'queued', error: null };50  entry.promise = new Promise((resolve, reject) => {51    waiting.push({ key, fn, resolve, reject, entry });52  });53  entry.promise.catch(() => {}); // observed by callers; avoid unhandled warnings54  jobs.set(key, entry);55  pump();56  return entry.promise;57}5859/** Status for a key: 'queued' | 'running' | 'done' | 'error' | null. */60export function jobStatus(key) {61  const entry = jobs.get(key);62  return entry ? { status: entry.status, error: entry.error ?? null } : null;63}6465/** Current queue depth (for /healthz). */66export function queueDepth() {67  return waiting.length + running;68}69