```typescript // src/workers/feed-worker.ts import Redis from 'ioredis'; import { BullMQEventSubscriber } from '../shared/infrastructure/events/BullMQEventSubscriber'; import { QueueNames } from '../shared/infrastructure/events/QueueConfig'; import { CardAddedToLibraryEventHandler } from '../modules/feeds/application/eventHandlers/CardAddedToLibraryEventHandler'; import { FeedService } from '../modules/feeds/application/services/FeedService'; async function startFeedWorker() { const redisConnection = new Redis({ host: process.env.REDIS_HOST, port: parseInt(process.env.REDIS_PORT || '6379'), }); // Create dependencies const feedService = new FeedService(/* dependencies */); // Create event subscriber for feeds queue const feedSubscriber = new BullMQEventSubscriber(redisConnection, { queueName: QueueNames.FEEDS, concurrency: 15 // Optional: override default concurrency }); // Register event handlers await feedSubscriber.subscribe( 'CardAddedToLibraryEvent', new CardAddedToLibraryEventHandler(feedService) ); // Start processing events await feedSubscriber.start(); console.log('Feed Worker started and listening to feeds queue'); // Optional: Graceful shutdown process.on('SIGTERM', async () => { console.log('Shutting down feed worker...'); await feedSubscriber.stop(); await redisConnection.quit(); process.exit(0); }); } startFeedWorker().catch(console.error); ``` And in your event publishing code: ```typescript // Example in a use case or command handler import { BullMQEventPublisher } from '../shared/infrastructure/events/BullMQEventPublisher'; class AddCardToLibraryUseCase { constructor( private eventPublisher: BullMQEventPublisher, private cardRepository: CardRepository ) {} async execute(command: AddCardToLibraryCommand): Promise<Result<void>> { // Create card, save to repository const card = Card.create(/* params */); await this.cardRepository.save(card); // Publish event to feeds queue const event = new CardAddedToLibraryEvent({ cardId: card.id, userId: command.userId }); return this.eventPublisher.publishEvents([event]); } } ``` Key improvements in this implementation: 1. **Queue-Specific Configuration**: Each queue can have its own concurrency, retry, and removal policies 2. **Extensible Routing**: Easy to add new queues and route events 3. **Contextual Logging**: Logs now include queue name for better traceability 4. **Flexible Subscriber Setup**: Can configure concurrency per worker When you're ready to add more queues (like notifications), you'll just: 1. Uncomment queue in `QueueConfig.ts` 2. Update `getTargetQueues()` in publisher 3. Create a new worker similar to the feed worker 4. Add event handlers for that queue The design allows for smooth scaling and extension of your event-driven architecture. Co-authored-by: aider (anthropic/claude-sonnet-4-20250514) <aider@aider.chat>