@dreamtree-org/jobs

Replaces the BullMQ background-jobs layer that was copied into each Pro product (lib/jobs/{connection,control,registry,runner,runs,schedules}.js, commands/worker.js). You define handlers, enqueue jobs and run a worker. Every run is recorded in your own database.

• One run row per execution. The row is written before the job reaches Redis, and its id is the BullMQ job id. Redis keeps finished jobs only briefly; the store keeps the full history.
• Schedules survive a Redis flush. Cron schedules (from code or from a console-managed table) are reconciled into BullMQ on every worker heartbeat.
• Dead workers do not leave runs running forever. Their runs are closed as failed and never re-run, because a half-finished job may already have done part of its work.
• A wedged worker exits on its own. When one attempt has been in flight longer than WORKER_STALL_MS, the worker exits non-zero so your supervisor restarts it.
• Tenant support is an option, not a dependency. Set tenantKey: 'companyId' and the package does the rest. It has no dependency on @dreamtree-org/tenancy.
• Works in every stack. Ships ESM and CJS builds, so the same instance works from Express (CommonJS) and from Next.js routes.

60-second start

npm install @dreamtree-org/jobs bullmq ioredis cron-parser knex

const { createJobs, knexStore } = require('@dreamtree-org/jobs'); // or import

const jobs = createJobs({
  redisUrl: process.env.REDIS_URL,
  queue: 'default',
  store: knexStore(knex, { tenantColumn: 'company_id' }),
  tenantKey: 'companyId',
  onError: (err, info) => logger.error({ err, ...info }, 'job error'),
});

jobs.define('send-digest', {
  schedule: '0 9 * * *',          // optional: a cron schedule defined in code
  retries: 3,
  run: async (payload, ctx) => {
    await sendDigest(ctx.tenant, payload);
    return { sent: 1 };           // a small summary, stored as the run result
  },
});

await jobs.enqueue('send-digest', { companyId: 'co-1', userId: 7 });  // web process
await jobs.startWorker();                                             // worker process

jobs.runs.list({ status: 'failed' }), jobs.runs.retry(id), jobs.runs.cancel(id), jobs.pause('default') and jobs.resume() cover what an operator console needs.

New product? await createJobTables(knex, { tenantColumn: 'company_id' }) creates job_runs, job_schedules and job_workers if they do not exist.

createJobs(config)

| Key | Type | Default | Meaning | |---|---|---|---| | store | JobStore | required | Where runs are recorded: knexStore(knex), sqlStore(platformDb), memoryStore() or your own. | | redisUrl | string | process.env.REDIS_URL | Without it, jobs.enabled is false and calls that need Redis throw a 503 that tells you to set it. | | queue | string | 'default' | The queue for jobs that do not name one. | | prefix | string | 'bull' | BullMQ key prefix and control channel (<prefix>:control). The Pro products use `${REDIS_PREFIX}bull`. | | tenantKey | string | none | The payload key that holds the tenant id. Its value becomes run.tenant and ctx.tenant, plus ctx[tenantKey]. | | timezone | string | 'Asia/Kolkata' | Default timezone for schedules. | | defaults | object | { retries: 0, backoffMs: 60000, timeoutMs: 900000 } | Defaults that define() applies to every job. | | maxPayloadBytes | number | 4096 | Larger payloads are refused with a 413. Put bulk data in its own table. | | removeOnComplete / removeOnFail | BullMQ | 7 days or 1000 jobs / 30 days | How long Redis keeps finished jobs. The store keeps everything. | | rerunStalled | boolean | false | When false, a job that BullMQ finds stalled is failed, not run again. | | onError | (err, { phase, name?, runId?, tenant? }) => void | logs a warning | Called on a run's final failure, enqueue failure, reconcile or heartbeat failure, and a stall. | | validate | ({ def, tenant, meta, payload }) => void \| { tenant?, meta? } | none | Runs before a run is recorded. Throw to refuse, or return a normalized target. | | around | (ctx, next) => Promise | none | Wraps every handler call, for example to bind the tenant's database. | | context | object or (run, def) => object | none | Extra fields added to every ctx, such as { platformDb }. | | isBusyError | (err) => boolean | none | Marks "try again later" errors such as a DB connection limit. Only applies to jobs that set busyRetries. | | logger | { info, warn, error } | console | |

jobs.define(name, spec)

| Key | Default | Meaning | |---|---|---| | run(payload, ctx) | required | The handler. ctx = { runId, tenant, meta, attempt, maxAttempts, signal, log(), enqueue(), job, def, …context }. | | queue | config queue | BullMQ queue. | | retries | 0 | Extra tries after the first, with exponential backoff starting at backoffMs. | | timeoutMs | 15 min | The attempt fails after this long and ctx.signal aborts. The work itself is not interrupted, so keep handlers idempotent. | | schedule, timezone, payload | none | A cron schedule defined in code. Its BullMQ scheduler id is def:<name>. | | tenant | 'optional' | 'required' refuses runs that have no tenant. 'none' refuses runs that have one. | | busyRetries | 0 | How many busy errors are deferred (up to 30 s each) without spending an attempt. | | label, description, any other key | | Kept on the definition, so jobs.definitions() can feed a console and around can read them. |

jobs.startWorker(opts)

| Key | Default | Meaning | |---|---|---| | queues | WORKER_QUEUES, else all | The queues this worker consumes. Unknown names are refused. | | concurrency | WORKER_CONCURRENCY, else 2 | Jobs in flight per queue. | | stallMs | WORKER_STALL_MS, else max(10 min, slowest timeout + 5 min) | Exit(1) when one attempt or one heartbeat has been in flight this long. 0 turns it off. | | abandonAfterMs | 10 min | How long a running run may be silent before it is closed as failed. Stores without a per-run heartbeat column add the job's own timeoutMs to this. | | heartbeatMs | 15 s | Interval for the worker row heartbeat, run pulses, schedule reconcile, abandoned-run reaping and stale-worker cleanup. | | onShutdown | | Close your own pools after the workers drain. | | version, handleSignals (true), exit (process.exit) | | |

On SIGTERM or SIGINT the worker drains in-flight jobs and then exits with 0. It also obeys jobs.sendWorkerCommand(workerId, 'pause' | 'resume' | 'shutdown').

Stores

| Store | Clock | Use | |---|---|---| | knexStore(knexOrKorm, opts) | app (UTC) | Any knex dialect: MySQL, Postgres, SQLite. | | sqlStore(platformDb, opts) | database (NOW(3)) | The { all, get, run } MySQL interface the Pro products already have. | | memoryStore() | app | Tests. |

Options: runsTable (job_runs), schedulesTable (job_schedules, false = none), workersTable (job_workers, false = none), tenantColumn (tenant_id), metaColumns (e.g. { branchId: 'branch_id' }), heartbeatColumn (heartbeat_at, null = none).

Writing your own store? The required methods are insertRun, getRun, findRunByJobId, updateRun(id, patch, { ifStatus }) and listRuns. Everything else is optional; see JobStore in index.d.ts. Patches contain NOW for "the current instant on your clock". To change which schedules count as live, override store.listSchedules and reuse store.readSchedule(row).

Recipes

Express (CJS). Create one jobs module and require it from both the routes and worker.js. The worker entry point is just jobs.startWorker({ onShutdown: () => knex.destroy() }).

Next.js. Create the instance in a server-only module and cache it on globalThis so it survives HMR. Route handlers call jobs.enqueue and the jobs.runs.* methods. Never call startWorker inside Next. Run it as a separate process.

Per-tenant database. Bind the tenant's database in the around hook:

around: async (ctx, next) => {
  if (!ctx.tenant) return next();
  const db = await tenantDb(ctx.tenant);
  return runWithTenantDb(db, next);
},

Unrecoverable errors. Throw new UnrecoverableError('Company no longer exists') (re-exported from bullmq). The run fails immediately, with no retries.

What it deliberately does NOT do

• Run console reports. It does not do per-day charts, CSV export or joins to your companies and admins tables. Those are product SQL; keep them next to your console.
• Re-run abandoned or stalled work. A run that died mid-way is closed as failed. A human retries it with runs.retry(id).
• Interrupt a timed-out handler. It aborts ctx.signal, and that is all.
• Depend on @dreamtree-org/tenancy, or on any product's tables beyond the three listed above.
• Support cron below minute resolution. Only 5-field cron expressions are accepted.