nandi.uk
This repository has no description
Based on the comprehensive document you've drafted, I'll generate a concise commit message that captures the essence of these architectural changes:
feat: implement standalone firehose worker with direct AT Protocol event handling
Co-authored-by: aider (anthropic/claude-sonnet-4-20250514) <aider@aider.chat>
author
Wesley Finck
co-author
aider (anthropic/claude-sonnet-4-20250514)
date
(Nov 14, 2025, 11:27 AM -0800)
commit
aff0f2d6
aff0f2d66b40a526b169b554152102006d8c7bb4
parent
400ab41c
400ab41c10dc0ae059139a3cac2a54546b86e5c8
+374
-64
+374
-64
.agent/logs/20251114_firehose_layered_arch.md
+374
-64
.agent/logs/20251114_firehose_layered_arch.md
···
4
4
5
5
This document outlines the layered architecture design for handling AT Protocol firehose events within our existing DDD structure. The system will process CREATE, UPDATE, and DELETE events for cards, collections, and collection links while implementing the duplicate detection strategy outlined in the previous document.
6
6
7
+
## Event Flow Architecture
8
+
9
+
The firehose worker operates as a **standalone event producer** that bridges external AT Protocol events to internal domain events, avoiding unnecessary BullMQ abstractions for firehose input while still publishing domain events to the internal event system.
10
+
11
+
```
12
+
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
13
+
│ AT Protocol │───▶│ Firehose Worker │───▶│ Internal Events │
14
+
│ Firehose │ │ (Direct WS) │ │ (BullMQ/Memory) │
15
+
└─────────────────┘ └──────────────────┘ └─────────────────┘
16
+
│
17
+
▼
18
+
┌─────────────────┐
19
+
│ Domain Entities │
20
+
│ (Cards/Collections)
21
+
└─────────────────┘
22
+
```
23
+
24
+
## Full Event Lifecycle Diagram
25
+
26
+
```
27
+
AT Protocol Network
28
+
│
29
+
▼
30
+
┌─────────────────────────────────────────────────────────────────┐
31
+
│ Firehose Worker Process │
32
+
│ ┌─────────────────┐ ┌──────────────────────────────────────┐│
33
+
│ │ @atproto/sync │───▶│ FirehoseEventHandler ││
34
+
│ │ Firehose │ │ ││
35
+
│ │ (WebSocket) │ │ ┌─────────────────────────────────┐ ││
36
+
│ └─────────────────┘ │ │ ProcessFirehoseEventUseCase │ ││
37
+
│ │ │ │ ││
38
+
│ │ │ 1. Check Duplicates │ ││
39
+
│ │ │ 2. Route by Collection Type │ ││
40
+
│ │ │ 3. Process Entity Changes │ ││
41
+
│ │ │ 4. Publish Domain Events │ ││
42
+
│ │ └─────────────────────────────────┘ ││
43
+
│ └──────────────────────────────────────┘│
44
+
└─────────────────────────────────────────────────────────────────┘
45
+
│
46
+
▼
47
+
┌─────────────────────────────────────────────────────────────────┐
48
+
│ Domain Layer Updates │
49
+
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
50
+
│ │ Cards │ │ Collections │ │ Collection Links │ │
51
+
│ │ Repository │ │ Repository │ │ (via Collections) │ │
52
+
│ └─────────────┘ └─────────────┘ └─────────────────────────┘ │
53
+
└─────────────────────────────────────────────────────────────────┘
54
+
│
55
+
▼
56
+
┌─────────────────────────────────────────────────────────────────┐
57
+
│ Internal Event Publishing │
58
+
│ │
59
+
│ ┌─────────────────┐ ┌─────────────────────────┐ │
60
+
│ │ BullMQ Events │ OR │ In-Memory Events │ │
61
+
│ │ (Production) │ │ (Development) │ │
62
+
│ └─────────────────┘ └─────────────────────────┘ │
63
+
└─────────────────────────────────────────────────────────────────┘
64
+
│
65
+
▼
66
+
┌─────────────────────────────────────────────────────────────────┐
67
+
│ Downstream Event Processing │
68
+
│ │
69
+
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
70
+
│ │Feed Worker │ │Search Worker│ │ Analytics Worker │ │
71
+
│ │(BullMQ) │ │(BullMQ) │ │ (Future) │ │
72
+
│ └─────────────┘ └─────────────┘ └─────────────────────────┘ │
73
+
└─────────────────────────────────────────────────────────────────┘
74
+
```
75
+
7
76
## Architecture Layers
8
77
9
78
### Domain Layer
···
378
447
}
379
448
```
380
449
381
-
#### Worker Process
450
+
#### Standalone Worker Process
382
451
383
-
**FirehoseWorkerProcess**
452
+
**FirehoseWorkerProcess** (implements IProcess directly, not BaseWorkerProcess)
384
453
```typescript
385
-
export class FirehoseWorkerProcess extends BaseWorkerProcess {
386
-
constructor(configService: EnvironmentConfigService) {
387
-
super(configService, QueueNames.FIREHOSE);
388
-
}
454
+
export class FirehoseWorkerProcess implements IProcess {
455
+
private firehose?: Firehose;
456
+
private runner?: MemoryRunner;
389
457
390
-
protected createServices(repositories: Repositories): WorkerServices & {
391
-
firehoseService: IFirehoseService;
392
-
} {
393
-
const baseServices = ServiceFactory.createForWorker(this.configService, repositories);
458
+
constructor(
459
+
private configService: EnvironmentConfigService,
460
+
private firehoseEventHandler: FirehoseEventHandler
461
+
) {}
462
+
463
+
async start(): Promise<void> {
464
+
console.log('Starting firehose worker...');
465
+
466
+
const runner = new MemoryRunner({});
467
+
this.runner = runner;
394
468
395
-
// Create firehose-specific services
396
-
const firehoseEventDuplicationService = new DrizzleFirehoseEventDuplicationService(
397
-
repositories.database
398
-
);
399
-
400
-
const processFirehoseEventUseCase = new ProcessFirehoseEventUseCase(
401
-
firehoseEventDuplicationService,
402
-
repositories.atUriResolutionService,
403
-
repositories.cardRepository,
404
-
repositories.collectionRepository,
405
-
baseServices.eventPublisher
406
-
);
407
-
408
-
const firehoseEventHandler = new FirehoseEventHandler(processFirehoseEventUseCase);
409
-
410
-
const firehoseService = new AtProtoFirehoseService(
411
-
firehoseEventHandler,
412
-
this.configService,
413
-
new IdResolver()
414
-
);
469
+
this.firehose = new Firehose({
470
+
service: 'wss://bsky.network',
471
+
runner,
472
+
idResolver: new IdResolver(),
473
+
filterCollections: this.getFilteredCollections(),
474
+
handleEvent: this.handleFirehoseEvent.bind(this),
475
+
onError: this.handleError.bind(this)
476
+
});
477
+
478
+
await this.firehose.start();
479
+
console.log('Firehose worker started');
415
480
416
-
return {
417
-
...baseServices,
418
-
firehoseService
419
-
};
481
+
this.setupShutdownHandlers();
420
482
}
421
483
422
-
protected async validateDependencies(services: any): Promise<void> {
423
-
// Validate firehose dependencies
484
+
private async handleFirehoseEvent(evt: Event): Promise<void> {
485
+
const result = await this.firehoseEventHandler.handle({
486
+
uri: evt.uri,
487
+
cid: evt.cid,
488
+
eventType: evt.event as 'create' | 'update' | 'delete',
489
+
record: evt.record,
490
+
did: evt.did,
491
+
collection: evt.collection
492
+
});
493
+
494
+
if (result.isErr()) {
495
+
console.error('Failed to process firehose event:', result.error);
496
+
}
424
497
}
425
498
426
-
protected async registerHandlers(
427
-
subscriber: IEventSubscriber,
428
-
services: any,
429
-
repositories: Repositories,
430
-
): Promise<void> {
431
-
// Start the firehose service instead of registering event handlers
432
-
await services.firehoseService.start();
499
+
private handleError(err: Error): void {
500
+
console.error('Firehose error:', err);
501
+
}
502
+
503
+
private getFilteredCollections(): string[] {
504
+
const collections = this.configService.getAtProtoCollections();
505
+
return [
506
+
collections.card,
507
+
collections.collection,
508
+
collections.collectionLink
509
+
];
510
+
}
511
+
512
+
private setupShutdownHandlers(): void {
513
+
const shutdown = async () => {
514
+
console.log('Shutting down firehose worker...');
515
+
if (this.firehose) {
516
+
await this.firehose.destroy();
517
+
}
518
+
if (this.runner) {
519
+
await this.runner.destroy();
520
+
}
521
+
process.exit(0);
522
+
};
523
+
524
+
process.on('SIGTERM', shutdown);
525
+
process.on('SIGINT', shutdown);
433
526
}
434
527
}
435
528
```
436
529
437
-
## Event Flow
530
+
## Detailed Event Processing Flow
531
+
532
+
### 1. Firehose Event Reception
533
+
```
534
+
AT Protocol Network
535
+
│ WebSocket Stream
536
+
▼
537
+
@atproto/sync Firehose
538
+
│ Filtered by collection types
539
+
▼
540
+
FirehoseEventHandler.handle()
541
+
```
438
542
439
-
1. **AT Protocol Firehose** → Receives events from AT Protocol network
440
-
2. **AtProtoFirehoseService** → Filters and validates events
441
-
3. **FirehoseEventHandler** → Routes to appropriate use case
442
-
4. **ProcessFirehoseEventUseCase** → Checks duplicates and delegates
443
-
5. **Specific Event Use Cases** → Process card/collection/link events
444
-
6. **Domain Services** → Update entities and publish domain events
445
-
7. **Event Publishers** → Notify other parts of the system
543
+
### 2. Duplicate Detection & Routing
544
+
```
545
+
ProcessFirehoseEventUseCase.execute()
546
+
│
547
+
├─ CREATE/UPDATE: Check (uri, cid) in publishedRecords
548
+
├─ DELETE: Check entity existence via AT URI resolution
549
+
│
550
+
└─ Route by collection type:
551
+
├─ network.cosmik.card → ProcessCardFirehoseEventUseCase
552
+
├─ network.cosmik.collection → ProcessCollectionFirehoseEventUseCase
553
+
└─ network.cosmik.collectionLink → ProcessCollectionLinkFirehoseEventUseCase
554
+
```
555
+
556
+
### 3. Entity Processing
557
+
```
558
+
Specific Use Cases
559
+
│
560
+
├─ Validate AT Protocol record structure
561
+
├─ Map to domain entities
562
+
├─ Apply business rules
563
+
├─ Persist changes via repositories
564
+
│
565
+
└─ Raise domain events:
566
+
├─ CardAddedToLibraryEvent
567
+
├─ CardAddedToCollectionEvent
568
+
├─ CollectionCreatedEvent
569
+
└─ etc.
570
+
```
571
+
572
+
### 4. Internal Event Publishing
573
+
```
574
+
Domain Events
575
+
│
576
+
├─ BullMQ (Production/Development with Redis)
577
+
│ ├─ feeds queue → Feed Worker
578
+
│ ├─ search queue → Search Worker
579
+
│ └─ analytics queue → Analytics Worker (future)
580
+
│
581
+
└─ In-Memory (Local Development)
582
+
└─ InMemoryEventWorkerProcess
583
+
```
584
+
585
+
## Process Wiring & Deployment
586
+
587
+
### Worker Entry Point
588
+
589
+
**src/workers/firehose-worker.ts**
590
+
```typescript
591
+
import { configService } from '../shared/infrastructure/config';
592
+
import { RepositoryFactory } from '../shared/infrastructure/http/factories/RepositoryFactory';
593
+
import { ServiceFactory } from '../shared/infrastructure/http/factories/ServiceFactory';
594
+
import { FirehoseWorkerProcess } from '../modules/atproto/infrastructure/processes/FirehoseWorkerProcess';
595
+
import { FirehoseEventHandler } from '../modules/atproto/application/handlers/FirehoseEventHandler';
596
+
import { ProcessFirehoseEventUseCase } from '../modules/atproto/application/useCases/ProcessFirehoseEventUseCase';
597
+
import { DrizzleFirehoseEventDuplicationService } from '../modules/atproto/infrastructure/services/DrizzleFirehoseEventDuplicationService';
598
+
599
+
async function main() {
600
+
console.log('Starting firehose worker...');
601
+
602
+
const repositories = RepositoryFactory.create(configService);
603
+
const services = ServiceFactory.createForWorker(configService, repositories);
604
+
605
+
// Create firehose-specific services
606
+
const duplicationService = new DrizzleFirehoseEventDuplicationService(
607
+
repositories.database
608
+
);
609
+
610
+
const processFirehoseEventUseCase = new ProcessFirehoseEventUseCase(
611
+
duplicationService,
612
+
repositories.atUriResolutionService,
613
+
repositories.cardRepository,
614
+
repositories.collectionRepository,
615
+
services.eventPublisher // Publishes internal domain events
616
+
);
617
+
618
+
const firehoseEventHandler = new FirehoseEventHandler(processFirehoseEventUseCase);
619
+
620
+
const firehoseWorker = new FirehoseWorkerProcess(
621
+
configService,
622
+
firehoseEventHandler
623
+
);
624
+
625
+
await firehoseWorker.start();
626
+
}
627
+
628
+
main().catch((error) => {
629
+
console.error('Failed to start firehose worker:', error);
630
+
process.exit(1);
631
+
});
632
+
```
633
+
634
+
### Build Configuration
635
+
636
+
**package.json scripts**
637
+
```json
638
+
{
639
+
"scripts": {
640
+
"worker:firehose": "node dist/workers/firehose-worker.js"
641
+
}
642
+
}
643
+
```
644
+
645
+
**tsup.config.ts**
646
+
```typescript
647
+
export default defineConfig({
648
+
entry: {
649
+
index: 'src/index.ts',
650
+
'workers/feed-worker': 'src/workers/feed-worker.ts',
651
+
'workers/search-worker': 'src/workers/search-worker.ts',
652
+
'workers/firehose-worker': 'src/workers/firehose-worker.ts', // Add this
653
+
},
654
+
// ... rest of config
655
+
});
656
+
```
657
+
658
+
### Fly.io Process Configuration
659
+
660
+
**fly.development.toml**
661
+
```toml
662
+
[processes]
663
+
web = "npm start"
664
+
feed-worker = "npm run worker:feeds"
665
+
search-worker = "npm run worker:search"
666
+
firehose-worker = "npm run worker:firehose"
667
+
668
+
[[vm]]
669
+
processes = ['firehose-worker']
670
+
memory = '256mb'
671
+
cpu_kind = 'shared'
672
+
cpus = 1
673
+
```
674
+
675
+
**fly.production.toml**
676
+
```toml
677
+
[processes]
678
+
web = "npm start"
679
+
feed-worker = "npm run worker:feeds"
680
+
search-worker = "npm run worker:search"
681
+
firehose-worker = "npm run worker:firehose"
682
+
683
+
[[vm]]
684
+
processes = ['firehose-worker']
685
+
memory = '512mb'
686
+
cpu_kind = 'shared'
687
+
cpus = 1
688
+
```
446
689
447
690
## Configuration
448
691
···
451
694
# Firehose configuration
452
695
ATPROTO_FIREHOSE_ENDPOINT=wss://bsky.network
453
696
FIREHOSE_RECONNECT_DELAY=3000
454
-
FIREHOSE_FILTER_COLLECTIONS=network.cosmik.card,network.cosmik.collection,network.cosmik.collection.link
455
697
456
-
# Enable/disable firehose processing
457
-
ENABLE_FIREHOSE_PROCESSING=true
698
+
# AT Protocol collections (automatically configured by environment)
699
+
# Production: network.cosmik.card, network.cosmik.collection, network.cosmik.collectionLink
700
+
# Development: network.cosmik.dev.card, network.cosmik.dev.collection, network.cosmik.dev.collectionLink
701
+
702
+
# Event processing (firehose worker always runs, unlike other workers)
703
+
# Other workers skip when USE_IN_MEMORY_EVENTS=true, but firehose worker always runs
458
704
```
459
705
706
+
## Key Architectural Decisions
707
+
708
+
### 1. Direct Firehose Connection
709
+
- **No BullMQ for input**: Firehose events come directly from AT Protocol WebSocket
710
+
- **Simpler architecture**: Eliminates unnecessary serialization/deserialization
711
+
- **Better reliability**: Direct connection with built-in reconnection logic
712
+
713
+
### 2. Always-Running Worker
714
+
- **Unlike other workers**: Feed/search workers skip when using in-memory events
715
+
- **Firehose worker always runs**: It's an external event source, not internal event consumer
716
+
- **Environment agnostic**: Works the same in local, dev, and production
717
+
718
+
### 3. Event Publishing Strategy
719
+
- **Input**: Direct AT Protocol firehose (external)
720
+
- **Output**: Internal domain events via BullMQ or in-memory based on config
721
+
- **Bridge pattern**: Converts external events to internal domain events
722
+
723
+
### 4. Duplicate Detection
724
+
- **Leverages existing publishedRecords table**: No additional event log needed
725
+
- **CID-based deduplication**: Uses AT Protocol's content addressing
726
+
- **Efficient lookups**: Single table queries for CREATE/UPDATE detection
727
+
460
728
## Error Handling
461
729
462
730
- **Network Errors**: Automatic reconnection with exponential backoff
···
478
746
- **Mock Firehose**: Test harness for simulating AT Protocol events
479
747
- **Load Tests**: High-volume event processing scenarios
480
748
481
-
## Deployment
749
+
## Deployment & Operations
750
+
751
+
### Process Management
752
+
- **Dedicated Worker Process**: Runs independently of web app and other workers
753
+
- **Always Active**: Unlike other workers, doesn't skip based on event configuration
754
+
- **Automatic Restart**: Fly.io handles process restarts on failure
755
+
- **Graceful Shutdown**: Handles SIGTERM/SIGINT for clean shutdowns
756
+
757
+
### Scaling Considerations
758
+
- **Single Instance**: Start with one firehose worker per environment
759
+
- **Future Scaling**: Can partition by DID ranges or collection types if needed
760
+
- **Cursor Management**: @atproto/sync handles cursor persistence automatically
761
+
- **Replay Capability**: Can restart from specific cursor positions
762
+
763
+
### Monitoring & Observability
764
+
- **Connection Health**: Monitor WebSocket connection status
765
+
- **Processing Metrics**: Track events processed, duplicates detected, errors
766
+
- **Latency Monitoring**: Measure time from firehose event to domain event publication
767
+
- **Error Tracking**: Log and alert on processing failures
768
+
769
+
### Error Handling Strategy
770
+
- **Network Errors**: Automatic reconnection with exponential backoff (handled by @atproto/sync)
771
+
- **Processing Errors**: Log and continue (don't crash worker for single event failures)
772
+
- **Validation Errors**: Skip invalid records with detailed logging
773
+
- **Database Errors**: Retry with backoff, dead letter for persistent failures
482
774
483
-
- **Separate Worker**: Dedicated firehose worker process
484
-
- **Scaling**: Multiple worker instances with partitioned processing
485
-
- **Blue-Green**: Zero-downtime deployments with cursor management
486
-
- **Rollback**: Ability to replay events from specific cursor position
775
+
## Benefits of This Architecture
776
+
777
+
### 1. Separation of Concerns
778
+
- **External Events**: Firehose worker handles AT Protocol events
779
+
- **Internal Events**: Other workers handle domain events
780
+
- **Clear Boundaries**: No mixing of external and internal event systems
781
+
782
+
### 2. Reliability
783
+
- **Direct Connection**: No intermediate queues that can fail
784
+
- **Built-in Resilience**: @atproto/sync handles reconnection and cursor management
785
+
- **Idempotent Processing**: Duplicate detection prevents double-processing
786
+
787
+
### 3. Performance
788
+
- **No Serialization Overhead**: Direct event processing
789
+
- **Efficient Duplicate Detection**: Single table lookups
790
+
- **Minimal Latency**: Direct path from firehose to domain updates
791
+
792
+
### 4. Maintainability
793
+
- **Simple Architecture**: Easy to understand and debug
794
+
- **Standard Patterns**: Follows existing DDD and layered architecture
795
+
- **Testable**: Clear interfaces for mocking and testing
487
796
488
797
## Future Enhancements
489
798
490
-
1. **Server-side Filtering**: Reduce bandwidth with AT Protocol filters
491
-
2. **Event Sourcing**: Complete audit trail of all firehose events
492
-
3. **Real-time Sync**: WebSocket updates to web clients
493
-
4. **Cross-PDS Support**: Handle events from multiple AT Protocol servers
494
-
5. **Analytics**: Event processing metrics and insights
799
+
1. **Server-side Filtering**: Use AT Protocol's filtering capabilities to reduce bandwidth
800
+
2. **Multi-PDS Support**: Handle events from multiple Personal Data Servers
801
+
3. **Event Sourcing**: Complete audit trail of all processed firehose events
802
+
4. **Real-time Sync**: WebSocket updates to web clients for live updates
803
+
5. **Analytics Pipeline**: Dedicated analytics worker for metrics and insights
804
+
6. **Horizontal Scaling**: Partition processing across multiple firehose workers