@repo/queue

A type-safe, schema-driven queue abstraction for Node.js.

• Job definitions as values
• Payload types inferred from schemas
• Queue-agnostic core with adapter-based backends
• Separate client and worker concerns
• Inspired by Laravel queues, built for TypeScript

Installation

npm install @repo/queue

Install a queue backend adapter as needed:

npm install pg-boss

Defining a Job

import { createJob } from "@repo/queue";
import { z } from "zod";

export const sendEmailsJob = createJob({
  name: "send-emails",
  input: z.object({
    email: z.string().email(),
    subject: z.string(),
    body: z.string(),
  }),
  dedupeKey: payload => `email:${payload.email}:${payload.subject}`,
  handler: async ({ email, subject, body }, ctx) => {
    // payload is fully typed
  },
  hooks: {
    onError: (err, payload, ctx) => {},
    onFailed: (err, payload, ctx) => {},
    onSuccess: (result, payload, ctx) => {},
  },
});

• payload type is inferred from the schema
• handler receives parsed/validated input
• dedupeKey is optional and can be static or derived from the typed payload
• hooks are fully typed
• onSuccess hook failures are logged and do not retry an already-successful job

Creating the Queue Runtime

import { getGlobalDb } from '$lib/db/managedDb.server'
import { createQueueRuntime } from "@repo/queue";
import { createPgBossAdapter } from "@repo/queue/adapters/pg-boss";
import { sendEmailsJob } from "./jobs/sendEmails.job";

export const queueRuntime = createQueueRuntime({
  queues: [
    { name: "low", priority: 0 },
    { name: "high", priority: 10 },
  ],
  adapter: createPgBossAdapter({
    connectionString: process.env.DATABASE_URL,
    getDb: getGlobalDb,
  }),
  workers: [
    {
      job: sendEmailsJob,
      concurrency: 10,
      queues: ["low"],
    },
  ],
  dev: {
    hot: import.meta.hot,
  },
});

Pass import.meta.hot in development modules that create the runtime. When Vite reloads that module, the queue runtime stops old workers before the new runtime starts fresh workers with updated job handlers. In dev, createJob() also preserves job object identity by name so existing runtime references can use updated handlers after a job module reload. If hot is missing, no HMR cleanup/restart is registered.

Running Workers

Start listening to the workers in Svelte.

// hooks.server.ts 
queueRuntime.startWorkers();

startWorkers() records that workers should be running. If the queue backend is unavailable during startup, the runtime logs the error and retries every 10 seconds until the workers start or stopWorkers() is called. The returned promise settles after the first attempt, so do not use it as a queue health check.

Queue UI

The package includes a superadmin-only queue UI for SvelteKit apps. It exposes named remote queries/commands and package-owned list/detail components.

// src/lib/queue/ui.remote.ts
import { queueRuntime } from '$lib/queue/runtime.server'
import { createQueueUiClient } from '@repo/queue/server'

export const {
  getActionableQueueJobsQuery,
  getQueueUiConfigQuery,
  getQueueJobDetailQuery,
  retryQueueJobCommand,
} = createQueueUiClient({
  runtime: queueRuntime,
})

<script lang="ts">
  import { getActionableQueueJobsQuery, getQueueUiConfigQuery, retryQueueJobCommand } from '$lib/queue/ui.remote'
  import { QueueUiList } from '@repo/queue/ui'
</script>

<QueueUiList {getActionableQueueJobsQuery} {getQueueUiConfigQuery} {retryQueueJobCommand} />

<script lang="ts">
  import { getQueueUiConfigQuery, getQueueJobDetailQuery, retryQueueJobCommand } from '$lib/queue/ui.remote'
  import { QueueUiDetail } from '@repo/queue/ui'

  const { params } = $props()
</script>

<QueueUiDetail jobId={params.id} {getQueueUiConfigQuery} {getQueueJobDetailQuery} {retryQueueJobCommand} />

• Access is restricted with isSuperadminGate and notImpersonatingGate.
• The overview is a URL-state DataGrid backed by adapter-owned job queries.
• The pg-boss adapter requires getDb for queue UI job data.
• The pg-boss adapter shows actionable jobs: created, retry, and failed.
• Failed jobs can be retried from the overview and detail header.
• Attempts are displayed as human attempts/max attempts. For pg-boss, this means retryCount + 1 / retryLimit + 1 after a job has started, and 0/maxAttempts while created.
• Clicking a job opens a job detail page with metadata, payload, and output.
• Filtering, sorting, and pagination use @repo/table/remote URL state.
• The overview filters queue, job, and state with dropdown options from the queue runtime config.
• QueueDef.id is optional and defaults to name.

Dispatching Jobs (Client)

// src/lib/queue/queueClient.server.ts
const queueClient = queueRuntime.createClient();

await queueClient.dispatch(
  sendEmailsJob,
  { email, subject, body },
);

• Dispatch is type-safe
• Payload must match the job schema
• Queue names are validated at compile time and before publish
• By default, dispatch captures the current @repo/auth session
• Workers run jobs inside the captured session context

There is a third argument you can pass for more context:

await queueClient.dispatch(
  sendEmailsJob,
  { email, subject, body },
  { maxAttempts: 5, queue: "low" }
);

Disable session capture for jobs that should run without a user session:

await queueClient.dispatch(
  sendEmailsJob,
  { email, subject, body },
  { session: false }
);

Jobs dispatched without a session still run inside an explicit no-session context. Inside the worker, getSession({ throwError: false }) returns undefined and getSession() throws a 401 instead of reading SvelteKit request state.

Use dedupeKey to prevent duplicate queued jobs. You can define it on the job, or override it per dispatch. Job-level keys can be static or derived from the typed payload:

export const sendEmailsJob = createJob({
  name: "send-emails",
  input: sendEmailsSchema,
  dedupeKey: payload => `email:${payload.email}:${payload.subject}`,
  handler: async (payload, ctx) => {},
});

export const dailyDigestJob = createJob({
  name: "daily-digest",
  input: dailyDigestSchema,
  dedupeKey: "daily-digest",
  handler: async (payload, ctx) => {},
});

await queueClient.dispatch(
  sendEmailsJob,
  { email, subject, body },
  { dedupeKey: `email:${email}:${subject}` }
);

Dispatching Jobs (Remote Form)

Create a SvelteKit remote form directly from a job:

export const sendEmailsForm = queueClient.jobAsForm(sendEmailsJob, {
  dedupeKey: "send-emails-form",
});

• Uses the job input schema as the form schema
• Dispatch options are the same typed options as dispatch
• The form handler dispatches the job and returns no value

Batch Dispatching

Dispatch multiple payloads with staggered delays to avoid overwhelming external APIs:

await queueClient.dispatchBatch(
  sendEmailsJob,
  users.map(u => ({ email: u.email, subject: "Hello", body: "..." })),
  { batchSize: 5, delayBetween: '10s' }
);

• First batch (5 jobs) dispatched immediately
• Second batch delayed 10s, third 20s, etc.
• If delay is also set, the batch delay is added to it
• batchSize defaults to 5, delayBetween defaults to '10s'
• Accepts all DispatchOptions (queue, maxAttempts, dedupeKey, etc.)
• delay and delayBetween support number (seconds) or string units: ms, s, m, h, d

Adapters

The core is queue-agnostic. Backends are provided via adapters.

Currently supported:

• pg-boss

Adapters are optional peer dependencies and only required if you use them.

Design Notes

• Jobs are first-class values, not strings
• Schemas define both runtime validation and TypeScript types
• No decorators, no DI container, no global registry
• Client and worker responsibilities are explicit