Delayed Task Scheduler With Priorities

interface Job {
  id: number;
  run: () => void | Promise<void>;
  priority: number;
  sequence: number;
  timer: ReturnType<typeof setTimeout> | null;
  settled: boolean;
}

export interface ScheduledHandle {
  id: number;
  cancel(): boolean;
}

export interface TaskSchedulerOptions {
  concurrency?: number;
  onError?: (error: unknown, id: number) => void;
}

export class TaskScheduler {
  private readonly jobs = new Map<number, Job>();
  private readonly ready: Job[] = [];
  private readonly idleWaiters: Array<() => void> = [];
  private readonly concurrency: number;
  private readonly onError: (error: unknown, id: number) => void;
  private nextId = 1;
  private sequence = 0;
  private running = 0;
  private closed = false;

  public constructor(options: TaskSchedulerOptions = {}) {
    const {
      concurrency = 1,
      onError = (error: unknown, id: number) => {
        console.error(`task ${id} failed`, error);
      },
    } = options;
    if (!Number.isInteger(concurrency) || concurrency < 1) {
      throw new RangeError("concurrency must be a positive integer");
    }
    if (typeof onError !== "function") {
      throw new TypeError("onError must be a function");
    }
    this.concurrency = concurrency;
    this.onError = onError;
  }

  public schedule(
    run: () => void | Promise<void>,
    options: { delayMs?: number; priority?: number } = {},
  ): ScheduledHandle {
    if (this.closed) throw new Error("scheduler is closed");
    if (typeof run !== "function") {
      throw new TypeError("run must be a function");
    }
    const { delayMs = 0, priority = 0 } = options;
    if (!Number.isFinite(delayMs) || delayMs < 0) {
      throw new RangeError("delayMs must be a non-negative finite number");
    }
    if (!Number.isFinite(priority)) {
      throw new RangeError("priority must be a finite number");
    }

    const id = this.nextId;
    this.nextId += 1;
    const job: Job = {
      id,
      run,
      priority,
      sequence: this.sequence,
      timer: null,
      settled: false,
    };
    this.sequence += 1;
    this.jobs.set(id, job);
    job.timer = setTimeout(() => {
      job.timer = null;
      if (job.settled) return;
      this.ready.push(job);
      this.flush();
    }, delayMs);

    return { id, cancel: () => this.cancel(id) };
  }

  public cancel(id: number): boolean {
    const job = this.jobs.get(id);
    if (job === undefined || job.settled) return false;
    const isWaiting = job.timer !== null || this.ready.includes(job);
    if (!isWaiting) return false;

    if (job.timer !== null) {
      clearTimeout(job.timer);
      job.timer = null;
    }
    const readyIndex = this.ready.indexOf(job);
    if (readyIndex >= 0) this.ready.splice(readyIndex, 1);
    job.settled = true;
    this.jobs.delete(id);
    this.resolveIdleIfNeeded();
    return true;
  }

  public drain(): Promise<void> {
    if (this.jobs.size === 0) return Promise.resolve();
    return new Promise<void>((resolve) => {
      this.idleWaiters.push(resolve);
    });
  }

  public close(): void {
    if (this.closed) return;
    this.closed = true;
    for (const job of [...this.jobs.values()]) {
      if (job.timer !== null) {
        clearTimeout(job.timer);
        job.timer = null;
      }
      const readyIndex = this.ready.indexOf(job);
      if (readyIndex >= 0) this.ready.splice(readyIndex, 1);
      job.settled = true;
    }
    this.jobs.clear();
    this.resolveIdleIfNeeded();
  }

  private flush(): void {
    this.ready.sort(
      (left, right) =>
        right.priority - left.priority || left.sequence - right.sequence,
    );
    while (this.running < this.concurrency && this.ready.length > 0) {
      const job = this.ready.shift();
      if (job === undefined || job.settled) continue;
      this.running += 1;
      Promise.resolve()
        .then(() => job.run())
        .catch((error: unknown) => {
          try {
            this.onError(error, job.id);
          } catch {
            // Error handlers must not break the scheduler loop.
          }
        })
        .finally(() => {
          this.running -= 1;
          job.settled = true;
          this.jobs.delete(job.id);
          this.resolveIdleIfNeeded();
          this.flush();
        });
    }
    this.resolveIdleIfNeeded();
  }

  private resolveIdleIfNeeded(): void {
    if (this.jobs.size > 0) return;
    const waiters = this.idleWaiters.splice(0);
    for (const resolve of waiters) resolve();
  }
}

async function main(): Promise<void> {
  const events: string[] = [];
  const scheduler = new TaskScheduler({
    concurrency: 1,
    onError: (error, id) => {
      events.push(
        `error ${id}: ${error instanceof Error ? error.message : String(error)}`,
      );
    },
  });

  scheduler.schedule(async () => {
    events.push("start");
    await new Promise((resolve) => setTimeout(resolve, 80));
    events.push("end");
  });
  scheduler.schedule(() => {
    events.push("low");
  }, { delayMs: 5, priority: 0 });
  scheduler.schedule(() => {
    events.push("high");
  }, { delayMs: 10, priority: 5 });
  scheduler.schedule(() => {
    events.push("never");
  }, { delayMs: 10, priority: 9 }).cancel();
  scheduler.schedule(() => {
    throw new Error("boom");
  }, { delayMs: 15, priority: 1 });

  await scheduler.drain();
  console.log(events);
  scheduler.close();
}

main().catch((error) => {
  console.error(error);
  process.exitCode = 1;
});