job-pro

job-pro is a broker-agnostic TypeScript framework for building reliable background job processors. It owns dispatching, retries, acknowledgement, polling, circuit breaking, workflow validation, state transitions, and structured observations. You retain ownership of your broker, persistence, payload validation, and business logic.

Status

This is the initial standalone extraction from a production processor runtime. It is designed as a small framework package rather than a hosted service or a broker implementation.

Install

Until it is published to npm, install it directly from GitHub:

npm install github:eodeluga/job-pro

Quick start

The complete runnable example is in examples/email-digest.

npm install
npm run example

Create a processor by extending JobProcessor. Its spec controls its queue, retry policy, polling interval, optional workflow dependencies, and concurrency ceiling.

import {
  JobProcessor,
  JobState,
  type JobExecutionResult,
  type JobProcessorContext,
} from 'job-pro'

type EmailDigestPayload = {
  recipient: string
  reportId: string
}

class EmailDigestProcessor extends JobProcessor<EmailDigestPayload> {
  public constructor() {
    super(JobProcessor.createSpec({
      failurePolicy: {
        action: 'retry',
        delaySeconds: 30,
        maxRetries: 3,
      },
      pollingMs: 1000,
      processorKey: 'email-digest',
    }))
  }

  public async execute(context: JobProcessorContext<EmailDigestPayload>): Promise<JobExecutionResult> {
    await sendDigest(context.job.payload)

    return {
      result: {
        status: JobState.completed,
      },
    }
  }
}

Register processors, supply the queue adapter and state store, then start the poller.

const registry = new JobProcessorRegistry()
const dispatcher = new JobDispatcher(registry, queueDriver, stateStore, observationSink)
const poller = new JobPoller(dispatcher, queueDriver, registry, observationSink, {
  circuitBreaker: {
    backoffMultiplier: 2,
    failureThreshold: 3,
    halfOpenMaxMessages: 1,
    maxOpenMs: 60000,
    openMs: 5000,
    windowMs: 30000,
  },
})

registry.register(new EmailDigestProcessor())
poller.start()

Use the envelope below when publishing jobs. jobId must be durable and idempotent in your application.

const job = {
  jobId: 'digest-001',
  payload: {
    recipient: 'developer@example.com',
    reportId: 'weekly-risk-report',
  },
  processorKey: 'email-digest',
}

What job-pro provides

Capability	Behaviour
Processor registry	Enforces unique processor keys and queue ownership; validates dependencies and cycle-free workflows on start.
Dispatcher	Validates envelopes, routes by processor key, records processing and terminal states, and acknowledges terminal messages.
Retry policy	Leaves expected failures on the queue until the receive-count limit; can change broker visibility for a delay.
Poller	Runs one non-overlapping polling loop per processor, limits batches, and exposes status for health endpoints.
Circuit breaker	Opens after a rolling failure threshold, probes in half-open mode, and backs off repeat probe failures.
Observations	Emits structured lifecycle, queue, and circuit events through a best-effort sink.
Queue abstraction	Supports enqueue, receive, acknowledgement, and visibility changes without coupling to a specific broker.

Deliberate boundaries

job-pro does not choose your broker, database, schema library, logger, monitoring service, or scheduler. Implement QueueDriver, JobStateStore, and optionally JobObservationSink at your application boundary. Validate each processor payload in execute using your preferred validator.

Workflows are declarative validation, not orchestration. A processor can enqueue the next job after successful work; dependencies protect startup from missing stages and invalid graphs.

Documentation

• docs/architecture.md explains the runtime model and delivery guarantees.
• docs/building-a-processor.md explains processor contracts, payload safety, and workflow chaining.
• docs/queue-adapters.md defines the broker adapter contract.
• docs/operations.md covers retries, circuits, observations, and safe shutdown.

Development

npm test
npm run typecheck
npm run lint
npm run build

The project uses @/… aliases in its source, with build-time rewriting so the compiled package uses portable relative imports.

Licence

MIT. See LICENSE.