Building a Job Queue System

1import { SolidisFeaturedClient } from '@vcms-io/solidis/featured';
2import { randomUUID } from 'node:crypto';
3
4export interface Job<T = any> {
5  id: string;
6  type: string;
7  data: T;
8  priority: number;
9  attempts: number;
10  maxAttempts: number;
11  createdAt: number;
12  processedAt?: number;
13}
14
15export class JobQueue {
16  private client: SolidisFeaturedClient;
17  private queueName: string;
18  private processingName: string;
19  private deadLetterName: string;
20
21  constructor(options: { host?: string; port?: number; queueName?: string } = {}) {
22    this.client = new SolidisFeaturedClient({
23      host: options.host || '127.0.0.1',
24      port: options.port || 6379,
25    });
26    this.queueName = options.queueName || 'jobs';
27    this.processingName = `${this.queueName}:processing`;
28    this.deadLetterName = `${this.queueName}:dead`;
29  }
30
31  async connect(): Promise<void> {
32    await this.client.connect();
33  }
34
35  disconnect(): void {
36    this.client.quit();
37  }
38
39  /**
40   * Add a job to the queue
41   */
42  async addJob<T>(
43    type: string,
44    data: T,
45    options: { priority?: number; maxAttempts?: number } = {}
46  ): Promise<string> {
47    const job: Job<T> = {
48      id: randomUUID(),
49      type,
50      data,
51      priority: options.priority || 0,
52      attempts: 0,
53      maxAttempts: options.maxAttempts || 3,
54      createdAt: Date.now(),
55    };
56
57    // Add to sorted set with priority as score (higher = more priority)
58    await this.client.zadd(this.queueName, job.priority, JSON.stringify(job));
59
60    return job.id;
61  }
62
63  /**
64   * Get next job from queue
65   */
66  async getNextJob(): Promise<Job | null> {
67    // Get highest priority job
68    const result = await this.client.zpopmax(this.queueName);
69
70    if (!result || result.length === 0) {
71      return null;
72    }
73
74    const job = JSON.parse(result[0].member) as Job;
75
76    // Move to processing queue
77    await this.client.hset(this.processingName, job.id, JSON.stringify(job));
78
79    return job;
80  }
81
82  /**
83   * Complete a job
84   */
85  async completeJob(jobId: string): Promise<void> {
86    await this.client.hdel(this.processingName, jobId);
87  }
88
89  /**
90   * Fail a job (retry or move to dead letter)
91   */
92  async failJob(jobId: string, error: string): Promise<void> {
93    const jobData = await this.client.hget(this.processingName, jobId);
94
95    if (!jobData) {
96      return;
97    }
98
99    const job = JSON.parse(jobData) as Job;
100    job.attempts++;
101
102    if (job.attempts >= job.maxAttempts) {
103      // Move to dead letter queue
104      await this.client.lpush(
105        this.deadLetterName,
106        JSON.stringify({ ...job, error })
107      );
108      await this.client.hdel(this.processingName, jobId);
109    } else {
110      // Retry: move back to main queue
111      await this.client.zadd(this.queueName, job.priority, JSON.stringify(job));
112      await this.client.hdel(this.processingName, jobId);
113    }
114  }
115
116  /**
117   * Get queue statistics
118   */
119  async getStats() {
120    const waiting = await this.client.zcard(this.queueName);
121    const processing = await this.client.hlen(this.processingName);
122    const dead = await this.client.llen(this.deadLetterName);
123
124    return { waiting, processing, dead };
125  }
126
127  /**
128   * Clear dead letter queue
129   */
130  async clearDeadLetters(): Promise<number> {
131    const length = await this.client.llen(this.deadLetterName);
132    if (length > 0) {
133      await this.client.del(this.deadLetterName);
134    }
135    return length;
136  }
137}

1import { JobQueue, Job } from './job-queue';
2
3type JobHandler<T = any> = (data: T) => Promise<void>;
4
5export class JobWorker {
6  private queue: JobQueue;
7  private handlers: Map<string, JobHandler>;
8  private running: boolean = false;
9  private pollInterval: number;
10
11  constructor(queue: JobQueue, pollInterval: number = 1000) {
12    this.queue = queue;
13    this.handlers = new Map();
14    this.pollInterval = pollInterval;
15  }
16
17  /**
18   * Register a job handler
19   */
20  register<T>(type: string, handler: JobHandler<T>): void {
21    this.handlers.set(type, handler);
22  }
23
24  /**
25   * Start processing jobs
26   */
27  async start(): Promise<void> {
28    this.running = true;
29    console.log('Worker started');
30
31    while (this.running) {
32      try {
33        const job = await this.queue.getNextJob();
34
35        if (job) {
36          await this.processJob(job);
37        } else {
38          // No jobs available, wait before polling again
39          await new Promise((resolve) =>
40            setTimeout(resolve, this.pollInterval)
41          );
42        }
43      } catch (error) {
44        console.error('Worker error:', error);
45        await new Promise((resolve) => setTimeout(resolve, 1000));
46      }
47    }
48  }
49
50  /**
51   * Stop processing jobs
52   */
53  stop(): void {
54    this.running = false;
55    console.log('Worker stopped');
56  }
57
58  private async processJob(job: Job): Promise<void> {
59    const handler = this.handlers.get(job.type);
60
61    if (!handler) {
62      console.error(`No handler for job type: ${job.type}`);
63      await this.queue.failJob(
64        job.id,
65        `No handler registered for type: ${job.type}`
66      );
67      return;
68    }
69
70    try {
71      console.log(`Processing job ${job.id} (type: ${job.type})`);
72      await handler(job.data);
73      await this.queue.completeJob(job.id);
74      console.log(`Completed job ${job.id}`);
75    } catch (error) {
76      const message = error instanceof Error ? error.message : 'Unknown error';
77      console.error(`Failed job ${job.id}:`, message);
78      await this.queue.failJob(job.id, message);
79    }
80  }
81}

1import { JobQueue } from './job-queue';
2import { JobWorker } from './job-worker';
3
4declare function sendEmail(to: string, subject: string): Promise<void>;
5declare function processImage(url: string): Promise<void>;
6declare function generateReport(userId: string): Promise<void>;
7
8// Create queue
9const queue = new JobQueue({
10  host: '127.0.0.1',
11  port: 6379,
12  queueName: 'myapp:jobs',
13});
14
15await queue.connect();
16
17// Create worker
18const worker = new JobWorker(queue);
19
20// Register job handlers
21worker.register('send-email', async (data: { to: string; subject: string }) => {
22  console.log(`Sending email to ${data.to}`);
23  await sendEmail(data.to, data.subject);
24});
25
26worker.register('process-image', async (data: { url: string }) => {
27  console.log(`Processing image: ${data.url}`);
28  await processImage(data.url);
29});
30
31worker.register('generate-report', async (data: { userId: string }) => {
32  console.log(`Generating report for user ${data.userId}`);
33  await generateReport(data.userId);
34});
35
36// Start worker
37const running = worker.start();
38
39// Add jobs from your application
40await queue.addJob('send-email', {
41  to: 'user@example.com',
42  subject: 'Welcome!',
43}, { priority: 5 });
44
45await queue.addJob('process-image', {
46  url: 'https://example.com/image.jpg',
47}, { priority: 3 });
48
49await queue.addJob('generate-report', {
50  userId: '123',
51}, { priority: 1, maxAttempts: 5 });
52
53// Get queue stats
54const stats = await queue.getStats();
55console.log('Queue stats:', stats);
56
57// Graceful shutdown
58process.on('SIGTERM', async () => {
59  worker.stop();
60  await running;
61  queue.disconnect();
62});