@lastshotlabs/slingshot-bullmq

npm install @lastshotlabs/slingshot-bullmq

Functions

bullmqAdapterOptionsSchema

Zod schema for BullMQAdapterOptions. Validated at adapter-creation time.

Remarks: The connection.port field must be a number, not a string. Passing a string (e.g. from an environment variable) will fail validation with a clear message.

Config Fields

Field	Description
`/**

• BullMQ/ioredis connection options. Must be a plain object (not a URL string).
• host must be a string and port must be a number when provided. / connection| BullMQ/ioredis connection options; must be a plain object (not a URL string) | |/* Event payload validation mode. Default| Event payload validation mode (default: "off") | |/** Queue name prefix. Default` | Queue name prefix (default: “slingshot:events”) |

Source: packages/slingshot-bullmq/src/bullmqAdapter.ts

createBullMQAdapter

Creates a SlingshotEventBus implementation backed by BullMQ and Redis.

Non-durable subscriptions (bus.on(event, handler)) behave identically to the in-process adapter — listeners are called in the same process, fire-and- forget. Durable subscriptions (bus.on(event, handler, { durable: true, name: 'worker-name' })) create a BullMQ Queue + Worker pair. Jobs survive process restarts and are retried up to opts.attempts times.

Durability caveat — if events fail to enqueue (Redis down) they are held in an in-memory pending buffer (up to 1 000 entries) and retried every 2 s. The buffer is process-local. Set walPath to opt into a JSON-lines write-ahead log; entries written there are replayed on adapter creation so events survive a process crash during a Redis outage.

function createBullMQAdapter(rawOpts: BullMQAdapterOptions & EventBusSerializationOptions & { onDrop?: (event: string, reason: BullMQAdapterDropReason) => void; /** * Optional structured logger (see `Logger` from * `@lastshotlabs/slingshot-core`). Defaults to a console-backed JSON * logger when omitted. All warn/error paths in the adapter route * through this logger with structured context fields. */ logger?: Logger; /** * Optional metrics sink. When provided, the adapter records publish / * consume / dlq counters, publish/consume durations, and pending-buffer * + worker-paused gauges so operators can wire ad-hoc dashboards * without log scraping. Defaults to a no-op emitter. */ metrics?: MetricsEmitter; },): AcknowledgedEventBus &

Source: packages/slingshot-bullmq/src/bullmqAdapter.ts

Classes

BullMQAdapterError

Errors thrown by the BullMQ event bus adapter.

Source: packages/slingshot-bullmq/src/errors.ts

DuplicateDurableSubscriptionError

Raised when a BullMQ durable subscription name is reused for the same event.

Source: packages/slingshot-bullmq/src/errors.ts

DurableSubscriptionNameRequiredError

Raised when a durable BullMQ subscription is registered without a required name.

Source: packages/slingshot-bullmq/src/errors.ts

DurableSubscriptionOffError

Raised when code tries to unregister a durable BullMQ subscription with off().

Source: packages/slingshot-bullmq/src/errors.ts

Interfaces

BullMQAdapterDropEvent

Detailed drop event surfaced via the onDropEvent callback.

Source: packages/slingshot-bullmq/src/bullmqAdapter.ts

BullMQAdapterHealth

Health snapshot for the BullMQ event bus adapter.

status is a coarse roll-up derived from the underlying signals:

• 'unhealthy' when buffered events have been dropped (bufferDroppedCount > 0) or the pending buffer has grown past 100 entries.
• 'degraded' when there is any pending-buffer pressure (pendingBufferSize > 0) or any worker has paused (workerPausedCount > 0) or any validation drops have been observed.
• 'healthy' otherwise.

Treat status as advisory — the raw fields are the source of truth.

Source: packages/slingshot-bullmq/src/bullmqAdapter.ts

Types

BullMQAdapterDropReason

Reason a durable event was dropped by createBullMQAdapter.

• 'buffer-full' — the in-memory pending buffer reached MAX_PENDING_BUFFER entries; the event was discarded immediately rather than buffered.
• 'max-attempts' — the buffered event exceeded MAX_ENQUEUE_ATTEMPTS consecutive Redis failures and was permanently discarded.
• 'enqueue-timeout' — the initial queue.add() exceeded enqueueTimeoutMs and was buffered for retry; raised as a drop signal so operators see the timeout, even though the event itself is not yet lost.
• 'permanent-error' — queue.add() rejected with a non-retryable error (e.g. EINVAL, WRONGTYPE). The event is not buffered.

Source: packages/slingshot-bullmq/src/bullmqAdapter.ts

BullMQAdapterOptions

Configuration options for createBullMQAdapter.

Remarks: Inferred from bullmqAdapterOptionsSchema.

Source: packages/slingshot-bullmq/src/bullmqAdapter.ts