nandi.uk
This repository has no description
1# Distributed Event System Implementation with BullMQ on Fly.io
2
3This document provides a comprehensive guide for implementing a distributed event system using BullMQ and Redis/Valkey on Fly.io, specifically designed for high-frequency social features like notifications and activity feeds.
4
5## Overview
6
7Our current in-memory domain event system works well for single-instance deployments, but as we scale to multiple regions and instances on Fly.io, we need a distributed approach. For applications with frequent URL additions that trigger notifications and social activity feeds, **BullMQ with Redis/Valkey is the recommended solution**.
8
9## Why BullMQ + Redis/Valkey?
10
11### Perfect for Social Features
12
13- **High Throughput**: Handle thousands of events per second for URL additions
14- **Real-time Processing**: Near-instant notifications and feed updates
15- **Reliable Delivery**: Built-in retry logic ensures notifications aren't lost
16- **Rate Limiting**: Prevent overwhelming external services (email, push notifications)
17- **Priority Queues**: Process notifications faster than feed updates
18
19### Technical Advantages
20
21- **Excellent TypeScript Support**: First-class TypeScript integration
22- **Rich Feature Set**: Delays, retries, rate limiting, job scheduling
23- **Great Monitoring**: Built-in dashboard and metrics
24- **Battle-tested**: Used in production by many high-scale applications
25- **Flexible Scaling**: Scale workers independently from web servers
26
27## Architecture Overview
28
29```
30┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
31│ Web Server │ │ Redis/Valkey │ │ Worker Nodes │
32│ │ │ │ │ │
33│ Domain Events ──┼───▶│ BullMQ Queues ├───▶│ Event Handlers │
34│ │ │ │ │ │
35│ - Card Added │ │ - Notifications │ │ - Send Emails │
36│ - URL Shared │ │ - Feeds │ │ - Update Feeds │
37│ - Collection │ │ - Analytics │ │ - Track Events │
38└─────────────────┘ └─────────────────┘ └─────────────────┘
39```
40
41## Process Flow: How Events Get Triggered
42
43Understanding how events flow from HTTP requests to worker processes is crucial for debugging and scaling:
44
45### 1. Web Process (HTTP Request Triggered)
46
47```
48HTTP Request → Use Case → Domain Logic → Event Publishing → HTTP Response
49 ↓ ↓ ↓ ↓ ↓
50POST /cards AddCardToLib card.addTo() publishEvents() 200 OK
51```
52
53**What happens:**
54
55- User makes HTTP request to your API
56- Express/Fastify routes to use case
57- Use case executes domain logic (adds domain events to aggregate)
58- Use case saves to database
59- Use case publishes events to Redis queue via `BullMQEventPublisher`
60- HTTP response returned immediately (don't wait for event processing)
61
62### 2. Redis Queue (Event Storage)
63
64```
65BullMQEventPublisher → Redis Queue → Job Storage
66 ↓ ↓ ↓
67 Serialize Event Store in Queue Wait for Worker
68```
69
70**What happens:**
71
72- Events are serialized to JSON
73- Stored in Redis with retry/priority configuration
74- BullMQ manages job lifecycle, retries, and failure handling
75
76### 3. Worker Process (Polling Triggered)
77
78```
79Worker Polling → Job Found → Event Handler → Job Complete
80 ↓ ↓ ↓ ↓
81 redis.poll() processJob() handler.handle() ack/nack
82```
83
84**What happens:**
85
86- Worker processes continuously poll Redis for new jobs
87- When job found, BullMQ calls your `processJob()` method
88- Event is reconstructed from serialized data
89- Your event handler is called with the reconstructed event
90- Job marked as complete or failed based on handler result
91
92### 4. Key Differences from Web Process
93
94| Aspect | Web Process | Worker Process |
95| ---------------- | -------------------- | ----------------------- |
96| **Trigger** | HTTP Request | Redis Job Available |
97| **Lifecycle** | Request/Response | Long-running polling |
98| **Scaling** | Scale with traffic | Scale with queue depth |
99| **Failure** | Return error to user | Retry job automatically |
100| **Dependencies** | Database, Redis | Database, External APIs |
101
102## Implementation Guide
103
104### Step 1: Install Dependencies
105
106```bash
107npm install bullmq ioredis
108npm install --save-dev @types/ioredis
109```
110
111### Step 2: Infrastructure Layer - Event Publisher
112
113Create the BullMQ event publisher that implements the `IEventPublisher` interface:
114
115```typescript
116// src/shared/infrastructure/events/BullMQEventPublisher.ts
117import { Queue } from 'bullmq';
118import Redis from 'ioredis';
119import { IEventPublisher } from '../../application/events/IEventPublisher';
120import { IDomainEvent } from '../../domain/events/IDomainEvent';
121import { Result, ok, err } from '../../core/Result';
122
123export class BullMQEventPublisher implements IEventPublisher {
124 private queues: Map<string, Queue> = new Map();
125
126 constructor(private redisConnection: Redis) {}
127
128 async publishEvents(events: IDomainEvent[]): Promise<Result<void>> {
129 try {
130 for (const event of events) {
131 await this.publishSingleEvent(event);
132 }
133 return ok(undefined);
134 } catch (error) {
135 return err(error as Error);
136 }
137 }
138
139 private async publishSingleEvent(event: IDomainEvent): Promise<void> {
140 const queueConfig = this.getQueueConfig(event);
141
142 if (!this.queues.has(queueConfig.name)) {
143 this.queues.set(
144 queueConfig.name,
145 new Queue(queueConfig.name, {
146 connection: this.redisConnection,
147 defaultJobOptions: queueConfig.options,
148 }),
149 );
150 }
151
152 const queue = this.queues.get(queueConfig.name)!;
153 await queue.add(event.constructor.name, {
154 eventType: event.constructor.name,
155 aggregateId: event.getAggregateId().toString(),
156 dateTimeOccurred: event.dateTimeOccurred.toISOString(),
157 // Serialize the event data
158 cardId: (event as any).cardId?.getValue?.()?.toString(),
159 curatorId: (event as any).curatorId?.value,
160 });
161 }
162
163 private getQueueConfig(event: IDomainEvent) {
164 // Route events to appropriate queues
165 if (event.constructor.name === 'CardAddedToLibraryEvent') {
166 return {
167 name: 'notifications',
168 options: {
169 priority: 1,
170 attempts: 5,
171 backoff: { type: 'exponential' as const, delay: 1000 },
172 removeOnComplete: 100,
173 removeOnFail: 50,
174 },
175 };
176 }
177
178 return {
179 name: 'events',
180 options: {
181 priority: 2,
182 attempts: 3,
183 backoff: { type: 'exponential' as const, delay: 2000 },
184 removeOnComplete: 50,
185 removeOnFail: 25,
186 },
187 };
188 }
189
190 async close(): Promise<void> {
191 await Promise.all(
192 Array.from(this.queues.values()).map((queue) => queue.close()),
193 );
194 }
195}
196```
197
198### Step 3: Infrastructure Layer - Event Subscriber
199
200Create the BullMQ event subscriber that implements the `IEventSubscriber` interface:
201
202```typescript
203// src/shared/infrastructure/events/BullMQEventSubscriber.ts
204import { Worker, Job } from 'bullmq';
205import Redis from 'ioredis';
206import {
207 IEventSubscriber,
208 IEventHandler,
209} from '../../application/events/IEventSubscriber';
210import { IDomainEvent } from '../../domain/events/IDomainEvent';
211import { CardAddedToLibraryEvent } from '../../../modules/cards/domain/events/CardAddedToLibraryEvent';
212import { CardId } from '../../../modules/cards/domain/value-objects/CardId';
213import { CuratorId } from '../../../modules/cards/domain/value-objects/CuratorId';
214
215export class BullMQEventSubscriber implements IEventSubscriber {
216 private workers: Worker[] = [];
217 private handlers: Map<string, IEventHandler<any>> = new Map();
218
219 constructor(private redisConnection: Redis) {}
220
221 async subscribe<T extends IDomainEvent>(
222 eventType: string,
223 handler: IEventHandler<T>,
224 ): Promise<void> {
225 this.handlers.set(eventType, handler);
226 }
227
228 async start(): Promise<void> {
229 // Start workers for different queues
230 const queues = ['notifications', 'events'];
231
232 for (const queueName of queues) {
233 const worker = new Worker(
234 queueName,
235 async (job: Job) => {
236 await this.processJob(job);
237 },
238 {
239 connection: this.redisConnection,
240 concurrency: queueName === 'notifications' ? 5 : 15,
241 },
242 );
243
244 worker.on('completed', (job) => {
245 console.log(`Job ${job.id} completed successfully`);
246 });
247
248 worker.on('failed', (job, err) => {
249 console.error(`Job ${job?.id} failed:`, err);
250 });
251
252 worker.on('error', (err) => {
253 console.error('Worker error:', err);
254 });
255
256 this.workers.push(worker);
257 }
258 }
259
260 async stop(): Promise<void> {
261 await Promise.all(this.workers.map((worker) => worker.close()));
262 this.workers = [];
263 }
264
265 private async processJob(job: Job): Promise<void> {
266 const eventData = job.data;
267 const eventType = eventData.eventType;
268
269 const handler = this.handlers.get(eventType);
270 if (!handler) {
271 console.warn(`No handler registered for event type: ${eventType}`);
272 return;
273 }
274
275 const event = this.reconstructEvent(eventData);
276 const result = await handler.handle(event);
277
278 if (result.isErr()) {
279 throw result.error;
280 }
281 }
282
283 private reconstructEvent(eventData: any): IDomainEvent {
284 if (eventData.eventType === 'CardAddedToLibraryEvent') {
285 const cardId = CardId.create(eventData.cardId).unwrap();
286 const curatorId = CuratorId.create(eventData.curatorId).unwrap();
287
288 const event = new CardAddedToLibraryEvent(cardId, curatorId);
289 (event as any).dateTimeOccurred = new Date(eventData.dateTimeOccurred);
290
291 return event;
292 }
293
294 throw new Error(`Unknown event type: ${eventData.eventType}`);
295 }
296}
297```
298
299### Step 4: Update Event Handler Registry
300
301Update your existing registry to publish events to the distributed system:
302
303```typescript
304// src/shared/infrastructure/events/EventHandlerRegistry.ts
305import { DomainEvents } from '../../domain/events/DomainEvents';
306import { CardAddedToLibraryEvent } from '../../../modules/cards/domain/events/CardAddedToLibraryEvent';
307import { IEventPublisher } from '../../application/events/IEventPublisher';
308
309export class EventHandlerRegistry {
310 constructor(private eventPublisher: IEventPublisher) {}
311
312 registerAllHandlers(): void {
313 // Register distributed event publishing
314 DomainEvents.register(async (event: CardAddedToLibraryEvent) => {
315 try {
316 await this.eventPublisher.publishEvents([event]);
317 } catch (error) {
318 console.error('Error publishing event to BullMQ:', error);
319 // Don't fail the main operation if event publishing fails
320 }
321 }, CardAddedToLibraryEvent.name);
322 }
323
324 clearAllHandlers(): void {
325 DomainEvents.clearHandlers();
326 }
327}
328```
329
330### Step 5: Fly.io Infrastructure Setup
331
332#### Set up Redis with Fly
333
334```bash
335# Create an Upstash Redis database via Fly
336fly redis create --name myapp-redis --region ord
337
338# For production, add replica regions for better performance
339fly redis create --name myapp-redis --region ord --replica-regions fra,iad
340
341# Check available plans if you need more capacity
342fly redis plans
343
344# View your Redis databases
345fly redis list
346```
347
348After creation, Fly will provide you with connection details. You can also check the status:
349
350```bash
351# Check Redis status and get connection info
352fly redis status myapp-redis
353
354# Connect to Redis for testing
355fly redis connect myapp-redis
356```
357
358#### Configure fly.toml for Worker Processes
359
360Fly.io supports multiple process types in a single app. Here's how to configure your `fly.toml` to run both web servers and worker processes:
361
362```toml
363# fly.toml
364app = "myapp"
365primary_region = "sea"
366
367[build]
368 dockerfile = "Dockerfile"
369
370# Define different process types
371[processes]
372 web = "npm start"
373 notification-worker = "npm run worker:notifications"
374 feed-worker = "npm run worker:feeds"
375
376[env]
377 NODE_ENV = "production"
378 # Redis URL will be automatically set by Fly when you attach the Redis database
379
380# HTTP service configuration - ONLY applies to web processes
381[http_service]
382 internal_port = 3000
383 force_https = true
384 auto_stop_machines = 'stop'
385 auto_start_machines = true
386 min_machines_running = 0
387 processes = ['web'] # Only web processes handle HTTP traffic
388
389# Default VM configuration for all processes
390[[vm]]
391 memory = '1gb'
392 cpu_kind = 'shared'
393 cpus = 1
394
395# Override VM settings for specific process types
396[[vm]]
397 processes = ['notification-worker']
398 memory = '512mb' # Workers typically need less memory
399 cpu_kind = 'shared'
400 cpus = 1
401
402[[vm]]
403 processes = ['feed-worker']
404 memory = '512mb'
405 cpu_kind = 'shared'
406 cpus = 1
407
408[deploy]
409 strategy = "rolling"
410```
411
412**Key Configuration Points:**
413
4141. **Process Types**: Define each worker as a separate process type
4152. **HTTP Service**: Only applies to web processes (workers don't need HTTP)
4163. **VM Configuration**: Can be customized per process type
4174. **Auto-scaling**: Workers can have different scaling rules than web processes
418
419#### Update package.json Scripts
420
421Add worker scripts to your `package.json`:
422
423```json
424// package.json
425{
426 "scripts": {
427 "start": "node dist/index.js",
428 "worker:notifications": "node dist/workers/notification-worker.js",
429 "worker:feeds": "node dist/workers/feed-worker.js",
430 "build": "tsup",
431 "dev": "concurrently \"tsc --watch\" \"nodemon dist/index.js\""
432 }
433}
434```
435
436**Important Notes:**
437
438- Workers and web processes use the same build output (`dist/`)
439- All processes are built together with `npm run build`
440- Each process type runs a different entry point
441
442```typescript
443// src/workers/notification-worker.ts
444import Redis from 'ioredis';
445import { BullMQEventSubscriber } from '../shared/infrastructure/events/BullMQEventSubscriber';
446import { CardAddedToLibraryEventHandler } from '../modules/notifications/application/eventHandlers/CardAddedToLibraryEventHandler';
447import { EnvironmentConfigService } from '../shared/infrastructure/config/EnvironmentConfigService';
448
449async function startNotificationWorker() {
450 console.log('Starting notification worker...');
451
452 const configService = new EnvironmentConfigService();
453
454 // Connect to Redis
455 const redisUrl = configService.get('REDIS_URL');
456 if (!redisUrl) {
457 throw new Error('REDIS_URL environment variable is required');
458 }
459
460 const redis = new Redis(redisUrl, {
461 maxRetriesPerRequest: 3,
462 retryDelayOnFailover: 100,
463 lazyConnect: true,
464 });
465
466 // Test Redis connection
467 try {
468 await redis.ping();
469 console.log('Connected to Redis successfully');
470 } catch (error) {
471 console.error('Failed to connect to Redis:', error);
472 process.exit(1);
473 }
474
475 // Create subscriber
476 const eventSubscriber = new BullMQEventSubscriber(redis);
477
478 // Create event handlers (wire up your services here)
479 const notificationHandler = new CardAddedToLibraryEventHandler(
480 notificationService,
481 );
482
483 // Register handlers
484 await eventSubscriber.subscribe(
485 'CardAddedToLibraryEvent',
486 notificationHandler,
487 );
488
489 // Start the worker - THIS IS WHAT TRIGGERS YOUR HANDLERS!
490 await eventSubscriber.start();
491
492 console.log('Notification worker started and listening for events...');
493
494 // Graceful shutdown
495 process.on('SIGTERM', async () => {
496 console.log('Shutting down notification worker...');
497 await eventSubscriber.stop();
498 await redis.quit();
499 process.exit(0);
500 });
501}
502
503startNotificationWorker().catch(console.error);
504```
505
506### Step 5: Service Factory Integration
507
508Update your service factory to use the distributed event system:
509
510```typescript
511// In your ServiceFactory
512export class ServiceFactory {
513 static create(
514 configService: EnvironmentConfigService,
515 repositories: Repositories,
516 ): Services {
517 // ... existing services
518
519 // Redis connection - Fly Redis provides a full connection URL
520 const redisUrl = configService.get('REDIS_URL');
521 if (!redisUrl) {
522 throw new Error('REDIS_URL environment variable is required');
523 }
524
525 const redis = new Redis(redisUrl, {
526 maxRetriesPerRequest: 3,
527 retryDelayOnFailover: 100,
528 lazyConnect: true,
529 });
530
531 // Event publisher
532 const eventPublisher = new BullMQEventPublisher(redis);
533
534 // Event handler registry
535 const eventHandlerRegistry = new EventHandlerRegistry(eventPublisher);
536 eventHandlerRegistry.registerAllHandlers();
537
538 return {
539 // ... existing services
540 eventHandlerRegistry,
541 eventPublisher,
542 };
543 }
544}
545```
546
547### Step 6: Create Use Cases with Event Publishing
548
549Update your use cases to extend `BaseUseCase` and publish events:
550
551```typescript
552// src/modules/cards/application/use-cases/AddCardToLibraryUseCase.ts
553import { BaseUseCase } from '../../../../shared/core/UseCase';
554import { Result, ok, err } from '../../../../shared/core/Result';
555import { IEventPublisher } from '../../../../shared/application/events/IEventPublisher';
556import { ICardRepository } from '../../domain/ICardRepository';
557
558interface AddCardToLibraryRequest {
559 cardId: string;
560 userId: string;
561}
562
563export class AddCardToLibraryUseCase extends BaseUseCase<
564 AddCardToLibraryRequest,
565 Result<void>
566> {
567 constructor(
568 private cardRepository: ICardRepository,
569 eventPublisher: IEventPublisher,
570 ) {
571 super(eventPublisher);
572 }
573
574 async execute(request: AddCardToLibraryRequest): Promise<Result<void>> {
575 // 1. Get the card
576 const cardResult = await this.cardRepository.findById(
577 CardId.create(request.cardId).unwrap(),
578 );
579 if (cardResult.isErr()) return err(cardResult.error);
580
581 const card = cardResult.value;
582 if (!card) return err(new Error('Card not found'));
583
584 // 2. Execute domain logic (adds events to aggregate)
585 const curatorId = CuratorId.create(request.userId).unwrap();
586 const addResult = card.addToLibrary(curatorId);
587 if (addResult.isErr()) return err(addResult.error);
588
589 // 3. Save to repository
590 const saveResult = await this.cardRepository.save(card);
591 if (saveResult.isErr()) return err(saveResult.error);
592
593 // 4. Publish events after successful save
594 const publishResult = await this.publishEventsForAggregate(card);
595 if (publishResult.isErr()) {
596 console.error('Failed to publish events:', publishResult.error);
597 // Don't fail the operation if event publishing fails
598 }
599
600 return ok(undefined);
601 }
602}
603```
604
605## Deployment Strategy
606
607### Step-by-Step Deployment Process
608
609#### 1. Create and Configure Redis
610
611```bash
612# Create Redis database with replicas for better performance
613fly redis create --name myapp-redis --region ord --replica-regions fra,iad
614
615# Attach Redis to your app (sets REDIS_URL environment variable)
616fly redis attach myapp-redis
617
618# Verify Redis connection details
619fly redis status myapp-redis
620```
621
622#### 2. Update Your Application Configuration
623
624Ensure your `fly.toml` includes worker processes:
625
626```toml
627[processes]
628 web = "npm start"
629 notification-worker = "npm run worker:notifications"
630 feed-worker = "npm run worker:feeds"
631
632[http_service]
633 processes = ['web'] # Only web processes handle HTTP
634```
635
636#### 3. Deploy All Processes
637
638```bash
639# Deploy the entire application (web + workers)
640fly deploy
641
642# Check deployment status
643fly status
644
645# View all running processes
646fly ps
647```
648
649#### 4. Scale Worker Processes
650
651```bash
652# Scale different process types independently
653fly scale count web=2 notification-worker=2 feed-worker=3
654
655# Scale to specific regions
656fly scale count web=2 --region ord
657fly scale count notification-worker=1 --region ord
658fly scale count notification-worker=1 --region fra
659fly scale count feed-worker=2 --region ord
660fly scale count feed-worker=1 --region fra
661
662# Check current scaling
663fly scale show
664```
665
666#### 5. Monitor Worker Health
667
668```bash
669# Monitor worker logs
670fly logs --process notification-worker
671fly logs --process feed-worker
672
673# Check specific worker instances
674fly logs --process notification-worker --region ord
675
676# Monitor Redis connectivity
677fly redis connect myapp-redis
678```
679
680### Understanding Fly.io Worker Deployment
681
682#### How Fly.io Handles Multiple Process Types
683
684**Single App, Multiple Process Types:**
685
686- All processes (web, workers) are part of the same Fly app
687- They share the same Docker image and environment variables
688- Each process type can be scaled independently
689- Workers run as long-lived background processes
690
691**Deployment Flow:**
692
6931. `fly deploy` builds one Docker image
6942. Fly creates different machine types based on `[processes]` config
6953. Each machine runs the specified command for its process type
6964. HTTP service only applies to processes listed in `[http_service].processes`
697
698#### Process Type Differences
699
700| Aspect | Web Process | Worker Processes |
701| ------------------- | --------------------------- | ------------------------------ |
702| **Command** | `npm start` | `npm run worker:notifications` |
703| **HTTP Traffic** | ✅ Receives HTTP requests | ❌ No HTTP traffic |
704| **Load Balancer** | ✅ Behind Fly proxy | ❌ Direct process |
705| **Health Checks** | HTTP endpoint | Process running |
706| **Scaling Trigger** | HTTP traffic | Queue depth |
707| **Auto-stop** | Can auto-stop when idle | Should run continuously |
708| **Regions** | Scale based on user traffic | Scale based on workload |
709
710#### Worker-Specific Configuration
711
712**Prevent Auto-stopping Workers:**
713
714```toml
715# In fly.toml - workers should not auto-stop
716[http_service]
717 auto_stop_machines = 'stop'
718 processes = ['web'] # Only web processes auto-stop
719
720# Or disable auto-stop entirely for workers
721[[vm]]
722 processes = ['notification-worker', 'feed-worker']
723 auto_stop_machines = false
724```
725
726**Worker Health Monitoring:**
727
728```bash
729# Workers don't have HTTP health checks
730# Monitor via logs and process status
731fly logs --process notification-worker --follow
732
733# Check if worker processes are running
734fly ps | grep worker
735
736# SSH into worker for debugging
737fly ssh console --process notification-worker
738```
739
740#### Environment Variables and Secrets
741
742**Shared Environment:**
743
744- All process types share the same environment variables
745- Redis URL, database URL, etc. are available to all processes
746- Secrets are shared across all process types
747
748```bash
749# Set environment variables for all processes
750fly secrets set DATABASE_URL="postgresql://..."
751fly secrets set REDIS_URL="redis://..."
752
753# Check environment in worker
754fly ssh console --process notification-worker
755env | grep REDIS_URL
756```
757
758#### Deployment Dependencies and Order
759
760**Critical Deployment Order:**
761
7621. **Redis First**: Must exist before workers can start
7632. **Database**: Must be accessible to both web and workers
7643. **Deploy**: All processes deploy together
7654. **Scale**: Scale workers after successful deployment
766
767```bash
768# 1. Ensure Redis exists
769fly redis status myapp-redis || fly redis create --name myapp-redis
770
771# 2. Ensure Redis is attached
772fly redis attach myapp-redis
773
774# 3. Deploy all processes
775fly deploy
776
777# 4. Verify all process types started
778fly ps
779
780# 5. Scale workers as needed
781fly scale count notification-worker=2 feed-worker=3
782```
783
784**Common Deployment Issues:**
785
7861. **Workers Exit Immediately**
787
788 ```bash
789 # Check worker logs for startup errors
790 fly logs --process notification-worker
791
792 # Common causes:
793 # - Missing REDIS_URL
794 # - Redis connection failed
795 # - Missing dependencies in package.json
796 ```
797
7982. **Workers Can't Connect to Redis**
799
800 ```bash
801 # Verify Redis attachment
802 fly redis status myapp-redis
803
804 # Test Redis connection from worker
805 fly ssh console --process notification-worker
806 node -e "const Redis = require('ioredis'); const r = new Redis(process.env.REDIS_URL); r.ping().then(console.log)"
807 ```
808
8093. **Workers Not Processing Jobs**
810
811 ```bash
812 # Check if jobs are being queued
813 fly redis connect myapp-redis
814 > KEYS *
815 > LLEN bull:notifications:waiting
816
817 # Check worker logs for processing
818 fly logs --process notification-worker --follow
819 ```
820
821### Worker Scaling Guidelines
822
823#### Notification Workers (External API Calls)
824
825```bash
826# Conservative scaling for external APIs
827fly scale count notification-worker=2 --region ord
828fly scale count notification-worker=1 --region fra
829
830# Monitor and adjust based on:
831# - External API rate limits
832# - Queue depth
833# - Processing time
834```
835
836**Configuration:**
837
838- **Concurrency**: 5-10 jobs per worker (respect API limits)
839- **Regions**: 2-3 regions max (avoid hitting API limits from too many IPs)
840- **Memory**: 512MB usually sufficient
841- **Scaling Trigger**: Queue depth > 100 jobs
842
843#### Feed Workers (Database Operations)
844
845```bash
846# More aggressive scaling for database operations
847fly scale count feed-worker=3 --region ord
848fly scale count feed-worker=2 --region fra
849
850# Scale based on:
851# - Database connection limits
852# - Queue processing speed
853# - Regional user distribution
854```
855
856**Configuration:**
857
858- **Concurrency**: 10-20 jobs per worker (database can handle more)
859- **Regions**: Match your user distribution
860- **Memory**: 512MB-1GB depending on data processing
861- **Scaling Trigger**: Queue depth > 50 jobs
862
863#### Scaling Commands Reference
864
865```bash
866# View current scaling
867fly scale show
868
869# Scale specific process types
870fly scale count web=2 notification-worker=2 feed-worker=3
871
872# Scale to specific regions
873fly scale count notification-worker=1 --region ord
874fly scale count notification-worker=1 --region fra
875
876# Scale with memory adjustments
877fly scale memory 1gb --process feed-worker
878fly scale memory 512mb --process notification-worker
879
880# Auto-scaling (if using Fly's auto-scaling features)
881fly autoscale set min=1 max=5 --process notification-worker
882fly autoscale set min=2 max=8 --process feed-worker
883```
884
885#### Monitoring Scaling Effectiveness
886
887```bash
888# Monitor queue depths
889fly redis connect myapp-redis
890> LLEN bull:notifications:waiting
891> LLEN bull:feeds:waiting
892
893# Monitor worker performance
894fly logs --process notification-worker | grep "Job.*completed"
895fly logs --process feed-worker | grep "processing time"
896
897# Check resource usage
898fly metrics --process notification-worker
899fly metrics --process feed-worker
900```
901
902### Monitoring and Observability
903
904#### BullMQ Dashboard
905
906```typescript
907// Optional: Add Bull Dashboard for monitoring
908import { createBullBoard } from '@bull-board/api';
909import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
910import { ExpressAdapter } from '@bull-board/express';
911
912const serverAdapter = new ExpressAdapter();
913serverAdapter.setBasePath('/admin/queues');
914
915const { addQueue } = createBullBoard({
916 queues: [new BullMQAdapter(notificationQueue), new BullMQAdapter(feedQueue)],
917 serverAdapter: serverAdapter,
918});
919
920app.use('/admin/queues', serverAdapter.getRouter());
921```
922
923#### Custom Metrics
924
925```typescript
926// Track event processing metrics
927export class EventMetrics {
928 private static eventCounts = new Map<string, number>();
929 private static processingTimes = new Map<string, number[]>();
930
931 static recordEvent(eventType: string, processingTime: number): void {
932 // Increment count
933 const count = this.eventCounts.get(eventType) || 0;
934 this.eventCounts.set(eventType, count + 1);
935
936 // Track processing time
937 const times = this.processingTimes.get(eventType) || [];
938 times.push(processingTime);
939 if (times.length > 100) times.shift(); // Keep last 100
940 this.processingTimes.set(eventType, times);
941 }
942
943 static getMetrics() {
944 const metrics: any = {};
945
946 for (const [eventType, count] of this.eventCounts) {
947 const times = this.processingTimes.get(eventType) || [];
948 const avgTime =
949 times.length > 0 ? times.reduce((a, b) => a + b, 0) / times.length : 0;
950
951 metrics[eventType] = {
952 count,
953 averageProcessingTime: avgTime,
954 recentProcessingTimes: times.slice(-10),
955 };
956 }
957
958 return metrics;
959 }
960}
961```
962
963## Testing Strategy
964
965### Unit Tests
966
967```typescript
968// Test event publishing
969describe('BullMQEventPublisher', () => {
970 it('should publish CardAddedToLibraryEvent to notifications queue', async () => {
971 const mockQueue = {
972 add: jest.fn().mockResolvedValue({}),
973 };
974
975 const publisher = new BullMQEventPublisher(redisConnection);
976 (publisher as any).queues.set('notifications', mockQueue);
977
978 const event = new CardAddedToLibraryEvent(
979 cardId,
980 curatorId,
981 CardTypeEnum.URL,
982 );
983 await publisher.publish(event);
984
985 expect(mockQueue.add).toHaveBeenCalledWith(
986 'CardAddedToLibraryEvent',
987 expect.objectContaining({
988 cardId,
989 curatorId,
990 cardType: CardTypeEnum.URL,
991 }),
992 );
993 });
994});
995```
996
997### Integration Tests
998
999```typescript
1000// Test end-to-end event flow
1001describe('Event Processing Integration', () => {
1002 it('should process CardAddedToLibraryEvent through the queue', async () => {
1003 const mockNotificationHandler = {
1004 handle: jest.fn().mockResolvedValue({}),
1005 };
1006
1007 // Publish event
1008 const event = new CardAddedToLibraryEvent(
1009 cardId,
1010 curatorId,
1011 CardTypeEnum.URL,
1012 );
1013 await eventPublisher.publish(event);
1014
1015 // Wait for processing
1016 await new Promise((resolve) => setTimeout(resolve, 1000));
1017
1018 // Verify handler was called
1019 expect(mockNotificationHandler.handle).toHaveBeenCalledWith(
1020 expect.objectContaining({
1021 cardId,
1022 curatorId,
1023 cardType: CardTypeEnum.URL,
1024 }),
1025 );
1026 });
1027});
1028```
1029
1030## Performance Optimization
1031
1032### Queue Configuration
1033
1034```typescript
1035// Optimize for your specific use case
1036const queueOptions = {
1037 // For high-frequency events
1038 defaultJobOptions: {
1039 removeOnComplete: 100, // Keep successful jobs for debugging
1040 removeOnFail: 50, // Keep failed jobs for analysis
1041 attempts: 3, // Retry failed jobs
1042 backoff: {
1043 type: 'exponential',
1044 delay: 2000, // Start with 2s delay
1045 },
1046 },
1047
1048 // Connection pooling
1049 connection: {
1050 ...redisConnection,
1051 maxRetriesPerRequest: 3,
1052 retryDelayOnFailover: 100,
1053 lazyConnect: true,
1054 },
1055};
1056```
1057
1058### Worker Optimization
1059
1060```typescript
1061// Optimize worker settings
1062const workerOptions = {
1063 concurrency: process.env.NODE_ENV === 'production' ? 10 : 5,
1064
1065 // Batch processing for better performance
1066 stalledInterval: 30000,
1067 maxStalledCount: 1,
1068
1069 // Memory management
1070 removeOnComplete: 100,
1071 removeOnFail: 50,
1072};
1073```
1074
1075## Troubleshooting
1076
1077### Common Issues
1078
10791. **Events not processing**: Check Redis connection and worker status
10802. **High memory usage**: Adjust `removeOnComplete` and `removeOnFail` settings
10813. **Slow processing**: Increase worker concurrency or add more workers
10824. **Failed jobs**: Check error logs and adjust retry settings
1083
1084### Debugging Commands
1085
1086```bash
1087# Check worker status
1088fly logs --app myapp --process notification-worker
1089
1090# Monitor Redis
1091fly redis connect myapp-redis
1092> MONITOR
1093
1094# Check Redis status and connection info
1095fly redis status myapp-redis
1096
1097# Proxy to Redis for local debugging
1098fly redis proxy myapp-redis
1099
1100# Check queue status
1101fly ssh console --app myapp
1102> npm run queue:status
1103```
1104
1105## Complete Deployment Checklist
1106
1107### Phase 1: Infrastructure Setup
1108
11091. **Create Redis Database**
1110
1111 ```bash
1112 fly redis create --name myapp-redis --region ord --replica-regions fra,iad
1113 fly redis attach myapp-redis
1114 fly redis status myapp-redis # Verify creation
1115 ```
1116
11172. **Update Application Configuration**
1118 - ✅ Add worker processes to `fly.toml`
1119 - ✅ Add worker scripts to `package.json`
1120 - ✅ Configure HTTP service for web processes only
1121 - ✅ Set appropriate VM resources for workers
1122
1123### Phase 2: Code Implementation
1124
11253. **Implement Event System**
1126 - ✅ Create `BullMQEventPublisher`
1127 - ✅ Create `BullMQEventSubscriber`
1128 - ✅ Update service factory to use distributed events
1129 - ✅ Create worker entry points
1130
11314. **Create Worker Files**
1132 ```bash
1133 # Ensure these files exist:
1134 ls src/workers/notification-worker.ts
1135 ls src/workers/feed-worker.ts
1136 ```
1137
1138### Phase 3: Deployment
1139
11405. **Deploy Application**
1141
1142 ```bash
1143 # Build and deploy all processes
1144 fly deploy
1145
1146 # Verify all process types are running
1147 fly ps
1148 fly status
1149 ```
1150
11516. **Verify Worker Startup**
1152
1153 ```bash
1154 # Check worker logs for successful startup
1155 fly logs --process notification-worker
1156 fly logs --process feed-worker
1157
1158 # Look for these messages:
1159 # "Connected to Redis successfully"
1160 # "Worker started and listening for events..."
1161 ```
1162
1163### Phase 4: Testing and Scaling
1164
11657. **Test Event Flow**
1166
1167 ```bash
1168 # Test Redis connection from workers
1169 fly ssh console --process notification-worker
1170 node -e "const Redis = require('ioredis'); const r = new Redis(process.env.REDIS_URL); r.ping().then(console.log)"
1171
1172 # Monitor Redis for job activity
1173 fly redis connect myapp-redis
1174 > MONITOR
1175 ```
1176
11778. **Scale Workers Based on Load**
1178
1179 ```bash
1180 # Start conservative
1181 fly scale count notification-worker=1 feed-worker=2
1182
1183 # Monitor and scale up as needed
1184 fly scale count notification-worker=2 feed-worker=3
1185
1186 # Add regional distribution
1187 fly scale count notification-worker=1 --region ord
1188 fly scale count notification-worker=1 --region fra
1189 ```
1190
1191### Phase 5: Monitoring and Optimization
1192
11939. **Set Up Monitoring**
1194
1195 ```bash
1196 # Monitor worker health
1197 fly logs --process notification-worker --follow
1198 fly logs --process feed-worker --follow
1199
1200 # Check Redis performance
1201 fly redis status myapp-redis
1202 fly redis dashboard myapp-redis
1203 ```
1204
120510. **Performance Optimization**
1206
1207 ```bash
1208 # Monitor queue depths
1209 fly redis connect myapp-redis
1210 > LLEN bull:notifications:waiting
1211 > LLEN bull:feeds:waiting
1212
1213 # Adjust scaling based on metrics
1214 fly scale count notification-worker=3 # If queue depth consistently high
1215 fly scale memory 1gb --process feed-worker # If memory usage high
1216 ```
1217
1218### Troubleshooting Common Issues
1219
1220**Workers Not Starting:**
1221
1222```bash
1223# Check for missing dependencies
1224fly logs --process notification-worker | grep "Error"
1225
1226# Verify Redis connection
1227fly ssh console --process notification-worker
1228echo $REDIS_URL
1229```
1230
1231**Jobs Not Processing:**
1232
1233```bash
1234# Check if jobs are being queued
1235fly redis connect myapp-redis
1236> KEYS bull:*
1237> LLEN bull:notifications:waiting
1238
1239# Verify worker registration
1240fly logs --process notification-worker | grep "subscribe"
1241```
1242
1243**High Memory Usage:**
1244
1245```bash
1246# Monitor resource usage
1247fly metrics --process notification-worker
1248fly metrics --process feed-worker
1249
1250# Scale memory if needed
1251fly scale memory 1gb --process feed-worker
1252```
1253
1254### Production Readiness Checklist
1255
1256- ✅ Redis database with replicas in multiple regions
1257- ✅ Worker processes configured in `fly.toml`
1258- ✅ Environment variables properly set
1259- ✅ Workers successfully connecting to Redis
1260- ✅ Event handlers processing jobs
1261- ✅ Monitoring and logging in place
1262- ✅ Scaling strategy defined
1263- ✅ Error handling and retry logic tested
1264
1265This comprehensive deployment guide ensures your distributed event system is properly configured and running on Fly.io with robust worker processes handling your social features at scale.