Background Jobs

warpkit-postgres uses a Postgres-backed job queue (jobs/failed_jobs/schedules tables in src/lib/db/schema.ts), replacing the old @bitclaw/jobs SQLite library dropped during the Postgres port. One implementation drains due work and evaluates cron schedules; two different callers invoke it depending on deploy target.

Architecture

enqueue('email:welcome', { userId, email, name })
         │
         ▼
    jobs table          ←── persisted in shared Postgres, same DB as everything else
         │
         ▼
   runTick(budgetMs)     ←── claims a batch (`FOR UPDATE SKIP LOCKED`), runs handlers,
         │                    evaluates due schedules in the same pass
         ├── success → marked done
         └── failure → retry (markJobFailed) or dead-letter (markJobDead, on NonRetryableError)

runTick() (src/features/jobs/tick.server.ts) is the single implementation. It has two callers:

Self-hosted: startWorkers() (src/features/jobs/workers.ts) calls runTick(30_000) on a 5s setInterval poll loop, started/stopped in server/start.ts alongside Sentry init and startup reconciliation , see "Why server/start.ts, not a Nitro plugin" below for why these boot-time side effects live there and not in a Nitro-plugin convention.
Cloudflare Workers: server/worker.ts's scheduled() export, invoked by a Cron Trigger (wrangler.jsonc's triggers.crons, "* * * * *" , Workers' finest granularity), calls runTick(25_000) once per firing, awaited (no lingering process after a Workers invocation returns). See docs/warpkit/deployment.md's Cloudflare Workers section for the full setup (Hyperdrive, nodejs_compat, env shim ordering).

Draining to a wall-clock budget rather than a fixed batch size matters for both targets: self-hosted's poll interval and a Cron Trigger's per-minute firing both need one invocation to clear whatever backlog has accumulated since the last one, not just a fixed row count.

Why `server/start.ts`, not a Nitro plugin

An earlier version of this codebase started workers from a server/plugins/jobs.ts file using defineNitroPlugin, on the assumption that Nitro auto-loads everything in server/plugins/ , a real feature of Nitro, but one that only activates when Nitro's own Vite plugin is registered with a scanDirs pointing at server/. This project's vite.config.ts never registered it (only @tanstack/react-start/plugin/vite was configured, which doesn't turn on Nitro's plugin scanner) , and this project's TanStack Start version doesn't depend on Nitro at all today. The result: startWorkers() inside that plugin file never ran, in dev, in bun run build, or in the built production bundle.

The fix was to drop the Nitro-plugin convention rather than wire up a missing Vite plugin. server/start.ts is the one file guaranteed to run in production (bun run start / bun cluster.ts) and never inspected by Vite's client bundler at all, so startWorkers()/stopWorkers() , plus Sentry init and startup reconciliation, which have the identical constraint , live there directly. This means none of these startup side effects run under bun run dev , there's no dev-mode equivalent currently. If you need to exercise job workers locally, run bun run build && bun run start. The Cloudflare Workers target has no equivalent gap: scheduled() is its own explicit entry point, invoked directly by wrangler dev --remote --test-scheduled.

Key Files

File	Purpose
`src/features/jobs/types.ts`	`AppJobs` type map , all job payloads
`src/features/jobs/job-types.ts`	`Job`, `JobStatus`, `NonRetryableError`, `AddJobOptions`, `Schedule`, and the other shape types the old `@bitclaw/jobs` package used to export
`src/features/jobs/queue.server.ts`	Plain async functions (`claimBatch`, `markJobDone`, `markJobFailed`, `markJobDead`, `reclaimExpiredLeases`, `releaseJob`, `enqueue`) operating on the `jobs`/`failed_jobs` tables
`src/features/jobs/enqueue.server.ts`	`enqueue(type, data, options?, database?)` , call from anywhere server-side, `await` it (async now; the old library's `.add()` was synchronous bun:sqlite)
`src/features/jobs/tick.server.ts`	`runTick(budgetMs)` , claims + runs due jobs, evaluates schedules, shared by both callers
`src/features/jobs/workers.ts`	Self-hosted's poll loop: `startWorkers()` / `stopWorkers()`
`src/features/jobs/rate-limiter.ts`	`tryConsumeRate(type, limit)` , in-memory per-type sliding-window throttle, checked inside `runTick()`
`src/features/jobs/scheduler.server.ts`	`SCHEDULES` registry, `evaluateDueSchedules()`, `ensureSchedulesSeeded()`, admin-facing `getScheduler()`
`src/features/jobs/handlers/`	One file per job type
`src/features/jobs/server/jobs-admin-logic.ts`	Pure logic for admin queries/mutations (injectable, testable)
`src/features/jobs/server/jobs-admin.queries.ts`	Admin server fns: stats, list, types, schedules
`src/features/jobs/server/jobs-admin.mutations.ts`	Admin server fns: cancel, retry, purge, pause/resume schedule
`server/start.ts`	Self-hosted entry point , boots the poll loop on server start, stops it on shutdown
`server/worker.ts`	Cloudflare Workers entry point , `scheduled()` export calls `runTick()` per Cron Trigger firing

Adding a New Job Type

1. Add to the type map

// src/features/jobs/types.ts
export type AppJobs = {
  'email:welcome': { userId: string; email: string; name: string | null };
  'email:receipt': { email: string; name: string | null; planName: string; amount: number; currency: string }; // new
};

2. Write the handler

// src/features/jobs/handlers/email-receipt.ts
import type { AppJobs } from '@/features/jobs/types';
import { sendReceiptEmail } from '@/server/email';
import type { Job } from '../job-types';
import { NonRetryableError } from '../job-types';

export const handleReceiptEmail = async (
  job: Job<AppJobs['email:receipt']>
): Promise<void> => {
  const result = await sendReceiptEmail(
    job.data.email,
    job.data.name,
    job.data.planName,
    job.data.amount,
    job.data.currency
  );
  if (!result.ok) {
    // Permanent failure (e.g. provider not configured) , skip retries
    if (result.code === 'EMAIL_PROVIDER_NOT_CONFIGURED') {
      throw new NonRetryableError(result.message);
    }
    // Transient failure , throw to trigger retry
    throw new Error(result.message);
  }
};

3. Register the handler

// src/features/jobs/tick.server.ts , add to the HANDLERS map
const HANDLERS: Record<string, Handler> = {
  // ...existing entries
  'email:receipt': handleReceiptEmail
};

There is no per-type worker/poll loop to register anymore , runTick() looks up every claimed job's handler in this one map.

4. Enqueue from anywhere server-side

import { enqueue } from '@/features/jobs/enqueue.server';

// Inside a server function, after a successful write , async now, await it:
await enqueue('email:receipt', { email: user.email, name: user.name, planName: 'Pro', amount: 2000, currency: 'usd' });

Pass a withUserTransaction tx handle as the 4th argument when the enqueue must be atomic with other writes in the same transaction.

Testing Job Handlers

Test handlers directly , no need to spin up runTick() or a real queue:

// src/features/jobs/handlers/email-receipt.test.ts
import { describe, expect, it } from 'bun:test';
import { http, HttpResponse } from 'msw';
import { mswServer } from '@/test/msw/server';
import { handleReceiptEmail } from './email-receipt';
import { makeJob as makeJobBase } from './jobs-test-fixtures';

const makeJob = (data?: Partial<AppJobs['email:receipt']>) =>
  makeJobBase('email:receipt', {
    email: 'test@example.com',
    name: 'Test User',
    planName: 'Pro',
    amount: 2000,
    currency: 'usd',
    ...data
  });

it('sends receipt email', async () => {
  await expect(handleReceiptEmail(makeJob())).resolves.toBeUndefined();
});

it('throws on transient failure so the next tick retries', async () => {
  mswServer.use(
    http.post('https://api.resend.com/emails', () =>
      HttpResponse.json({ name: 'internal_server_error', message: 'Service unavailable' }, { status: 500 })
    )
  );
  await expect(handleReceiptEmail(makeJob())).rejects.toThrow();
});

jobs-test-fixtures.ts's makeJob(type, data) builds a well-typed Job<AppJobs[K]> fixture , reuse it rather than hand-rolling the Job shape. MSW lifecycle is set up globally in src/test/setup.ts.

Dead-Letter Queue

Jobs that exhaust all retries (or throw NonRetryableError) are moved to the failed_jobs table via markJobDead. Inspect with a Postgres client against DATABASE_URL:

SELECT * FROM failed_jobs ORDER BY failed_at DESC LIMIT 20;

Or via the admin UI: /dashboard/admin/scheduled-jobs and the jobs admin queries (getJobStats, listFailedAdminJobs) surface the same data without a direct DB connection.

Rate Limiting

Per-type throttling is preserved from the old library's maxRate worker option, now as an in-process sliding-window limiter (src/features/jobs/rate-limiter.ts's tryConsumeRate), checked inside runTick() before running each claimed job. A job that's rate-limited is released back to pending for the next tick to retry, rather than burning a retry attempt. Known limitation on Cloudflare Workers: the limiter is in-memory and resets every Cron Trigger invocation, since each firing is a fresh isolate , accepted, not a bug, same gap the code's own comments already documented before the Workers port existed.

Scheduler (Cron Jobs)

All schedule definitions live in src/features/jobs/scheduler.server.ts's SCHEDULES array , upserted into the shared schedules table by ensureSchedulesSeeded() (so the admin page shows the full list even before the first tick), evaluated by evaluateDueSchedules(db, now) at the end of every runTick() call.

// src/features/jobs/scheduler.server.ts (trimmed , see the real file for the
// full set and their individual cadence-reasoning comments)
export const SCHEDULES: Array<{
  name: string;
  type: keyof AppJobs;
  cron: string;
  data: Record<string, never>;
  overlap: boolean;
}> = [
  {
    name: 'reengagement-scan',
    type: 'email:reengagement-scan',
    cron: '0 9 * * *', // daily at 9am
    data: {},
    overlap: false // skip if previous run still in progress
  }
  // ...plus 'reconcile-deletions', 'snapshot-mrr', and 'reconcile-billing'
];

evaluateDueSchedules uses cronDueBetween (any matching minute since the schedule's lastRunAt, not just an exact-minute match), so a tick that runs late or infrequently still catches a schedule due sometime in the gap , this matters most for self-hosted's 5s poll (fine-grained) vs Workers' per-minute Cron Trigger firing (coarser).

Both reconciler-style schedules run exclusively through this table, never called directly from a boot hook or from scheduled(): reconcile-deletions (daily, account:reconcile-deletions) and reconcile-billing (daily, billing:reconcile , added specifically to avoid ~1440 stripe.subscriptions.retrieve() calls/day per subscription that a per-minute cadence would cost). See the SCHEDULES array's own comments for the full reasoning per entry.

Adding a scheduled job

Add payload type to types.ts:
```
'my:scan': Record<string, never>
```
Create handler in handlers/my-scan.ts. Use dynamic imports for @/lib/db to avoid client bundle leak:
```
import type { Job } from '../job-types';

export const handleMyScan = async (_job: Job<Record<string, never>>): Promise<void> => {
  const [{ db }, { users }] = await Promise.all([
    import('@/lib/db'),
    import('@/lib/db/schema')
  ]);
  // ... do work
};
```
Why dynamic imports? tick.server.ts statically imports every handler to build the HANDLERS map, and tick.server.ts itself gets pulled into both boot paths. Vite's import-protection blocks **/lib/db/** from reaching the client bundle. Dynamic imports inside the handler body are excluded from the client bundle trace.
Register in tick.server.ts's HANDLERS map:
```
'my:scan': handleMyScan
```

Add to SCHEDULES in scheduler.server.ts:

{ name: 'my-scan', type: 'my:scan', cron: '0 3 * * *', data: {}, overlap: false }

Admin UI

/dashboard/admin/scheduled-jobs shows all registered schedules with enabled/paused status, cron expression, last/next run times, and a pause/resume toggle per schedule.

Server fns: getSchedulesList (query), pauseScheduleAdmin / resumeScheduleAdmin (mutations) in src/features/jobs/server/.

Query key: adminSchedulesQueryKey(). Query options: schedulesQueryOptions exported from src/features/jobs/index.ts.

Pause / resume

import { pauseScheduleAdmin, resumeScheduleAdmin } from '@/features/jobs';

await pauseScheduleAdmin({ data: { name: 'reengagement-scan' } });
await resumeScheduleAdmin({ data: { name: 'reengagement-scan' } });

State persists in the schedules table across restarts and across Cron Trigger invocations, no code changes needed.

Delayed Jobs

Pass runAt in the options to schedule a job for future execution. claimBatch only claims rows whose runAt has passed.

import { enqueue } from '@/features/jobs/enqueue.server';

// Run 5 minutes from now , allows dependent state to settle first
await enqueue(
  'cleanup:expired-sessions',
  { userId },
  { runAt: new Date(Date.now() + 5 * 60 * 1000) }
);

When to use: cleanup or follow-up jobs that depend on external state settling before they run , DNS propagation, cache TTL expiry, a deployment completing elsewhere. Fire-and-forget: the caller returns immediately; the job runs once runAt arrives and the next tick claims it.

When not to use: jobs that must complete before the response returns. Those belong inline in the server function or in a domain event handler.

Email preference gating

withEmailPreferenceGate(settingKey, handler, testDb?) (src/features/jobs/handlers/with-email-preference-gate.ts) wraps a handler to skip it when the user has opted out of a given setting (checked via getSetting against the shared settings table). Default is opt-in , no setting row means send, so a new user isn't silently excluded before ever visiting settings. The testDb parameter is an injectable override (real-default pattern, not a mock.module seam) so tests can pass a withTestDb transaction handle instead of the ambient shared db.

'email:onboarding-day3': withEmailPreferenceGate('marketing_emails', handleOnboardingDay3),

Not every handler is gated this way , email:trial-expiring is intentionally ungated in tick.server.ts's HANDLERS map, since it communicates an impending loss of paid access rather than a promotional nudge.

Fire-and-Forget Async (IIFE Pattern)

For long-running work that doesn't need job-queue persistence, the IIFE pattern lets a server function return immediately while background work continues:

// In a server function, inside withUserTransaction (or right after it commits)
async function startLongOperation(userId: string, runId: string, runFn: RunFn) {
  await db.insert(runs).values({ id: runId, userId, status: 'running' });

  // Fire and forget , response returns, this continues in background
  (async () => {
    try {
      const result = await runFn();
      await db.update(runs).set({ status: 'completed', result }).where(eq(runs.id, runId));
    } catch (error: unknown) {
      const message = error instanceof Error ? error.message : String(error);
      await db.update(runs).set({ status: 'failed', error: message }).where(eq(runs.id, runId));
    }
  })();

  return ok({ runId });
}

The caller polls for completion by querying the run status (e.g. with refetchInterval in TanStack Query).

Use this when:

Work is long-running (seconds to minutes) but ephemeral , results written to DB
Failure is acceptable and recoverable , mark status 'failed' in the catch block
No retry logic needed , one attempt is enough
Work doesn't need to survive a server restart (and doesn't need to survive a Cloudflare Workers isolate ending mid-flight either , this pattern is self-hosted-only in practice, since a Workers invocation has no guarantee of continuing to run after the response is returned)

Use the job queue instead when:

Work must survive server restarts or isolate boundaries (jobs persist in Postgres)
Retry logic is needed (markJobFailed's retry, NonRetryableError for dead-letter)
Work needs scheduling (runAt) or rate limiting (tryConsumeRate)
Work must be auditable or inspectable (failed_jobs table, admin UI)

External Process Execution (Bun.spawn)

For CLI tools or system utilities on self-hosted, Bun.spawn is the standard , no npm package needed. Not available on Cloudflare Workers (workerd has no process-spawning API) , if a handler needs this, it can only run via self-hosted's poll loop caller, not the Workers Cron Trigger:

const proc = Bun.spawn(['dig', '+short', domain, 'A', '@1.1.1.1'], {
  stdout: 'pipe',
  stderr: 'pipe',
});
const output = await new Response(proc.stdout).text();
await proc.exited;
const ips = output.trim().split('\n').filter(line => /^\d+\.\d+\.\d+\.\d+$/.test(line));

Gate ordering: Bun.spawn is an external call , run it outside withUserTransaction. Write the result to the DB inside the transaction after the process completes. (Same rule as network calls , don't hold a transaction open while waiting for I/O.)