PaperClipAI/server/src/services/scoped-deadline-work.ts

import { randomUUID } from "node:crypto";
import type { DeadlineReconciliationSnapshot } from "./deadline-reconciliation.js";
import { takeCompanyWorkBatch } from "./company-work-batch.js";
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
import type { DeliveryQueue } from "./delivery-work-notifications.js";

type ScheduledEntry = { id: string; companyId: string; dueAt: number };

/** Coalesced scope reconstruction and fair deadline dispatch. Domain owners
 * supply persistence, eligibility, and execution; this helper owns no SQL. */
export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
  options: {
    queue: DeliveryQueue;
    batchSize?: number;
    minimumPassIntervalMs?: number;
    concurrentChecks?: {
      limit: number;
      key(entry: Entry): string | readonly string[];
      // A pass otherwise ends only after its slowest check. Owners whose checks
      // can legitimately run for minutes may let idle slots keep admitting
      // entries that become due meanwhile, so one slow resource cannot hold
      // every other resource's saved deadline until it finishes. check() must
      // revalidate each entry against durable state, because an entry admitted
      // this way is not reconstructed first.
      admitDueWhileActive?: boolean;
    };
    reconstructionIntervalMs?: number | ((scope: Scope | undefined) => number);
    beforePass?(): Promise<void>;
    decodeScope(hint: string): Scope | undefined;
    matches(entry: Entry, scope?: Scope): boolean;
    load(scope?: Scope, signal?: AbortSignal): Promise<Entry[] | { rows: Entry[] | null; complete: boolean }>;
    // Owners may retain pending queue priority while refreshing eligibility.
    mergeEntry?(pending: Entry, loaded: Entry): Entry;
    priority?(entry: Entry): number;
    check(entry: Entry, signal?: AbortSignal): Promise<Entry | null>;
    onStop?(): void;
  },
  canRun: () => boolean,
) {
  if (options.concurrentChecks && (!Number.isInteger(options.concurrentChecks.limit) || options.concurrentChecks.limit < 1)) {
    throw new RangeError("Concurrent check limit must be a positive integer");
  }
  const ownerRevision = randomUUID();
  let revision = 0, running = false, stopped = false, lastRunFailed = false;
  const dirty = new Set<string>(["*"]);
  const versions = new Map<string, number>();
  const scans = new Map<string, { version: number; seen: Set<string> }>();
  const entries = new Map<string, Entry>();
  const failures = new Map<string, { count: number; retryAt: number }>();
  const scopeFailures = new Map<string, { count: number; retryAt: number }>();
  const reconstructAfter = new Map<string, number>();
  let earliest = 0;
  let lastCompany: string | null = null;
  function failed(map: typeof failures, id: string) {
    const count = Math.min((map.get(id)?.count ?? 0) + 1, 5);
    map.set(id, { count, retryAt: Date.now() + Math.min(300_000, 30_000 * 2 ** (count - 1)) });
  }
  function dirtyScopes() {
    return [...dirty].filter(hint => hint !== "*" || !scopeFailures.has(hint))
      .filter(hint => !scans.has(hint) || scopeFailures.has(hint))
      .filter(hint => (reconstructAfter.get(hint) ?? 0) <= Date.now() || scopeFailures.has(hint))
      .map(hint => { try { return options.decodeScope(hint); } catch { return undefined; } });
  }
  function next() {
    if (!canRun()) return dirty.size > 0 || entries.size > 0 ? Date.now() + 30_000 : null;
    let nextAt: number | null = null;
    const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); };
    for (const hint of dirty) take(Math.max(earliest, reconstructAfter.get(hint) ?? 0, scopeFailures.get(hint)?.retryAt ?? 0));
    const unavailable = dirtyScopes();
    for (const entry of entries.values()) if (!unavailable.some(scope => options.matches(entry, scope)))
      take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0));
    return nextAt;
  }
  const worker = coordinator.register(options.queue, {
    retryMs: 30_000,
    pausePassiveMaintenance: true,
    onStop: () => { stopped = true; revision++; options.onStop?.(); },
    onSettled: hints => { revision++; for (const hint of hints ?? ["*"]) {
      dirty.add(hint); versions.set(hint, (versions.get(hint) ?? 0) + 1);
    } },
    run: async signal => {
      revision++; running = true;
      try {
        if (!canRun() || Date.now() < earliest) return;
        for (const [hint, until] of reconstructAfter) {
          if (until <= Date.now()) reconstructAfter.delete(hint);
        }
        const ready = [...dirty].filter(hint => Math.max(reconstructAfter.get(hint) ?? 0, scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now());
        if (ready.length === 0 && ![...entries.values()].some(entry => Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now())) return;
        earliest = Date.now() + (options.minimumPassIntervalMs ?? 1000);
        try { await options.beforePass?.(); } catch (error) {
          for (const hint of ready) failed(scopeFailures, hint);
          for (const entry of entries.values()) if (entry.dueAt <= Date.now()) failed(failures, entry.id);
          throw error;
        }
        const errors: unknown[] = [];
        const requested = ready.includes("*")
          ? ["*", ...ready.filter(hint => hint !== "*")].slice(0, 25)
          : ready.slice(0, 25);
        for (const hint of requested) dirty.delete(hint);
        for (let index = 0; index < requested.length; index++) {
          if (signal.aborted || !canRun()) { for (const hint of requested.slice(index)) dirty.add(hint); return; }
          const hint = requested[index];
          let scope: Scope | undefined;
          let decoded = false;
          try {
            scope = options.decodeScope(hint); decoded = true;
            const scan = scans.get(hint) ?? { version: versions.get(hint) ?? 0, seen: new Set<string>() };
            scans.set(hint, scan);
            const loaded = await options.load(scope, signal);
            const result = Array.isArray(loaded) ? { rows: loaded, complete: true } : loaded;
            for (const entry of result.rows ?? []) {
              const pending = entries.get(entry.id);
              entries.set(entry.id, pending && options.mergeEntry ? options.mergeEntry(pending, entry) : entry);
              for (const [other, progress] of scans) if (options.matches(entry, options.decodeScope(other))) progress.seen.add(entry.id);
            }
            if (!result.complete) { dirty.add(hint); continue; }
            if (result.rows !== null) for (const entry of entries.values()) if (options.matches(entry, scope) && !scan.seen.has(entry.id)) {
              entries.delete(entry.id); failures.delete(entry.id);
            }
            scans.delete(hint);
            const reconstructionInterval = typeof options.reconstructionIntervalMs === "function"
              ? options.reconstructionIntervalMs(scope) : (options.reconstructionIntervalMs ?? 0);
            if (reconstructionInterval > 0) {
              reconstructAfter.set(hint, Date.now() + reconstructionInterval);
            }
            if ((versions.get(hint) ?? 0) !== scan.version) dirty.add(hint); else versions.delete(hint);
            if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint);
          } catch (error) {
            const retryHint = decoded ? hint : "*";
            if (!decoded) { scans.delete(hint); versions.delete(hint); }
            dirty.add(retryHint); failed(scopeFailures, retryHint); errors.push(error);
          }
        }
        const unavailable = dirtyScopes();
        const due = [...entries.values()].filter(entry => !unavailable.some(scope => options.matches(entry, scope))
          && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now())
          .sort((a, b) => (options.priority?.(b) ?? 0) - (options.priority?.(a) ?? 0)
            || a.dueAt - b.dueAt || a.id.localeCompare(b.id));
        const batch = takeCompanyWorkBatch(due, options.batchSize ?? 25, lastCompany); lastCompany = batch.lastCompany;
        // An owner may share more than one resource (for example a task and an
        // endpoint). Merge intersecting keys before dispatch so waiting siblings
        // do not occupy the bounded check slots ahead of independent work.
        const parents = batch.entries.map((_, index) => index);
        const root = (index: number): number => {
          while (parents[index] !== index) index = parents[index];
          return index;
        };
        const owners = new Map<string, number>();
        for (const [index, entry] of batch.entries.entries()) {
          const configured = options.concurrentChecks?.key(entry) ?? "";
          const keys = typeof configured === "string" ? [configured] : configured;
          for (const key of keys) {
            const previous = owners.get(key);
            if (previous === undefined) owners.set(key, index);
            else parents[root(index)] = root(previous);
          }
        }
        const groups = new Map<number, Entry[]>();
        for (const [index, entry] of batch.entries.entries()) {
          const key = root(index);
          const group = groups.get(key) ?? [];
          group.push(entry); groups.set(key, group);
        }
        const pending = [...groups.values()];
        let nextGroup = 0;
        const admitDueWhileActive = options.concurrentChecks?.admitDueWhileActive === true;
        const keysOf = (entry: Entry) => {
          const configured = options.concurrentChecks?.key(entry) ?? "";
          return typeof configured === "string" ? [configured] : [...configured];
        };
        // Entries owned by this pass (queued in a group or running) and the
        // resource keys they hold. Every batch group reserves its keys until the
        // whole group finishes, so admission can neither overlap a queued
        // sibling nor jump ahead of it. Admission never overlaps either set.
        const owned = new Set(batch.entries.map(entry => entry.id));
        const busyKeys = new Map<string, number>();
        const reserve = (keys: readonly string[]) => {
          for (const key of keys) busyKeys.set(key, (busyKeys.get(key) ?? 0) + 1);
        };
        const release = (keys: readonly string[]) => {
          for (const key of keys) {
            const count = (busyKeys.get(key) ?? 1) - 1;
            if (count > 0) busyKeys.set(key, count); else busyKeys.delete(key);
          }
        };
        const groupKeys = pending.map(group => [...new Set(group.flatMap(keysOf))]);
        for (const keys of groupKeys) reserve(keys);
        const lastChecked = new Map<string, number>();
        const spacing = options.minimumPassIntervalMs ?? 1000;
        let running = 0;
        let changed: (() => void) | null = null;
        const notify = () => { const resolve = changed; changed = null; resolve?.(); };
        const checkEntry = async (entry: Entry) => {
          running += 1;
          try {
            const current = await options.check(entry, signal);
            if (current) entries.set(entry.id, current); else entries.delete(entry.id);
            failures.delete(entry.id);
          } catch (error) { failed(failures, entry.id); errors.push(error); }
          finally {
            running -= 1;
            lastChecked.set(entry.id, Date.now());
            notify();
          }
        };
        const nextAdmission = () => {
          // Mid-pass settlements mark scopes dirty, but reconstruction waits for
          // the next pass. Admission therefore relies on check() revalidating
          // the entry against durable state, as the option requires.
          const now = Date.now();
          let wakeAt: number | null = null;
          let candidate: Entry | null = null;
          for (const item of entries.values()) {
            if (owned.has(item.id)) continue;
            if (keysOf(item).some(key => busyKeys.has(key))) continue;
            const at = Math.max(item.dueAt, failures.get(item.id)?.retryAt ?? 0, (lastChecked.get(item.id) ?? -Infinity) + spacing);
            if (at > now) { wakeAt = wakeAt === null ? at : Math.min(wakeAt, at); continue; }
            if (!candidate || item.dueAt < candidate.dueAt || (item.dueAt === candidate.dueAt && item.id < candidate.id)) candidate = item;
          }
          return { candidate, wakeAt };
        };
        const waitForChange = (wakeAt: number | null) => new Promise<void>(resolve => {
          const timer = wakeAt === null ? undefined : setTimeout(finish, Math.max(0, wakeAt - Date.now()));
          function finish() {
            if (timer !== undefined) clearTimeout(timer);
            signal.removeEventListener("abort", finish);
            if (changed === finish) changed = null;
            resolve();
          }
          const previous = changed;
          changed = previous ? () => { previous(); finish(); } : finish;
          signal.addEventListener("abort", finish, { once: true });
        });
        const checkGroups = async () => {
          while (!signal.aborted && canRun()) {
            const index = nextGroup++;
            const group = pending[index];
            if (group) {
              try {
                for (const entry of group) {
                  if (signal.aborted || !canRun()) return;
                  try { await checkEntry(entry); } finally { owned.delete(entry.id); }
                }
              } finally { release(groupKeys[index]); notify(); }
              continue;
            }
            if (!admitDueWhileActive) return;
            const { candidate, wakeAt } = nextAdmission();
            if (candidate) {
              const keys = keysOf(candidate);
              owned.add(candidate.id); reserve(keys);
              try { await checkEntry(candidate); } finally { owned.delete(candidate.id); release(keys); notify(); }
              continue;
            }
            // Nothing is admissible now. Stay available only while another
            // check still holds this pass open; otherwise the pass ends and the
            // coordinator schedules the next saved deadline as usual.
            if (running === 0) return;
            await waitForChange(wakeAt);
          }
        };
        // With admission enabled, spare slots start too: a pass that begins with
        // one slow group must still admit other resources that come due later.
        const workers = admitDueWhileActive ? options.concurrentChecks!.limit
          : Math.min(options.concurrentChecks?.limit ?? 1, pending.length);
        const checks = await Promise.allSettled(Array.from({ length: workers }, checkGroups));
        for (const result of checks) if (result.status === "rejected") errors.push(result.reason);
        if (errors.length) throw new AggregateError(errors, `${options.queue} could not settle`);
        lastRunFailed = false;
      } catch (error) { lastRunFailed = true; throw error; }
      finally { running = false; revision++; }
    },
    // Retain scopes/deadlines for resume without retaining a settled write's
    // idle hold during drain. The coordinator still fences unresolved XIDs.
    hasPending: async () => canRun() && dirty.size > 0,
    nextRunAt: next,
    nextRunAtOnError: next,
  });
  function invalidate(hint = "*") {
    revision++;
    dirty.add(hint); versions.set(hint, (versions.get(hint) ?? 0) + 1);
    worker.wake();
  }
  return { ...worker,
    reconciliationSnapshot: (): DeadlineReconciliationSnapshot => ({
      revision: `${ownerRevision}:${revision}`,
      reconciled: !stopped && !running && !lastRunFailed && dirty.size === 0 && scans.size === 0
        && failures.size === 0 && scopeFailures.size === 0,
    }),
    wake() { revision++; worker.wake(); },
    invalidate,
    refresh(hint = "*") {
      reconstructAfter.delete(hint);
      invalidate(hint);
    },
  };
}