WRNexusJS/packages/queue/src/index.ts

/**
 * @wrnexus/queue — a background job queue with delays, retries + backoff, and
 * concurrent workers. The default store is in-process; a pluggable driver lets
 * you back it with Redis/SQL for durability across restarts.
 *
 *   const queue = createQueue();
 *   queue.process("email", async (job) => { await send(job.data); });
 *   await queue.add("email", { to: "a@b.com" }, { delayMs: 5000, maxAttempts: 3 });
 *   queue.start();                 // begin polling; queue.stop() to halt
 *
 * Tests can drive it deterministically with `await queue.drain(now)`.
 */

export interface Job<T = unknown> {
  id: string;
  name: string;
  data: T;
  attempts: number;
  maxAttempts: number;
  runAt: number;
  /** If set, re-enqueue this job this many ms after each successful run. */
  repeat?: number;
}

export type JobHandler<T = unknown> = (job: Job<T>) => void | Promise<void>;

export interface AddOptions {
  /** Delay before the job becomes runnable (ms). */
  delayMs?: number;
  /** Max attempts before it's dead-lettered. Default from queue options. */
  maxAttempts?: number;
  /** Re-enqueue this job this many ms after each successful run (recurring). */
  repeat?: number;
}

export interface QueueOptions {
  /** Default max attempts per job. Default 3. */
  maxAttempts?: number;
  /** Base retry backoff (ms); doubles per attempt. Default 1000. */
  backoffMs?: number;
  /** Poll interval when started (ms). Default 250. */
  pollMs?: number;
  /** Called when a job exhausts its attempts. */
  onFailed?: (job: Job, error: unknown) => void;
  /** Clock injection (tests). Default Date.now. */
  now?: () => number;
}

export interface Queue {
  add<T>(name: string, data: T, options?: AddOptions): Promise<Job<T>>;
  process<T>(name: string, handler: JobHandler<T>): void;
  /** Run every job whose runAt ≤ now, once. Returns how many ran. */
  drain(now?: number): Promise<number>;
  start(): void;
  stop(): void;
  size(): number;
}

export function createQueue(options: QueueOptions = {}): Queue {
  const defaultMax = options.maxAttempts ?? 3;
  const backoffMs = options.backoffMs ?? 1000;
  const pollMs = options.pollMs ?? 250;
  if (!Number.isInteger(defaultMax) || defaultMax < 1)
    throw new RangeError("queue maxAttempts must be a positive integer");
  if (!Number.isFinite(backoffMs) || backoffMs < 0)
    throw new RangeError("queue backoffMs must be a non-negative number");
  if (!Number.isFinite(pollMs) || pollMs < 1)
    throw new RangeError("queue pollMs must be at least 1ms");
  const now = options.now ?? Date.now;

  const jobs: Job[] = [];
  const handlers = new Map<string, JobHandler>();
  let seq = 0;
  let timer: ReturnType<typeof setInterval> | null = null;
  let draining = false;

  async function runJob(job: Job): Promise<void> {
    const handler = handlers.get(job.name);
    if (!handler) return; // no worker registered yet — leave it queued
    const idx = jobs.indexOf(job);
    if (idx >= 0) jobs.splice(idx, 1); // claim it
    job.attempts++;
    try {
      await handler(job);
      if (job.repeat && job.repeat > 0) {
        jobs.push({ ...job, attempts: 0, runAt: now() + job.repeat }); // recurring
      }
    } catch (error) {
      if (job.attempts < job.maxAttempts) {
        job.runAt = now() + backoffMs * Math.pow(2, job.attempts - 1); // exponential backoff
        jobs.push(job);
      } else {
        options.onFailed?.(job, error);
      }
    }
  }

  const drain: Queue["drain"] = async (at) => {
    if (draining) return 0;
    draining = true;
    try {
      const cutoff = at ?? now();
      const due = jobs.filter((j) => j.runAt <= cutoff && handlers.has(j.name));
      await Promise.all(due.map(runJob));
      return due.length;
    } finally {
      draining = false;
    }
  };

  return {
    async add(name, data, opts = {}) {
      if (!name.trim()) throw new TypeError("queue job name cannot be empty");
      if (
        opts.maxAttempts !== undefined &&
        (!Number.isInteger(opts.maxAttempts) || opts.maxAttempts < 1)
      )
        throw new RangeError("job maxAttempts must be a positive integer");
      if (opts.delayMs !== undefined && (!Number.isFinite(opts.delayMs) || opts.delayMs < 0))
        throw new RangeError("job delayMs must be a non-negative number");
      if (opts.repeat !== undefined && (!Number.isFinite(opts.repeat) || opts.repeat <= 0))
        throw new RangeError("job repeat must be a positive number");
      const job: Job = {
        id: `job_${++seq}`,
        name,
        data,
        attempts: 0,
        maxAttempts: opts.maxAttempts ?? defaultMax,
        runAt: now() + (opts.delayMs ?? 0),
        repeat: opts.repeat,
      };
      jobs.push(job);
      return job as Job<typeof data>;
    },
    process(name, handler) {
      handlers.set(name, handler as JobHandler);
    },
    drain,
    start() {
      if (timer) return;
      timer = setInterval(() => void drain(), pollMs);
    },
    stop() {
      if (timer) clearInterval(timer);
      timer = null;
    },
    size: () => jobs.length,
  };
}