DentalManagementMH06/apps/Backend/src/queue/inProcessQueue.ts

/**
 * A lightweight in-process async job queue.
 * No Redis, no external dependencies — just Node.js Promises.
 *
 * Features:
 *  - Configurable concurrency limit
 *  - Non-blocking add() — returns a jobId immediately
 *  - Job status tracking in-memory
 *  - onComplete / onFail callbacks for WebSocket notifications
 */
import { randomUUID } from "crypto";

export type JobStatus = "queued" | "active" | "completed" | "failed";

export interface QueueJob<T = any> {
  id: string;
  data: T;
  status: JobStatus;
  result?: any;
  error?: string;
}

type Processor<T> = (job: QueueJob<T>) => Promise<any>;

export class InProcessQueue<T = any> {
  private concurrency: number;
  private running = 0;
  private waitQueue: Array<() => void> = [];
  private jobs = new Map<string, QueueJob<T>>();

  constructor(concurrency = 1) {
    this.concurrency = concurrency;
  }

  /**
   * Enqueue a job. Returns the jobId immediately; processing starts
   * as soon as a concurrency slot is free.
   */
  add(data: T, processor: Processor<T>): string {
    const id = randomUUID();
    const job: QueueJob<T> = { id, data, status: "queued" };
    this.jobs.set(id, job);
    // Fire-and-forget — errors are captured in job.error
    this._run(job, processor).catch(() => {});
    return id;
  }

  getJob(id: string): QueueJob<T> | undefined {
    return this.jobs.get(id);
  }

  /** How many jobs are waiting for a slot. */
  get waiting() {
    return this.waitQueue.length;
  }

  private async _run(job: QueueJob<T>, processor: Processor<T>) {
    // Block until a concurrency slot is available
    if (this.running >= this.concurrency) {
      await new Promise<void>((resolve) => this.waitQueue.push(resolve));
    }

    this.running++;
    job.status = "active";

    try {
      job.result = await processor(job);
      job.status = "completed";
    } catch (err: any) {
      job.status = "failed";
      job.error = err?.message ?? String(err);
    } finally {
      this.running--;
      // Wake up next waiting job
      const next = this.waitQueue.shift();
      if (next) next();
    }
  }
}