@torkjs/queue

Ergonomic BullMQ integration for Tork — Redis required, type-native, no decorators. Workers are declared as service classes (same shape as Tork cron services): static queue + static jobs, with DI via Injectable.

bun add @torkjs/queue bullmq

Worker services

import { Injectable, Logger } from "@torkjs/core";
import { type JobMap, type Job } from "@torkjs/queue";
import { Mailer } from "./mailer.ts";

class EmailWorker extends Injectable({ mailer: Mailer, log: Logger }) {
  static queue = "emails";                          // string, or { name, concurrency, ... }
  static jobs = {                                    // method → job name ("*" = catch-all)
    welcome: "welcome",
    any: "*",
  } satisfies JobMap<EmailWorker>;

  async welcome(job: Job<{ to: string }>) {
    await this.mailer.send(job.data.to);             // injected dep
    this.log.info({ to: job.data.to }, "welcome sent");
  }
  async any(job: Job) { /* handles every other job in the queue */ }
}

Bind workers to the app lifecycle (built on start, gracefully closed on shutdown):

import { queue } from "@torkjs/queue/tork";
app.use(queue({ workers: [EmailWorker] }));

A thrown handler error is not swallowed — it propagates to BullMQ, whose retry/backoff handles it (failures are also logged via the app Logger).

Producers

import { createQueue } from "@torkjs/queue";

const emails = createQueue<{ to: string }>("emails");
await emails.add("welcome", { to: "[email protected]" }, {
  attempts: 3,
  backoff: { type: "exponential", delay: 1000 },
});

Connection

BullMQ runs on Bun's native redis client (createBunRedisClient — no ioredis):

import { bunConnection, createQueue } from "@torkjs/queue";
const q = createQueue("emails", { connection: bunConnection(process.env.REDIS_URL) });

bunConnection() defaults to REDIS_URL, then redis://localhost:6379.