ravenflow

Typed background jobs for Node.js, on BullMQ. Redis holds the work in flight; MongoDB holds the backlogs and the history.

• Types end to end. A send is checked against the job's declaration.
• A client and a worker. Your API sends work, a worker process runs it, and both share one config.
• Two triggers. A clock, and a spool: a backlog in MongoDB drained at the job's pace.
• What production asks for. Chains, deduplication, retries, a rate held across the fleet, levers to pause a job, and a record of every execution.

Install

npm install ravenflow

Requires Node.js 20.19+, Redis and MongoDB. For development:

docker run -d -p 6379:6379 redis:7
docker run -d -p 27017:27017 mongo:7

Create and run a job

1. Declare it. A job is a handler from a typed payload to a typed result. The config registers it under a name.

// config.ts
import { defineConfig, defineJob } from 'ravenflow';

export const sendWelcomeEmail = defineJob<{ to: string; name: string }, { messageId: string }>({
  handler: async ({ params, logger }) => {
    logger.info({ to: params.to }, 'sending the welcome email');
    const { id } = await emailTransport.send({ to: params.to, subject: `Welcome, ${params.name}` });
    return { messageId: id };
  },
});

export const config = defineConfig({
  defaultQueue: 'default',
  jobs: { sendWelcomeEmail },
});

2. Send it from any process: your API, a CLI, another job.

// client.ts
import { createRavenflowClient } from 'ravenflow';
import { config } from './config';

const client = createRavenflowClient(config, {
  redis: { host: 'localhost', port: 6379 },
  mongo: 'mongodb://localhost:27017/myapp',
});

await client.jobs.sendWelcomeEmail.send({ to: '[email protected]', name: 'Ada' });

3. Run it in a worker. A worker serves one queue.

// worker.ts
import { createRavenflowWorker } from 'ravenflow';
import { config } from './config';

const worker = createRavenflowWorker(config, {
  redis: { host: 'localhost', port: 6379 },
  mongo: 'mongodb://localhost:27017/myapp',
  queue: 'default',
  concurrency: 10,
});

await worker.start();
process.once('SIGTERM', () => void worker.close());

The handler receives { params, logger, signal, job, continueWith, deduplicationId }. What it returns is the job's result; what it throws fails the job.

A send takes options: { delayMs, priority, deduplicationId }.

Triggers

Without a trigger, a client sends the job. A trigger feeds it for you.

A clock

import { defineJob } from 'ravenflow';

export const cleanupTempFiles = defineJob<void, { removed: number }>({
  trigger: { cron: '0 3 * * *' },
  handler: async () => ({ removed: await purgeTempDir() }),
});

One occurrence at a time across the fleet. A worker registers the schedules of its queue's jobs when it starts. A clock has nothing to pass, so the payload must be void or all-optional: anything else does not compile.

A spool

import { defineJob } from 'ravenflow';

export const enrichDocument = defineJob<PendingDocument, void>({
  pace: { perSecond: 10 },
  trigger: { spool: { name: 'pending_documents' } },
  handler: async ({ params }) => enrich(params),
});

await client.jobs.enrichDocument.pushMany([
  { url: 'https://example.com/a', tenantId: 'acme' },
  { url: 'https://example.com/b', tenantId: 'globex' },
]);

Items wait in MongoDB (rav:spool-backlog:<name>). Every 30 seconds, the worker serving the job's queue releases at most one tick's work at the job's pace into Redis, and an item leaves the backlog once its run ends. Options: tickMs, canPull, expireUnclaimedAfterMs.

Advanced

Chains. A handler that returns continueWith(nextParams) ends its link and queues the next one, each link a job of its own with its own retries.

import { defineJob } from 'ravenflow';

export const syncCatalog = defineJob<{ cursor?: string; synced?: number }, { synced: number }>({
  trigger: { cron: '0 2 * * *' },
  handler: async ({ params, continueWith }) => {
    const page = await api.fetchPage(params.cursor);
    await db.products.upsertMany(page.items);
    const synced = (params.synced ?? 0) + page.items.length;
    if (page.next) return continueWith({ cursor: page.next, synced });
    return { synced };
  },
});

Deduplication. deduplicate: true keeps one run of the job in flight at a time, deduplicate: (params) => id one per id, and send(params, { deduplicationId }) chooses for one send. A duplicate is ignored, and send resolves with the job already holding the id.

Retries. attempts: 5, backoff: { type: 'exponential', delayMs: 1_000 }. A handler runs at least once, not exactly once: make it safe to run twice.

Pace. pace: { perSecond: 2 } holds a rate for the whole job, across every worker process. A job that comes too early waits its turn without spending an attempt.

Levers. Every job has client.jobs.<name>.controls:

| Lever | On | Effect | | --- | --- | --- | | pauseExecution() / resumeExecution() | every job | Jobs are deferred, not lost, and no attempt is spent. | | pauseTrigger() / resumeTrigger() | a clock or a spool | Occurrences are skipped; a spool stops draining. | | pauseIngest() / resumeIngest() | a spool | push throws a SpoolIngestPausedError. | | purgeBacklog() | a spool | Drops the items never claimed. |

Job results. A worker writes one record per execution to MongoDB (rav:job-results): the params, the result or the error and its causes, what the handler logged, how long it ran. Kept 14 days, and counted per minute into rav:aggregated-results.

Withdrawal. A worker that cannot reach MongoDB for 30 seconds closes itself, sets process.exitCode to 1 and calls withdrawal.onWithdraw(reason). Let your process manager restart it.

Configuration

| defineConfig | Default | | --- | --- | | defaultQueue, jobs | required | | logger: { level, pretty } | debug and readable; info and JSON with NODE_ENV=production | | queuesRetention: { completedCount, failedCount } | 1000 each | | jobResultsRetentionHours | 336 | | aggregatedResultsRetentionHours | 2160 |

| createRavenflowWorker | Default | | --- | --- | | redis, mongo, queue | required | | concurrency | 5 | | withdrawal: { afterMs, onWithdraw } | 30 000 ms |

License

MIT