Getting Started

End-to-end setup: define a job, create a runtime, start workers, dispatch work.

1. Define a Job

// src/lib/jobs/sendEmails.job.ts
import { createJob } from "@repo/queue";
import { z } from "zod";

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

2. Create the Runtime

// src/lib/queue/runtime.server.ts
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,
  }),
  workers: [
    {
      job: sendEmailsJob,
      concurrency: 10,
      queues: ["low"],
    },
  ],
});

3. Start Workers

Call startWorkers() once at server startup. In SvelteKit, use hooks.server.ts:

// hooks.server.ts
import { queueRuntime } from "$lib/queue/runtime.server";

queueRuntime.startWorkers();

4. Dispatch Jobs

Create a client and dispatch type-safe payloads:

// src/lib/queue/queueClient.server.ts
import { queueRuntime } from "./runtime.server";

const queueClient = queueRuntime.createClient();

await queueClient.dispatch(sendEmailsJob, {
  email: "user@example.com",
  subject: "Welcome",
  body: "Hello!",
});

Dispatch accepts an optional third argument for advanced options:

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