Queue

@tulip-systems/queue is the low-level task execution atom. Use it when the application wants to say: do this work later, durably, and with retries handled by the adapter.

Queue tasks are command-like. A task represents an instruction, not a fact that already happened.

When to use Queue

Use Queue for work such as:

• sending an email
• processing an import
• syncing a file provider
• dispatching an event listener
• running slow side effects outside the request path

Do not use Queue as the domain language for facts such as project.created or invoice.paid. Those are events. Queue can still execute the work caused by those events.

Entry Points

Define tasks and task handler collections from the package entrypoint:

import { createQueueHandlers, defineTask } from "@tulip-systems/queue";

Create the server runtime from the server entrypoint:

import { Queue } from "@tulip-systems/queue/server";

Use the QStash adapter when jobs should be delivered through QStash:

import { qstashAdapter } from "@tulip-systems/queue/qstash";

Define a Task

Tasks are stable contracts. The task name is stored in queued jobs and used during callback dispatch.

// app/admin/queue/_lib/tasks.ts
import { defineTask } from "@tulip-systems/queue";
import { z } from "zod";

export const sendWelcomeEmailTask = defineTask("email.send-welcome", {
  schema: z.object({
    userId: z.uuid(),
    email: z.email(),
  }),
});

The schema validates payloads before publishing and again before executing the handler.

Initialize Queue

Create one queue runtime for the app. The adapter owns delivery. The queue runtime owns task payload validation and callback dispatch.

// server/queue/init.ts
import { url } from "@tulip-systems/core";
import { qstashAdapter } from "@tulip-systems/queue/qstash";
import { Queue } from "@tulip-systems/queue/server";

export const queue = Queue.init({
  adapter: qstashAdapter({
    baseUrl: process.env.QSTASH_URL!,
    token: process.env.QSTASH_TOKEN!,
    endpoint: url("/api/queue/dispatch"),
    currentSigningKey: process.env.QSTASH_CURRENT_SIGNING_KEY!,
    nextSigningKey: process.env.QSTASH_NEXT_SIGNING_KEY!,
    queueName: "app-jobs",
    label: "demo-code",
  }),
});

endpoint must be the public URL QStash will call. The same endpoint is used to verify QStash callback signatures.

Enqueue a Task

Call queue.enqueue(task, payload, options?) from server code.

import { queue } from "@/server/queue/init";
import { sendWelcomeEmailTask } from "./tasks";

await queue.enqueue(
  sendWelcomeEmailTask,
  {
    userId: user.id,
    email: user.email,
  },
  {
    dedupeKey: `welcome-${user.id}`,
    delay: "30s",
  },
);

dedupeKey and delay are passed to the adapter. With QStash, the dedupe key becomes the QStash deduplication id, so keep it provider-safe.

Handle Callbacks

Each queue callback route receives a collection of queue handlers.

// server/queue/handlers.ts
import { createQueueHandlers } from "@tulip-systems/queue";
import { sendWelcomeEmailTask } from "@/app/admin/queue/_lib/tasks";

export const queueHandlers = createQueueHandlers([
  sendWelcomeEmailTask.handle("send-welcome-email", async ({ payload, job }) => {
    await emailProvider.send({
      to: payload.email,
      subject: "Welcome",
      idempotencyKey: job.id,
    });
  }),
]);

Then expose the route handler:

// app/api/queue/dispatch/route.ts
import { queueHandlers } from "@/server/queue/handlers";
import { queue } from "@/server/queue/init";

export const POST = queue.handleCallback({ handlers: queueHandlers });

The callback flow is:

1. The adapter verifies the request.
2. Queue finds the handler by task name.
3. The task schema validates the job payload.
4. The handler receives { payload, job }.
5. A thrown error returns 500, allowing the adapter to retry.

One Handler Per Task

Queue tasks are command-like, so each task should have one handler in a callback route.

createQueueHandlers([
  sendWelcomeEmailTask.handle("send-welcome-email", async () => {}),
]);

If you need fan-out to multiple side effects, model the source as an event and use @tulip-systems/events to enqueue one listener job per side effect.

Queue vs Events

Queue is the durable execution layer. Events are the domain fact layer.

Failure Behavior

Callback failures are intentionally visible to the adapter:

• unknown task returns 404
• invalid payload returns 400
• handler failure returns 500

For retryable work, throw from the handler. For non-retryable domain decisions, handle the case inside the handler and return normally.

Current Adapter

Tulip currently ships with a QStash adapter:

import { qstashAdapter } from "@tulip-systems/queue/qstash";

The adapter supports:

• QStash publishing
• callback signature verification
• delay headers
• dedupe headers
• queue name headers
• labels for filtering logs

Related Pages