Class Eddyq

The migration line this client was built for (default: "main").

Register a fixed-interval schedule ({ every: ms }). Fires every intervalMs milliseconds — no cron expression required. Skip-missed semantics: a delayed fire doesn't catch up, matching the cron path and the Redis backend.

Upsert a cron schedule. Jobs of kind with the given payload will be enqueued automatically each time the cron fires. Passing the same name updates the schedule in place.

Cron syntax is a 6- or 7-field sec min hour day month dayOfWeek [year] expression (the cron crate's dialect — note the leading seconds field).

await queue.addSchedule(
  "daily-report",
  "0 0 8 * * *",           // every day at 08:00:00 UTC
  "report.generate",
  { scope: "daily" },
  { priority: 5 },
);
Copy

Cancel a pending job. Returns true if cancelled, false if the job doesn't exist or is already running / finalized (handlers must cooperate to stop a running job — eddyq can't abort it for you).

Ad-hoc retention sweep. Deletes up to limit finalized jobs in state older than graceMs milliseconds. state is one of "completed" | "failed" | "cancelled". Returns the number of rows actually deleted.

Close the underlying Postgres pool and release any retained NAPI ThreadsafeFunctions. Call on shutdown.

Defensive in two ways:

• If shutdown() was never called and the queue is Running, we run an internal Abandon shutdown (drops runtime without awaiting handlers, leaves DB rows for heartbeat-sweep recovery). This isn't graceful — callers should await queue.shutdown() first — but it guarantees close() can't hang the process.
• The abort TSFN is dropped if shutdown() didn't already.

After close(), no further calls on this Eddyq instance are valid.

Enqueue a job. payload is serialized as JSON and passed to the worker registered for kind.

Enqueue a batch of jobs and (optionally) a callback that fires when every item reaches terminal state. Native fan-in primitive — replaces the per-app counter table workaround for "run X after these N jobs."

The callback's payload gets a namespaced envelope: { _eddyq_batch: { batchId, total, completed, failed, cancelled, durationMs }, ...userPayload }. Handler branches on failed / cancelled counts to decide what success vs partial-failure means in its domain.

Items skipped via uniqueKey dedup do not count toward the batch's total — they belong to the batch that originally enqueued them. The returned skipped reports the count for the caller's logging.

Enqueue a batch of jobs in a single round-trip. Uses a Postgres UNNEST-based INSERT, so 1000 jobs cost roughly one statement instead of one per job. Mixed kind within a batch is supported.

Returns aggregate counts (inserted, skipped); per-job ids are not surfaced — use a stable uniqueKey per item if you need to correlate results back to your own domain objects, or fall back to enqueue() when you need the auto-generated id.

Batch size is capped at 5,000 items per call — split larger workloads client-side.

Job counts grouped by (queue, state). One SQL query — use as the landing query for a dashboard.

Every group that has an explicit row (cap, pause, rate-limit state).

Paginated job listing with optional filters. Defaults: limit=50, offset=0. Limit caps at 500.

Every named queue that has an explicit row (concurrency cap, pause state, etc.). Queues with no row are implicitly unlimited and not returned here — use getStats() to see all queues with live jobs.

Every registered cron schedule.

Apply all pending schema migrations.

Roll back up to max_steps migrations.

Full migration status (all known versions, applied or pending).

Remove a schedule. Returns true if a row was deleted.

Register a handler invoked when shutdown() is called. The handler's reason arg is a human-readable string; the JS ergonomics layer (lib.cjs) uses this to broadcast .abort() to all in-flight AbortControllers, so user handlers observing call.signal can bail.

Most users don't call this directly — lib.cjs wires it automatically.

Cap concurrent running jobs in group_key. Jobs with enqueue(..., { groupKey }) respect this cap across all workers.

Token-bucket rate limit: at most count jobs may start per periodMs milliseconds in this group.

Cap total running jobs on a named queue across all worker processes.

Set a default per-job timeout (milliseconds) for this named queue. Pass null to clear.

Toggle a schedule on or off without deleting it. Returns true if a row was updated.

Set worker concurrency (max in-flight jobs in this process). Default 10. Must be called before start().

Stop the worker runtime. Signals any registered abort handler first, then waits up to gracefulTimeoutMs (default 30 000) for in-flight jobs to finish before forcibly cancelling the runtime tasks. Admin methods remain usable after shutdown — call close() to release the DB pool entirely.

On return, all NAPI ThreadsafeFunction references this binding holds are dropped (handler TSFNs via the worker registry, plus the abort TSFN). That releases their libuv ref counts so Node's event loop can drain naturally — without that drop, a Nest app.close() on SIGTERM would call this method but the process would still hang until the orchestrator force-killed it.

options.mode:

• "drain" (default) — graceful: stop claiming new jobs, fire AbortSignal to in-flight handlers, await up to gracefulTimeoutMs. Use for routine deploys.
• "force" — fast: abort runtime tasks immediately and proactively reclaim rows this pod was processing (set running→pending) so other pods pick them up without waiting for heartbeat sweep. Use when SIGKILL is imminent (Kubernetes grace period almost up).
• "abandon" — last-resort: drop runtime, leave rows alone. The heartbeat sweep on another pod will recover after staleAfter. Use only on panic exits.

Start the worker runtime. Handlers registered via work() begin processing jobs from Postgres. Fetch/sweep/scheduler loops run until shutdown() is called.

Pending-migration guard. Before starting, checks that every migration the binary knows about has been applied. If any are missing, start() errors out with a clear message instead of booting workers that will trip on missing columns. Pass { skipMigrationCheck: true } to override (e.g. when schema is managed by a separate deploy step).

Async so the internal tokio::spawns land on napi's tokio runtime.

Subscribe this worker to specific named queues. Default ["default"]. Must be called before start().

Reconcile DB schedules against a code-declared list. Each entry is upserted; any DB schedule whose name is not in declared is deleted. Idempotent — safe to run on every boot. Use this when schedules are declared in module config (e.g. EddyqModule.forRoot({ schedules })).

Register an async JS handler for a job kind. Call once per kind before start(). The handler receives a JobCall object and should resolve on success or throw/reject to trigger a retry.

await queue.work("send.email", async ({ payload, id, attempt }) => {
  await sendgrid.send(payload);
});
Copy

Connect to Postgres and construct a client. Does not run migrations — call migrate() on first boot (or on app deploy).