@job-kit/core

A production-ready, trigger-agnostic, storage-agnostic job lifecycle engine. Zero runtime dependencies. Designed to be attached directly to a host framework (strapi.jobEngine = engine) and driven or inspected from a UI: engine.state, engine.pause(), engine.inFlightCount, etc.

Philosophy

• Names developers already recognize: onError, onCompleted, onClaimed, retry, limits, lease.
• Every hook receives a rich context object — including a reference to the engine itself, so a hook (an audit log, a metrics collector, a UI panel) can inspect or control the engine without needing it wired in separately.
• Config is scoped by domain (limits, lease, retry, recovery, hooks) instead of a flat list of parameters.
• Every config leaf may be a getter, so it can be sourced from env vars, a config service, or a feature flag — the engine reads these live on every cycle, never caches them.
• The engine is a small state machine you can observe (engine.state), pause (engine.pause()), resume (engine.resume()), and query (engine.inFlightCount).

Quick usage

import { JobEngine, exponentialBackoff } from '@job-kit/core';

const engine = new JobEngine({
  jobType: 'payment-webhook',
  storage: myStorageAdapter,
  execute: async (job) => {
    // job.data is your domain payload; job.leaseToken proves you own this attempt.
    await chargeCard(job.data);
  },
  limits: {
    get maxConcurrency() {
      return featureFlags.get('payment-webhook.concurrency') ?? 10;
    },
    batchSize: 50,
  },
  lease: {
    durationMs: 30_000,
    heartbeatIntervalMs: 10_000,
  },
  retry: {
    policy: { maxAttempts: 5, backoffMs: exponentialBackoff(1_000, 60_000) },
    onError: async (ctx) => {
      if (ctx.error.name === 'ValidationError') return 'dead_letter';
      if (ctx.error.name === 'CardDeclined') return 'fail';
      return 'retry';
    },
  },
  recovery: {
    staleIntervalMs: 60_000,
  },
  hooks: {
    onClaimed: (ctx) => auditLog.write('claimed', ctx),
    onStarted: (ctx) => metrics.timer(`${ctx.jobType}.duration`).start(ctx.idempotencyKey),
    onCompleted: (ctx) => metrics.timer(`${ctx.jobType}.duration`).stop(ctx.idempotencyKey),
    onFailed: (ctx) => alerting.page(`${ctx.jobType} failed`, ctx),
    onDeadLetter: (ctx) => archive.store(ctx),
    onEngineError: (error, info) => logger.error(error, info),
  },
});

// Attach to a host framework and let its lifecycle drive the trigger.
strapi.jobEngine = engine;

// Any trigger just calls tick(). This example uses an interval, but this
// package has no idea an interval is being used — a Kafka consumer, a
// RabbitMQ handler, a cron tick, an HTTP endpoint, or a "run now" button
// in an admin UI would all just call the same method.
setInterval(() => void engine.tick(), 500);

// Run once at startup, and periodically (see recovery.staleIntervalMs),
// to reclaim work orphaned by a crash.
await engine.recoverStale();
setInterval(() => void engine.recoverStale(), 60_000);

// From a UI or shutdown hook:
engine.pause(); // stop claiming new work, let in-flight work finish
engine.resume(); // start claiming again
await engine.shutdown(10_000); // stop for good, on SIGTERM/app shutdown

Config reference

| Scope | Field | Default | Notes | | ---------- | ---------------------------------------------------------------------------------------------------------------- | ------------------------------- | ----------------------------------------------------- | | limits | maxConcurrency | 10 | In-flight execution cap for this engine instance | | limits | batchSize | 50 | Candidates fetched per cycle | | lease | durationMs | 30_000 | How long a claim is valid before it's stale | | lease | heartbeatIntervalMs | off | Renews the lease while executing; omit for short jobs | | retry | policy | 5 attempts, exponential backoff | { maxAttempts, backoffMs(attempt) } | | retry | onError | always 'retry' | Returns 'retry' \| 'fail' \| 'dead_letter' | | recovery | staleIntervalMs | 60_000 | Informational — you own the timer, see below | | recovery | batchSize | 100 | Stale jobs recovered per recoverStale() call | | hooks | onClaimed/onStarted/onCompleted/onFailed/onDeadLetter/onRetryScheduled/onRecovered/onEngineError | none | All optional, all isolated from crashing the engine |

Every field above can be written as a getter:

limits: {
   get maxConcurrency() {
      return process.env.JOB_CONCURRENCY ? Number(process.env.JOB_CONCURRENCY) : 10;
   },
},

The state machine

active ──pause()──► paused ──resume()──► active
   │                                        │
   └──────────────── shutdown() ───────────►┤
                                             ▼
                                        shutting_down ──► stopped

• active — the default; tick() claims and executes work.
• paused — tick() is a no-op; in-flight work keeps running to completion. Resumable.
• shutting_down → stopped — terminal. shutdown(gracePeriodMs) waits for in-flight work, then releases anything still running so another worker (or this one, after a restart) can pick it up immediately rather than waiting out the lease.

engine.state, engine.inFlightCount are readable at any time — the intended basis for a UI status panel.

Retry verdicts

retry.onError(ctx) returns one of:

• 'retry' — try again later, subject to retry.policy.maxAttempts. If attempts are exhausted, the engine treats this the same as 'fail' — it does not silently promote it to 'dead_letter'; only an explicit 'dead_letter' verdict routes there.
• 'fail' — terminal now. Fires hooks.onFailed.
• 'dead_letter' — terminal, routed to hooks.onDeadLetter instead of onFailed, for errors that need archival/manual review rather than an alert.

If retry.onError is omitted, every error is treated as 'retry' (matching the historical default of "retry everything").

What's intentionally NOT in this package

• Any storage technology (Strapi, SQL, Redis...) — implement JobStorage.
• Any trigger mechanism (interval, Kafka, RabbitMQ, cron, HTTP) — implement Trigger, or just call engine.tick() from wherever.
• Timers of its own. recovery.staleIntervalMs is a documented convention for you to wire up (setInterval, a cron job, a Kafka consumer on a schedule topic...) — the engine takes no timer dependency.
• True idempotency of side effects (e.g. "never charge a card twice") — the engine guarantees duplicate execution is rare and detectable via lease fencing; use ctx.idempotencyKey to make your own side effects idempotent where it matters.

Package layout

src/
  types.ts     JobStorage/Trigger contracts, ManagedJob, scoped config interfaces
  context.ts   staged per-attempt context, EngineHandle, JobEngineHooks
  errors.ts    error serialization + hook-failure isolation
  engine.ts    JobEngine — the state machine + lifecycle logic
  index.ts     public exports
test/
  inMemoryStorage.ts   in-memory JobStorage, for testing the engine ONLY
  engine.test.ts       28 tests

Known limitation

hooks.onCancelled is defined in the JobEngineHooks type for API completeness, but is not currently fired by the engine: cancellation (storage.cancel(...)) happens outside the claim flow the engine observes (a cancelled pending job simply stops appearing in findEligible results), so there's no natural point in today's engine where it would detect the transition to raise the hook. Firing it would require either a dedicated cancellation-scan (mirroring recoverStale()) or a storage-contract addition — left as an open question rather than implemented speculatively.