@molecule/api-cron-bullmq

Auto-generated, AI-first package reference for the molecule.dev ecosystem. It is written to be read by coding agents as much as by people, and is generated from this package's source — edit src/index.ts JSDoc, not this file.

BullMQ cron scheduling provider for molecule.dev.

Implements the CronProvider interface using BullMQ repeatable jobs backed by Redis. The schedule (the repeatable job definition in Redis) is persistent and distributed — it survives process restarts and produces ticks that any worker process sharing the queue can pick up. The handler function itself, and this bond's local list()/pause state, are NOT persisted — see the remarks below.

Quick Start

import { setProvider, schedule } from '@molecule/api-cron'
import { createProvider } from '@molecule/api-cron-bullmq'

const provider = createProvider({
  connection: { host: 'localhost', port: 6379 },
  onError: (error) => console.error('cron Redis connection error', error),
})
setProvider(provider)

await schedule('cleanup', '0 3 * * *', async () => {
  console.log('Nightly cleanup')
})

Type

provider

Installation

npm install @molecule/api-cron-bullmq @molecule/api-bond @molecule/api-cron bullmq

API

Interfaces

BullMQCronConfig

Configuration options for the BullMQ cron provider.

interface BullMQCronConfig {
  /** Redis connection options. */
  connection: RedisConnectionOptions

  /** Queue name prefix. Defaults to `'molecule-cron'`. */
  queueName?: string

  /** Default timezone for all jobs. */
  timezone?: string

  /**
   * Called whenever the underlying BullMQ queue or worker emits a
   * connection-level `'error'` event (e.g. Redis unreachable). These errors
   * are always logged via the bonded logger regardless of this callback —
   * use it for additional handling (alerting, metrics, a fail-fast exit).
   */
  onError?: (error: Error) => void
}

RedisConnectionOptions

Redis connection options for BullMQ.

interface RedisConnectionOptions {
  /** Redis host. Defaults to `'localhost'`. */
  host?: string

  /** Redis port. Defaults to `6379`. */
  port?: number

  /** Redis password. */
  password?: string

  /** Redis database number. */
  db?: number

  /**
   * Full Redis connection URL. Takes precedence over `host`/`port`/`password`/
   * `db` (ioredis parses the URL first, so those fields only fill gaps the URL
   * leaves). Use the `rediss://` scheme for a TLS endpoint — it enables TLS
   * automatically, same as passing `tls`.
   */
  url?: string

  /**
   * TLS options for the Redis connection, forwarded to the ioredis client
   * BullMQ builds. Pass `true` (shorthand for `{}`, ioredis' "TLS with
   * defaults") to enable TLS, or a `tls.ConnectionOptions` object
   * (`ca`/`cert`/`key`/`rejectUnauthorized`/`servername`/…) for a managed
   * Redis that requires it. Omit (or `false`) for a plaintext connection. A
   * `rediss://` `url` also enables TLS on its own.
   */
  tls?: boolean | TLSConnectionOptions
}

Functions

createProvider(config)

Creates a BullMQ cron provider.

function createProvider(config: BullMQCronConfig): CronProvider

• config — Provider configuration including Redis connection.

Returns: A CronProvider backed by BullMQ repeatable jobs.

Core Interface

Implements @molecule/api-cron interface.

Injection Notes

Requirements

Peer dependencies:

• @molecule/api-bond ^1.0.1
• @molecule/api-cron ^1.0.1

Runtime Dependencies

• @molecule/api-bond

• @molecule/api-cron

• bullmq

• schedule() must be called for every job on every process boot — including after a restart. The repeatable job scheduler lives in Redis and keeps ticking across restarts, but the JavaScript handler function passed to schedule() only lives in this process's memory. A tick for a job this process hasn't (re-)registered logs a warning ("...has no registered handler in this process...") and no-ops rather than silently dropping the tick.

• list() and getStatus-style reads only reflect jobs registered on THIS process — they do NOT query Redis for jobs registered by other worker processes. cancel() still works for a job unknown to this process's memory (it falls through to queue.removeJobScheduler).

• pause()/resume() are cluster-wide: a paused flag is written to Redis and checked by every worker sharing the queue on every tick, so pausing on one process stops execution on all of them (not just the caller). Note this only affects the handler running — BullMQ still records a normal 'completed' entry for the skipped tick; that's expected, not a hidden error.

• Worker/job errors are never swallowed: a thrown handler is logged (with the job name) AND rethrown so BullMQ marks that occurrence 'failed' — the repeatable schedule itself keeps ticking (a transient failure does not kill the job, same semantics as the node-cron bond and real crontab). Queue/worker-level connection errors (e.g. Redis unreachable) are logged with an actionable message instead of the silent hang you'd otherwise get while ioredis retries the connection indefinitely; pass onError for additional handling.

• CronOptions.noOverlap: true is emulated per-worker-process (an in-memory running flag) — it prevents a slow handler from overlapping itself on the SAME worker, but does not coordinate across multiple distributed worker processes running the same job concurrently.

E2E Tests

Integration checklist — drive the real flow (no mocks), adapt each item to this app's actual scheduled jobs, and check every box off one by one. A box you can't check is an integration bug to fix — not a skip:

• [ ] Every job the app defines is REGISTERED at startup: after bootstrap, list() returns each one (or its schedule() call ran without error) — a job that never registers never fires.
• [ ] Each job's REAL side effect happens when it runs — the digest row is written, the cleanup deletes, the report is generated. Trigger it now with runNow(jobId) (or invoke the handler directly) and assert the effect; never stub the body. COUNTERPARTY: the sandbox process is short-lived, so a real timed tick may never arrive — that is expected. Verify by direct invocation, not by waiting minutes for the schedule to fire.
• [ ] Re-running a job is safe: invoke it twice and confirm no double effect (no double-charge, double-send, or duplicate row) — the handler is idempotent or guards its own re-entry.
• [ ] A failing job is observable, not swallowed: force the handler to throw and confirm the error is logged/surfaced and the job's status reflects it.
• [ ] The cadence is correct: read each job's cron expression and confirm it matches the intended schedule (nightly, hourly, …) — verify by reading it, not by waiting for a tick.
• [ ] Any user-facing trigger is locked down: if the app exposes a manual "run now" or schedule-management endpoint, only an authorized caller can hit it — an anonymous request can't fire jobs or register arbitrary schedules.