BullMQStore — BullMQ Redis-backed queue

BullMQStore — BullMQ Redis-backed queue Overview BullMQ is a Redis-backed job/task queue, not a database. A Queue object adds jobs to a named queue, and a Worker object pulls jobs from that queue and runs a handler. There is no query language and no SQL-shaped access pattern. Because none of that fits the SQL-shaped Dialect interface, BullMQStore implements the minimal NoSqlStore marker interface (src/nosql/store.ts) — connection lifecycle plus a getClient() escape hatch — and exposes BullMQ's real add/process/getJob operations (enqueue/process/getJob/remove/stats) directly rather than forcing them into a query(sql) shape. Identity: Property | Value | --------- | ---------- | name | 'bullmq' | library | 'bullmq' | Queue-scoped caching Each BullMQ queue is a distinct object bound to a name. This store keeps a Map of one Queue per queue name (created and cached on first use) plus a Map of one Worker per queue name (created on process()). A second enqueue() on the same name reuses the cached Queue; a second process() on the same name returns the existing Worker. Both maps are closed and cleared on disconnect(). Lazy loading — not a hard dependency bullmq is not a hard dependency of this package. It is an optional peer, loaded lazily via require('bullmq') inside connect() rather than a top-level import. Importing this module therefore does not require the driver to be installed — it is only needed when a BullMQ store is actually connected. BullMQ has no single connection to open; connect() just resolves the driver module, and per-queue Redis connections open lazily when the first Queue/Worker for a name is constructed. Injected client BullMQStoreOptions accepts a pre-built client — the bullmq module itself, or any stand-in exposing Queue and Worker constructors. When provided, connect() adopts it directly and does not require('bullmq'). This is how the test suite injects mocks (no driver, no network). Connection Build a store from connection options and call connect(): The connection options are passed to every Queue/Worker the store constructs. Failures resolving the driver are wrapped in a ConnectionError (a message containing ECONNREFUSED is normalized to 'Connection refused'). Option | Type | Purpose | ------------ | ----- | --------------------------------------------------------------------------------------------------------- | connection | any | ioredis-style connection options ({ host, port } or { url }) passed to every Queue/Worker. Defaults to { host: '127.0.0.1', port: 6379 }. | client | any | A pre-built bullmq module (or stand-in with Queue/Worker). When set, the driver is not required. | Injected-client form Methods Driver failures are wrapped in a DatabaseError (message BullMQ <action> failed: ..., preserving the original error). Using the store before connect() (or after disconnect()) throws a ConnectionError. Lifecycle Method | Signature | Behavior | ------------- | ----------------------------- | ------------------------------------------------------------------------------------------------------------- | connect | connect(): Promise<void> | Adopts an injected client, otherwise lazy-requires bullmq. Idempotent when already connected. Wraps failures in ConnectionError. | disconnect | disconnect(): Promise<void> | close()s every cached Worker then every cached Queue, clears both maps, and drops the driver. | isConnected | isConnected(): boolean | true only when connected and a driver is present. | getClient | getClient(): any | Returns the underlying bullmq driver module (with Queue/Worker). Throws ConnectionError if not connected. | Queue operations Method | Signature | Behavior | --------- | ----------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------- | enqueue | enqueue(queue: string, jobData: unknown, opts?: BullMQEnqueueOptions): Promise<string> | Gets (or creates + caches) the Queue for queue and calls queue.add(name ?? queue, jobData, jobOpts), where name is pulled out of opts and the rest passed through as BullMQ JobsOptions. Returns the assigned job id. | process | process(queue: string, handler: (job: any) => any \| Promise<any>, opts?: any): Promise<any> | Returns the cached Worker for queue if present, otherwise constructs new Worker(queue, handler, { connection, ...opts }), caches it, and returns it. | getJob | getJob(queue: string, id: string): Promise<any> | Delegates to queue.getJob(id); null if the job no longer exists. | remove | remove(queue: string, id: string): Promise<void> | Fetches the job via queue.getJob(id) and calls job.remove() when present. | stats | stats(queue: string): Promise<any> | Returns job counts by state via queue.getJobCounts() (waiting/active/completed/failed/...). | Supporting type: Example Verification status Unit / mock-verified only. The tests in tests/nosql/bullmq.test.ts are fully mock-driven: a fake driver module whose Queue and Worker constructors are Jest spies (with spied add/getJob/getJobCounts/close on instances) is injected via BullMQStoreOptions.client, so connect() adopts it and skips require('bullmq'). The real bullmq package is not installed, and there is no live Redis and no network in the test run. What this proves: name/library identity and injected-driver adoption without loading the driver. Lifecycle: ConnectionError when using the store before connecting. enqueue constructing a Queue with the connection opts and calling add(name, data, jobOpts), returning the id; caching one Queue per name across enqueues. process constructing a Worker with the handler and caching one per name (repeat returns the same worker). getJob delegating to queue.getJob; remove fetching then calling job.remove(); stats returning getJobCounts(). Error handling: add failures wrapped in DatabaseError. What this does not prove: live execution against real Redis via BullMQ. Constructor/method shapes are verified against the driver's documented contract, but end-to-end execution over the wire has not been exercised here. Related reading All data stores — the full catalogue, grouped by purpose Database types — where this sits among the 22 categories