@repo/vqs -- Vercel Queue System

A job queueing abstraction built on Vercel Queues. Provides deduplication, status tracking, and retry logic out of the box.

Core Concepts

Jobbable

Jobbable<TParams, TResult> is the abstract base class for all queue jobs. Extend it to define a job:

import { z } from "zod";
import { Jobbable, type JobConfig } from "@repo/vqs";

const syncPatientParamsSchema = z.object({
  patientId: z.number(),
});

type SyncPatientParams = z.infer<typeof syncPatientParamsSchema>;

export class SyncPatientJobbable extends Jobbable<SyncPatientParams, void> {
  readonly config: JobConfig = {
    topic: "sync-patient",       // Vercel Queue topic name (must be unique)
    maxAttempts: 3,              // Max delivery attempts before acknowledging
    visibilityTimeoutSeconds: 300, // How long the message is hidden after delivery
    ttlHours: 24,                // How long the dedup key stays active
  };

  validate(params: unknown): params is SyncPatientParams {
    return syncPatientParamsSchema.safeParse(params).success;
  }

  jobKey(params: SyncPatientParams): string | null {
    return `sync-patient:${params.patientId}`;
  }

  async run(params: SyncPatientParams): Promise<void> {
    // Your job logic here
  }
}
Copy

Required Methods

Optional Methods

Enqueuing Jobs

Call performLater() to enqueue a job:

const job = new SyncPatientJobbable();

// Basic enqueue
await job.performLater({ patientId: 123 });

// With options
await job.performLater({ patientId: 123 }, {
  delaySeconds: 60,           // Delay delivery
  idempotencyKey: "custom",   // Override the idempotency key
});
Copy

performLater handles deduplication automatically -- if a job with the same jobKey already exists and hasn't expired (based on ttlHours), the enqueue is skipped.

createQueueHandler

Wires a Jobbable to a Next.js route handler for Vercel Queues:

// app/api/queue/sync-patient/route.ts
import { createQueueHandler } from "@repo/vqs";
import { SyncPatientJobbable } from "~/shared/jobs/sync-patient.jobbable";

export const POST = createQueueHandler(SyncPatientJobbable);
Copy

The handler takes care of:

• Validating the incoming message
• Running the job with status tracking (running -> success / failed)
• Retry logic with exponential backoff
• Error logging

Adding a New Job

1. Create the Jobbable in apps/web/shared/jobs/<name>.jobbable.ts
2. Create the route at apps/web/app/api/queue/<topic>/route.ts:

   import { createQueueHandler } from "@repo/vqs";
   import { MyJobbable } from "~/shared/jobs/my.jobbable";
   
   export const POST = createQueueHandler(MyJobbable);
   Copy

3. Register in vercel.json (apps/web/vercel.json):

   {
     "functions": {
       "app/api/queue/<topic>/route.ts": {
         "experimentalTriggers": [
           {
             "type": "queue/v2beta",
             "topic": "<topic>",
             "retryAfterSeconds": 300,
             "initialDelaySeconds": 0,
             "maxDeliveries": 3,
             "maxConcurrency": 50
           }
         ]
       }
     }
   }
   Copy

Important: Vercel Queues requires one topic per route file. Each Jobbable gets its own route.

Queue Drivers (VQS_DRIVER)

performLater hands the message to a QueueDriver chosen once per process from VQS_DRIVER:

VQS_DRIVER=local is for local dev and E2E, where there is no Vercel Queue or OIDC token to call back into. It replaces vercel env pull for local job runs. Semantics differ from the real queue:

• single delivery, no retries (maxAttempts is ignored; a failing job is logged as Local job failed and not rethrown)
• delaySeconds is ignored (logged at info)
• dedup via job_executions and status tracking work exactly as in production
• a job that enqueues another job runs the child through the same driver

The local driver is refused when VERCEL_ENV is production or preview: assertLocalDriverAllowed (exported from @repo/vqs/driver-env) throws LocalQueueDriverForbiddenError at boot from apps/web/instrumentation.ts and again on the first performLater.

Retry Behavior

When a job fails, retryWithMaxAttempts determines the backoff:

Formula: (10 * deliveryCount)^2 seconds. After maxAttempts is reached, the message is acknowledged (deleted).

Status Tracking

All jobs are tracked in the job_executions table:

• Enqueue: Records enqueued_id, key, payload, optional entityId, and ttlHours
• Running: Status set to "running" when the handler picks up the message
• Success/Failed: Status updated after run() completes or throws