@nogaree/priority-queue

A lightweight, TypeScript-first priority queue built on a binary heap — with optional async stores (Redis / SQL), method decorators, and worker-loop utilities.

• Zero dependencies. The built-in Redis/SQL stores take your client as a parameter — nothing is bundled.
• Fully tree-shakeable. Built with tsdown as pure ESM with "sideEffects": false, so bundlers drop everything you don't import. Import just PriorityQueue and that's all you ship. Subpath entries (/redis, /sql) keep things minimal even without a bundler.
• Composable. Every store plugs into the same AsyncPriorityQueue, so decorators, take, consume, and drain work identically in-memory, on Redis, or on SQL.

Installation

npm install @nogaree/priority-queue

Entry points

| Import | Contents | | ------------------------------- | --------------------------- | | @nogaree/priority-queue | Everything (tree-shakeable) | | @nogaree/priority-queue/redis | RedisStore only | | @nogaree/priority-queue/sql | SqlStore only |

Usage

The queue is comparator-driven — the same API works as a min-heap, max-heap, or any custom priority.

Min-heap (smallest value first)

import { PriorityQueue } from '@nogaree/priority-queue';

const pq = new PriorityQueue<number>((a, b) => a - b);

pq.push(3);
pq.push(1);
pq.push(2);

pq.peek(); // 1  (read without removing)
pq.size; // 3

pq.pop(); // 1
pq.pop(); // 2
pq.pop(); // 3

Max-heap (largest value first)

const pq = new PriorityQueue<number>((a, b) => b - a);

pq.push(3);
pq.push(1);
pq.push(2);

pq.pop(); // 3

Custom objects

type Task = { name: string; priority: number };

const pq = new PriorityQueue<Task>((a, b) => a.priority - b.priority);

pq.push({ name: 'low', priority: 10 });
pq.push({ name: 'high', priority: 1 });

pq.pop(); // { name: 'high', priority: 1 }

Method Decorators

Attach a queue to class methods declaratively with UseQueue.

Requirements: TypeScript 5.0+, no experimentalDecorators flag. NestJS 10+ users: remove "experimentalDecorators": true from tsconfig to enable Stage 3 decorator support.

Action.Push — auto-push the return value

The decorated method runs normally, and its return value is automatically pushed to the queue.

import { PriorityQueue, Action, UseQueue } from '@nogaree/priority-queue';

type Task = { name: string; priority: number };

const queue = new PriorityQueue<Task>((a, b) => a.priority - b.priority);

class TaskScheduler {
  @UseQueue({ action: Action.Push, queue })
  schedule(name: string): Task {
    return { name, priority: Math.random() };
    // return value is automatically pushed to queue
  }
}

const scheduler = new TaskScheduler();
scheduler.schedule('A');
scheduler.schedule('B');

queue.size; // 2

Action.Pop — inject the top item as the last argument

Before the method body runs, queue.pop() is called and the result is injected as the last parameter. The caller does not pass that argument. If the queue is empty, the injected value is undefined.

const queue = new PriorityQueue<Task>((a, b) => a.priority - b.priority);
queue.push({ name: 'low', priority: 10 });
queue.push({ name: 'high', priority: 1 });

class TaskWorker {
  @UseQueue({ action: Action.Pop, queue })
  process(popped: Task | undefined): void {
    if (!popped) return;
    console.log(`Processing: ${popped.name}`);
  }
}

const worker = new TaskWorker();
worker.process(); // queue.pop() runs first → Processing: high
worker.process(); // queue.pop() runs first → Processing: low
worker.process(); // queue is empty          → (nothing logged)

If the method has other parameters, they are passed normally and the popped value is appended last:

class TaskWorker {
  @UseQueue({ action: Action.Pop, queue })
  process(label: string, popped: Task | undefined): void {
    console.log(label, popped?.name);
    return;
  }
}

worker.process('next'); // label = 'next', popped = queue.pop()

Dynamic queue

Pass a function instead of a queue instance to resolve the queue at call time:

let activeQueue = new PriorityQueue<Task>((a, b) => a.priority - b.priority);

class Worker {
  @UseQueue({ action: Action.Pop, queue: () => activeQueue })
  process(popped: Task | undefined): void { ... }
}

// Swap the queue at runtime — the decorator picks it up automatically
activeQueue = anotherQueue;

Action.From — replace the entire queue with the method's return value (iterable)

When the decorated method returns an iterable (e.g. an array), the queue is replaced wholesale with that result. Internally uses heapify (O(n)), which is more efficient than pushing elements one by one (O(n log n)).

Execution order:

1. The method runs.
2. The queue's contents are fully replaced by the returned iterable (previous contents are discarded).
3. The return value is passed through to the caller unchanged.

import { PriorityQueue, Action, UseQueue } from '@nogaree/priority-queue';

type Task = { name: string; priority: number };

const queue = new PriorityQueue<Task>((a, b) => a.priority - b.priority);

class TaskLoader {
  @UseQueue({ action: Action.From, queue })
  load(): Task[] {
    // the entire return value is loaded into the queue
    return [
      { name: 'C', priority: 30 },
      { name: 'A', priority: 10 },
      { name: 'B', priority: 20 },
    ];
  }
}

const loader = new TaskLoader();
loader.load();
// queue now holds all 3 items in heapified order

queue.pop(); // { name: 'A', priority: 10 }
queue.pop(); // { name: 'B', priority: 20 }
queue.pop(); // { name: 'C', priority: 30 }

Calling load() again resets the queue from scratch — previous contents are gone:

loader.load(); // queue is re-initialized with 3 items
queue.size; // 3

Difference from Action.Push:

| | Action.Push | Action.From | | ----------------------- | ------------------------- | ------------------------------- | | Return type | T (single value) | Iterable<T> (multiple values) | | Existing queue contents | Preserved (item appended) | Replaced | | Complexity | O(log n) | O(n) heapify |

Push + Pop pipeline

const queue = new PriorityQueue<Task>((a, b) => a.priority - b.priority);

class Pipeline {
  @UseQueue({ action: Action.Push, queue })
  produce(name: string, priority: number): Task {
    return { name, priority }; // return value is pushed to queue
  }

  @UseQueue({ action: Action.Pop, queue })
  consume(popped: Task | undefined): void {
    if (popped) console.log(`Processing: ${popped.name}`);
  }
}

const p = new Pipeline();
p.produce('A', 30);
p.produce('B', 10);
p.produce('C', 20);

p.consume(); // queue.pop() runs first → Processing: B  (priority 10)
p.consume(); // queue.pop() runs first → Processing: C  (priority 20)
p.consume(); // queue.pop() runs first → Processing: A  (priority 30)

Built-in stores

Real workloads usually move the queue into a central store. Two ready-made stores ship with the package — both are thin, dependency-free adapters over a client you provide, and both plug straight into AsyncPriorityQueue so all the utilities above (decorators, take, consume, drain, …) keep working unchanged.

Import them from the root (tree-shaken away when unused) or from their subpath:

import { RedisStore } from '@nogaree/priority-queue/redis';
import { SqlStore } from '@nogaree/priority-queue/sql';

RedisStore — Redis sorted set

Backed by a sorted set (ZADD / ZPOPMIN). An ioredis client satisfies the required command interface as-is; any other client works with a thin wrapper.

import Redis from 'ioredis';
import { AsyncPriorityQueue } from '@nogaree/priority-queue';
import { RedisStore } from '@nogaree/priority-queue/redis';

type Job = { id: string; priority: number };

const queue = new AsyncPriorityQueue(
  new RedisStore<Job>({
    client: new Redis(),
    key: 'jobs',
    score: (job) => job.priority, // lower score pops first
  })
);

await queue.push({ id: 'a', priority: 2 });
await queue.push({ id: 'b', priority: 1 });
await queue.pop(); // { id: 'b', priority: 1 }

Options:

| Option | Description | | ------------- | ----------------------------------------------------------------- | | client | Redis client exposing zadd, zpopmin, zrange, zcard, del | | key | Sorted-set key backing the queue | | score | (value) => number — priority score, lower pops first | | serialize | Member serializer (default JSON.stringify) | | deserialize | Member deserializer (default JSON.parse) |

A sorted set is a set: two values that serialize to the same string collapse into one entry. Include a unique id in the payload if duplicates must survive.

SqlStore — any SQL database

Backed by a plain table (id auto-increment PK, priority float, payload text). You provide an executor function that runs a parameterized statement and resolves with the rows, so it works with pg, mysql2, better-sqlite3, or any driver:

import pg from 'pg';
import { AsyncPriorityQueue } from '@nogaree/priority-queue';
import { SqlStore } from '@nogaree/priority-queue/sql';

const pool = new pg.Pool();

const store = new SqlStore<Job>({
  execute: (sql, params) => pool.query(sql, params).then((r) => r.rows),
  dialect: 'postgres', // 'postgres' | 'mysql' | 'sqlite'
  table: 'jobs', // default 'priority_queue'
  score: (job) => job.priority,
});

await store.setup(); // CREATE TABLE IF NOT EXISTS + index — or run your own migration

const queue = new AsyncPriorityQueue(store);
await queue.push({ id: 'a', priority: 2 });
await queue.pop(); // { id: 'a', priority: 2 }

Executor adapters for common drivers:

// pg
(sql, params) => pool.query(sql, params).then((r) => r.rows);
// mysql2 (promise API)
(sql, params) => pool.query(sql, params).then(([rows]) => rows);
// better-sqlite3
(sql, params) => Promise.resolve(db.prepare(sql).all(...params));

Concurrency notes per dialect:

• postgres — pop is one atomic DELETE … RETURNING with FOR UPDATE SKIP LOCKED; safe for many concurrent consumers.
• sqlite — pop is one DELETE … RETURNING (requires SQLite 3.35+).
• mysql — pop is a select-then-delete pair; wrap the executor in a transaction if multiple consumers pop concurrently.

Custom stores (QueueStore)

If the built-ins don't fit (different data layout, another backend, extra features like pub/sub wake-ups), implement the QueueStore<T> contract yourself and wrap it in an AsyncPriorityQueue. All utilities keep working on top of it.

Every store method may be sync or async (Awaitable<V> = V | Promise<V>):

interface QueueStore<T> {
  push(value: T): Awaitable<void>;
  pop(): Awaitable<T | undefined>;
  peek(): Awaitable<T | undefined>;
  size(): Awaitable<number>;
  clear(): Awaitable<void>;
  reset?(iterable: Iterable<T>): Awaitable<void>; // optional batch replace
  subscribe?(onItem: () => void): () => void; // optional new-item notification
}

Ordering is the store's responsibility — a Redis sorted set orders by score, so no comparator is needed at the queue level.

import { AsyncPriorityQueue, type QueueStore } from '@nogaree/priority-queue';

class MyStore implements QueueStore<Job> {
  // push / pop / peek / size / clear ...
}

const queue = new AsyncPriorityQueue(new MyStore());

HeapStore<T> — the in-memory default

HeapStore wraps the binary heap PriorityQueue in the QueueStore contract. Use it as a drop-in local store (e.g. in tests) and swap it for a Redis/SQL store in production without touching the surrounding code:

import { AsyncPriorityQueue, HeapStore } from '@nogaree/priority-queue';

const queue = new AsyncPriorityQueue(new HeapStore<number>((a, b) => a - b));
await queue.push(3);
await queue.pop(); // 3

AsyncPriorityQueue<T> API

All methods return promises. Unlike the sync PriorityQueue, size and isEmpty are methods, not getters — a central store must be queried each time.

| Method | Description | | --------------------------------- | ------------------------------------------------------------------ | | push(value): Promise<void> | Delegates to store.push | | pop(): Promise<T \| undefined> | Delegates to store.pop | | peek(): Promise<T \| undefined> | Delegates to store.peek | | size(): Promise<number> | Delegates to store.size | | isEmpty(): Promise<boolean> | size() === 0 | | clear(): Promise<void> | Delegates to store.clear | | reset(iterable): Promise<void> | Uses store.reset if implemented, otherwise clear + push loop | | drain(): Promise<T[]> | Pops everything in priority order | | take(options?): Promise<T> | Blocking pop — waits until an item is available | | consume(options?) | Async iterator over take for worker loops | | static from(iterable, store) | Builds a queue on the store and fills it via reset |

Decorators with an async queue

UseQueue accepts an AsyncPriorityQueue anywhere it accepts a PriorityQueue. With a sync queue the decorated method behaves exactly as before; with an async queue the wrapped method returns a Promise:

const queue = new AsyncPriorityQueue(
  new RedisStore<Job>({ client: redis, key: 'jobs', score: (j) => j.priority })
);

class Worker {
  @UseQueue({ action: Action.Push, queue })
  schedule(id: string): Job {
    return { id, priority: Math.random() };
  }

  @UseQueue({ action: Action.Pop, queue })
  process(popped: Job | undefined): void {
    if (popped) console.log(`Processing: ${popped.id}`);
  }
}

const worker = new Worker();
await worker.schedule('a'); // resolves to the Job after the store push completes
await worker.process(); // store.pop() is awaited, then injected as the last argument

Consuming as a worker

pop() returns undefined on an empty queue — fine for one-shot reads, but a worker wants to wait for the next item. AsyncPriorityQueue ships two consumption primitives so you never have to hand-roll a polling loop.

take(options?) — blocking pop

Resolves as soon as an item is available. On an empty queue it waits: via the store's subscribe hook when implemented (instant wake-up), otherwise by polling.

const queue = new AsyncPriorityQueue(new HeapStore<Job>((a, b) => a.priority - b.priority));

const job = await queue.take(); // resolves when something is pushed

Options:

| Option | Description | | -------------- | --------------------------------------------------------------------- | | signal | AbortSignal — cancel the wait; take rejects with signal.reason | | pollInterval | Polling interval in ms for stores without subscribe (default 100) |

const controller = new AbortController();
setTimeout(() => controller.abort(), 5000);

try {
  const job = await queue.take({ signal: controller.signal });
} catch {
  // aborted — no job within 5s
}

consume(options?) — worker loop with for await

An async iterator built on take. Aborting the signal ends the loop cleanly (no throw), so shutdown is just controller.abort():

const controller = new AbortController();
process.on('SIGTERM', () => controller.abort());

for await (const job of queue.consume({ signal: controller.signal })) {
  await handle(job); // items arrive in priority order, as they become available
}
// loop exits here after abort

subscribe — instant wake-up for your store

take/consume fall back to polling, which works with any store (including the built-in RedisStore/SqlStore). If your backend can signal "a new item arrived", implement the optional subscribe hook in a custom store and waiting consumers wake immediately instead:

class MyRedisStore implements QueueStore<Job> {
  // ...push/pop/peek/size/clear as before, plus:

  async push(job: Job) {
    await this.redis.zadd(this.key, job.priority, JSON.stringify(job));
    await this.redis.publish(`${this.key}:new`, '1');
  }

  subscribe(onItem: () => void): () => void {
    const sub = this.redis.duplicate();
    sub.subscribe(`${this.key}:new`, onItem);
    return () => sub.unsubscribe(`${this.key}:new`);
  }
}

The built-in HeapStore implements subscribe, so local queues never poll.

wait: true — blocking pop in decorators

Action.Pop normally injects undefined when the queue is empty. With wait: true the decorated method waits for the next item instead (requires an AsyncPriorityQueue):

class Worker {
  @UseQueue({ action: Action.Pop, queue, wait: true })
  async process(popped: Job): Promise<void> {
    // called only once an item is available — popped is never undefined
  }
}

API

new PriorityQueue<T>(compare: (a: T, b: T) => number)

Creates a new priority queue. The comparator follows the same convention as Array.prototype.sort: if compare(a, b) < 0, a comes out first.

push(value: T): void

Adds a value to the queue. O(log n).

pop(): T | undefined

Removes and returns the highest-priority value. Returns undefined if the queue is empty. O(log n).

peek(): T | undefined

Returns the highest-priority value without removing it. Returns undefined if the queue is empty. O(1).

size: number

The number of elements currently in the queue.

isEmpty: boolean

true if the queue has no elements.

pq.isEmpty; // true
pq.push(1);
pq.isEmpty; // false

clear(): void

Removes all elements from the queue. O(1).

pq.push(1);
pq.push(2);
pq.clear();
pq.size; // 0

drain(): T[]

Removes and returns all elements in priority order. The queue is empty after this call. O(n log n).

const pq = new PriorityQueue<number>((a, b) => a - b);
pq.push(3);
pq.push(1);
pq.push(2);

pq.drain(); // [1, 2, 3]
pq.isEmpty; // true

reset(iterable: Iterable<T>): void

Replaces the queue's contents with the given iterable. Keeps the existing comparator and uses heapify internally. O(n).

const pq = new PriorityQueue<number>((a, b) => a - b);
pq.push(99);

pq.reset([3, 1, 2]);
pq.drain(); // [1, 2, 3]  — 99 is gone

static PriorityQueue.from<T>(iterable: Iterable<T>, compare: (a: T, b: T) => number): PriorityQueue<T>

Creates a queue from an existing iterable. Uses the heapify algorithm internally, so it runs in O(n) — faster than pushing elements one by one (O(n log n)).

const pq = PriorityQueue.from([3, 1, 4, 1, 5, 9], (a, b) => a - b);

pq.peek(); // 1
pq.drain(); // [1, 1, 3, 4, 5, 9]

License

MIT