Providers

A provider is the queue backend that accepts enqueued jobs and delivers them to the processor registered by BackgroundJobRuntime. Two providers ship with UDIOT. The provider abstraction is defined by the JobProvider interface, so custom backends can be plugged in by implementing that interface.

JobProvider Interface

interface JobProvider {
  capabilities: JobProviderCapabilities;
  initialize?(options?: Record<string, unknown>): Promise<void> | void;
  registerProcessor(processor: JobProcessor): void;
  enqueue<TPayload = unknown>(options: EnqueueJobOptions<TPayload>): Promise<ScheduledJob>;
  drain?(): Promise<void>;
  getStats?(): Promise<JobQueueStats>;
}

type JobProcessor = (job: ScheduledJob) => Promise<void>;

BackgroundJobRuntime calls registerProcessor during construction and initialize (if defined) when runtime.start() is called. drain is called on runtime.stop() to flush the queue before the process exits.

InMemoryJobProvider

Stores and executes jobs in the Node.js process. Execution is deferred with setTimeout according to the job's scheduledAt timestamp. No external dependencies are required.

Use when: running tests, local development, or any scenario where persistence and durability are not required.

import { InMemoryJobProvider } from '@tagna/udiot';

const provider = new InMemoryJobProvider();

Capabilities

Constructor: no arguments.

enqueue behaviour: assigns a sequential ID ({jobName}-{counter}), schedules a setTimeout for the delay, then calls the registered processor when the timer fires.

drain behaviour: clears all pending timers. Jobs that have not yet executed are discarded.

getStats behaviour: returns in-memory counters for waiting, active, completed, failed, and delayed since the provider was constructed.

BullMQJobProvider

Wraps BullMQ to provide a Redis-backed, persistent job queue. BullMQ handles durability, distributed workers, retries at the queue level, and cron scheduling natively.

Use when: deploying to production with multiple worker processes, needing job persistence across restarts, or requiring rate limiting.

import { BullMQJobProvider } from '@tagna/udiot';

const provider = new BullMQJobProvider({
  queueName?: string;          // Default: 'udiot-background-jobs'
  connection?: RedisOptions;   // BullMQ/ioredis connection options
  concurrency?: number;        // Worker concurrency; default 5
  defaultJobOptions?: JobsOptions; // BullMQ JobsOptions applied to every enqueue
});

Constructor Options

Capabilities

initialize behaviour: creates the BullMQ Queue, Worker, and QueueEvents instances. Must be called before enqueue. BackgroundJobRuntime.start() calls this automatically.

enqueue behaviour: calculates delay from scheduledAt - Date.now(), sets removeOnComplete: 100 and removeOnFail: 1000 by default, then calls queue.add(). Returns a ScheduledJob with the BullMQ job ID.

drain behaviour: closes the worker, queue, and queue events connections in parallel.

BackgroundJobRuntime

The engine that bridges your job definitions to a provider.

import { BackgroundJobRuntime } from '@tagna/udiot';

const runtime = new BackgroundJobRuntime({
  provider?: JobProvider;      // Default: InMemoryJobProvider
  hooks?: JobRuntimeHooks;     // onLifecycle, onError callbacks
  logger?: JobLogger;          // Default: console-based logger
});

Methods

scheduleJob Options

interface ScheduleJobOptions {
  tenantId?: string;
  runAt?: number;              // Unix ms; overrides the computed next-run time
  metadata?: Record<string, unknown>;
}

Every scheduleJob call wraps the execution in a jobs.process OpenTelemetry span and emits jobs.scheduled to the metrics registry.

Telemetry

The runtime emits the following metrics via the UDIOT metrics registry:

JobScheduler

Manages recurring schedules. Reads all JobDefinition entries that have a schedule and sets up setTimeout timers. After each execution the timer is re-created for the next occurrence.

import { JobScheduler } from '@tagna/udiot';

const scheduler = new JobScheduler(runtime, {
  logger?: JobLogger;     // Default: delegates to getLogger()
  timezone?: string;      // Fallback IANA timezone for cron expressions without one
});

Methods

Cron expressions are parsed with cron-parser. The scheduler calls expression.next().getTime() to determine the delay for the next setTimeout.

Basic Usage Example

import 'reflect-metadata';
import {
  BackgroundJobRuntime,
  JobScheduler,
  BullMQJobProvider,
  AIBackgroundJob,
  AISchedule,
  JobContext,
} from '@tagna/udiot';

@AISchedule({ kind: 'cron', expression: '*/5 * * * *', timezone: 'UTC' })
@AIBackgroundJob({ name: 'HeartbeatJob', category: 'system', priority: 'low' })
class HeartbeatJob {
  async execute(context: JobContext): Promise<void> {
    context.logger.info('Heartbeat OK');
  }
}

// Production setup
const provider = new BullMQJobProvider({
  connection: { host: 'redis', port: 6379 },
  concurrency: 3,
});

const runtime = new BackgroundJobRuntime({
  provider,
  hooks: {
    onLifecycle: (event) => {
      if (event.type === 'failed') {
        console.error(`Job ${event.jobName} failed on attempt ${event.attempt}`, event.error);
      }
    },
  },
});

const scheduler = new JobScheduler(runtime, { timezone: 'UTC' });

await runtime.start();
await scheduler.start();

// Graceful shutdown
process.on('SIGTERM', async () => {
  await scheduler.stop();
  await runtime.stop();
  process.exit(0);
});

Testing Patterns

Using InMemoryJobProvider in unit tests

import {
  BackgroundJobRuntime,
  InMemoryJobProvider,
  registerJobDefinition,
  resetJobRegistry,
} from '@tagna/udiot';

let runtime: BackgroundJobRuntime;
let provider: InMemoryJobProvider;

beforeEach(() => {
  resetJobRegistry();
  provider = new InMemoryJobProvider();
  runtime = new BackgroundJobRuntime({ provider });
});

afterEach(async () => {
  await runtime.stop();
});

test('ProcessInvoiceJob sends payment', async () => {
  // Register the definition manually so the import side-effect is controlled
  registerJobDefinition({
    name: 'ProcessInvoiceJob',
    target: ProcessInvoiceJob,
    category: 'data',
    priority: 'normal',
    retry: { strategy: 'none' },
  });

  await runtime.start();
  await runtime.scheduleJob('ProcessInvoiceJob', { invoiceId: 'inv_001', amount: 50 });

  // InMemoryJobProvider uses setTimeout; wait for it to fire
  await new Promise(resolve => setTimeout(resolve, 10));

  const stats = await provider.getStats();
  expect(stats.completed).toBe(1);
  expect(stats.failed).toBe(0);
});

Verifying a job was enqueued without executing it

Override the provider's enqueue method or wrap it with a spy:

const enqueuedJobs: string[] = [];
const enqueueOrig = provider.enqueue.bind(provider);
provider.enqueue = async (opts) => {
  enqueuedJobs.push(opts.definition.name);
  return enqueueOrig(opts);
};

Integration Notes

• BullMQJobProvider requires bullmq as a peer dependency. Install it separately: npm install bullmq.
• A single BullMQJobProvider instance creates one BullMQ Worker with concurrency slots. To scale horizontally, run multiple Node.js processes each with their own provider connected to the same Redis queue.
• BackgroundJobRuntime is not tied to a specific provider at construction time. Call runtime.setProvider(newProvider) to swap providers after construction, for example during a blue/green deployment.

Pitfalls

• Calling runtime.scheduleJob() before runtime.start() will still call provider.enqueue(), but BullMQJobProvider will throw because its queue is not yet initialised. Always await runtime.start() first.
• InMemoryJobProvider.drain() cancels pending timers, meaning jobs that have been enqueued but not yet executed are silently dropped. In test environments, ensure all timers have fired before calling runtime.stop().
• The default BullMQ options set removeOnComplete: 100 and removeOnFail: 1000. Adjust defaultJobOptions if you need to retain more history for auditing.