feat: queue plugin

Author: muniter

src/core/queue.ts

@@ -1,5 +1,6 @@
 import { EventEmitter } from 'events';
 import type { JobStatus, JobMeta, QueueMessage, QueueEvent, JobData, JobOptions, BaseJobRequest } from '../interfaces/job.ts';
+import type { QueuePlugin, QueueOptions } from '../interfaces/plugin.ts';
 
 /**
  * Abstract queue class providing event-based job processing.
@@ -29,16 +30,23 @@ import type { JobStatus, JobMeta, QueueMessage, QueueEvent, JobData, JobOptions,
  */
 export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends BaseJobRequest<any> = BaseJobRequest<any>> extends EventEmitter {
   protected ttrDefault = 300;
+  protected plugins: QueuePlugin[];
+  protected pluginDisposers: Array<() => Promise<void>> = [];
+  public readonly name?: string;
 
   /**
    * Creates a new Queue instance.
    * 
    * @param options - Configuration options
    * @param options.ttrDefault - Default time-to-run for jobs in seconds (default: 300)
+   * @param options.name - Optional name for the queue
+   * @param options.plugins - Array of plugins to use with this queue
    */
-  constructor(options: { ttrDefault?: number } = {}) {
+  constructor(options: QueueOptions = {}) {
     super();
     if (options.ttrDefault) this.ttrDefault = options.ttrDefault;
+    this.name = options.name;
+    this.plugins = options.plugins || [];
   }
 
 
@@ -147,22 +155,111 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
    * ```
    */
   async run(repeat: boolean = false, timeout: number = 0): Promise<void> {
-    const canContinue = () => true;
+    const disposers = [...this.pluginDisposers];
 
-    while (canContinue()) {
-      const message = await this.reserve(timeout);
-      
-      if (!message) {
-        if (!repeat) break;
-        if (timeout > 0) {
-          await this.sleep(timeout * 1000);
+    // 1. Initialize plugins if not already initialized
+    if (this.pluginDisposers.length === 0) {
+      for (const plugin of this.plugins) {
+        if (plugin.init) {
+          const dispose = await plugin.init({ queue: this as any, queueName: this.name });
+          if (dispose) {
+            this.pluginDisposers.push(dispose);
+            disposers.push(dispose);
+          }
         }
-        continue;
       }
+    }
+
+    try {
+      // 2. Main processing loop (enhancing existing loop)
+      let stopped = false;
+      
+      while (!stopped) {
+        // Check if any plugin wants to stop
+        try {
+          for (const plugin of this.plugins) {
+            if (plugin.beforePoll) {
+              const result = await plugin.beforePoll();
+              if (result === 'stop') {
+                stopped = true;
+                break;
+              }
+            }
+          }
+        } catch (error) {
+          console.error('Plugin beforePoll error:', error);
+          // Continue polling despite plugin error
+        }
+        if (stopped) break;
+
+        const message = await this.reserve(timeout);
+        if (!message) {
+          if (!repeat) break;
+          if (timeout > 0) {
+            await this.sleep(timeout * 1000);
+          }
+          continue;
+        }
 
-      const success = await this.handleMessage(message);
-      if (success) {
-        await this.release(message);
+        // 3. Pre-execution hooks
+        try {
+          for (const plugin of this.plugins) {
+            if (plugin.beforeJob) {
+              await plugin.beforeJob(message);
+            }
+          }
+        } catch (error) {
+          console.error('Plugin beforeJob error:', error);
+          // Continue processing despite plugin error
+        }
+
+        // 4. Execute job (with plugin hooks)
+        let success = false;
+        let jobError: unknown;
+        
+        // We need to track errors for plugins, but handleMessage catches them internally
+        // So we'll set up an event listener to capture the error
+        let capturedError: unknown;
+        const errorListener = (event: QueueEvent) => {
+          if (event.type === 'afterError' && event.id === message.id) {
+            capturedError = event.error;
+          }
+        };
+        
+        this.once('afterError', errorListener);
+        
+        try {
+          success = await this.handleMessage(message);
+          jobError = capturedError; // Will be undefined if no error
+        } catch (error) {
+          // This shouldn't happen since handleMessage catches errors
+          jobError = error;
+          success = false;
+        } finally {
+          this.removeListener('afterError', errorListener);
+        }
+
+        // 5. Post-execution hooks
+        try {
+          for (const plugin of this.plugins) {
+            if (plugin.afterJob) {
+              await plugin.afterJob(message, jobError);
+            }
+          }
+        } catch (error) {
+          console.error('Plugin afterJob error:', error);
+          // Don't let plugin errors affect job completion
+        }
+
+        // Complete the job if successful
+        if (success) {
+          await this.release(message);
+        }
+      }
+    } finally {
+      // 6. Cleanup only if we initialized in this run
+      for (const dispose of disposers.reverse()) {
+        await dispose();
       }
     }
   }
@@ -176,6 +273,7 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
    */
   protected async handleMessage(message: QueueMessage): Promise<boolean> {
     try {
+      // Parse the job data (this may have been modified by plugins)
       const jobData: JobData = JSON.parse(message.payload);
       const { name, payload } = jobData;
 

src/drivers/db.ts

@@ -1,11 +1,12 @@
 import { Queue } from '../core/queue.ts';
 import type { JobStatus, JobMeta, QueueMessage, DbJobRequest } from '../interfaces/job.ts';
 import type { DatabaseAdapter, QueueJobRecord } from '../interfaces/database.ts';
+import type { QueueOptions } from '../interfaces/plugin.ts';
 
 export class DbQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, DbJobRequest<any>> {
   constructor(
     private db: DatabaseAdapter,
-    options: { ttrDefault?: number } = {}
+    options: QueueOptions = {}
   ) {
     super(options);
   }

src/drivers/file.ts

@@ -3,6 +3,7 @@ import { open } from 'fs/promises';
 import path from 'path';
 import { Queue } from '../core/queue.js';
 import type { QueueMessage, JobMeta, JobStatus, FileJobRequest } from '../interfaces/job.js';
+import type { QueueOptions } from '../interfaces/plugin.js';
 
 interface IndexData {
   lastId: number;
@@ -17,7 +18,7 @@ interface ReservedInfo {
   attempt: number;
 }
 
-interface FileQueueOptions {
+interface FileQueueOptions extends QueueOptions {
   path: string;
   dirMode?: number;
   fileMode?: number;
@@ -30,7 +31,7 @@ export class FileQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Fil
   private indexPath: string;
 
   constructor(options: FileQueueOptions) {
-    super();
+    super(options);
     this.path = path.resolve(options.path);
     this.dirMode = options.dirMode ?? 0o755;
     this.fileMode = options.fileMode;

src/drivers/sqs.ts

@@ -16,6 +16,7 @@ import type {
   QueueMessage,
   SqsJobRequest,
 } from "../interfaces/job.ts";
+import type { QueueOptions } from "../interfaces/plugin.ts";
 
 export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
   TJobMap,
@@ -24,7 +25,7 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
   constructor(
     private client: SQSClient,
     private queueUrl: string,
-    options: { ttrDefault?: number } = {}
+    options: QueueOptions = {}
   ) {
     super(options);
   }

src/interfaces/plugin.ts

@@ -0,0 +1,91 @@
+import type { Queue } from '../core/queue.ts';
+import type { QueueMessage } from './job.ts';
+
+/**
+ * Plugin interface for extending queue functionality.
+ * 
+ * Plugins can hook into the queue lifecycle to add features like:
+ * - Task protection (ECS, Kubernetes)
+ * - Metrics collection
+ * - Distributed tracing
+ * - Circuit breaking
+ * - Job enrichment
+ */
+export interface QueuePlugin {
+  /**
+   * Called once when the queue starts processing.
+   * Use this hook to initialize resources, connections, or state.
+   * Return a cleanup function that will be called on shutdown.
+   * 
+   * @param ctx - Context containing the queue instance and optional name
+   * @returns Optional cleanup function
+   */
+  init?(ctx: { queue: Queue; queueName?: string }): Promise<(() => Promise<void>) | void>;
+
+  /**
+   * Called before each poll/reserve attempt.
+   * Use this hook to control whether the queue should continue polling for jobs.
+   * 
+   * Examples:
+   * - Check if environment is being drained (ECS, Kubernetes)
+   * - Implement circuit breaker logic
+   * - Check resource availability
+   * 
+   * @returns 'stop' to gracefully shut down processing, 'continue' or void to continue
+   */
+  beforePoll?(): Promise<'continue' | 'stop' | void>;
+
+  /**
+   * Called after a job is reserved but before execution.
+   * Use this hook to prepare for job processing using job details.
+   * 
+   * Examples:
+   * - Acquire task protection using job TTR
+   * - Start distributed tracing spans
+   * - Enrich job with metadata
+   * - Track job start metrics
+   * 
+   * Note: Once a job is reserved, it will be processed. This hook cannot reject jobs.
+   * Use beforePoll() if you need to stop processing entirely.
+   * 
+   * @param job - The reserved job message with full context
+   */
+  beforeJob?(job: QueueMessage): Promise<void>;
+
+  /**
+   * Called after job execution (success or failure).
+   * Use this hook for cleanup and post-processing.
+   * 
+   * Examples:
+   * - Release task protection
+   * - End distributed tracing spans
+   * - Record completion metrics
+   * - Clean up resources
+   * 
+   * @param job - The job that was processed
+   * @param error - Error if job failed, undefined if successful
+   */
+  afterJob?(job: QueueMessage, error?: unknown): Promise<void>;
+}
+
+/**
+ * Extended queue options interface that includes plugin support.
+ */
+export interface QueueOptions {
+  /**
+   * Default time-to-run for jobs in seconds.
+   */
+  ttrDefault?: number;
+  
+  /**
+   * Optional name for the queue (used in plugin context).
+   */
+  name?: string;
+  
+  /**
+   * Array of plugins to use with this queue.
+   * Plugins are initialized when the queue starts processing
+   * and their hooks are called during the job lifecycle.
+   */
+  plugins?: QueuePlugin[];
+}