Priority Task Queue With Starvation Aging

export interface PriorityQueueOptions {
  concurrency?: number;
  agingMs?: number;
  now?: () => number;
}

interface InternalTask {
  id: string;
  priority: number;
  enqueuedAt: number;
  run: () => Promise<unknown>;
  resolve: (value: unknown) => void;
  reject: (error: unknown) => void;
}

export class PriorityTaskQueue {
  private readonly tasks: InternalTask[] = [];
  private readonly knownIds = new Set<string>();
  private readonly idleWaiters: Array<() => void> = [];
  private readonly concurrency: number;
  private readonly agingMs: number;
  private readonly now: () => number;
  private active = 0;

  constructor(options: PriorityQueueOptions = {}) {
    const concurrency = options.concurrency ?? 4;
    if (!Number.isInteger(concurrency) || concurrency < 1) {
      throw new RangeError("concurrency must be a positive integer");
    }
    const agingMs = options.agingMs ?? 5_000;
    if (!Number.isFinite(agingMs) || agingMs <= 0) {
      throw new RangeError("agingMs must be a positive finite number");
    }
    this.concurrency = concurrency;
    this.agingMs = agingMs;
    this.now = options.now ?? Date.now;
  }

  get size(): number {
    return this.tasks.length;
  }

  get activeCount(): number {
    return this.active;
  }

  add<T>(id: string, priority: number, run: () => Promise<T>): Promise<T> {
    if (typeof id !== "string" || id.length === 0) {
      throw new TypeError("id must be a non-empty string");
    }
    if (!Number.isFinite(priority)) throw new TypeError("priority must be finite");
    if (typeof run !== "function") throw new TypeError("run must be a function");
    if (this.knownIds.has(id)) throw new Error("duplicate task id: " + id);
    this.knownIds.add(id);

    return new Promise<T>((resolve, reject) => {
      this.tasks.push({
        id,
        priority,
        enqueuedAt: this.now(),
        run,
        resolve: (value) => resolve(value as T),
        reject,
      });
      this.pump();
    });
  }

  onIdle(): Promise<void> {
    if (this.active === 0 && this.tasks.length === 0) return Promise.resolve();
    return new Promise<void>((resolve) => {
      this.idleWaiters.push(resolve);
    });
  }

  private pick(): InternalTask | undefined {
    if (this.tasks.length === 0) return undefined;
    const now = this.now();
    let bestIndex = 0;
    let bestScore = Number.NEGATIVE_INFINITY;
    for (let index = 0; index < this.tasks.length; index += 1) {
      const task = this.tasks[index];
      const waited = Math.max(0, now - task.enqueuedAt);
      const aging = Math.floor(waited / this.agingMs);
      const score = task.priority + aging;
      if (score > bestScore) {
        bestScore = score;
        bestIndex = index;
      }
    }
    return this.tasks.splice(bestIndex, 1)[0];
  }

  private pump(): void {
    while (this.active < this.concurrency) {
      const task = this.pick();
      if (!task) {
        this.maybeIdle();
        return;
      }
      this.active += 1;
      Promise.resolve()
        .then(task.run)
        .then(task.resolve, task.reject)
        .finally(() => {
          this.active -= 1;
          this.knownIds.delete(task.id);
          this.maybeIdle();
          this.pump();
        });
    }
  }

  private maybeIdle(): void {
    if (this.active !== 0 || this.tasks.length !== 0) return;
    const waiters = this.idleWaiters.splice(0);
    for (const resolve of waiters) resolve();
  }
}