nandi.uk
This repository has no description
1# Firehose Event Handlers Implementation Guide
2
3## Overview
4
5This guide outlines how to enhance the firehose event processors to handle AT Protocol events by creating, updating, and deleting domain aggregates while maintaining clean separation between domain models and AT Protocol record types.
6
7## Architecture Approach
8
9Instead of adding AT Protocol dependencies to domain models, we'll:
10
111. **Use existing factory methods** - Leverage `CardFactory` and `Collection.create()`
122. **Create mapper services** - Convert AT Protocol records to domain input DTOs
133. **Enhance use cases** - Handle full aggregate lifecycle in event processors
144. **Maintain domain purity** - Keep domain models free of AT Protocol dependencies
15
16## Implementation Steps
17
18### Step 1: Create AT Protocol Record Mappers
19
20Create mapper services that convert AT Protocol records to domain input objects:
21
22```typescript
23// src/modules/atproto/infrastructure/mappers/ATProtoRecordToDomainMapper.ts
24export class ATProtoRecordToDomainMapper {
25 static cardRecordToCardInput(record: CardRecord, atUri: string): ICardInput {
26 // Convert AT Protocol card record to CardFactory input
27 }
28
29 static collectionRecordToCollectionProps(
30 record: CollectionRecord,
31 atUri: string,
32 ): CollectionCreateProps {
33 // Convert AT Protocol collection record to Collection.create() input
34 }
35}
36```
37
38### Step 2: Enhance Event Processors
39
40Update the firehose event processors to:
41
42#### For Card Events:
43
44- **Create**: Parse record → Create via CardFactory → Mark as published → Save → Publish events
45- **Update**: Find existing → Parse record → Update content → Update published ID → Save → Publish events
46- **Delete**: Resolve ID → Delete from repo
47
48#### For Collection Events:
49
50- **Create**: Parse record → Create via Collection.create() → Mark as published → Save → Publish events
51- **Update**: Find existing → Parse record → Update details → Update published ID → Save → Publish events
52- **Delete**: Resolve ID → Delete from repo
53
54### Step 3: Add Domain Methods
55
56Add these methods to support external updates:
57
58#### Card Domain:
59
60```typescript
61// In Card.ts
62public updateContentFromText(text: string): Result<void, CardValidationError> {
63 // For NOTE cards - update text content
64}
65
66public updatePublishedRecordId(publishedRecordId: PublishedRecordId): void {
67 // Update the published record ID from firehose events
68}
69```
70
71#### Collection Domain:
72
73```typescript
74// In Collection.ts
75public updateDetailsFromExternal(name: string, description?: string): Result<void, CollectionValidationError> {
76 // Update collection details from external source
77}
78
79public updatePublishedRecordId(publishedRecordId: PublishedRecordId): void {
80 // Update the published record ID from firehose events
81}
82```
83
84### Step 4: Enhance AT URI Resolution Service
85
86Extend `IAtUriResolutionService` to support bidirectional mapping:
87
88```typescript
89// Add methods to store AT URI → Domain ID mappings
90storeCardMapping(atUri: string, cardId: CardId): Promise<Result<void>>;
91storeCollectionMapping(atUri: string, collectionId: CollectionId): Promise<Result<void>>;
92```
93
94### Step 5: Event Publishing Pattern
95
96Each processor should follow this pattern:
97
98```typescript
99// 1. Process the record
100const aggregate = // ... create or update aggregate
101 // 2. Save to repository
102 await this.repository.save(aggregate);
103
104// 3. Publish domain events
105const events = aggregate.domainEvents;
106const publishResult = await this.eventPublisher.publishEvents(events);
107aggregate.clearEvents(); // Clear after publishing
108```
109
110### Step 6: Error Handling Strategy
111
112- **Parsing errors**: Log and skip (don't fail the firehose)
113- **Domain validation errors**: Log and skip
114- **Repository errors**: Log and potentially retry
115- **Event publishing errors**: Log but don't fail the operation
116
117### Step 7: Testing Strategy
118
119Create integration tests that:
120
121- Mock firehose events with real AT Protocol record structures
122- Verify aggregates are created/updated correctly
123- Verify domain events are published
124- Test error scenarios
125
126## Key Benefits
127
1281. **Domain Purity**: Domain models remain free of AT Protocol dependencies
1292. **Reusability**: Existing factory methods and validation logic are reused
1303. **Consistency**: Same creation/update paths as internal operations
1314. **Event Driven**: Proper domain events are published for downstream processing
1325. **Testability**: Easy to test with mock AT Protocol records
133
134## Files to Modify
135
1361. `ProcessCardFirehoseEventUseCase.ts` - Enhanced card event handling
1372. `ProcessCollectionFirehoseEventUseCase.ts` - Enhanced collection event handling
1383. `ProcessCollectionLinkFirehoseEventUseCase.ts` - Enhanced collection link handling
1394. `Card.ts` - Add update methods for external sources
1405. `Collection.ts` - Add update methods for external sources
1416. `IAtUriResolutionService.ts` - Add mapping storage methods
142
143## New Files to Create
144
1451. `ATProtoRecordToDomainMapper.ts` - Record to domain input conversion
1462. Integration tests for each event processor
147
148This approach maintains clean architecture while enabling full firehose event processing capabilities.