A TypeScript queue system inspired by Yii2-Queue architecture
npm install adapter-queueA TypeScript queue system inspired by Yii2-Queue architecture, providing a clean abstraction for job processing with multiple storage backends and event-based job handling.
- Driver-based architecture: Swap between DB, SQS, and File drivers seamlessly
- Event-based jobs: Register handlers for job types without complex classes
- Type-safe API: Full TypeScript support with driver-specific option validation
- Multiple backends: Database, Amazon SQS, and File storage drivers
- Event system: Hook into queue lifecycle events
``bash`
pnpm add adapter-queue
For SQS support:
`bash`
pnpm add adapter-queue @aws-sdk/client-sqs
`typescript
import { FileQueue } from 'adapter-queue';
// Define your job types with TypeScript
interface MyJobs {
'send-email': { to: string; subject: string; body: string };
'resize-image': { url: string; width: number; height: number };
'generate-report': { type: string; period: string };
}
const queue = new FileQueue
// Register type-safe handlers
queue.setHandlers({
'send-email': async ({ payload }) => {
// payload is automatically typed as { to: string; subject: string; body: string }
console.log(Sending email to ${payload.to}: ${payload.subject});Resizing image ${payload.url} to ${payload.width}x${payload.height}
await sendEmail(payload.to, payload.subject, payload.body);
},
'resize-image': async ({ payload }) => {
// payload is automatically typed as { url: string; width: number; height: number }
console.log();Generating ${payload.type} report for ${payload.period}
await resizeImage(payload.url, payload.width, payload.height);
},
'generate-report': async ({ payload }) => {
// Handle report generation
console.log();`
}
});
`typescript
// Simple job addition
await queue.addJob('send-email', {
payload: {
to: 'user@example.com',
subject: 'Welcome!',
body: 'Thanks for signing up!'
}
});
// Job with options (TTR supported by all drivers)
await queue.addJob('resize-image', {
payload: {
url: 'https://example.com/image.jpg',
width: 800,
height: 600
},
ttr: 300 // 5 minute timeout
});
// Job with delay (supported by File and SQS drivers)
await queue.addJob('generate-report', {
payload: {
type: 'monthly',
period: 'December 2024'
},
delay: 60, // 1 minute delay
ttr: 600 // 10 minute timeout
});
`
`typescript
// Start processing jobs
await queue.run(true, 3); // Run continuously, poll every 3 seconds
// Or process jobs once and exit
await queue.run(false);
`
A file-based queue that stores jobs as individual files with JSON index tracking. Perfect for development and single-server applications.
`typescript
import { FileQueue } from 'adapter-queue';
const queue = new FileQueue
path: './queue-data', // Directory to store queue files
dirMode: 0o755, // Directory permissions (optional)
fileMode: 0o644 // File permissions (optional)
});
// Supports: TTR, Delay
// Does not support: Priority
await queue.addJob('send-email', {
payload: { to: 'user@example.com', subject: 'Test', body: 'File queue test' },
ttr: 300,
delay: 60
});
`
Use any database that implements the DatabaseAdapter interface:
`typescript
import { DbQueue } from 'adapter-queue';
// You provide the database adapter implementation
const dbAdapter = new YourDatabaseAdapter(); // implements DatabaseAdapter
const queue = new DbQueue
// Supports: TTR, Delay, Priority (depends on adapter implementation)
await queue.addJob('send-email', {
payload: { to: 'user@example.com', subject: 'Test', body: 'DB queue test' },
ttr: 300,
delay: 60,
priority: 5
});
`
Amazon SQS integration with native delay support:
`typescript
import { SQSClient } from '@aws-sdk/client-sqs';
import { SqsQueue } from 'adapter-queue';
const sqsClient = new SQSClient({ region: 'us-east-1' });
const queue = new SqsQueue
sqsClient,
'https://sqs.us-east-1.amazonaws.com/123456789/my-queue'
);
// Supports: TTR, Delay
// Does not support: Priority (SQS FIFO queues would be needed)
await queue.addJob('send-email', {
payload: { to: 'user@example.com', subject: 'Test', body: 'SQS test' },
ttr: 300,
delay: 60
// priority: 5 // ❌ TypeScript error - not supported by SQS driver
});
`
The library provides compile-time type safety for both payloads and driver-specific options:
`typescript
interface MyJobs {
'send-email': { to: string; subject: string; body: string };
}
const fileQueue = new FileQueue
const sqsQueue = new SqsQueue
// ✅ Payload is type-checked
await fileQueue.addJob('send-email', {
payload: { to: 'user@example.com', subject: 'Test', body: 'Hello' }
});
// ✅ TTR and delay work with FileQueue
await fileQueue.addJob('send-email', {
payload: { to: 'user@example.com', subject: 'Test', body: 'Hello' },
ttr: 300,
delay: 60
});
// ❌ TypeScript error - FileQueue doesn't support priority
await fileQueue.addJob('send-email', {
payload: { to: 'user@example.com', subject: 'Test', body: 'Hello' },
priority: 5 // Error!
});
// ✅ SqsQueue supports delay but not priority
await sqsQueue.addJob('send-email', {
payload: { to: 'user@example.com', subject: 'Test', body: 'Hello' },
delay: 30 // Works
});
// ❌ TypeScript error - SqsQueue doesn't support priority
await sqsQueue.addJob('send-email', {
payload: { to: 'user@example.com', subject: 'Test', body: 'Hello' },
priority: 5 // Error!
});
`
`typescript
// Job lifecycle events
queue.on('beforePush', (event) => {
console.log('About to add job:', event.name, event.payload);
});
queue.on('afterPush', (event) => {
console.log('Job added with ID:', event.id);
});
queue.on('beforeExec', (event) => {
console.log('Starting job:', event.id, event.name);
});
queue.on('afterExec', (event) => {
console.log('Job completed:', event.id, 'Result:', event.result);
});
queue.on('afterError', (event) => {
console.error('Job failed:', event.id, 'Error:', event.error);
});
`
To create your own database driver, implement the DatabaseAdapter interface:
`typescript
import { DatabaseAdapter, QueueJobRecord, JobMeta, JobStatus } from 'adapter-queue';
export class YourDatabaseAdapter implements DatabaseAdapter {
async insertJob(payload: Buffer, meta: JobMeta): Promise
// Insert job into your database
// Return unique job ID
}
async reserveJob(timeout: number): Promise
// Find and reserve next available job
// Handle delay, priority, TTR logic
// Return job record or null
}
async completeJob(id: string): Promise
// Mark job as completed
}
async releaseJob(id: string): Promise
// Release job back to queue (for retry)
}
async failJob(id: string, error: string): Promise
// Mark job as failed
}
async getJobStatus(id: string): Promise
// Return 'waiting' | 'reserved' | 'done' | 'failed'
}
}
`
- addJob - Add job to queuesetHandlers(handlers: JobHandlers
- - Register all job handlers with type safetyrun(repeat?: boolean, timeout?: number): Promise
- - Start processing jobsstatus(id: string): Promise
- - Get job status
- All drivers: { ttr?: number } (time-to-run in seconds){ ttr?, delay?, priority? }
- DbQueue: (depends on adapter implementation){ ttr?, delay? }
- SqsQueue: (uses SQS DelaySeconds){ ttr?, delay? }
- FileQueue: (implements delay functionality)
`typescript
interface JobMap {
'job-name': { / payload type / };
'another-job': { / payload type / };
}
// Jobs are defined as TypeScript interfaces, not classes
// Handlers are registered with queue.setHandlers()
`
The queue system supports plugins to extend functionality. Plugins can hook into the queue lifecycle to add features like task protection, metrics collection, distributed tracing, and more.
Prevents job loss during ECS container termination by automatically acquiring and releasing ECS Task Protection based on job activity.
Why ECS Task Protection?
In ECS environments, containers can be terminated during:
- Auto-scaling scale-in events
- Rolling deployments
- Spot instance interruptions
- Manual task stopping
Without protection, in-flight jobs are lost when the container terminates. ECS Task Protection prevents this by marking tasks as "protected" from termination while they're processing jobs.
How this plugin helps:
- Automatic: No manual protection management - activated only when needed
- Efficient: Protection is acquired when jobs start, released when idle
- Safe: Detects ECS draining and gracefully stops accepting new work
- Reliable: Auto-renews protection for long-running jobs
`bash`
pnpm add adapter-queue
`typescript
import { SQSQueue } from 'adapter-queue/sqs';
import { SQSClient } from '@aws-sdk/client-sqs';
import { EcsProtectionManager, ecsTaskProtection } from 'adapter-queue/plugins/ecs-protection-manager';
// Create protection manager (share across all queues in your app)
const protectionManager = new EcsProtectionManager();
const queue = new SQSQueue({
client: new SQSClient({ region: 'us-east-1' }),
queueUrl: process.env.SQS_QUEUE_URL!,
name: 'email-queue',
onFailure: 'delete', // or 'leaveInQueue'
plugins: [ecsTaskProtection({
manager: protectionManager,
defaultProtectionTimeout: 600 // 10 minutes default
})]
});
await queue.run(true, 3);
// Clean up when shutting down
await protectionManager.cleanup();
`
Features:
- Smart Protection Management: Acquires protection before polling, maintains it while jobs are active
- Reference Counting: Tracks active jobs per queue, only releases protection when all jobs complete
- TTR-Aware: Automatically extends protection for long-running jobs based on their TTR
- Draining Detection: Detects when ECS is draining and gracefully stops accepting new jobs
- Zero Dependencies: Uses built-in Node.js fetch API
- Configurable Logging: Integrate with your existing logging system
Custom Logger Example:
`typescript
import pino from 'pino';
const logger = pino();
const protectionManager = new EcsProtectionManager({
logger: {
log: (message) => logger.info(message),
warn: (message) => logger.warn(message),
error: (message, error) => logger.error({ error }, message)
}
});
`
Multiple Queues:
`typescript
// Use the same protection manager across all queues
const protectionManager = new EcsProtectionManager();
const emailQueue = new SQSQueue({
client: new SQSClient({ region: 'us-east-1' }),
queueUrl: process.env.EMAIL_QUEUE_URL!,
name: 'email-queue',
onFailure: 'delete',
plugins: [ecsTaskProtection({ manager: protectionManager })]
});
const imageQueue = new SQSQueue({
client: new SQSClient({ region: 'us-east-1' }),
queueUrl: process.env.IMAGE_QUEUE_URL!,
name: 'image-queue',
onFailure: 'delete',
plugins: [ecsTaskProtection({ manager: protectionManager })] // Same manager, different plugin instance
});
// Both queues coordinate protection through the shared manager
await Promise.all([
emailQueue.run(true),
imageQueue.run(true)
]);
`
⚠️ Important: Only create one EcsProtectionManager instance per application/container. Multiple instances will conflict and break protection coordination.
Plugins implement the QueuePlugin interface and can hook into these lifecycle events:
- init?() - Called once when queue starts, return cleanup functionbeforePoll?()
- - Called before polling for jobs, can return 'stop' to gracefully shut downbeforeJob?()
- - Called after job is reserved but before executionafterJob?()
- - Called after job completion (success or failure)
`typescript
import { QueuePlugin } from 'adapter-queue';
function customPlugin(): QueuePlugin {
return {
async init({ queue }) {
console.log(Plugin initialized for queue: ${queue.name});Starting job ${job.id}
return async () => console.log('Plugin cleanup');
},
async beforeJob(job) {
console.log();Job ${job.id} failed:
},
async afterJob(job, error) {
if (error) {
console.error(, error);Job ${job.id} completed
} else {
console.log();`
}
}
};
}
Run the test suite:
`bash`
pnpm test
Build the project:
`bash`
pnpm run build
Type checking:
`bash``
pnpm run lint
The library uses an event-based architecture:
1. Abstract Queue - Common interface and job processing logic
2. Drivers - Storage-specific implementations (DB, SQS, File)
3. Event Handlers - Functions that process specific job types
4. Type Safety - Compile-time validation of payloads and options
5. Events - Lifecycle hooks for monitoring and cross-cutting concerns
Benefits:
- Swap drivers without changing job code
- Add new storage backends easily
- Type-safe job payloads and options
- Test with mock implementations
- Scale workers independently
- Monitor via events
MIT