@aion2hub/shared-queue-utils

Shared Redis queue utilities for Aion2Hub event-driven RSS processing system.

Features

- Queue Management: Create, configure, and manage Redis-based queues
- Worker Processing: Concurrent job processing with retry logic and error handling
- Monitoring: Comprehensive queue health monitoring and metrics
- Event-Driven Architecture: Support for RSS processing pipeline events
- Dead Letter Queues: Automatic handling of failed jobs
- Priority Queues: Support for job prioritization
- Delayed Jobs: Schedule jobs for future execution
- Batch Processing: Process multiple jobs together for efficiency

Installation

``bash
npm install @aion2hub/shared-queue-utils
`

Quick Start

$3

`typescript
import { QueueManager, QUEUE_NAMES } from '@aion2hub/shared-queue-utils';

const queueManager = new QueueManager('redis://localhost:6379');

// Publish a job
await queueManager.publishJob(QUEUE_NAMES.RSS_SCRAPED, {
feedId: 'feed-123',
articles: [...]
}, {
priority: 1,
correlationId: 'scrape-job-456'
});

// Get queue metrics
const metrics = await queueManager.getQueueMetrics(QUEUE_NAMES.RSS_SCRAPED);
console.log(Queue depth: ${metrics.depth});
`

$3

`typescript
import { QueueWorker, createQueueLogger } from '@aion2hub/shared-queue-utils';

const logger = createQueueLogger('rss-processor');

const worker = new QueueWorker(
'redis://localhost:6379',
{
queueName: QUEUE_NAMES.RSS_TRANSLATE,
concurrency: 3,
batchSize: 10,
timeout: 60000
},
async (jobs) => {
// Process batch of translation jobs
for (const job of jobs) {
await translateArticles(job.data);
}
},
logger
);

await worker.start();
`

$3

`typescript
import { QueueMonitor, createQueueLogger } from '@aion2hub/shared-queue-utils';

const logger = createQueueLogger('queue-monitor');
const monitor = new QueueMonitor('redis://localhost:6379', logger);

// Start monitoring with 30-second intervals
monitor.startMonitoring(30000);

// Get current health status
const health = await monitor.getHealthStatus();
console.log(System health: ${health.overall});
`

Queue Types

The system defines several predefined queues for RSS processing:

- rss_scraped: Raw parsed article data from RSS feeds
- rss_translate: Articles ready for translation processing
- rss_classify: Articles ready for AI classification
- rss_ingest: Articles ready for database ingestion
- *_dlq: Dead letter queues for failed jobs

Event Types

$3


`typescript
interface RSSScrapedEvent {
jobId: string;
correlationId: string;
feedId: string;
feedUrl: string;
feedName: string;
timestamp: Date;
articles: Array<{
title: string;
content: string;
url: string;
publishedAt: Date;
author?: string;
language: string;
description?: string;
}>;
metadata: {
priority: number;
retryCount: number;
maxRetries: number;
feedCategory?: string;
};
}
`

$3

`typescript
interface RSSTranslateEvent {
jobId: string;
correlationId: string;
feedId: string;
articles: Array<{
id: string;
title: string;
content: string;
description?: string;
url: string;
publishedAt: Date;
author?: string;
language: string;
needsTranslation: boolean;
}>;
metadata: {
priority: number;
retryCount: number;
maxRetries: number;
};
}
`

Configuration

Default queue configurations are provided but can be customized:

`typescript
import { DEFAULT_QUEUE_CONFIGS, QueueConfig } from '@aion2hub/shared-queue-utils';

const customConfig: QueueConfig = {
...DEFAULT_QUEUE_CONFIGS.rss_scraped,
maxLength: 5000,
retryAttempts: 5
};
`

Error Handling

The system provides comprehensive error handling:

- Automatic Retries: Failed jobs are automatically retried with exponential backoff
- Dead Letter Queues: Jobs that exceed retry limits are moved to dead letter queues
- Circuit Breakers: Workers can be paused/resumed for maintenance
- Timeout Handling: Jobs that exceed timeout limits are automatically failed

Monitoring

Built-in monitoring provides:

- Queue depth and processing metrics
- Health status for all queues
- Redis connection monitoring
- Error rate tracking
- Performance metrics

Best Practices

1. Use Correlation IDs: Track related jobs across the pipeline
2. Set Appropriate Timeouts: Configure timeouts based on job complexity
3. Monitor Queue Health: Set up alerts for queue depth and error rates
4. Handle Failures Gracefully: Implement proper error handling in processors
5. Use Batch Processing: Process multiple jobs together when possible
6. Configure Dead Letter Queues: Always configure DLQs for failed job analysis

Development

`bash


Install dependencies

npm install

Build the package

npm run build

Run tests

npm test

Watch for changes

npm run dev
``

License

MIT