Documentation - Packages - @vorsteh-queue/adapter-drizzle

The Drizzle adapter provides PostgreSQL support using Drizzle ORM, offering excellent TypeScript integration and performance.

Installation #

npm install @vorsteh-queue/core @vorsteh-queue/adapter-drizzle

pnpm add @vorsteh-queue/core @vorsteh-queue/adapter-drizzle

bun add @vorsteh-queue/core @vorsteh-queue/adapter-drizzle

yarn add @vorsteh-queue/core @vorsteh-queue/adapter-drizzle

Quick Start #

import { drizzle } from "drizzle-orm/node-postgres"
import { Pool } from "pg"

import { PostgresQueueAdapter } from "@vorsteh-queue/adapter-drizzle"
import { Queue } from "@vorsteh-queue/core"

const pool = new Pool({ connectionString: "postgresql://..." })
const db = drizzle(pool)
const adapter = new PostgresQueueAdapter(db)
const queue = new Queue(adapter)

Supported providers #

PGlite
Postgres.JS
Node Progress

Database Setup #

Schema #

The adapter includes a pre-defined schema:

import { queueJobsTable } from "@vorsteh-queue/adapter-drizzle"

// Use in your schema file
export { queueJobsTable }

If you don't want to use the pre-defined schema, you can create your own schema definition.

queue-jobs.ts

import { sql } from "drizzle-orm"
import { index, integer, jsonb, pgTable, timestamp, uuid, varchar } from "drizzle-orm/pg-core"

export const queueJobs = pgTable(
  "queue_jobs",
  {
    id: uuid("id")
      .ryKey()
      .default(sql`gen_random_uuid()`),
    queueName: varchar("queue_name", { length: 255 }).notNull(),
    name: varchar("name", { length: 255 }).notNull(),
    payload: jsonb("payload").notNull(),
    status: varchar("status", { length: 50 }).notNull(),
    priority: integer("priority").notNull(),
    attempts: integer("attempts").default(0).notNull(),
    maxAttempts: integer("max_attempts").notNull(),
    createdAt: timestamp("created_at", { withTimezone: true, mode: "date" })
      .notNull()
      .default(sql`timezone('utc', now())`),
    processAt: timestamp("process_at", { withTimezone: true, mode: "date" }).notNull(),
    processedAt: timestamp("processed_at", { withTimezone: true, mode: "date" }),
    ompletedAt: timestamp("completed_at", { withTimezone: true, mode: "date" }),
    failedAt: timestamp("failed_at", { withTimezone: true, mode: "date" }),
    error: jsonb("error"),
    result: jsonb("result"),
    progress: integer("progress").default(0),
    cron: varchar("cron", { length: 255 }),
    repeatEvery: integer("repeat_every"),
    repeatLimit: integer("repeat_limit"),
    repeatCount: integer("repeat_count").default(0),
    timeout: jsonb("timeout"),
  },
  (table) => [
    index("idx_queue_jobs_status_priority").on(
      table.queueName,
      table.status,
      table.priority,
      table.createdAt,
    ),
    index("idx_queue_jobs_process_at").on(table.processAt),
  ],
)

export type QueueJob = typeof queueJobs.$inferSelect
export type InsertQueueJob = typeof queueJobs.$inferInsert

Migration #

To run the migrations, we recommend to use drizzle-kit.

Drizzle Kit is a CLI tool for managing SQL database migrations with Drizzle.

Drizzle Adapter