@intelligo-dev/jobs

Install

pnpm add @intelligo-dev/jobs drizzle-orm

drizzle-orm is a peer. The queue is one table, jobs, in the framework’s schema: intelligo migrate from @intelligo-dev/cli creates it, and the connection is the DATABASE_URL that @intelligo-dev/core reads.

Use

Enqueue from anywhere that can reach the database:

import { enqueue } from "@intelligo-dev/jobs";

await enqueue({
  kind: "digest.send",
  workspaceId, // or null: a job no workspace owns (an anonymous request's work)
  payload: { userId },
  runAt: new Date(Date.now() + 60_000), // optional: not before
  maxAttempts: 5, // optional: 3 by default
});

Drain from a cron route, a long-running worker or a script — handlers are keyed by kind:

import { drain } from "@intelligo-dev/jobs";

const result = await drain(
  {
    "digest.send": async (job) => {
      await sendDigest(job.payload);
    },
  },
  { limit: 25, deadlineMs: 45_000 }
);
// { claimed, succeeded, failed, unhandled, deferred }

Drain on a schedule

The framework runs no worker. The maintenance route that intelligo add maintenance writes prunes the queue but does not drain it, because only the application has the handlers. Give the drain a route of its own, behind the same CRON_SECRET, and schedule it beside maintenance:

import { timingSafeEqual } from "node:crypto";

import { drain } from "@intelligo-dev/jobs";

import { composeIntelligo } from "@/lib/intelligo";
import { handlers } from "@/lib/jobs";

export const maxDuration = 60;

function authorized(request: Request): boolean {
  const secret = process.env.CRON_SECRET;
  const header = request.headers.get("authorization") ?? "";
  if (!secret || secret.length < 32) return false;
  const expected = Buffer.from(`Bearer ${secret}`);
  const given = Buffer.from(header);
  return given.length === expected.length && timingSafeEqual(given, expected);
}

export async function GET(request: Request) {
  if (!authorized(request)) return new Response(null, { status: 403 });
  composeIntelligo();
  return Response.json(
    await drain(handlers, { limit: 25, deadlineMs: 45_000 })
  );
}

Recurring work is a handler that enqueues its own next run:

"digest.daily": async (job) => {
  await sendDailyDigests();
  await enqueue({
    kind: "digest.daily",
    workspaceId: null,
    payload: {},
    runAt: new Date(Date.now() + 24 * 60 * 60_000),
  });
},

In development, call the route by hand or in a loop: curl -H "Authorization: Bearer $CRON_SECRET" localhost:4002/api/cron/jobs.

What the queue guarantees

• No double-processing. Claiming is FOR UPDATE SKIP LOCKED, so several workers drain the same queue and each takes different rows. A worker writes to a job only while it still holds the claim: a job another worker claimed again while it waited in a batch is skipped, and a late outcome never overwrites the newer one.
• A recorded outcome. A handler runs once per attempt: a failed status write is retried rather than counted as a handler failure, and one job’s failed write does not stop the rest of the batch.
• Retry with backoff. A handler that throws puts the job back, one more minute later per attempt; after maxAttempts it is failed and keeps its last error. listFailedJobs() reads them, newest first.
• A dead worker does not strand a job. One still running thirty minutes after its handler started is claimed again, so a handler must finish well inside that — until it has used its attempts: a handler that keeps killing its worker ends failed.
• A drain fits its time limit. With deadlineMs, jobs claimed but not started by then go back without spending an attempt (deferred).
• An unknown kind costs nothing. A job with no handler in this worker goes back without spending an attempt, a minute behind the jobs the worker can run — another deployment may know it.

pruneJobs(before, { failedBefore }) deletes succeeded jobs finished before before and failed ones finished before failedBefore (the same cutoff when omitted), in bounded batches, and returns how many. postgresJobQueue is the same enqueue and drain as one JobQueue value; a different queue implements that type, and call sites stay as they are.

Entry points

• @intelligo-dev/jobs
• @intelligo-dev/jobs/db

npm · source