refactor: streamline job handling and improve logging in queues

Author: muniter

example-queue-project/package.json

@@ -9,9 +9,9 @@
     "lint": "tsc --noEmit",
     "dev": "tsx src/index.ts",
     "start": "node dist/index.js",
-    "add-job": "tsx src/add-job.ts",
-    "process-jobs": "tsx src/process-jobs.ts",
-    "demo": "echo 'Run \"npm run process-jobs\" in one terminal, then \"npm run add-job\" in another!'"
+    "push": "tsx src/index.ts --push",
+    "run": "tsx src/index.ts --run",
+    "demo": "echo 'Run \"npm run run\" in one terminal, then \"npm run push\" in another!'"
   },
   "dependencies": {
     "@muniter/queue": "file:..",

example-queue-project/src/email-queue.ts

@@ -1,8 +1,8 @@
 import { FileQueue } from "@muniter/queue";
 
 interface EmailJobs {
-  'welcome-email': { to: string; name: string };
-  'notification': { to: string; subject: string; body: string };
+    'welcome-email': { to: string; name: string };
+    'notification': { to: string; subject: string; body: string };
 }
 
 
@@ -28,4 +28,16 @@ emailQueue.onJob('notification', async (payload) => {
     console.log(`Sending notification email to ${to}: ${subject}`);
     await new Promise(resolve => setTimeout(resolve, 500));
     console.log(`Notification sent successfully`);
-});
+});
+
+emailQueue.on('beforeExec', (event) => {
+    console.log(`\n[emailQueue][${new Date().toISOString()}] Starting ${event.name} job ${event.id}...`);
+});
+
+emailQueue.on('afterExec', (event) => {
+    console.log(`[emailQueue][${new Date().toISOString()}] Email job ${event.id} (${event.name}) completed successfully`);
+});
+
+emailQueue.on('afterError', (event) => {
+    console.error(`[emailQueue][${new Date().toISOString()}] Email job ${event.id} (${event.name}) failed:`, event.error);
+});

example-queue-project/src/general-queue.ts

@@ -34,4 +34,17 @@ generalQueue.onJob('generate-report', async (payload) => {
     }
 
     console.log(`Report generated: ${type} for ${period}`);
+});
+
+// Add event listeners for both queues
+generalQueue.on('beforeExec', (event) => {
+    console.log(`\n[generalQueue][${new Date().toISOString()}] Starting ${event.name} job ${event.id}...`);
+});
+
+generalQueue.on('afterExec', (event) => {
+    console.log(`[generalQueue][${new Date().toISOString()}] Job ${event.id} (${event.name}) completed successfully`);
+});
+
+generalQueue.on('afterError', (event) => {
+    console.error(`[generalQueue][${new Date().toISOString()}] Job ${event.id} (${event.name}) failed:`, event.error);
 });

example-queue-project/src/index.ts

@@ -1,4 +1,4 @@
-import { initializeDatabase, run } from './database.js';
+import { initializeDatabase } from './database.js';
 import { emailQueue } from './email-queue.js';
 import { generalQueue } from './general-queue.js';
 import { parseArgs } from 'util';
@@ -37,53 +37,20 @@ async function push() {
   });
   console.log(`Image job added with ID: ${imageJobId}`);
 
-  const reportJobId = await generalQueue.addJob('generate-report', {
+  await generalQueue.addJob('generate-report', {
     payload: {
       type: 'sales',
       period: 'Q4-2023'
     },
     delay: 2
   });
-  console.log(`Delayed report job added with ID: ${reportJobId}`);
-
-  // Add event listeners for both queues
-  generalQueue.on('beforeExec', (event) => {
-    console.log(`\n[generalQueue][${new Date().toISOString()}] Starting ${event.name} job ${event.id}...`);
-  });
-
-  generalQueue.on('afterExec', (event) => {
-    console.log(`[generalQueue][${new Date().toISOString()}] Job ${event.id} (${event.name}) completed successfully`);
-  });
-
-  generalQueue.on('afterError', (event) => {
-    console.error(`[generalQueue][${new Date().toISOString()}] Job ${event.id} (${event.name}) failed:`, event.error);
-  });
-
-  emailQueue.on('beforeExec', (event) => {
-    console.log(`\n[emailQueue][${new Date().toISOString()}] Starting ${event.name} job ${event.id}...`);
-  });
-
-  emailQueue.on('afterExec', (event) => {
-    console.log(`[emailQueue][${new Date().toISOString()}] Email job ${event.id} (${event.name}) completed successfully`);
-  });
-
-  emailQueue.on('afterError', (event) => {
-    console.error(`[emailQueue][${new Date().toISOString()}] Email job ${event.id} (${event.name}) failed:`, event.error);
-  });
-
-  console.log('\nStarting workers...');
-
-  process.on('SIGINT', () => {
-    console.log('\nShutting down gracefully...');
-    process.exit(0);
-  });
 }
 
 async function run() {
   console.log('Starting queue workers...');
   await Promise.allSettled([
-    emailQueue.run(true),
-    generalQueue.run(true)
+    emailQueue.run(true, 1),
+    generalQueue.run(true, 1),
   ])
   console.log('Queue workers are running. Press Ctrl+C to exit.');
 }
@@ -105,7 +72,7 @@ if (import.meta.url === `file://${process.argv[1]}`) {
     console.log('  --run          Run the queue workers (default: false)');
     process.exit(0);
   }
-  
+
   if (args.values.push) {
     await push();
   } else if (args.values.run) {