Queue & Workers

The @saasframe/queue package provides a multi-strategy job queue system for reliable background processing. It integrates seamlessly with the Events system to handle persistent events asynchronously.

Overview

The queue system supports two strategies:

Strategy	Backend	Use Case	Persistence
`local`	File system (JSON)	Development, single-instance deployments	`.queue/<name>/queue.json`
`async`	Redis (BullMQ)	Production, multi-instance deployments	Redis sorted sets

Quick Start

import { createQueue } from '@saasframe/queue'

// Create a queue
const queue = createQueue<{ userId: string }>('notifications', 'local')

// Add a job to the queue
await queue.enqueue({ userId: '123' })

// Process jobs
await queue.process(async (job, ctx) => {
  console.log(`Processing job ${ctx.jobId}:`, job.payload)
})

Queue Strategies

Local Strategy

The local strategy stores jobs in JSON files, making it ideal for development and simple deployments without Redis.

Production Warning

Never use the local strategy in production. It is filesystem-based and does not support distributed systems, concurrent access from multiple processes, or high availability. Always use the async (Redis/BullMQ) strategy for production deployments.

import { createQueue } from '@saasframe/queue'

const queue = createQueue<MyJobData>('my-queue', 'local', {
  baseDir: '.queue' // Optional, defaults to '.queue'
})

File structure:

.queue/
  my-queue/
    queue.json   # Array of queued jobs
    state.json   # Processing state (last processed ID)

Characteristics:

Synchronous processing via queue.process()
Jobs persist across restarts
No external dependencies
Single-process only (not suitable for distributed systems)

Async Strategy (BullMQ)

The async strategy uses Redis via BullMQ for production-grade job processing with concurrent workers.

import { createQueue } from '@saasframe/queue'

const queue = createQueue<MyJobData>('my-queue', 'async', {
  connection: {
    url: 'redis://localhost:6379',
    // Or use individual options:
    // host: 'localhost',
    // port: 6379,
    // password: 'secret'
  },
  concurrency: 5 // Number of concurrent job processors
})

Environment variables:

REDIS_URL or QUEUE_REDIS_URL - Redis connection URL (fallback if not provided in options)

Characteristics:

Persistent job storage in Redis
Automatic retries with exponential backoff
Concurrent job processing
Distributed across multiple instances
Job prioritization and scheduling

Queue Interface

All queues implement the same interface regardless of strategy:

interface Queue<T> {
  readonly name: string
  readonly strategy: 'local' | 'async'

  // Add a job to the queue
  enqueue(data: T): Promise<string>

  // Process jobs from the queue
  process(handler: JobHandler<T>, options?: ProcessOptions): Promise<ProcessResult>

  // Remove all jobs
  clear(): Promise<{ removed: number }>

  // Close the queue and release resources
  close(): Promise<void>

  // Get current job counts by status
  getJobCounts(): Promise<{
    waiting: number
    active: number
    completed: number
    failed: number
  }>
}

Job Handler

The job handler receives the job data and a context object:

type JobHandler<T> = (job: QueuedJob<T>, ctx: JobContext) => Promise<void> | void

type QueuedJob<T> = {
  id: string
  payload: T
  createdAt: string
  metadata?: Record<string, unknown>
}

type JobContext = {
  jobId: string
  attemptNumber: number  // 1-based attempt count
  queueName: string
}

Example:

await queue.process(async (job, ctx) => {
  console.log(`[${ctx.queueName}] Processing job ${ctx.jobId} (attempt ${ctx.attemptNumber})`)

  const { userId } = job.payload
  await sendNotification(userId)
})

Running Workers

For production deployments, run dedicated worker processes that continuously process jobs.

Unified Entrypoint (Development Only)

Open Saasframe provides a unified entrypoint that runs both the Next.js app and all queue workers automatically:

# Development mode with workers (default)
yarn dev

When you run this command, the system automatically:

Starts the Next.js application
Spawns a worker process that handles ALL discovered queues

Production Warning

Never use the unified entrypoint with auto-spawned workers in production. In production, you should:

Set AUTO_SPAWN_WORKERS=false
Run worker processes separately using yarn start:workers or yarn saasframe queue worker --all

This separation allows you to:

Scale workers independently from the web application
Deploy workers on dedicated machines
Restart workers without affecting the main application
Monitor and manage worker processes separately

Controlling Worker Auto-Spawn

Use the AUTO_SPAWN_WORKERS environment variable to control this behavior:

# .env
AUTO_SPAWN_WORKERS=true  # Default: spawn workers automatically

Set to false for production deployments where you want to run workers separately for scaling:

AUTO_SPAWN_WORKERS=false yarn start

SF_AUTO_SPAWN_WORKERS is an optional Open Saasframe-prefixed alias. The legacy AUTO_SPAWN_WORKERS always wins when both are set, so existing deployments do not change behavior.

Lazy Worker Auto-Spawn (Memory-Sensitive Dev)

The default eager mode starts a single saasframe queue worker --all process whose runner instantiates a per-queue runtime even when most queues have no jobs. Each runner adds polling timers (local) or BullMQ Worker + Redis resources (async) to the process tree, so the dev memory monitor's Memory ... RSS (peak ...) line scales with the number of enabled modules rather than actual workload.

Lazy mode replaces the eager worker with a lightweight watcher process. The watcher probes each discovered queue for ready jobs without importing handler code or creating BullMQ workers. By default, the first ready job on a queue triggers saasframe queue worker <queueName>; queues that never receive jobs stay completely idle.

For memory-sensitive local development, set SF_AUTO_SPAWN_WORKERS_LAZY_MODE=shared. In that mode the watcher still stays idle until the first ready job appears, but it starts one shared saasframe queue worker --all process instead of one process per active queue. This keeps queue behavior intact while avoiding several full CLI bootstraps when scheduler-driven jobs make many queues ready at once.

# Enable lazy auto-spawn when bypassing the dev wrapper
SF_AUTO_SPAWN_WORKERS_LAZY=true saasframe server dev

The monorepo and standalone yarn dev wrappers set SF_AUTO_SPAWN_WORKERS_LAZY=true, SF_AUTO_SPAWN_WORKERS_LAZY_MODE=shared, and SF_AUTO_SPAWN_SCHEDULER_LAZY=true automatically for local memory reduction. Direct saasframe server dev and production saasframe server start keep the eager default until these variables are set explicitly. Use SF_AUTO_SPAWN_WORKERS_LAZY=false SF_AUTO_SPAWN_SCHEDULER_LAZY=false yarn dev to force the historical eager mode.

Lazy mode env variables:

Variable	Default	Purpose
`SF_AUTO_SPAWN_WORKERS_LAZY`	`false`	Enables the lazy supervisor. Ignored when workers are disabled by `AUTO_SPAWN_WORKERS=false`.
`SF_AUTO_SPAWN_WORKERS_LAZY_MODE`	`per-queue`	`per-queue` starts one worker process per queue after that queue receives a job. `shared` starts one `queue worker --all` process after the first ready job.
`SF_AUTO_SPAWN_WORKERS_LAZY_POLL_MS`	`1000`	Probe interval. Clamped to a minimum of `250`.
`SF_AUTO_SPAWN_WORKERS_LAZY_RESTART`	`true`	Restart a per-queue worker if it exits unexpectedly while jobs remain pending.
`SF_AUTO_SPAWN_SCHEDULER_LAZY`	`false`	Enables lazy local scheduler startup. Ignored when the scheduler is disabled by `AUTO_SPAWN_SCHEDULER=false`.
`SF_AUTO_SPAWN_SCHEDULER_LAZY_POLL_MS`	`1000`	Enabled-schedule probe interval. Clamped to a minimum of `250`.
`SF_AUTO_SPAWN_SCHEDULER_LAZY_RESTART`	`true`	Restart the scheduler process if it exits unexpectedly while enabled schedules remain.

Trade-offs:

The first job on a cold queue pays one poll cycle plus the worker startup time before it starts processing.
Probes are read-only and never invoke handlers. A probe failure (filesystem error, Redis unreachable) is logged at most once per queue per minute and keeps the queue idle until the next successful probe.
Lazy mode does not change worker handler contracts. Existing workers/*.ts metadata, runWorker, and saasframe queue worker <queueName> continue to work unchanged.
Lazy scheduler mode does not change schedule execution. It still launches the existing saasframe scheduler start process after an enabled schedule exists.
Production deployments should still prefer AUTO_SPAWN_WORKERS=false plus separately managed worker processes — lazy mode targets memory-sensitive dev and small unified deployments.

The dev runtime's 🧠 Memory ... RSS (peak ...) line measures the process-tree RSS of the app runtime, so any auto-spawned workers and the scheduler are counted in that figure. Lazy mode reduces idle RSS because no per-queue runner exists until a job triggers one and no local scheduler process exists until a schedule is enabled. The number rises again as queues and the scheduler become active.

Available Scripts

Script	Description
`yarn dev`	Development mode with workers
`yarn start`	Production mode with workers
`yarn dev:app`	Development mode (app only)
`yarn start:app`	Production mode (app only)
`yarn start:workers`	Run workers only (all queues)

Running All Workers via CLI

The --all flag processes jobs from all discovered queues in a single process:

# Process all queues
yarn saasframe queue worker --all

# Via npm script
yarn start:workers

This starts workers for all registered queues (e.g., events, fulltext-indexing, vector-indexing).

Running a Single Queue

For fine-grained control, run workers for specific queues:

# Start a worker for a specific queue
yarn saasframe queue worker events

# With custom concurrency
yarn saasframe queue worker events --concurrency=5

Using `runWorker` Programmatically

import { runWorker } from '@saasframe/queue/worker'

await runWorker({
  queueName: 'events',
  handler: async (job, ctx) => {
    console.log(`Processing ${ctx.jobId}:`, job.payload)
  },
  connection: { url: process.env.REDIS_URL },
  concurrency: 5,
  gracefulShutdown: true // Handle SIGTERM/SIGINT
})

The worker will:

Connect to Redis and start a BullMQ worker
Process jobs continuously with the specified concurrency
Gracefully shutdown on SIGTERM/SIGINT signals

Routed Handlers

For queues with multiple job types, use createRoutedHandler:

import { runWorker, createRoutedHandler } from '@saasframe/queue/worker'

const handler = createRoutedHandler({
  'user.created': async (job, ctx) => {
    await sendWelcomeEmail(job.payload.email)
  },
  'order.placed': async (job, ctx) => {
    await notifyWarehouse(job.payload.orderId)
  },
  'payment.received': async (job, ctx) => {
    await updateAccountBalance(job.payload)
  },
})

await runWorker({
  queueName: 'events',
  handler,
  concurrency: 10,
})

Jobs must have a type field in their payload to route correctly.

Integration with Events

Persistent events (persistent: true) are automatically queued for async processing. The event system uses the queue package under the hood.

Subscriber with Persistent Events

// src/modules/orders/subscribers/order-created.ts
export const metadata = {
  event: 'order.created',
  persistent: true, // This event will be queued
}

export default async function handle(
  payload: { orderId: string; total: number },
  ctx: { resolve: <T>(name: string) => T }
) {
  const emailService = ctx.resolve('emailService')
  await emailService.sendOrderConfirmation(payload.orderId)
}

Processing Persistent Events

Persistent events are processed by running a queue worker via the CLI:

# Start a worker to process events continuously
yarn saasframe queue worker events

# With custom concurrency
yarn saasframe queue worker events --concurrency=5

# Check queue status
yarn saasframe queue status events

# Clear all queued events
yarn saasframe queue clear events

# Emit an event (for testing)
yarn saasframe events emit order.created '{"id":123}' --persistent

Configuration

Strategy Selection

Set the queue strategy via environment variable:

# Use local file-based queue (default)
EVENTS_STRATEGY=local

# Use Redis-backed queue
EVENTS_STRATEGY=redis

Multi-Instance Safety Guard

saasframe server start runs a boot-time guard that catches single-instance infrastructure strategies before they cause silent production incidents. Three defaults are only safe for a single process:

Env var	Single-instance default	Multi-instance-safe value
`CACHE_STRATEGY`	`memory` (stale ACLs after a privilege revocation)	`redis`
`QUEUE_STRATEGY`	`local` (duplicate job processing — emails, webhooks, indexing)	`async`
`RATE_LIMIT_STRATEGY`	`memory` (limits multiplied by the instance count)	`redis`

The guard is purely additive and opt-in via a topology hint:

Declare a multi-instance topology with SF_MULTI_INSTANCE=1 (or SF_INSTANCE_COUNT=<n> with n > 1). In production (NODE_ENV=production), start then refuses to boot while any strategy above is left on its single-instance value.
Without the hint, production logs a prominent warning but still starts, so existing single-instance deployments are unchanged.
Override with SF_ALLOW_SINGLE_INSTANCE_STRATEGIES=1 to downgrade the hard failure to a warning when you knowingly accept the risks.
Development and single-instance production boots are never affected.

Redis Configuration

# Primary Redis URL
REDIS_URL=redis://localhost:6379

# Or use events-specific URL
EVENTS_REDIS_URL=redis://events-redis:6379

# Queue-specific URL
QUEUE_REDIS_URL=redis://queue-redis:6379

Best Practices

1. Use Persistent Events for Side Effects

// Good: Email sending should be async
await bus.emitEvent('user.registered', { userId }, { persistent: true })

// Good: Inline for immediate state updates
await bus.emitEvent('cache.invalidated', { key }, { persistent: false })

2. Make Handlers Idempotent

Jobs may be retried on failure. Design handlers to be safely re-executed:

async function handlePayment(job, ctx) {
  const { paymentId } = job.payload

  // Check if already processed
  const existing = await db.findPayment(paymentId)
  if (existing.status === 'completed') {
    return // Already processed, skip
  }

  await processPayment(paymentId)
}

3. Set Appropriate Concurrency

// CPU-bound tasks: match CPU cores
createQueue('image-processing', 'async', { concurrency: 4 })

// I/O-bound tasks: higher concurrency
createQueue('api-calls', 'async', { concurrency: 20 })

// Rate-limited external APIs: lower concurrency
createQueue('email-sending', 'async', { concurrency: 2 })

4. Monitor Job Counts

const counts = await queue.getJobCounts()
console.log(`Waiting: ${counts.waiting}, Active: ${counts.active}, Failed: ${counts.failed}`)

if (counts.failed > 100) {
  alertOps('High failure rate in queue')
}

Type Definitions

// Queue strategy types
type QueueStrategyType = 'local' | 'async'

// Options for local strategy
type LocalQueueOptions = {
  baseDir?: string // Default: '.queue'
}

// Options for async strategy
type AsyncQueueOptions = {
  connection?: {
    url?: string
    host?: string
    port?: number
    password?: string
  }
  concurrency?: number // Default: 1
}

// Process options
type ProcessOptions = {
  limit?: number // Max jobs to process (local strategy only)
}

// Process result
type ProcessResult = {
  processed: number
  failed: number
  lastJobId?: string
}

Worker Auto-Discovery

Open Saasframe automatically discovers workers from modules following a naming convention. This allows you to define workers alongside your module code.

File Convention

Place worker files in your module's workers/ directory with the .worker.ts suffix:

src/modules/<module>/workers/
  └── <queue-name>.worker.ts

# or in packages
packages/<package>/src/modules/<module>/workers/
  └── <queue-name>.worker.ts

Worker File Structure

Each worker file must export:

metadata - Worker configuration (WorkerMeta type)
default - Handler function

// src/modules/notifications/workers/email.worker.ts
import type { QueuedJob, JobContext, WorkerMeta } from '@saasframe/queue'

// Required: Export metadata for auto-discovery
export const metadata: WorkerMeta = {
  queue: 'email-notifications',
  concurrency: parseInt(process.env.WORKERS_EMAIL_CONCURRENCY || '2', 10),
}

type EmailJobPayload = {
  to: string
  subject: string
  body: string
}

// Required: Export default handler function
export default async function handle(
  job: QueuedJob<EmailJobPayload>,
  ctx: JobContext
): Promise<void> {
  const { to, subject, body } = job.payload
  console.log(`[email-worker] Sending email to ${to}: ${subject}`)
  // Your job processing logic here
}

WorkerMeta Type

type WorkerMeta = {
  /** Queue name this worker processes */
  queue: string
  /** Optional unique identifier (defaults to <module>:workers:<filename>) */
  id?: string
  /** Worker concurrency (default: 1) */
  concurrency?: number
}

Per-Queue Concurrency

Control concurrency per queue via environment variables:

# Format: WORKERS_<QUEUE_NAME>_CONCURRENCY
WORKERS_EVENTS_CONCURRENCY=5
WORKERS_EMAIL_NOTIFICATIONS_CONCURRENCY=2
WORKERS_FULLTEXT_INDEXING_CONCURRENCY=10

Existing Workers

Open Saasframe includes these built-in workers:

Worker	Queue	Purpose
`events.worker.ts`	`events`	Dispatches persistent events to subscribers
`fulltext-index.worker.ts`	`fulltext-indexing`	Indexes documents for Meilisearch
`vector-index.worker.ts`	`vector-indexing`	Generates embeddings for vector search

Overview​

Quick Start​

Queue Strategies​

Local Strategy​

Async Strategy (BullMQ)​

Queue Interface​

Job Handler​

Running Workers​

Unified Entrypoint (Development Only)​

Controlling Worker Auto-Spawn​

Lazy Worker Auto-Spawn (Memory-Sensitive Dev)​

Available Scripts​

Running All Workers via CLI​

Running a Single Queue​

Using runWorker Programmatically​

Routed Handlers​

Integration with Events​

Subscriber with Persistent Events​

Processing Persistent Events​

Configuration​

Strategy Selection​

Multi-Instance Safety Guard​

Redis Configuration​

Best Practices​

1. Use Persistent Events for Side Effects​

2. Make Handlers Idempotent​

3. Set Appropriate Concurrency​

4. Monitor Job Counts​

Type Definitions​

Worker Auto-Discovery​

File Convention​

Worker File Structure​

WorkerMeta Type​

Per-Queue Concurrency​

Existing Workers​