Creating Workers

Workers are background processes that handle asynchronous job processing. Open Saasframe provides a worker discovery system that automatically registers workers defined in your modules.

Quick Start

Create a worker by adding a .worker.ts file to your module's workers/ directory:

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

export const metadata: WorkerMeta = {
  queue: 'my-queue',
  concurrency: 2,
}

export default async function handle(
  job: QueuedJob<{ message: string }>,
  ctx: JobContext
): Promise<void> {
  console.log(`Processing job ${ctx.jobId}:`, job.payload.message)
}

Run the worker:

# Run all workers (recommended)
yarn dev

# Or run specific queue
yarn saasframe queue worker my-queue

File Structure

Workers follow a strict naming convention for auto-discovery:

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

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

The file name (minus .worker.ts) typically matches the queue name, though you can specify any queue name in the metadata.

Worker Anatomy

Every worker file must export two things:

1. Metadata Export

import type { WorkerMeta } from '@saasframe/queue'

export const metadata: WorkerMeta = {
  queue: 'my-queue',           // Required: queue name to process
  id: 'my-module:my-worker',   // Optional: unique ID (auto-generated if omitted)
  concurrency: 5,              // Optional: parallel job processing (default: 1)
}

2. Default Handler Function

import type { QueuedJob, JobContext } from '@saasframe/queue'

export default async function handle(
  job: QueuedJob<MyPayloadType>,
  ctx: JobContext
): Promise<void> {
  // Process the job
}

Job and Context Types

QueuedJob

type QueuedJob<T> = {
  id: string           // Unique job identifier
  payload: T           // Your job data
  createdAt: string    // ISO timestamp
  metadata?: Record<string, unknown>
}

JobContext

type JobContext = {
  jobId: string        // Same as job.id
  attemptNumber: number // 1-based retry count
  queueName: string    // Queue being processed
  resolve: <T>(name: string) => T // DI container access
}

Accessing Services via DI

Use ctx.resolve() to access services from the dependency injection container:

export default async function handle(
  job: QueuedJob<{ userId: string }>,
  ctx: JobContext & { resolve: <T>(name: string) => T }
): Promise<void> {
  // Resolve services from DI container
  const emailService = ctx.resolve<EmailService>('emailService')
  const userRepo = ctx.resolve<UserRepository>('userRepository')

  const user = await userRepo.findById(job.payload.userId)
  await emailService.sendWelcome(user.email)
}

Environment Variables

Per-Queue Concurrency

Override default concurrency via environment variables:

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

Use dashes converted to underscores for queue names with hyphens.

Auto-Spawn Control

# Enable/disable automatic worker spawning (default: true)
AUTO_SPAWN_WORKERS=true

# Set to false to run workers separately
AUTO_SPAWN_WORKERS=false

Running Workers

Unified Entrypoint (Development Only)

By default, yarn dev automatically spawns workers for all queues:

# Start app + all workers (development)
yarn dev

# Start app only (no workers)
AUTO_SPAWN_WORKERS=false yarn dev

Production Warning

Never use auto-spawned workers in production. The unified entrypoint is designed for development convenience only. In production:

Set AUTO_SPAWN_WORKERS=false in your environment
Run workers as separate processes using yarn start:workers

This allows proper scaling, monitoring, and independent restarts of worker processes.

CLI Commands

# Run all discovered workers
yarn saasframe queue worker --all
yarn start:workers

# Run specific queue
yarn saasframe queue worker events
yarn saasframe queue worker fulltext-indexing --concurrency=10

# Check queue status
yarn saasframe queue status events

# Clear queue
yarn saasframe queue clear events

Production Deployment

For production, run workers separately for better scaling:

# Main app (disable auto-spawn)
AUTO_SPAWN_WORKERS=false yarn start

# Worker process (can run multiple)
yarn start:workers

# Or specific queues on different machines
yarn saasframe queue worker events --concurrency=10
yarn saasframe queue worker fulltext-indexing --concurrency=5

Complete Example

Here's a full example of a notification worker:

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

// Queue name constant for reuse
export const NOTIFICATIONS_QUEUE = 'notifications'

// Read concurrency from environment
const envConcurrency = process.env.WORKERS_NOTIFICATIONS_CONCURRENCY
const DEFAULT_CONCURRENCY = 3

// Metadata for auto-discovery
export const metadata: WorkerMeta = {
  queue: NOTIFICATIONS_QUEUE,
  concurrency: envConcurrency ? parseInt(envConcurrency, 10) : DEFAULT_CONCURRENCY,
}

// Job payload type
type NotificationPayload = {
  type: 'email' | 'sms' | 'push'
  userId: string
  template: string
  data: Record<string, unknown>
}

// Extended context with DI access
type HandlerContext = JobContext & {
  resolve: <T>(name: string) => T
}

// Main handler function
export default async function handle(
  job: QueuedJob<NotificationPayload>,
  ctx: HandlerContext
): Promise<void> {
  const { type, userId, template, data } = job.payload

  console.log(`[notifications] Processing ${type} notification for user ${userId}`)

  // Resolve services from DI
  const notificationService = ctx.resolve<NotificationService>('notificationService')

  try {
    switch (type) {
      case 'email':
        await notificationService.sendEmail(userId, template, data)
        break
      case 'sms':
        await notificationService.sendSms(userId, template, data)
        break
      case 'push':
        await notificationService.sendPush(userId, template, data)
        break
    }

    console.log(`[notifications] Successfully sent ${type} to user ${userId}`)
  } catch (error) {
    console.error(`[notifications] Failed to send ${type} to user ${userId}:`, error)
    throw error // Re-throw to trigger retry
  }
}

Enqueueing Jobs

To add jobs to your queue, use the queue factory:

import { createQueue } from '@saasframe/queue'
import { NOTIFICATIONS_QUEUE } from './workers/notifications.worker'

// Get queue strategy from environment
const strategy = process.env.QUEUE_STRATEGY === 'async' ? 'async' : 'local'

const queue = createQueue<NotificationPayload>(NOTIFICATIONS_QUEUE, strategy, {
  connection: { url: process.env.REDIS_URL },
})

// Enqueue a job
await queue.enqueue({
  type: 'email',
  userId: '123',
  template: 'welcome',
  data: { name: 'John' },
})

Production Warning

Never use the local queue strategy in production. The local strategy is filesystem-based and:

Does not support distributed systems
Cannot handle concurrent access from multiple processes
Provides no high availability or persistence guarantees

Always use QUEUE_STRATEGY=async with Redis in production environments.

Best Practices

1. Make Handlers Idempotent

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

export default async function handle(job, ctx) {
  const { paymentId } = job.payload

  // Check if already processed
  const payment = await paymentRepo.findById(paymentId)
  if (payment.status === 'completed') {
    return // Skip - already processed
  }

  await processPayment(paymentId)
}

2. Handle Errors Gracefully

export default async function handle(job, ctx) {
  try {
    await processJob(job.payload)
  } catch (error) {
    if (error instanceof RetryableError) {
      throw error // Will be retried
    }

    // Log and swallow non-retryable errors
    console.error('Non-retryable error:', error)
  }
}

3. Use Appropriate Concurrency

// CPU-bound: match CPU cores
export const metadata: WorkerMeta = { queue: 'image-processing', concurrency: 4 }

// I/O-bound: higher concurrency
export const metadata: WorkerMeta = { queue: 'api-calls', concurrency: 20 }

// Rate-limited APIs: lower concurrency
export const metadata: WorkerMeta = { queue: 'email-sending', concurrency: 2 }

4. Log Progress

export default async function handle(job, ctx) {
  console.log(`[${ctx.queueName}] Starting job ${ctx.jobId} (attempt ${ctx.attemptNumber})`)

  await processJob(job.payload)

  console.log(`[${ctx.queueName}] Completed job ${ctx.jobId}`)
}

Built-in Workers

Open Saasframe includes these workers:

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

Queue & Workers Overview - Queue strategies and configuration
Events System - Event publishing and subscribers

Quick Start​

File Structure​

Worker Anatomy​

1. Metadata Export​

2. Default Handler Function​

Job and Context Types​

QueuedJob​

JobContext​

Accessing Services via DI​

Environment Variables​

Per-Queue Concurrency​

Auto-Spawn Control​

Running Workers​

Unified Entrypoint (Development Only)​

CLI Commands​

Production Deployment​

Complete Example​

Enqueueing Jobs​

Best Practices​

1. Make Handlers Idempotent​

2. Handle Errors Gracefully​

3. Use Appropriate Concurrency​

4. Log Progress​

Built-in Workers​

Related Documentation​