SERVER ECOSYSTEM

Background Queues

@rasenganjs/queue adds background job queues — the Controller equivalent for async work, mirroring @rasenganjs/ws's Gateway pattern exactly.

Terminal
npm install @rasenganjs/queue

Registering the Plugin

main.ts
import { bootstrap } from '@rasenganjs/server';
import { createQueuePlugin } from '@rasenganjs/queue';
import appModule from './app.module';
 
bootstrap((app) => {
  app.registerPlugin(createQueuePlugin()); // defaults to MemoryQueueAdapter
  app.registerModule(appModule); // may declare queues: [EmailQueue]
});

Writing a Queue

email.queue.ts
import { Queue, JobRouter, type JobHandler } from '@rasenganjs/queue';
 
export class EmailQueue extends Queue {
  name = 'emails';
 
  constructor(private mailer: MailerService) {
    super();
  }
 
  jobs(router: JobRouter) {
    router.process('welcome', this.sendWelcome, {
      attempts: 3,
      backoff: 5_000,
    });
  }
 
  sendWelcome: JobHandler<{ userId: string }> = async (job) => {
    await this.mailer.sendWelcome(job.data.userId);
    // resolve -> job completes; throw -> retried per backoff policy
  };
}
queue.module.ts
import { defineModule } from '@rasenganjs/server';
import { EmailQueue } from './email.queue';
 
export default defineModule({
  queues: [EmailQueue],
  providers: [MailerService],
});

Like Gateway, Queue extends Provider — constructor injection and lifecycle hooks work identically to any other provider.

Enqueuing Jobs

Inject the queue anywhere it's exported and visible, and call .add():

Enqueuing from an HTTP controller
export class SignupController extends Controller {
  constructor(private emailQueue: EmailQueue) {
    super();
  }
 
  routes(router: Router) {
    router.post('/signup', this.signup);
  }
 
  signup: RouteHandler = async (ctx) => {
    const user = await createUser(ctx.body);
    await this.emailQueue.add('welcome', { userId: user.id });
    return ctx.res.json(user);
  };
}

Delayed and Recurring Jobs

Delayed job
await this.emailQueue.add('reminder', { userId }, { delay: 60_000 }); // fires in 1 minute
Recurring job
await this.reportQueue.add(
  'dailyDigest',
  {},
  { repeat: { every: 86_400_000 } }
); // every 24h

A { repeat } registration is idempotent by jobKey (derived from the job name + data, or an explicit repeat.key) — calling .add() again with the same spec doesn't create a duplicate schedule. It also resolves with the recurring job's jobKey instead of a fresh id.

Retry Semantics

At-least-once processing: resolving the handler completes the job, throwing retries with exponential backoff until attempts is exhausted, then the job moves to a dead-letter state.

Inspecting and retrying dead jobs
const dead = await this.emailQueue.getDead();
await this.emailQueue.retryDead(dead[0].id); // resets attempt count to 1, back to waiting

ProcessOptions

OptionDefaultDescription
attempts1Max attempts before dead-lettering
backoff0Base delay (ms) for exponential retry backoff
concurrency1Max jobs of this name processed concurrently

Scaling with Redis

MemoryQueueAdapter (the default) is in-process and loses jobs on restart — dev only. RedisQueueAdapter persists across restarts and is safe across multiple processes sharing one queue:

Redis-backed queue
import { createQueuePlugin, RedisQueueAdapter } from '@rasenganjs/queue';
import Redis from 'ioredis';
 
const client = new Redis();
const adapter = new RedisQueueAdapter({
  client,
  blockingClient: client.duplicate(), // reserve()'s BLMOVE needs its own connection
});
 
app.registerPlugin(createQueuePlugin({ adapter }));

QueuePluginOptions

OptionDefaultDescription
adapterMemoryQueueAdapterJob storage shared by every queue this plugin registers
workertruePass false for a produce-only process — .add() still works, nothing is ever reserved/processed there
stallTimeout30_000msHow long a reserved job may go without completing before the sweeper reclaims it as stalled
sweepInterval5_000msHow often the sweeper promotes due delayed/repeat jobs and reclaims stalled reservations

worker: false is a deployment-topology choice, not a code change — useful for splitting a producer-only web process from dedicated worker processes that actually run job handlers.

WebSocket Gateways
Validation Adapters