Skip to content

ARCHITECTURE.md

Latest commit

 

History

History
921 lines (751 loc) · 24.8 KB

ARCHITECTURE.md

File metadata and controls

921 lines (751 loc) · 24.8 KB
Raw

Architecture Documentation

Overview

adapter-queue is a TypeScript-based job queue system inspired by Yii2-Queue, designed with a clean abstraction layer that enables seamless switching between multiple storage backends. The system uses an event-driven architecture where jobs are defined as TypeScript interfaces (not classes) and handlers are registered as functions.

Core Design Principles

  1. Type Safety First: Jobs are defined as TypeScript interfaces in a JobMap, providing compile-time validation
  2. Storage Agnostic: Abstract queue interface allows switching between backends without changing job code
  3. Event-Driven: Rich event system for monitoring job lifecycle and building plugins
  4. Plugin System: Extensible architecture supporting features like ECS task protection, metrics, and tracing
  5. Minimal Boilerplate: Interface-based jobs are simpler than class-based systems

System Architecture

High-Level Component Diagram

graph TB
    subgraph "Application Layer"
        APP[Application Code]
        JOBDEF[Job Type Definitions<br/>JobMap Interface]
        HANDLERS[Job Handlers<br/>Functions]
    end

    subgraph "Core Queue System"
        QUEUE[Abstract Queue Class]
        EVENTS[Event Emitter<br/>beforePush, afterPush<br/>beforeExec, afterExec<br/>afterError]
        PLUGIN[Plugin System]
    end

    subgraph "Driver Layer"
        DB[DbQueue]
        SQS[SqsQueue]
        FILE[FileQueue]
        MEM[InMemoryQueue]
        REDIS[RedisQueue]
        SQLITE[SQLiteQueue]
    end

    subgraph "Storage Backends"
        DBADAPTER[Database Adapter<br/>Interface]
        SQSCLIENT[AWS SQS Client]
        FILESYSTEM[File System]
        MEMORY[In-Memory Maps]
        REDISCLIENT[Redis Client]
        SQLITEDB[SQLite Database]
    end

    subgraph "Plugin Implementations"
        ECSPLUGIN[ECS Protection Plugin]
        CUSTOMP[Custom Plugins]
    end

    APP --> JOBDEF
    APP --> HANDLERS
    APP --> QUEUE

    QUEUE --> EVENTS
    QUEUE --> PLUGIN
    HANDLERS --> QUEUE

    QUEUE -.implements.-> DB
    QUEUE -.implements.-> SQS
    QUEUE -.implements.-> FILE
    QUEUE -.implements.-> MEM
    QUEUE -.implements.-> REDIS
    QUEUE -.implements.-> SQLITE

    DB --> DBADAPTER
    SQS --> SQSCLIENT
    FILE --> FILESYSTEM
    MEM --> MEMORY
    REDIS --> REDISCLIENT
    SQLITE --> SQLITEDB

    PLUGIN --> ECSPLUGIN
    PLUGIN --> CUSTOMP

    style QUEUE fill:#e1f5ff
    style EVENTS fill:#fff3e0
    style PLUGIN fill:#f3e5f5
Loading

Class Hierarchy

classDiagram
    class Queue~TJobMap~ {
        <<abstract>>
        #ttrDefault: number
        #plugins: QueuePlugin[]
        #handlers: Map~string, Function~
        #supportsLongPolling: boolean
        +addJob(name, request) Promise~string~
        +setHandlers(handlers) void
        +setHandler(jobName, handler) void
        +run(repeat, timeout) Promise~void~
        #pushMessage(payload, meta)* Promise~string~
        #reserve(timeout)* Promise~QueueMessage~
        #completeJob(message)* Promise~void~
        #failJob(message, error)* Promise~void~
        +status(id)* Promise~JobStatus~
    }

    class DbQueue~TJobMap~ {
        -db: DatabaseAdapter
        +adapter: DatabaseAdapter
        #pushMessage(payload, meta) Promise~string~
        #reserve(timeout) Promise~QueueMessage~
        #completeJob(message) Promise~void~
        #failJob(message, error) Promise~void~
        +status(id) Promise~JobStatus~
    }

    class SqsQueue~TJobMap~ {
        -client: SQSClient
        -queueUrl: string
        -onFailure: "delete" | "leaveInQueue"
        #pushMessage(payload, meta) Promise~string~
        #reserve(timeout) Promise~QueueMessage~
        #completeJob(message) Promise~void~
        #failJob(message, error) Promise~void~
        +status(id) Promise~JobStatus~
    }

    class FileQueue~TJobMap~ {
        -path: string
        -indexPath: string
        -touchIndex(callback) Promise~T~
        #pushMessage(payload, meta) Promise~string~
        #reserve(timeout) Promise~QueueMessage~
        #completeJob(message) Promise~void~
        #failJob(message, error) Promise~void~
        +status(id) Promise~JobStatus~
        +clear() Promise~void~
    }

    class InMemoryQueue~TJobMap~ {
        -jobs: Map~string, Record~
        -waitingJobs: string[]
        -reservedJobs: Set~string~
        -delayedJobs: Map~string, Timeout~
        #pushMessage(payload, meta) Promise~string~
        #reserve(timeout) Promise~QueueMessage~
        #completeJob(message) Promise~void~
        #failJob(message, error) Promise~void~
        +status(id) Promise~JobStatus~
        +clear() void
        +getStats() Stats
    }

    class RedisQueue~TJobMap~ {
        -redis: RedisClient
        -messagesKey: string
        -waitingKey: string
        -delayedKey: string
        -reservedKey: string
        #pushMessage(payload, meta) Promise~string~
        #reserve(timeout) Promise~QueueMessage~
        #completeJob(message) Promise~void~
        #failJob(message, error) Promise~void~
        +status(id) Promise~JobStatus~
        +clear() Promise~void~
    }

    class SQLiteQueue~TJobMap~ {
        -adapter: SQLiteDatabaseAdapter
        #pushMessage(payload, meta) Promise~string~
        #reserve(timeout) Promise~QueueMessage~
        #completeJob(message) Promise~void~
        #failJob(message, error) Promise~void~
        +status(id) Promise~JobStatus~
    }

    Queue <|-- DbQueue
    Queue <|-- SqsQueue
    Queue <|-- FileQueue
    Queue <|-- InMemoryQueue
    Queue <|-- RedisQueue
    DbQueue <|-- SQLiteQueue

    class DatabaseAdapter {
        <<interface>>
        +insertJob(payload, meta) Promise~string~
        +reserveJob(timeout) Promise~QueueJobRecord~
        +completeJob(id) Promise~void~
        +releaseJob(id) Promise~void~
        +failJob(id, error) Promise~void~
        +getJobStatus(id) Promise~JobStatus~
    }

    DbQueue --> DatabaseAdapter
Loading

Job Lifecycle

Job State Machine

stateDiagram-v2
    [*] --> Waiting: addJob()
    [*] --> Delayed: addJob(delaySeconds)

    Delayed --> Waiting: delay expires
    Waiting --> Reserved: reserve()

    Reserved --> Done: completeJob()
    Reserved --> Failed: failJob()
    Reserved --> Waiting: TTR expires

    Done --> [*]
    Failed --> [*]

    note right of Waiting
        Job is ready to be processed
        Sorted by priority (if supported)
    end note

    note right of Reserved
        Job is being processed
        TTR timer is active
        Will auto-recover if TTR expires
    end note

    note right of Delayed
        Job waiting for delay period
        Not available for reservation
    end note
Loading

Job Processing Flow

sequenceDiagram
    participant App as Application
    participant Q as Queue
    participant P as Plugin
    participant D as Driver
    participant S as Storage
    participant H as Handler

    Note over App,H: Job Addition Phase
    App->>Q: addJob(name, payload, options)
    Q->>P: emit beforePush event
    Q->>D: pushMessage(payload, meta)
    D->>S: store job with metadata
    S-->>D: return job ID
    D-->>Q: job ID
    Q->>P: emit afterPush event
    Q-->>App: job ID

    Note over App,H: Job Processing Phase
    App->>Q: run(repeat, timeout)

    loop Main Processing Loop
        Q->>P: beforePoll() hook

        alt Plugin returns "stop"
            P-->>Q: stop
            Q-->>App: exit gracefully
        else Plugin returns "continue"
            P-->>Q: continue

            Q->>D: reserve(timeout)
            D->>S: fetch next available job

            alt Job TTR expired
                S->>S: recover job to waiting
            end

            S-->>D: job or null
            D-->>Q: QueueMessage or null

            alt No job available
                Q->>Q: sleep or exit based on repeat flag
            else Job available
                Q->>P: beforeJob(message) hook
                Q->>P: emit beforeExec event

                Q->>H: execute handler(jobContext, queue)

                alt Handler succeeds
                    H-->>Q: success
                    Q->>P: emit afterExec event
                    Q->>D: completeJob(message)
                    D->>S: mark job as done
                else Handler throws error
                    H-->>Q: throw error
                    Q->>P: emit afterError event
                    Q->>D: failJob(message, error)
                    D->>S: mark job as failed
                end

                Q->>P: afterJob(message, error?) hook
            end
        end
    end
Loading

Data Flow Architecture

Job Data Serialization

flowchart LR
    subgraph "Application Layer"
        PAYLOAD[Job Payload<br/>TypeScript Object]
    end

    subgraph "Queue Core"
        META[Job Metadata<br/>JobMeta]
        MSG[Queue Message<br/>QueueMessage]
    end

    subgraph "Driver Layer"
        SER[Serialization]
    end

    subgraph "Storage Layer"
        DB_STORE[(Database<br/>Buffer)]
        SQS_STORE[(SQS<br/>JSON String)]
        FILE_STORE[(File<br/>JSON)]
        MEM_STORE[(Memory<br/>Object)]
        REDIS_STORE[(Redis<br/>JSON String)]
    end

    PAYLOAD --> META
    META --> MSG
    MSG --> SER

    SER --> DB_STORE
    SER --> SQS_STORE
    SER --> FILE_STORE
    SER --> MEM_STORE
    SER --> REDIS_STORE

    style SER fill:#ffe0b2
Loading

Driver-Specific Storage Strategies

Database Queue (DbQueue)

graph TB
    subgraph "DatabaseAdapter Interface"
        INSERT[insertJob<br/>Serialize payload to Buffer]
        RESERVE[reserveJob<br/>Atomic SELECT + UPDATE]
        COMPLETE[completeJob<br/>Mark as done]
        FAIL[failJob<br/>Mark as failed]
    end

    subgraph "Implementation Examples"
        SQLITE[SQLite<br/>JSONB payload<br/>Indexed by status+priority]
        POSTGRES[PostgreSQL<br/>JSONB payload<br/>FOR UPDATE SKIP LOCKED]
        MYSQL[MySQL<br/>JSON payload<br/>Transaction-based]
    end

    INSERT --> SQLITE
    INSERT --> POSTGRES
    INSERT --> MYSQL

    RESERVE --> SQLITE
    RESERVE --> POSTGRES
    RESERVE --> MYSQL
Loading

SQS Queue

graph TB
    subgraph "SQS Operations"
        SEND[SendMessage<br/>DelaySeconds<br/>MessageAttributes]
        RECEIVE[ReceiveMessage<br/>WaitTimeSeconds<br/>Long Polling]
        DELETE[DeleteMessage<br/>ReceiptHandle]
        VISIBILITY[ChangeMessageVisibility<br/>Implement TTR]
    end

    subgraph "Message Format"
        BODY[MessageBody<br/>JSON payload]
        ATTR[MessageAttributes<br/>name, ttr, priority]
        RECEIPT[ReceiptHandle<br/>For deletion]
    end

    SEND --> BODY
    SEND --> ATTR
    RECEIVE --> RECEIPT
    DELETE --> RECEIPT
    VISIBILITY --> RECEIPT
Loading

File Queue

graph TB
    subgraph "File Structure"
        INDEX[queue.index.json<br/>Job metadata]
        LOCK[queue.index.json.lock<br/>File locking]
        JOB1[job1.data]
        JOB2[job2.data]
        JOB3[job3.data]
    end

    subgraph "Index Content"
        LASTID[lastId: counter]
        WAIT[waiting: Array<id, ttr>]
        DELAY[delayed: Array<id, ttr, time>]
        RES[reserved: Array<id, ttr, attempt, time>]
    end

    INDEX --> LASTID
    INDEX --> WAIT
    INDEX --> DELAY
    INDEX --> RES

    LOCK -.locks.-> INDEX
Loading

Redis Queue

graph TB
    subgraph "Redis Data Structures"
        COUNTER[queue:default:id<br/>INCR counter]
        HASH[queue:default:messages<br/>HSET id => JSON]
        WAITING[queue:default:waiting<br/>ZSET sorted by priority]
        DELAYED[queue:default:delayed<br/>ZSET sorted by time]
        RESERVED[queue:default:reserved<br/>ZSET sorted by expire time]
        ATTEMPTS[queue:default:attempts<br/>HASH id => count]
    end

    COUNTER --> HASH
    HASH --> WAITING
    HASH --> DELAYED
    HASH --> RESERVED
    HASH --> ATTEMPTS
Loading

In-Memory Queue

graph TB
    subgraph "In-Memory Data Structures"
        JOBS[Map<id, InMemoryJobRecord>]
        WAIT[waitingJobs: string[]<br/>sorted by priority]
        RES[reservedJobs: Set<string>]
        DEL[delayedJobs: Map<id, Timeout>]
        TTR[ttrTimeouts: Map<id, Timeout>]
    end

    JOBS --> WAIT
    JOBS --> RES
    JOBS --> DEL
    JOBS --> TTR

    DEL -.setTimeout.-> WAIT
    TTR -.setTimeout.-> WAIT
Loading

Plugin System

Plugin Architecture

graph TB
    subgraph "Plugin Lifecycle"
        INIT[init<br/>Called once on queue start<br/>Returns cleanup function]
        POLL[beforePoll<br/>Called before each reserve<br/>Can return 'stop']
        BEFORE[beforeJob<br/>Called after reserve<br/>Before execution]
        AFTER[afterJob<br/>Called after execution<br/>Success or failure]
    end

    subgraph "Plugin Examples"
        ECS[ECS Protection<br/>Task protection management]
        METRICS[Metrics<br/>Performance tracking]
        TRACE[Tracing<br/>Distributed tracing]
        CIRCUIT[Circuit Breaker<br/>Failure handling]
    end

    INIT --> POLL
    POLL --> BEFORE
    BEFORE --> AFTER

    ECS -.implements.-> INIT
    ECS -.implements.-> POLL
    ECS -.implements.-> BEFORE
    ECS -.implements.-> AFTER

    METRICS -.implements.-> BEFORE
    METRICS -.implements.-> AFTER

    TRACE -.implements.-> BEFORE
    TRACE -.implements.-> AFTER

    CIRCUIT -.implements.-> POLL
    CIRCUIT -.implements.-> AFTER
Loading

ECS Protection Plugin Flow

sequenceDiagram
    participant Q as Queue
    participant Plugin as ECS Plugin
    participant Manager as Protection Manager
    participant ECS as ECS Agent

    Note over Q,ECS: Initialization
    Q->>Plugin: init({ queue })
    Plugin-->>Q: return cleanup function

    Note over Q,ECS: Processing Loop
    loop Main Loop
        Q->>Plugin: beforePoll()
        Plugin->>Manager: attemptProtect(ttrSeconds)

        alt Not draining
            Manager->>ECS: PUT /task-protection/v1/state<br/>ProtectionEnabled: true
            ECS-->>Manager: 200 OK
            Manager-->>Plugin: true
            Plugin-->>Q: "continue"

            Q->>Q: reserve job
            Q->>Plugin: beforeJob(message)
            Plugin->>Plugin: jobsInProgress.add(id)

            alt Job needs more protection
                Plugin->>Manager: attemptProtect(jobTtr)
                Manager->>ECS: PUT with extended ExpiresInMinutes
            end

            Q->>Q: execute handler
            Q->>Plugin: afterJob(message, error?)
            Plugin->>Plugin: jobsInProgress.delete(id)

            alt No jobs in progress
                Plugin->>Manager: attemptRelease()
                Manager->>ECS: PUT /task-protection/v1/state<br/>ProtectionEnabled: false
            end

        else ECS is draining
            Manager->>ECS: PUT /task-protection/v1/state
            ECS-->>Manager: Error (task draining)
            Manager->>Manager: draining = true
            Manager-->>Plugin: false
            Plugin-->>Q: "stop"
            Q-->>Q: exit gracefully
        end
    end
Loading

Event System

Event Types and Data Flow

graph LR
    subgraph "Queue Events"
        BP[beforePush<br/>name, payload, meta]
        AP[afterPush<br/>id, name, payload, meta]
        BE[beforeExec<br/>id, name, payload, meta]
        AE[afterExec<br/>id, name, payload, meta, result]
        ERR[afterError<br/>id, name, payload, meta, error]
    end

    subgraph "Event Listeners"
        LOG[Logger]
        METRICS[Metrics Collector]
        MONITOR[Monitoring System]
        ALERT[Alert System]
    end

    BP --> LOG
    AP --> METRICS
    BE --> MONITOR
    AE --> METRICS
    ERR --> ALERT

    style ERR fill:#ffcdd2
    style AE fill:#c8e6c9
Loading

Type System

Core Type Relationships

classDiagram
    class JobMap {
        <<interface>>
        'job-name': PayloadType
    }

    class JobMeta {
        name: string
        ttr?: number
        delaySeconds?: number
        priority?: number
        pushedAt?: Date
        reservedAt?: Date
        doneAt?: Date
        receiptHandle?: string
    }

    class QueueMessage {
        id: string
        payload: unknown
        meta: JobMeta
    }

    class JobContext~T~ {
        id: string
        payload: T
        meta: JobMeta
        pushedAt?: Date
        reservedAt?: Date
    }

    class JobHandler~T~ {
        <<function>>
        (job: JobContext~T~, queue: Queue) => Promise~void~
    }

    class JobHandlers~TJobMap~ {
        <<mapped type>>
        [K in keyof TJobMap]: JobHandler~TJobMap[K]~
    }

    class BaseJobRequest~TPayload~ {
        payload: TPayload
        ttr?: number
    }

    class WithPriority {
        priority?: number
    }

    class WithDelay {
        delaySeconds?: number
    }

    QueueMessage --> JobMeta
    JobContext --> JobMeta
    JobHandler --> JobContext
    JobHandlers --> JobHandler
    JobHandlers --> JobMap
    BaseJobRequest --|> WithPriority
    BaseJobRequest --|> WithDelay
Loading

Driver Feature Matrix

Supported Features by Driver

graph TB
    subgraph "Feature Support Matrix"
        direction TB

        FEATURES[["<b>Features</b>"]]
        PRIORITY["Priority Queue"]
        DELAY["Delayed Jobs"]
        TTR["TTR Recovery"]
        STATUS["Job Status"]
        PERSIST["Persistence"]
        LONGPOLL["Long Polling"]

        style FEATURES fill:#e3f2fd
    end

    subgraph "DbQueue"
        DB_PRIORITY[✓ Priority]
        DB_DELAY[✓ Delay]
        DB_TTR[✓ TTR]
        DB_STATUS[✓ Status]
        DB_PERSIST[✓ Persist]
        DB_POLL[✗ Long Poll]
    end

    subgraph "SqsQueue"
        SQS_PRIORITY[✗ Priority]
        SQS_DELAY[✓ Delay 0-900s]
        SQS_TTR[✓ TTR via Visibility]
        SQS_STATUS[✗ Status]
        SQS_PERSIST[✓ Persist]
        SQS_POLL[✓ Long Poll]
    end

    subgraph "FileQueue"
        FILE_PRIORITY[✗ Priority]
        FILE_DELAY[✓ Delay]
        FILE_TTR[✓ TTR]
        FILE_STATUS[✓ Status]
        FILE_PERSIST[✓ Persist]
        FILE_POLL[✗ Long Poll]
    end

    subgraph "InMemoryQueue"
        MEM_PRIORITY[✓ Priority]
        MEM_DELAY[✓ Delay]
        MEM_TTR[✓ TTR]
        MEM_STATUS[✓ Status]
        MEM_PERSIST[✗ Persist]
        MEM_POLL[✗ Long Poll]
    end

    subgraph "RedisQueue"
        REDIS_PRIORITY[✓ Priority]
        REDIS_DELAY[✓ Delay]
        REDIS_TTR[✓ TTR]
        REDIS_STATUS[✓ Status]
        REDIS_PERSIST[✓ Persist]
        REDIS_POLL[✗ Long Poll]
    end

    PRIORITY -.-> DB_PRIORITY
    DELAY -.-> DB_DELAY
    TTR -.-> DB_TTR
    STATUS -.-> DB_STATUS
    PERSIST -.-> DB_PERSIST
    LONGPOLL -.-> DB_POLL

    PRIORITY -.-> SQS_PRIORITY
    DELAY -.-> SQS_DELAY
    TTR -.-> SQS_TTR
    STATUS -.-> SQS_STATUS
    PERSIST -.-> SQS_PERSIST
    LONGPOLL -.-> SQS_POLL

    PRIORITY -.-> FILE_PRIORITY
    DELAY -.-> FILE_DELAY
    TTR -.-> FILE_TTR
    STATUS -.-> FILE_STATUS
    PERSIST -.-> FILE_PERSIST
    LONGPOLL -.-> FILE_POLL

    PRIORITY -.-> MEM_PRIORITY
    DELAY -.-> MEM_DELAY
    TTR -.-> MEM_TTR
    STATUS -.-> MEM_STATUS
    PERSIST -.-> MEM_PERSIST
    LONGPOLL -.-> MEM_POLL

    PRIORITY -.-> REDIS_PRIORITY
    DELAY -.-> REDIS_DELAY
    TTR -.-> REDIS_TTR
    STATUS -.-> REDIS_STATUS
    PERSIST -.-> REDIS_PERSIST
    LONGPOLL -.-> REDIS_POLL
Loading

Key Design Patterns

1. Template Method Pattern

The abstract Queue class defines the job processing workflow in run(), while concrete drivers implement storage-specific operations (pushMessage, reserve, completeJob, failJob).

2. Strategy Pattern

Different drivers implement the same queue interface with different storage strategies (database, file, SQS, Redis, memory).

3. Observer Pattern

The event emitter system allows external code to observe and react to queue events without coupling.

4. Plugin Pattern

The plugin system enables extending queue functionality without modifying core code.

5. Adapter Pattern

The DatabaseAdapter interface allows different database implementations to work with DbQueue.

6. Factory Pattern

Each driver provides a factory-like constructor that sets up the appropriate storage backend.

Implementation Details

Job ID Generation

  • DbQueue/SQLiteQueue: Auto-increment from database
  • SqsQueue: AWS SQS MessageId
  • FileQueue: Incrementing counter in index file
  • InMemoryQueue: Incrementing counter in memory
  • RedisQueue: Redis INCR command

Concurrency Control

  • DbQueue: Database transactions and row-level locking
  • SqsQueue: SQS visibility timeout
  • FileQueue: File-based locking with .lock files
  • InMemoryQueue: Single-threaded by nature (Node.js)
  • RedisQueue: Redis atomic operations (ZREM, ZADD)

TTR (Time To Run) Implementation

  • DbQueue: expire_time column checked on reserve
  • SqsQueue: ChangeMessageVisibility API
  • FileQueue: reserved array with expiration timestamps
  • InMemoryQueue: setTimeout for automatic recovery
  • RedisQueue: Sorted set with expiration score

Priority Ordering

  • DbQueue/SQLiteQueue: ORDER BY priority DESC, push_time ASC
  • SqsQueue: Not supported (FIFO only)
  • FileQueue: Not supported (FIFO only)
  • InMemoryQueue: In-memory sorted array by priority
  • RedisQueue: Sorted set with priority-based score

Performance Considerations

Scalability

  1. DbQueue: Limited by database connection pool and row locking
  2. SqsQueue: Highly scalable, AWS-managed infrastructure
  3. FileQueue: Limited by file I/O and locking overhead
  4. InMemoryQueue: Very fast but limited by process memory
  5. RedisQueue: Highly scalable, limited by Redis cluster capacity

Recommended Use Cases

  • DbQueue: Single application instance, transactional consistency needed
  • SqsQueue: Multi-region distributed systems, high throughput
  • FileQueue: Development, testing, simple single-server deployments
  • InMemoryQueue: Testing, development, short-lived jobs
  • RedisQueue: High throughput with persistence, multiple workers

Extension Points

Creating a Custom Driver

  1. Extend the Queue<TJobMap> class
  2. Implement abstract methods:
    • pushMessage(payload, meta): Promise<string>
    • reserve(timeout): Promise<QueueMessage | null>
    • completeJob(message): Promise<void>
    • failJob(message, error): Promise<void>
    • status(id): Promise<JobStatus>
  3. Set supportsLongPolling = true if applicable

Creating a Custom Plugin

  1. Implement the QueuePlugin interface:
    • init?(ctx): Promise<(() => Promise<void>) | void>
    • beforePoll?(): Promise<'continue' | 'stop' | void>
    • beforeJob?(job): Promise<void>
    • afterJob?(job, error?): Promise<void>
  2. Use the plugin via the plugins option

Creating a Custom DatabaseAdapter

  1. Implement the DatabaseAdapter interface:
    • insertJob(payload, meta): Promise<string>
    • reserveJob(timeout): Promise<QueueJobRecord | null>
    • completeJob(id): Promise<void>
    • releaseJob(id): Promise<void>
    • failJob(id, error): Promise<void>
    • getJobStatus(id): Promise<JobStatus | null>
  2. Pass to DbQueue constructor

Security Considerations

Input Validation

  • Job names are validated as strings in addJob()
  • Payloads are serialized safely using JSON.stringify()
  • No arbitrary code execution from job payloads

Error Handling

  • Custom QueueError class for queue-specific errors
  • Error events emitted via afterError event
  • Handlers can throw errors safely without crashing the worker

Resource Limits

  • InMemoryQueue has maxJobs limit to prevent memory leaks
  • TTR prevents jobs from running indefinitely
  • File queue uses file locking to prevent race conditions

Testing Strategy

Unit Testing

  • Mock drivers for testing job handlers
  • InMemoryQueue for fast, isolated tests
  • Plugin testing with dedicated instances

Integration Testing

  • Use SQLiteQueue with in-memory database (:memory:)
  • Use testcontainers for Redis/PostgreSQL testing
  • Mock AWS SQS client for SQS testing

Example Test Structure

import { InMemoryQueue } from 'adapter-queue/drivers/memory';

describe('Job Processing', () => {
  let queue: InMemoryQueue<MyJobs>;

  beforeEach(() => {
    queue = new InMemoryQueue({ name: 'test' });
    queue.setHandlers({
      'send-email': async ({ payload }) => {
        // Test implementation
      }
    });
  });

  it('should process jobs', async () => {
    await queue.addJob('send-email', {
      payload: { to: 'test@example.com' }
    });
    await queue.run();
    expect(queue.getStats().done).toBe(1);
  });
});

Glossary

  • TTR (Time To Run): Maximum time a job can run before being automatically recovered
  • Job: A unit of work with a name and payload
  • Queue: Storage and processing system for jobs
  • Driver: Backend-specific implementation of the queue interface
  • Handler: Function that executes when a job is processed
  • JobMap: TypeScript interface mapping job names to payload types
  • Plugin: Extension that hooks into the queue lifecycle
  • Reserve: Atomically claim a job for processing
  • Visibility Timeout: SQS term equivalent to TTR
  • DatabaseAdapter: Interface for custom database backends