nandi.uk / semble
This repository has no description
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

formatting and linting

author
Wesley Finck date (Nov 15, 2025, 12:03 PM -0800) commit 724d6f21 parent 170ae92e
+388 -187
+16 -9
.agent/logs/20251114_firehose_event_handlers.md
··· 8 8 9 9 Instead of adding AT Protocol dependencies to domain models, we'll: 10 10 11 - 1. **Use existing factory methods** - Leverage `CardFactory` and `Collection.create()` 11 + 1. **Use existing factory methods** - Leverage `CardFactory` and `Collection.create()` 12 12 2. **Create mapper services** - Convert AT Protocol records to domain input DTOs 13 13 3. **Enhance use cases** - Handle full aggregate lifecycle in event processors 14 14 4. **Maintain domain purity** - Keep domain models free of AT Protocol dependencies ··· 25 25 static cardRecordToCardInput(record: CardRecord, atUri: string): ICardInput { 26 26 // Convert AT Protocol card record to CardFactory input 27 27 } 28 - 29 - static collectionRecordToCollectionProps(record: CollectionRecord, atUri: string): CollectionCreateProps { 28 + 29 + static collectionRecordToCollectionProps( 30 + record: CollectionRecord, 31 + atUri: string, 32 + ): CollectionCreateProps { 30 33 // Convert AT Protocol collection record to Collection.create() input 31 34 } 32 35 } ··· 37 40 Update the firehose event processors to: 38 41 39 42 #### For Card Events: 43 + 40 44 - **Create**: Parse record → Create via CardFactory → Mark as published → Save → Publish events 41 - - **Update**: Find existing → Parse record → Update content → Update published ID → Save → Publish events 45 + - **Update**: Find existing → Parse record → Update content → Update published ID → Save → Publish events 42 46 - **Delete**: Resolve ID → Delete from repo 43 47 44 48 #### For Collection Events: 49 + 45 50 - **Create**: Parse record → Create via Collection.create() → Mark as published → Save → Publish events 46 51 - **Update**: Find existing → Parse record → Update details → Update published ID → Save → Publish events 47 52 - **Delete**: Resolve ID → Delete from repo ··· 51 56 Add these methods to support external updates: 52 57 53 58 #### Card Domain: 59 + 54 60 ```typescript 55 61 // In Card.ts 56 62 public updateContentFromText(text: string): Result<void, CardValidationError> { ··· 63 69 ``` 64 70 65 71 #### Collection Domain: 66 - ```typescript 72 + 73 + ```typescript 67 74 // In Collection.ts 68 75 public updateDetailsFromExternal(name: string, description?: string): Result<void, CollectionValidationError> { 69 76 // Update collection details from external source ··· 91 98 ```typescript 92 99 // 1. Process the record 93 100 const aggregate = // ... create or update aggregate 94 - 95 - // 2. Save to repository 96 - await this.repository.save(aggregate); 101 + // 2. Save to repository 102 + await this.repository.save(aggregate); 97 103 98 104 // 3. Publish domain events 99 105 const events = aggregate.domainEvents; ··· 111 117 ### Step 7: Testing Strategy 112 118 113 119 Create integration tests that: 120 + 114 121 - Mock firehose events with real AT Protocol record structures 115 122 - Verify aggregates are created/updated correctly 116 123 - Verify domain events are published ··· 127 134 ## Files to Modify 128 135 129 136 1. `ProcessCardFirehoseEventUseCase.ts` - Enhanced card event handling 130 - 2. `ProcessCollectionFirehoseEventUseCase.ts` - Enhanced collection event handling 137 + 2. `ProcessCollectionFirehoseEventUseCase.ts` - Enhanced collection event handling 131 138 3. `ProcessCollectionLinkFirehoseEventUseCase.ts` - Enhanced collection link handling 132 139 4. `Card.ts` - Add update methods for external sources 133 140 5. `Collection.ts` - Add update methods for external sources
+30 -10
.agent/logs/20251114_firehose_event_handling.md
··· 7 7 ## Current Implementation 8 8 9 9 ### Cards 10 + 10 11 - **Create**: `ATProtoCardPublisher.publishCardToLibrary()` creates/updates records, stores `PublishedRecordId` in card's library membership 11 12 - **Update**: Uses `putRecord()` with existing rkey when card already has a published record 12 13 - **Delete**: `unpublishCardFromLibrary()` deletes the AT Protocol record 13 14 14 15 ### Collections 16 + 15 17 - **Create**: `ATProtoCollectionPublisher.publish()` creates/updates collection records 16 18 - **Update**: Uses `putRecord()` with existing rkey when collection has `publishedRecordId` 17 19 - **Delete**: `unpublish()` deletes the collection record 18 20 19 21 ### Collection Links 22 + 20 23 - **Create**: `publishCardAddedToCollection()` creates link records 21 24 - **Delete**: `unpublishCardAddedToCollection()` deletes link records 22 25 - **No Update**: Collection links don't have update operations 23 26 24 27 ### Storage Pattern 28 + 25 29 All use the `publishedRecords` table with: 30 + 26 31 ```sql 27 32 id: uuid (primary key) 28 33 uri: text (AT URI) ··· 34 39 ## Duplicate Detection Strategy 35 40 36 41 ### For CREATE Events 42 + 37 43 ✅ **Works as designed**: Check if `(uri, cid)` exists in `publishedRecords` table 44 + 38 45 - If exists → already processed, skip 39 46 - If not exists → process the create 40 47 41 - ### For UPDATE Events 48 + ### For UPDATE Events 49 + 42 50 ⚠️ **Needs verification**: Check if `(uri, cid)` exists in `publishedRecords` table 51 + 43 52 - **Key question**: When we update a collection/card, does the new CID get stored? 44 53 - Looking at the code: Yes, `putRecord()` operations should generate new CIDs, and these should be stored 45 54 46 55 ### For DELETE Events 56 + 47 57 ⚠️ **Needs implementation**: More complex logic required 58 + 48 59 1. Find all `publishedRecords` with matching `uri` (ignore CID) 49 60 2. Use `ATUri.create(uri)` to determine entity type from collection field 50 61 3. Query appropriate table to see if entity still exists ··· 54 65 ## Changes Needed 55 66 56 67 ### 1. Verify UPDATE CID Storage 68 + 57 69 Need to confirm that when we call `putRecord()`, the new CID is captured and stored. Looking at the publishers: 58 70 59 71 ```typescript ··· 70 82 **Required change**: Ensure `putRecord()` responses update the `publishedRecords` table with new CID. 71 83 72 84 ### 2. Enhance AT URI Resolution for DELETE Detection 85 + 73 86 Expand `IAtUriResolutionService` to handle all entity types: 74 87 75 88 ```typescript ··· 78 91 resolveCardId(atUri: string): Promise<Result<CardId | null>>; 79 92 resolveCollectionId(atUri: string): Promise<Result<CollectionId | null>>; 80 93 // Add collection link resolution 81 - resolveCollectionLinkId(atUri: string): Promise<Result<{collectionId: CollectionId, cardId: CardId} | null>>; 94 + resolveCollectionLinkId( 95 + atUri: string, 96 + ): Promise<Result<{ collectionId: CollectionId; cardId: CardId } | null>>; 82 97 } 83 98 84 99 enum AtUriResourceType { ··· 89 104 ``` 90 105 91 106 ### 3. Add Collection Type Detection 107 + 92 108 Enhance `ATUri` or create a helper to determine entity type from collection field: 93 109 94 110 ```typescript ··· 103 119 ``` 104 120 105 121 ### 4. Implement DELETE Detection Service 122 + 106 123 ```typescript 107 124 interface IDeleteDetectionService { 108 125 hasBeenDeleted(atUri: string): Promise<Result<boolean>>; ··· 113 130 // 1. Find all publishedRecords with matching URI 114 131 const records = await this.findPublishedRecordsByUri(atUri); 115 132 if (records.length === 0) return ok(true); // No records = was deleted 116 - 133 + 117 134 // 2. Determine entity type from AT URI 118 135 const atUriResult = ATUri.create(atUri); 119 136 if (atUriResult.isErr()) return err(atUriResult.error); 120 - 137 + 121 138 const entityType = atUriResult.value.getEntityType(this.configService); 122 139 123 140 // 3. Check if entity still exists 124 141 switch (entityType) { 125 142 case AtUriResourceType.COLLECTION: 126 - const collectionId = await this.atUriResolver.resolveCollectionId(atUri); 143 + const collectionId = 144 + await this.atUriResolver.resolveCollectionId(atUri); 127 145 return ok(collectionId === null); 128 146 case AtUriResourceType.CARD: 129 147 const cardId = await this.atUriResolver.resolveCardId(atUri); 130 148 return ok(cardId === null); 131 149 case AtUriResourceType.COLLECTION_LINK: 132 - const linkInfo = await this.atUriResolver.resolveCollectionLinkId(atUri); 150 + const linkInfo = 151 + await this.atUriResolver.resolveCollectionLinkId(atUri); 133 152 return ok(linkInfo === null); 134 153 } 135 154 } ··· 137 156 ``` 138 157 139 158 ### 5. Clean Up Published Records on DELETE 159 + 140 160 Modify delete operations to remove `publishedRecords` entries: 141 161 142 162 ```typescript 143 163 // In delete use cases and services 144 164 async deleteCollection(collectionId: CollectionId) { 145 165 const collection = await this.repository.findById(collectionId); 146 - 166 + 147 167 // Delete the collection 148 168 await this.repository.delete(collectionId); 149 - 169 + 150 170 // Clean up published records 151 171 if (collection.publishedRecordId) { 152 172 await this.cleanupPublishedRecord(collection.publishedRecordId); 153 173 } 154 - 174 + 155 175 // Clean up collection link published records 156 176 for (const link of collection.cardLinks) { 157 177 if (link.publishedRecordId) { ··· 164 184 ## Implementation Priority 165 185 166 186 1. **High Priority**: Verify and fix UPDATE CID storage 167 - 2. **High Priority**: Implement card and collection link AT URI resolution 187 + 2. **High Priority**: Implement card and collection link AT URI resolution 168 188 3. **Medium Priority**: Add DELETE detection service 169 189 4. **Low Priority**: Clean up published records on delete (nice to have, but detection works without it) 170 190
+103 -49
.agent/logs/20251114_firehose_layered_arch.md
··· 80 80 #### New Domain Concepts 81 81 82 82 **AtUriResourceType Enum** 83 + 83 84 ```typescript 84 85 export enum AtUriResourceType { 85 86 CARD = 'card', ··· 89 90 ``` 90 91 91 92 **Enhanced ATUri Domain Object** 93 + 92 94 ```typescript 93 95 export class ATUri extends ValueObject<ATUriProps> { 94 96 // ... existing methods ... 95 - 96 - public getEntityType(configService: EnvironmentConfigService): AtUriResourceType { 97 + 98 + public getEntityType( 99 + configService: EnvironmentConfigService, 100 + ): AtUriResourceType { 97 101 const collections = configService.getAtProtoCollections(); 98 102 if (this.collection === collections.card) return AtUriResourceType.CARD; 99 - if (this.collection === collections.collection) return AtUriResourceType.COLLECTION; 100 - if (this.collection === collections.collectionLink) return AtUriResourceType.COLLECTION_LINK; 103 + if (this.collection === collections.collection) 104 + return AtUriResourceType.COLLECTION; 105 + if (this.collection === collections.collectionLink) 106 + return AtUriResourceType.COLLECTION_LINK; 101 107 throw new Error(`Unknown collection type: ${this.collection}`); 102 108 } 103 109 } ··· 106 112 #### Enhanced Domain Services 107 113 108 114 **Expanded IAtUriResolutionService** 115 + 109 116 ```typescript 110 117 export interface IAtUriResolutionService { 111 118 resolveAtUri(atUri: string): Promise<Result<AtUriResolutionResult | null>>; 112 119 resolveCardId(atUri: string): Promise<Result<CardId | null>>; 113 120 resolveCollectionId(atUri: string): Promise<Result<CollectionId | null>>; 114 - resolveCollectionLinkId(atUri: string): Promise<Result<{collectionId: CollectionId, cardId: CardId} | null>>; 121 + resolveCollectionLinkId( 122 + atUri: string, 123 + ): Promise<Result<{ collectionId: CollectionId; cardId: CardId } | null>>; 115 124 } 116 125 ``` 117 126 118 127 **New IFirehoseEventDuplicationService** 128 + 119 129 ```typescript 120 130 export interface IFirehoseEventDuplicationService { 121 131 hasEventBeenProcessed( 122 - atUri: string, 123 - cid: string | null, 124 - operation: FirehoseEventType 132 + atUri: string, 133 + cid: string | null, 134 + operation: FirehoseEventType, 125 135 ): Promise<Result<boolean>>; 126 - 136 + 127 137 hasBeenDeleted(atUri: string): Promise<Result<boolean>>; 128 138 } 129 139 ``` ··· 133 143 #### New Domain Events 134 144 135 145 **FirehoseEventProcessed** 146 + 136 147 ```typescript 137 148 export class FirehoseEventProcessed extends DomainEvent { 138 149 constructor( ··· 140 151 public readonly cid: string | null, 141 152 public readonly eventType: FirehoseEventType, 142 153 public readonly entityType: AtUriResourceType, 143 - public readonly entityId?: string 154 + public readonly entityId?: string, 144 155 ) { 145 156 super(); 146 157 } ··· 150 161 #### New Use Cases 151 162 152 163 **ProcessFirehoseEventUseCase** 164 + 153 165 ```typescript 154 166 export interface ProcessFirehoseEventDTO { 155 167 atUri: string; ··· 165 177 private cardRepository: ICardRepository, 166 178 private collectionRepository: ICollectionRepository, 167 179 private eventPublisher: IEventPublisher, 168 - private configService: EnvironmentConfigService 180 + private configService: EnvironmentConfigService, 169 181 ) {} 170 182 171 183 async execute(request: ProcessFirehoseEventDTO): Promise<Result<void>> { ··· 178 190 ``` 179 191 180 192 **ProcessCardFirehoseEventUseCase** 193 + 181 194 ```typescript 182 195 export class ProcessCardFirehoseEventUseCase { 183 196 async execute(request: { ··· 192 205 ``` 193 206 194 207 **ProcessCollectionFirehoseEventUseCase** 208 + 195 209 ```typescript 196 210 export class ProcessCollectionFirehoseEventUseCase { 197 211 async execute(request: { ··· 206 220 ``` 207 221 208 222 **ProcessCollectionLinkFirehoseEventUseCase** 223 + 209 224 ```typescript 210 225 export class ProcessCollectionLinkFirehoseEventUseCase { 211 226 async execute(request: { ··· 222 237 #### Event Handlers 223 238 224 239 **FirehoseEventHandler** 240 + 225 241 ```typescript 226 242 export class FirehoseEventHandler { 227 243 constructor( 228 - private processFirehoseEventUseCase: ProcessFirehoseEventUseCase 244 + private processFirehoseEventUseCase: ProcessFirehoseEventUseCase, 229 245 ) {} 230 246 231 247 async handle(event: AtProtoFirehoseEvent): Promise<Result<void>> { ··· 233 249 atUri: event.uri, 234 250 cid: event.cid, 235 251 eventType: event.eventType, 236 - record: event.record 252 + record: event.record, 237 253 }); 238 254 } 239 255 } ··· 242 258 #### New Application Services 243 259 244 260 **IFirehoseService** 261 + 245 262 ```typescript 246 263 export interface IFirehoseService { 247 264 start(): Promise<void>; ··· 255 272 #### AT Protocol Firehose Integration 256 273 257 274 **AtProtoFirehoseService** 275 + 258 276 ```typescript 259 277 export class AtProtoFirehoseService implements IFirehoseService { 260 278 constructor( 261 279 private firehoseEventHandler: FirehoseEventHandler, 262 280 private configService: EnvironmentConfigService, 263 - private idResolver: IdResolver 281 + private idResolver: IdResolver, 264 282 ) {} 265 283 266 284 async start(): Promise<void> { 267 285 const runner = new MemoryRunner({}); 268 - 286 + 269 287 this.firehose = new Firehose({ 270 288 service: 'wss://bsky.network', 271 289 runner, 272 290 idResolver: this.idResolver, 273 291 filterCollections: this.getFilteredCollections(), 274 292 handleEvent: this.handleFirehoseEvent.bind(this), 275 - onError: this.handleError.bind(this) 293 + onError: this.handleError.bind(this), 276 294 }); 277 295 278 296 await this.firehose.start(); ··· 285 303 eventType: evt.event as 'create' | 'update' | 'delete', 286 304 record: evt.record, 287 305 did: evt.did, 288 - collection: evt.collection 306 + collection: evt.collection, 289 307 }); 290 308 291 309 if (result.isErr()) { ··· 298 316 return [ 299 317 collections.card, 300 318 collections.collection, 301 - collections.collectionLink 319 + collections.collectionLink, 302 320 ]; 303 321 } 304 322 } 305 323 ``` 306 324 307 325 **DrizzleFirehoseEventDuplicationService** 326 + 308 327 ```typescript 309 - export class DrizzleFirehoseEventDuplicationService implements IFirehoseEventDuplicationService { 328 + export class DrizzleFirehoseEventDuplicationService 329 + implements IFirehoseEventDuplicationService 330 + { 310 331 constructor( 311 332 private db: PostgresJsDatabase, 312 333 private atUriResolver: IAtUriResolutionService, 313 - private configService: EnvironmentConfigService 334 + private configService: EnvironmentConfigService, 314 335 ) {} 315 336 316 337 async hasEventBeenProcessed( 317 - atUri: string, 318 - cid: string | null, 319 - operation: FirehoseEventType 338 + atUri: string, 339 + cid: string | null, 340 + operation: FirehoseEventType, 320 341 ): Promise<Result<boolean>> { 321 342 try { 322 343 // For CREATE/UPDATE: check if (uri, cid) exists 323 344 if (operation === 'create' || operation === 'update') { 324 345 if (!cid) return ok(false); 325 - 346 + 326 347 const result = await this.db 327 348 .select({ id: publishedRecords.id }) 328 349 .from(publishedRecords) 329 350 .where( 330 - and( 331 - eq(publishedRecords.uri, atUri), 332 - eq(publishedRecords.cid, cid) 333 - ) 351 + and(eq(publishedRecords.uri, atUri), eq(publishedRecords.cid, cid)), 334 352 ) 335 353 .limit(1); 336 - 354 + 337 355 return ok(result.length > 0); 338 356 } 339 - 357 + 340 358 // For DELETE: use more complex logic 341 359 return this.hasBeenDeleted(atUri); 342 360 } catch (error) { ··· 367 385 // 3. Check if entity still exists 368 386 switch (entityType) { 369 387 case AtUriResourceType.COLLECTION: 370 - const collectionId = await this.atUriResolver.resolveCollectionId(atUri); 388 + const collectionId = 389 + await this.atUriResolver.resolveCollectionId(atUri); 371 390 return ok(collectionId === null); 372 391 case AtUriResourceType.CARD: 373 392 const cardId = await this.atUriResolver.resolveCardId(atUri); 374 393 return ok(cardId === null); 375 394 case AtUriResourceType.COLLECTION_LINK: 376 - const linkInfo = await this.atUriResolver.resolveCollectionLinkId(atUri); 395 + const linkInfo = 396 + await this.atUriResolver.resolveCollectionLinkId(atUri); 377 397 return ok(linkInfo === null); 378 398 default: 379 399 return err(new Error(`Unknown entity type: ${entityType}`)); ··· 386 406 ``` 387 407 388 408 **Enhanced DrizzleAtUriResolutionService** 409 + 389 410 ```typescript 390 411 export class DrizzleAtUriResolutionService implements IAtUriResolutionService { 391 412 // ... existing methods ... ··· 395 416 const cardResult = await this.db 396 417 .select({ id: cards.id }) 397 418 .from(cards) 398 - .innerJoin(publishedRecords, eq(cards.publishedRecordId, publishedRecords.id)) 419 + .innerJoin( 420 + publishedRecords, 421 + eq(cards.publishedRecordId, publishedRecords.id), 422 + ) 399 423 .where(eq(publishedRecords.uri, atUri)) 400 424 .limit(1); 401 425 ··· 415 439 } 416 440 417 441 async resolveCollectionLinkId( 418 - atUri: string 419 - ): Promise<Result<{collectionId: CollectionId, cardId: CardId} | null>> { 442 + atUri: string, 443 + ): Promise<Result<{ collectionId: CollectionId; cardId: CardId } | null>> { 420 444 try { 421 445 const linkResult = await this.db 422 446 .select({ 423 447 collectionId: collectionCards.collectionId, 424 - cardId: collectionCards.cardId 448 + cardId: collectionCards.cardId, 425 449 }) 426 450 .from(collectionCards) 427 - .innerJoin(publishedRecords, eq(collectionCards.publishedRecordId, publishedRecords.id)) 451 + .innerJoin( 452 + publishedRecords, 453 + eq(collectionCards.publishedRecordId, publishedRecords.id), 454 + ) 428 455 .where(eq(publishedRecords.uri, atUri)) 429 456 .limit(1); 430 457 ··· 432 459 return ok(null); 433 460 } 434 461 435 - const collectionIdResult = CollectionId.createFromString(linkResult[0]!.collectionId); 462 + const collectionIdResult = CollectionId.createFromString( 463 + linkResult[0]!.collectionId, 464 + ); 436 465 const cardIdResult = CardId.createFromString(linkResult[0]!.cardId); 437 466 438 467 if (collectionIdResult.isErr()) { ··· 444 473 445 474 return ok({ 446 475 collectionId: collectionIdResult.value, 447 - cardId: cardIdResult.value 476 + cardId: cardIdResult.value, 448 477 }); 449 478 } catch (error) { 450 479 return err(error as Error); ··· 456 485 #### Standalone Worker Process 457 486 458 487 **FirehoseWorkerProcess** (implements IProcess directly, not BaseWorkerProcess) 488 + 459 489 ```typescript 460 490 export class FirehoseWorkerProcess implements IProcess { 461 491 private firehose?: Firehose; ··· 463 493 464 494 constructor( 465 495 private configService: EnvironmentConfigService, 466 - private firehoseEventHandler: FirehoseEventHandler 496 + private firehoseEventHandler: FirehoseEventHandler, 467 497 ) {} 468 498 469 499 async start(): Promise<void> { ··· 471 501 472 502 const runner = new MemoryRunner({}); 473 503 this.runner = runner; 474 - 504 + 475 505 this.firehose = new Firehose({ 476 506 service: 'wss://bsky.network', 477 507 runner, 478 508 idResolver: new IdResolver(), 479 509 filterCollections: this.getFilteredCollections(), 480 510 handleEvent: this.handleFirehoseEvent.bind(this), 481 - onError: this.handleError.bind(this) 511 + onError: this.handleError.bind(this), 482 512 }); 483 513 484 514 await this.firehose.start(); ··· 494 524 eventType: evt.event as 'create' | 'update' | 'delete', 495 525 record: evt.record, 496 526 did: evt.did, 497 - collection: evt.collection 527 + collection: evt.collection, 498 528 }); 499 529 500 530 if (result.isErr()) { ··· 511 541 return [ 512 542 collections.card, 513 543 collections.collection, 514 - collections.collectionLink 544 + collections.collectionLink, 515 545 ]; 516 546 } 517 547 ··· 536 566 ## Detailed Event Processing Flow 537 567 538 568 ### 1. Firehose Event Reception 569 + 539 570 ``` 540 571 AT Protocol Network 541 572 │ WebSocket Stream ··· 547 578 ``` 548 579 549 580 ### 2. Duplicate Detection & Routing 581 + 550 582 ``` 551 583 ProcessFirehoseEventUseCase.execute() 552 584 │ ··· 560 592 ``` 561 593 562 594 ### 3. Entity Processing 595 + 563 596 ``` 564 597 Specific Use Cases 565 598 │ ··· 576 609 ``` 577 610 578 611 ### 4. Internal Event Publishing 612 + 579 613 ``` 580 614 Domain Events 581 615 │ ··· 593 627 ### Worker Entry Point 594 628 595 629 **src/workers/firehose-worker.ts** 630 + 596 631 ```typescript 597 632 import { configService } from '../shared/infrastructure/config'; 598 633 import { RepositoryFactory } from '../shared/infrastructure/http/factories/RepositoryFactory'; ··· 612 647 const duplicationService = new DrizzleFirehoseEventDuplicationService( 613 648 repositories.database, 614 649 repositories.atUriResolutionService, 615 - configService 650 + configService, 616 651 ); 617 652 618 653 const processFirehoseEventUseCase = new ProcessFirehoseEventUseCase( ··· 621 656 repositories.cardRepository, 622 657 repositories.collectionRepository, 623 658 services.eventPublisher, // Publishes internal domain events 624 - configService 659 + configService, 625 660 ); 626 661 627 - const firehoseEventHandler = new FirehoseEventHandler(processFirehoseEventUseCase); 628 - 662 + const firehoseEventHandler = new FirehoseEventHandler( 663 + processFirehoseEventUseCase, 664 + ); 665 + 629 666 const firehoseWorker = new FirehoseWorkerProcess( 630 667 configService, 631 - firehoseEventHandler 668 + firehoseEventHandler, 632 669 ); 633 670 634 671 await firehoseWorker.start(); ··· 643 680 ### Build Configuration 644 681 645 682 **package.json scripts** 683 + 646 684 ```json 647 685 { 648 686 "scripts": { ··· 652 690 ``` 653 691 654 692 **tsup.config.ts** 693 + 655 694 ```typescript 656 695 export default defineConfig({ 657 696 entry: { ··· 667 706 ### Fly.io Process Configuration 668 707 669 708 **fly.development.toml** 709 + 670 710 ```toml 671 711 [processes] 672 712 web = "npm start" ··· 682 722 ``` 683 723 684 724 **fly.production.toml** 725 + 685 726 ```toml 686 727 [processes] 687 728 web = "npm start" ··· 699 740 ## Configuration 700 741 701 742 **Environment Variables** 743 + 702 744 ```bash 703 745 # Firehose configuration 704 746 ATPROTO_FIREHOSE_ENDPOINT=wss://bsky.network ··· 715 757 ## Key Architectural Decisions 716 758 717 759 ### 1. Direct Firehose Connection 760 + 718 761 - **No BullMQ for input**: Firehose events come directly from AT Protocol WebSocket 719 762 - **Simpler architecture**: Eliminates unnecessary serialization/deserialization 720 763 - **Better reliability**: Direct connection with built-in reconnection logic 721 764 722 765 ### 2. Always-Running Worker 766 + 723 767 - **Unlike other workers**: Feed/search workers skip when using in-memory events 724 768 - **Firehose worker always runs**: It's an external event source, not internal event consumer 725 769 - **Environment agnostic**: Works the same in local, dev, and production 726 770 727 771 ### 3. Event Publishing Strategy 772 + 728 773 - **Input**: Direct AT Protocol firehose (external) 729 774 - **Output**: Internal domain events via BullMQ or in-memory based on config 730 775 - **Bridge pattern**: Converts external events to internal domain events 731 776 732 777 ### 4. Duplicate Detection 778 + 733 779 - **Leverages existing publishedRecords table**: No additional event log needed 734 780 - **CID-based deduplication**: Uses AT Protocol's content addressing 735 781 - **Efficient lookups**: Single table queries for CREATE/UPDATE detection ··· 758 804 ## Deployment & Operations 759 805 760 806 ### Process Management 807 + 761 808 - **Dedicated Worker Process**: Runs independently of web app and other workers 762 809 - **Always Active**: Unlike other workers, doesn't skip based on event configuration 763 810 - **Automatic Restart**: Fly.io handles process restarts on failure 764 811 - **Graceful Shutdown**: Handles SIGTERM/SIGINT for clean shutdowns 765 812 766 813 ### Scaling Considerations 814 + 767 815 - **Single Instance**: Start with one firehose worker per environment 768 816 - **Future Scaling**: Can partition by DID ranges or collection types if needed 769 817 - **Cursor Management**: @atproto/sync handles cursor persistence automatically 770 818 - **Replay Capability**: Can restart from specific cursor positions 771 819 772 820 ### Monitoring & Observability 821 + 773 822 - **Connection Health**: Monitor WebSocket connection status 774 823 - **Processing Metrics**: Track events processed, duplicates detected, errors 775 824 - **Latency Monitoring**: Measure time from firehose event to domain event publication 776 825 - **Error Tracking**: Log and alert on processing failures 777 826 778 827 ### Error Handling Strategy 828 + 779 829 - **Network Errors**: Automatic reconnection with exponential backoff (handled by @atproto/sync) 780 830 - **Processing Errors**: Log and continue (don't crash worker for single event failures) 781 831 - **Validation Errors**: Skip invalid records with detailed logging ··· 784 834 ## Benefits of This Architecture 785 835 786 836 ### 1. Separation of Concerns 837 + 787 838 - **External Events**: Firehose worker handles AT Protocol events 788 839 - **Internal Events**: Other workers handle domain events 789 840 - **Clear Boundaries**: No mixing of external and internal event systems 790 841 791 842 ### 2. Reliability 843 + 792 844 - **Direct Connection**: No intermediate queues that can fail 793 845 - **Built-in Resilience**: @atproto/sync handles reconnection and cursor management 794 846 - **Idempotent Processing**: Duplicate detection prevents double-processing 795 847 796 848 ### 3. Performance 849 + 797 850 - **No Serialization Overhead**: Direct event processing 798 851 - **Efficient Duplicate Detection**: Single table lookups 799 852 - **Minimal Latency**: Direct path from firehose to domain updates 800 853 801 854 ### 4. Maintainability 855 + 802 856 - **Simple Architecture**: Easy to understand and debug 803 857 - **Standard Patterns**: Follows existing DDD and layered architecture 804 858 - **Testable**: Clear interfaces for mocking and testing
+19 -4
.agent/logs/20251115_firehose_event_handlers.md
··· 16 16 ### Core Pattern 17 17 18 18 Each firehose processor will: 19 + 19 20 1. Parse the AT URI and extract relevant data from the record 20 21 2. Use `IAtUriResolutionService` to resolve AT URIs to domain entities 21 22 3. Call existing use cases with an optional `publishedRecordId` parameter ··· 24 25 ### Use Case Modifications 25 26 26 27 Add optional `publishedRecordId?: PublishedRecordId` parameter to these use cases: 28 + 27 29 - `AddUrlToLibraryUseCase` 28 - - `CreateCollectionUseCase` 30 + - `CreateCollectionUseCase` 29 31 - `UpdateCollectionUseCase` 30 32 - `DeleteCollectionUseCase` 31 33 - `RemoveCardFromLibraryUseCase` ··· 37 39 ### URL Card Events 38 40 39 41 #### Create 42 + 40 43 - Extract URL and curator DID from record 41 44 - Call `AddUrlToLibraryUseCase` with `publishedRecordId` from event 42 45 43 - #### Update 46 + #### Update 47 + 44 48 - Skip for now (URL cards don't typically update) 45 49 46 50 #### Delete 51 + 47 52 - Resolve AT URI to card ID 48 53 - Call `RemoveCardFromLibraryUseCase` with `publishedRecordId` to skip unpublishing 49 54 50 55 ### Note Card Events 51 56 52 57 #### Create/Update 58 + 53 59 - Extract parent card AT URI, note text, and curator DID 54 60 - Resolve parent card AT URI to card ID 55 61 - Call `UpdateUrlCardAssociationsUseCase` with note and `publishedRecordId` 56 62 57 63 #### Delete 64 + 58 65 - Resolve AT URI to card ID 59 66 - Verify it's a note card 60 67 - Call `RemoveCardFromLibraryUseCase` with `publishedRecordId` ··· 62 69 ### Collection Events 63 70 64 71 #### Create 72 + 65 73 - Extract name, description, curator DID from record 66 74 - Call `CreateCollectionUseCase` with `publishedRecordId` from event 67 75 68 76 #### Update 77 + 69 78 - Resolve AT URI to collection ID 70 79 - Extract updated fields from record 71 80 - Call `UpdateCollectionUseCase` with `publishedRecordId` 72 81 73 82 #### Delete 83 + 74 84 - Resolve AT URI to collection ID 75 85 - Call `DeleteCollectionUseCase` with `publishedRecordId` to skip unpublishing 76 86 77 87 ### Collection Link Events 78 88 79 89 #### Create 90 + 80 91 - Extract collection and card strong refs from record 81 92 - Resolve AT URIs to collection ID and card ID 82 93 - Call `UpdateUrlCardAssociationsUseCase` with `addToCollections` and link `publishedRecordId` 83 94 84 95 #### Delete 96 + 85 97 - Extract collection and card strong refs from record 86 - - Resolve AT URIs to collection ID and card ID 98 + - Resolve AT URIs to collection ID and card ID 87 99 - Call `UpdateUrlCardAssociationsUseCase` with `removeFromCollections` and link `publishedRecordId` 88 100 89 101 ## Key Benefits ··· 96 108 ## Files to Modify 97 109 98 110 ### Use Cases (add optional publishedRecordId parameter): 111 + 99 112 - `AddUrlToLibraryUseCase.ts` 100 113 - `CreateCollectionUseCase.ts` 101 - - `UpdateCollectionUseCase.ts` 114 + - `UpdateCollectionUseCase.ts` 102 115 - `DeleteCollectionUseCase.ts` 103 116 - `RemoveCardFromLibraryUseCase.ts` 104 117 - `UpdateUrlCardAssociationsUseCase.ts` (more complex - add collection link published record IDs) 105 118 106 119 ### Firehose Processors (simplify to call existing use cases): 120 + 107 121 - `ProcessCardFirehoseEventUseCase.ts` 108 122 - `ProcessCollectionFirehoseEventUseCase.ts` 109 123 - `ProcessCollectionLinkFirehoseEventUseCase.ts` 110 124 111 125 ### Services (add mapping storage methods): 126 + 112 127 - `IAtUriResolutionService.ts` (already has the methods we need) 113 128 114 129 This approach maintains clean separation of concerns while maximizing code reuse and minimizing the complexity of firehose event handling.
+38 -9
.agent/logs/20251115_optional_publishing_event_handling.md
··· 9 9 **Approach**: Extend the DTO with optional published record IDs for different operations. 10 10 11 11 **Pros**: 12 + 12 13 - Single interface, maintains existing API 13 14 - Minimal code changes 14 15 - Handles all scenarios in one place 15 16 16 17 **Cons**: 18 + 17 19 - Makes the DTO more complex 18 20 - Mixed concerns (normal operations vs firehose events) 19 21 - Hard to understand which parameters apply to which operations 20 22 21 23 **Implementation**: 24 + 22 25 ```typescript 23 26 export interface UpdateUrlCardAssociationsDTO { 24 27 cardId: string; ··· 38 41 **Approach**: Create dedicated use cases for firehose event processing that handle the publishing logic differently. 39 42 40 43 **Pros**: 44 + 41 45 - Clear separation of concerns 42 46 - Easier to understand and maintain 43 47 - Can optimize for firehose-specific requirements 44 48 - No impact on existing use cases 45 49 46 50 **Cons**: 51 + 47 52 - Code duplication 48 53 - More files to maintain 49 54 - Need to keep business logic in sync 50 55 51 56 **Implementation**: 57 + 52 58 ```typescript 53 59 // New use case specifically for firehose events 54 60 export class ProcessNoteCardFirehoseEventUseCase { 55 - async execute(request: ProcessNoteCardFirehoseEventDTO): Promise<Result<void>> { 61 + async execute( 62 + request: ProcessNoteCardFirehoseEventDTO, 63 + ): Promise<Result<void>> { 56 64 // Handle note card creation/update with pre-existing published record ID 57 65 // Skip all publishing operations 58 66 // Reuse domain services but with different flow ··· 60 68 } 61 69 62 70 export class ProcessCollectionLinkFirehoseEventUseCase { 63 - async execute(request: ProcessCollectionLinkFirehoseEventDTO): Promise<Result<void>> { 71 + async execute( 72 + request: ProcessCollectionLinkFirehoseEventDTO, 73 + ): Promise<Result<void>> { 64 74 // Handle collection link operations with pre-existing published record ID 65 75 // Skip publishing operations 66 76 } ··· 72 82 **Approach**: Add a context parameter that controls publishing behavior throughout the operation chain. 73 83 74 84 **Pros**: 85 + 75 86 - Clean separation of concerns 76 87 - Reuses existing logic 77 88 - Easy to extend for other contexts 78 89 - Minimal API changes 79 90 80 91 **Cons**: 92 + 81 93 - Requires threading context through multiple layers 82 94 - Could be forgotten in new code paths 83 95 84 96 **Implementation**: 97 + 85 98 ```typescript 86 99 export enum OperationContext { 87 100 USER_INITIATED = 'user_initiated', 88 101 FIREHOSE_EVENT = 'firehose_event', 89 - SYSTEM_MIGRATION = 'system_migration' 102 + SYSTEM_MIGRATION = 'system_migration', 90 103 } 91 104 92 105 export interface UpdateUrlCardAssociationsDTO { ··· 109 122 **Approach**: Modify domain services to accept publishing control parameters. 110 123 111 124 **Pros**: 125 + 112 126 - Centralized publishing control 113 127 - Reuses existing use case logic 114 128 - Clean separation at service level 115 129 116 130 **Cons**: 131 + 117 132 - Changes to multiple service interfaces 118 133 - Could make services more complex 119 134 120 135 **Implementation**: 136 + 121 137 ```typescript 122 138 // Modify CardLibraryService 123 139 async addCardToLibrary( ··· 129 145 } 130 146 ): Promise<Result<Card, ...>> 131 147 132 - // Modify CardCollectionService 148 + // Modify CardCollectionService 133 149 async addCardToCollections( 134 150 card: Card, 135 151 collectionIds: CollectionId[], ··· 146 162 **Approach**: Create different variants of the use case based on the execution context. 147 163 148 164 **Pros**: 165 + 149 166 - Clean separation without code duplication 150 167 - Easy to test different behaviors 151 168 - Flexible for future contexts 152 169 153 170 **Cons**: 171 + 154 172 - More complex setup 155 173 - Factory pattern overhead 156 174 157 175 **Implementation**: 176 + 158 177 ```typescript 159 178 export class UpdateUrlCardAssociationsUseCaseFactory { 160 - static createForUserOperation(dependencies): UpdateUrlCardAssociationsUseCase { 161 - return new UpdateUrlCardAssociationsUseCase(dependencies, { publishingEnabled: true }); 179 + static createForUserOperation( 180 + dependencies, 181 + ): UpdateUrlCardAssociationsUseCase { 182 + return new UpdateUrlCardAssociationsUseCase(dependencies, { 183 + publishingEnabled: true, 184 + }); 162 185 } 163 - 164 - static createForFirehoseEvent(dependencies): UpdateUrlCardAssociationsUseCase { 165 - return new UpdateUrlCardAssociationsUseCase(dependencies, { publishingEnabled: false }); 186 + 187 + static createForFirehoseEvent( 188 + dependencies, 189 + ): UpdateUrlCardAssociationsUseCase { 190 + return new UpdateUrlCardAssociationsUseCase(dependencies, { 191 + publishingEnabled: false, 192 + }); 166 193 } 167 194 } 168 195 ``` ··· 170 197 ## Recommended Approach: Option 3 (Context-Based Publishing Control) 171 198 172 199 **Rationale**: 200 + 173 201 - Provides clean separation without major architectural changes 174 202 - Reuses existing business logic 175 203 - Easy to understand and maintain ··· 177 205 - Minimal impact on existing code 178 206 179 207 **Implementation Strategy**: 208 + 180 209 1. Add `OperationContext` enum and optional context parameter to DTOs 181 210 2. Thread context through service calls 182 211 3. Modify services to check context before publishing
+2 -2
src/modules/atproto/application/handlers/FirehoseEventHandler.ts
··· 12 12 13 13 export class FirehoseEventHandler { 14 14 constructor( 15 - private processFirehoseEventUseCase: ProcessFirehoseEventUseCase 15 + private processFirehoseEventUseCase: ProcessFirehoseEventUseCase, 16 16 ) {} 17 17 18 18 async handle(event: AtProtoFirehoseEvent): Promise<Result<void>> { ··· 20 20 atUri: event.uri, 21 21 cid: event.cid, 22 22 eventType: event.eventType, 23 - record: event.record 23 + record: event.record, 24 24 }); 25 25 } 26 26 }
+4 -1
src/modules/atproto/application/useCases/ProcessCardFirehoseEventUseCase.ts
··· 11 11 UrlContent, 12 12 } from '../../infrastructure/lexicon/types/network/cosmik/card'; 13 13 import { AddUrlToLibraryUseCase } from '../../../cards/application/useCases/commands/AddUrlToLibraryUseCase'; 14 - import { UpdateUrlCardAssociationsUseCase, OperationContext } from '../../../cards/application/useCases/commands/UpdateUrlCardAssociationsUseCase'; 14 + import { 15 + UpdateUrlCardAssociationsUseCase, 16 + OperationContext, 17 + } from '../../../cards/application/useCases/commands/UpdateUrlCardAssociationsUseCase'; 15 18 import { RemoveCardFromLibraryUseCase } from '../../../cards/application/useCases/commands/RemoveCardFromLibraryUseCase'; 16 19 17 20 export interface ProcessCardFirehoseEventDTO {
+6 -3
src/modules/atproto/application/useCases/ProcessCollectionFirehoseEventUseCase.ts
··· 175 175 const collectionIdResult = 176 176 await this.atUriResolutionService.resolveCollectionId(request.atUri); 177 177 if (collectionIdResult.isErr()) { 178 - console.warn(`Failed to resolve collection ID: ${collectionIdResult.error.message}`); 178 + console.warn( 179 + `Failed to resolve collection ID: ${collectionIdResult.error.message}`, 180 + ); 179 181 return ok(undefined); 180 182 } 181 183 ··· 200 202 return ok(undefined); 201 203 } 202 204 203 - console.log(`Successfully deleted collection: ${result.value.collectionId}`); 205 + console.log( 206 + `Successfully deleted collection: ${result.value.collectionId}`, 207 + ); 204 208 } 205 209 206 210 return ok(undefined); ··· 209 213 return ok(undefined); 210 214 } 211 215 } 212 - 213 216 }
+54 -20
src/modules/atproto/application/useCases/ProcessCollectionLinkFirehoseEventUseCase.ts
··· 6 6 import { PublishedRecordId } from '../../../cards/domain/value-objects/PublishedRecordId'; 7 7 import { ATUri } from '../../domain/ATUri'; 8 8 import { Record as CollectionLinkRecord } from '../../infrastructure/lexicon/types/network/cosmik/collectionLink'; 9 - import { UpdateUrlCardAssociationsUseCase, OperationContext } from '../../../cards/application/useCases/commands/UpdateUrlCardAssociationsUseCase'; 9 + import { 10 + UpdateUrlCardAssociationsUseCase, 11 + OperationContext, 12 + } from '../../../cards/application/useCases/commands/UpdateUrlCardAssociationsUseCase'; 10 13 11 14 export interface ProcessCollectionLinkFirehoseEventDTO { 12 15 atUri: string; ··· 15 18 record?: CollectionLinkRecord; 16 19 } 17 20 18 - export class ProcessCollectionLinkFirehoseEventUseCase implements UseCase<ProcessCollectionLinkFirehoseEventDTO, Result<void>> { 21 + export class ProcessCollectionLinkFirehoseEventUseCase 22 + implements UseCase<ProcessCollectionLinkFirehoseEventDTO, Result<void>> 23 + { 19 24 constructor( 20 25 private atUriResolutionService: IAtUriResolutionService, 21 26 private updateUrlCardAssociationsUseCase: UpdateUrlCardAssociationsUseCase, 22 27 ) {} 23 28 24 - async execute(request: ProcessCollectionLinkFirehoseEventDTO): Promise<Result<void>> { 29 + async execute( 30 + request: ProcessCollectionLinkFirehoseEventDTO, 31 + ): Promise<Result<void>> { 25 32 try { 26 - console.log(`Processing collection link firehose event: ${request.atUri} (${request.eventType})`); 33 + console.log( 34 + `Processing collection link firehose event: ${request.atUri} (${request.eventType})`, 35 + ); 27 36 28 37 switch (request.eventType) { 29 38 case 'create': ··· 32 41 return await this.handleCollectionLinkDelete(request); 33 42 case 'update': 34 43 // Collection links don't typically have update operations 35 - console.log(`Collection link update event for ${request.atUri} (unusual)`); 44 + console.log( 45 + `Collection link update event for ${request.atUri} (unusual)`, 46 + ); 36 47 break; 37 48 } 38 49 ··· 46 57 request: ProcessCollectionLinkFirehoseEventDTO, 47 58 ): Promise<Result<void>> { 48 59 if (!request.record || !request.cid) { 49 - console.warn('Collection link create event missing record or cid, skipping'); 60 + console.warn( 61 + 'Collection link create event missing record or cid, skipping', 62 + ); 50 63 return ok(undefined); 51 64 } 52 65 ··· 62 75 const curatorDid = atUriResult.value.did.value; 63 76 64 77 // Resolve collection and card from strong refs 65 - const collectionId = await this.atUriResolutionService.resolveCollectionId( 66 - request.record.collection.uri, 67 - ); 78 + const collectionId = 79 + await this.atUriResolutionService.resolveCollectionId( 80 + request.record.collection.uri, 81 + ); 68 82 const cardId = await this.atUriResolutionService.resolveCardId( 69 83 request.record.card.uri, 70 84 ); 71 85 72 86 if (collectionId.isErr() || !collectionId.value) { 73 - console.warn(`Failed to resolve collection: ${request.record.collection.uri}`); 87 + console.warn( 88 + `Failed to resolve collection: ${request.record.collection.uri}`, 89 + ); 74 90 return ok(undefined); 75 91 } 76 92 ··· 85 101 }); 86 102 87 103 const collectionLinkMap = new Map<string, PublishedRecordId>(); 88 - collectionLinkMap.set(collectionId.value.getStringValue(), publishedRecordId); 104 + collectionLinkMap.set( 105 + collectionId.value.getStringValue(), 106 + publishedRecordId, 107 + ); 89 108 90 109 const result = await this.updateUrlCardAssociationsUseCase.execute({ 91 110 cardId: cardId.value.getStringValue(), ··· 98 117 }); 99 118 100 119 if (result.isErr()) { 101 - console.warn(`Failed to add card to collection: ${result.error.message}`); 120 + console.warn( 121 + `Failed to add card to collection: ${result.error.message}`, 122 + ); 102 123 return ok(undefined); 103 124 } 104 125 ··· 125 146 const curatorDid = atUriResult.value.did.value; 126 147 127 148 // Handle collection link deletion if we have it in our system 128 - const linkInfoResult = await this.atUriResolutionService.resolveCollectionLinkId(request.atUri); 149 + const linkInfoResult = 150 + await this.atUriResolutionService.resolveCollectionLinkId( 151 + request.atUri, 152 + ); 129 153 if (linkInfoResult.isErr()) { 130 - console.warn(`Failed to resolve collection link: ${linkInfoResult.error.message}`); 154 + console.warn( 155 + `Failed to resolve collection link: ${linkInfoResult.error.message}`, 156 + ); 131 157 return ok(undefined); 132 158 } 133 - 159 + 134 160 if (linkInfoResult.value) { 135 - console.log(`Collection link deleted externally: ${request.atUri}, removing from our system`); 136 - 161 + console.log( 162 + `Collection link deleted externally: ${request.atUri}, removing from our system`, 163 + ); 164 + 137 165 const publishedRecordId = PublishedRecordId.create({ 138 166 uri: request.atUri, 139 167 cid: request.cid || 'deleted', ··· 142 170 const result = await this.updateUrlCardAssociationsUseCase.execute({ 143 171 cardId: linkInfoResult.value.cardId.getStringValue(), 144 172 curatorId: curatorDid, 145 - removeFromCollections: [linkInfoResult.value.collectionId.getStringValue()], 173 + removeFromCollections: [ 174 + linkInfoResult.value.collectionId.getStringValue(), 175 + ], 146 176 context: OperationContext.FIREHOSE_EVENT, 147 177 }); 148 178 149 179 if (result.isErr()) { 150 - console.warn(`Failed to remove card from collection: ${result.error.message}`); 180 + console.warn( 181 + `Failed to remove card from collection: ${result.error.message}`, 182 + ); 151 183 return ok(undefined); 152 184 } 153 185 154 - console.log(`Successfully removed card from collection from firehose event`); 186 + console.log( 187 + `Successfully removed card from collection from firehose event`, 188 + ); 155 189 } 156 190 157 191 return ok(undefined);
-1
src/modules/atproto/domain/ATUri.ts
··· 119 119 public equals(other: ATUri): boolean { 120 120 return this.props.value === other.props.value; 121 121 } 122 - 123 122 }
+4 -4
src/modules/atproto/domain/services/IFirehoseEventDuplicationService.ts
··· 4 4 5 5 export interface IFirehoseEventDuplicationService { 6 6 hasEventBeenProcessed( 7 - atUri: string, 8 - cid: string | null, 9 - operation: FirehoseEventType 7 + atUri: string, 8 + cid: string | null, 9 + operation: FirehoseEventType, 10 10 ): Promise<Result<boolean>>; 11 - 11 + 12 12 hasBeenDeleted(atUri: string): Promise<Result<boolean>>; 13 13 }
+3 -3
src/modules/atproto/infrastructure/mappers/CollectionLinkMapper.ts
··· 6 6 type CollectionLinkRecordDTO = Record; 7 7 8 8 export class CollectionLinkMapper { 9 - static readonly collectionLinkType = new EnvironmentConfigService() 10 - .getAtProtoCollections().collectionLink; 9 + static readonly collectionLinkType = 10 + new EnvironmentConfigService().getAtProtoCollections().collectionLink; 11 11 12 12 static toCreateRecordDTO( 13 13 cardLink: CardLink, 14 14 collectionPublishedRecordId: PublishedRecordIdProps, 15 15 cardPublishedRecordId: PublishedRecordIdProps, 16 - originalCardPublishedRecordId?: PublishedRecordIdProps 16 + originalCardPublishedRecordId?: PublishedRecordIdProps, 17 17 ): CollectionLinkRecordDTO { 18 18 const record: CollectionLinkRecordDTO = { 19 19 $type: this.collectionLinkType as any,
+24 -20
src/modules/atproto/infrastructure/services/DrizzleFirehoseEventDuplicationService.ts
··· 1 1 import { eq, and } from 'drizzle-orm'; 2 2 import { PostgresJsDatabase } from 'drizzle-orm/postgres-js'; 3 - import { 4 - IFirehoseEventDuplicationService, 5 - FirehoseEventType 3 + import { 4 + IFirehoseEventDuplicationService, 5 + FirehoseEventType, 6 6 } from '../../domain/services/IFirehoseEventDuplicationService'; 7 7 import { IAtUriResolutionService } from '../../../cards/domain/services/IAtUriResolutionService'; 8 8 import { ATUri } from '../../domain/ATUri'; ··· 10 10 import { Result, ok, err } from 'src/shared/core/Result'; 11 11 import { EnvironmentConfigService } from 'src/shared/infrastructure/config/EnvironmentConfigService'; 12 12 13 - export class DrizzleFirehoseEventDuplicationService implements IFirehoseEventDuplicationService { 13 + export class DrizzleFirehoseEventDuplicationService 14 + implements IFirehoseEventDuplicationService 15 + { 14 16 constructor( 15 17 private db: PostgresJsDatabase, 16 18 private atUriResolver: IAtUriResolutionService, 17 - private configService: EnvironmentConfigService 19 + private configService: EnvironmentConfigService, 18 20 ) {} 19 21 20 22 async hasEventBeenProcessed( 21 - atUri: string, 22 - cid: string | null, 23 - operation: FirehoseEventType 23 + atUri: string, 24 + cid: string | null, 25 + operation: FirehoseEventType, 24 26 ): Promise<Result<boolean>> { 25 27 try { 26 28 // For CREATE/UPDATE: check if (uri, cid) exists 27 29 if (operation === 'create' || operation === 'update') { 28 30 if (!cid) return ok(false); 29 - 31 + 30 32 const result = await this.db 31 33 .select({ id: publishedRecords.id }) 32 34 .from(publishedRecords) 33 35 .where( 34 - and( 35 - eq(publishedRecords.uri, atUri), 36 - eq(publishedRecords.cid, cid) 37 - ) 36 + and(eq(publishedRecords.uri, atUri), eq(publishedRecords.cid, cid)), 38 37 ) 39 38 .limit(1); 40 - 39 + 41 40 return ok(result.length > 0); 42 41 } 43 - 42 + 44 43 // For DELETE: use more complex logic 45 44 return this.hasBeenDeleted(atUri); 46 45 } catch (error) { ··· 71 70 72 71 // 3. Check if entity still exists based on collection type 73 72 switch (collection) { 74 - case collections.collection: 75 - const collectionIdResult = await this.atUriResolver.resolveCollectionId(atUri); 73 + case collections.collection: { 74 + const collectionIdResult = 75 + await this.atUriResolver.resolveCollectionId(atUri); 76 76 if (collectionIdResult.isErr()) { 77 77 return err(collectionIdResult.error); 78 78 } 79 79 return ok(collectionIdResult.value === null); 80 - case collections.card: 80 + } 81 + case collections.card: { 81 82 const cardIdResult = await this.atUriResolver.resolveCardId(atUri); 82 83 if (cardIdResult.isErr()) { 83 84 return err(cardIdResult.error); 84 85 } 85 86 return ok(cardIdResult.value === null); 86 - case collections.collectionLink: 87 - const linkInfoResult = await this.atUriResolver.resolveCollectionLinkId(atUri); 87 + } 88 + case collections.collectionLink: { 89 + const linkInfoResult = 90 + await this.atUriResolver.resolveCollectionLinkId(atUri); 88 91 if (linkInfoResult.isErr()) { 89 92 return err(linkInfoResult.error); 90 93 } 91 94 return ok(linkInfoResult.value === null); 95 + } 92 96 default: 93 97 return err(new Error(`Unknown collection type: ${collection}`)); 94 98 }
+7 -10
src/modules/cards/application/useCases/commands/AddUrlToLibraryUseCase.ts
··· 192 192 193 193 // Update note card in library (handles save and republish) 194 194 const updateNoteResult = 195 - await this.cardLibraryService.updateCardInLibrary(noteCard, curatorId); 195 + await this.cardLibraryService.updateCardInLibrary( 196 + noteCard, 197 + curatorId, 198 + ); 196 199 if (updateNoteResult.isErr()) { 197 200 // Propagate authentication errors 198 - if ( 199 - updateNoteResult.error instanceof AuthenticationError 200 - ) { 201 + if (updateNoteResult.error instanceof AuthenticationError) { 201 202 return err(updateNoteResult.error); 202 203 } 203 - if ( 204 - updateNoteResult.error instanceof AppError.UnexpectedError 205 - ) { 204 + if (updateNoteResult.error instanceof AppError.UnexpectedError) { 206 205 return err(updateNoteResult.error); 207 206 } 208 - return err( 209 - new ValidationError(updateNoteResult.error.message), 210 - ); 207 + return err(new ValidationError(updateNoteResult.error.message)); 211 208 } 212 209 213 210 // Update noteCard reference to the one returned by the service
+2 -1
src/modules/cards/application/useCases/commands/CreateCollectionUseCase.ts
··· 90 90 collection.markAsPublished(request.publishedRecordId); 91 91 } else { 92 92 // Publish collection normally 93 - const publishResult = await this.collectionPublisher.publish(collection); 93 + const publishResult = 94 + await this.collectionPublisher.publish(collection); 94 95 if (publishResult.isErr()) { 95 96 // Propagate authentication errors 96 97 if (publishResult.error instanceof AuthenticationError) {
+5 -1
src/modules/cards/application/useCases/commands/DeleteCollectionUseCase.ts
··· 97 97 } 98 98 99 99 // Handle unpublishing - skip if publishedRecordId provided (firehose event) 100 - if (!request.publishedRecordId && collection.isPublished && collection.publishedRecordId) { 100 + if ( 101 + !request.publishedRecordId && 102 + collection.isPublished && 103 + collection.publishedRecordId 104 + ) { 101 105 const unpublishResult = await this.collectionPublisher.unpublish( 102 106 collection.publishedRecordId, 103 107 );
+6 -2
src/modules/cards/application/useCases/commands/RemoveCardFromLibraryUseCase.ts
··· 106 106 if (removeFromLibraryResult.error instanceof AuthenticationError) { 107 107 return err(removeFromLibraryResult.error); 108 108 } 109 - if (removeFromLibraryResult.error instanceof AppError.UnexpectedError) { 109 + if ( 110 + removeFromLibraryResult.error instanceof AppError.UnexpectedError 111 + ) { 110 112 return err(removeFromLibraryResult.error); 111 113 } 112 - return err(new ValidationError(removeFromLibraryResult.error.message)); 114 + return err( 115 + new ValidationError(removeFromLibraryResult.error.message), 116 + ); 113 117 } 114 118 115 119 const updatedCard = removeFromLibraryResult.value;
+1 -1
src/modules/cards/application/useCases/commands/UpdateCollectionUseCase.ts
··· 119 119 if (request.publishedRecordId) { 120 120 // Update published record ID with provided value 121 121 collection.markAsPublished(request.publishedRecordId); 122 - 122 + 123 123 // Save collection with updated published record ID 124 124 const saveUpdatedResult = 125 125 await this.collectionRepository.save(collection);
+48 -31
src/modules/cards/application/useCases/commands/UpdateUrlCardAssociationsUseCase.ts
··· 19 19 export enum OperationContext { 20 20 USER_INITIATED = 'user_initiated', 21 21 FIREHOSE_EVENT = 'firehose_event', 22 - SYSTEM_MIGRATION = 'system_migration' 22 + SYSTEM_MIGRATION = 'system_migration', 23 23 } 24 24 25 25 export interface UpdateUrlCardAssociationsDTO { ··· 164 164 } 165 165 166 166 // Determine service options based on context 167 - const isFirehoseEvent = request.context === OperationContext.FIREHOSE_EVENT; 168 - const noteCardOptions = isFirehoseEvent && request.publishedRecordIds?.noteCard ? { 169 - skipPublishing: true, 170 - publishedRecordId: request.publishedRecordIds.noteCard, 171 - } : undefined; 167 + const isFirehoseEvent = 168 + request.context === OperationContext.FIREHOSE_EVENT; 169 + const noteCardOptions = 170 + isFirehoseEvent && request.publishedRecordIds?.noteCard 171 + ? { 172 + skipPublishing: true, 173 + publishedRecordId: request.publishedRecordIds.noteCard, 174 + } 175 + : undefined; 172 176 173 177 // Update note card in library (handles save and republish) 174 178 const updateNoteResult = 175 - await this.cardLibraryService.updateCardInLibrary(noteCard, curatorId, noteCardOptions); 179 + await this.cardLibraryService.updateCardInLibrary( 180 + noteCard, 181 + curatorId, 182 + noteCardOptions, 183 + ); 176 184 if (updateNoteResult.isErr()) { 177 185 // Propagate authentication errors 178 - if ( 179 - updateNoteResult.error instanceof AuthenticationError 180 - ) { 186 + if (updateNoteResult.error instanceof AuthenticationError) { 181 187 return err(updateNoteResult.error); 182 188 } 183 - if ( 184 - updateNoteResult.error instanceof AppError.UnexpectedError 185 - ) { 189 + if (updateNoteResult.error instanceof AppError.UnexpectedError) { 186 190 return err(updateNoteResult.error); 187 191 } 188 - return err( 189 - new ValidationError(updateNoteResult.error.message), 190 - ); 192 + return err(new ValidationError(updateNoteResult.error.message)); 191 193 } 192 194 193 195 // Update noteCard reference to the one returned by the service ··· 221 223 } 222 224 223 225 // Determine service options based on context 224 - const isFirehoseEvent = request.context === OperationContext.FIREHOSE_EVENT; 225 - const noteCardOptions = isFirehoseEvent && request.publishedRecordIds?.noteCard ? { 226 - skipPublishing: true, 227 - publishedRecordId: request.publishedRecordIds.noteCard, 228 - } : undefined; 226 + const isFirehoseEvent = 227 + request.context === OperationContext.FIREHOSE_EVENT; 228 + const noteCardOptions = 229 + isFirehoseEvent && request.publishedRecordIds?.noteCard 230 + ? { 231 + skipPublishing: true, 232 + publishedRecordId: request.publishedRecordIds.noteCard, 233 + } 234 + : undefined; 229 235 230 236 // Add note card to library using domain service 231 237 const addNoteCardToLibraryResult = 232 - await this.cardLibraryService.addCardToLibrary(noteCard, curatorId, noteCardOptions); 238 + await this.cardLibraryService.addCardToLibrary( 239 + noteCard, 240 + curatorId, 241 + noteCardOptions, 242 + ); 233 243 if (addNoteCardToLibraryResult.isErr()) { 234 244 // Propagate authentication errors 235 245 if ( ··· 273 283 } 274 284 275 285 // Determine service options based on context 276 - const isFirehoseEvent = request.context === OperationContext.FIREHOSE_EVENT; 277 - const collectionOptions = isFirehoseEvent && request.publishedRecordIds?.collectionLinks ? { 278 - skipPublishing: true, 279 - publishedRecordIds: request.publishedRecordIds.collectionLinks, 280 - } : undefined; 286 + const isFirehoseEvent = 287 + request.context === OperationContext.FIREHOSE_EVENT; 288 + const collectionOptions = 289 + isFirehoseEvent && request.publishedRecordIds?.collectionLinks 290 + ? { 291 + skipPublishing: true, 292 + publishedRecordIds: request.publishedRecordIds.collectionLinks, 293 + } 294 + : undefined; 281 295 282 296 const addToCollectionsResult = 283 297 await this.cardCollectionService.addCardToCollections( ··· 335 349 } 336 350 337 351 // Determine service options based on context 338 - const isFirehoseEvent = request.context === OperationContext.FIREHOSE_EVENT; 339 - const collectionOptions = isFirehoseEvent ? { 340 - skipPublishing: true, 341 - } : undefined; 352 + const isFirehoseEvent = 353 + request.context === OperationContext.FIREHOSE_EVENT; 354 + const collectionOptions = isFirehoseEvent 355 + ? { 356 + skipPublishing: true, 357 + } 358 + : undefined; 342 359 343 360 const removeFromCollectionsResult = 344 361 await this.cardCollectionService.removeCardFromCollections(
+7 -2
src/modules/cards/domain/services/CardCollectionService.ts
··· 70 70 71 71 // Handle publishing based on options 72 72 if (options?.skipPublishing && options?.publishedRecordIds) { 73 - const publishedRecordId = options.publishedRecordIds.get(collectionId.getStringValue()); 73 + const publishedRecordId = options.publishedRecordIds.get( 74 + collectionId.getStringValue(), 75 + ); 74 76 if (publishedRecordId) { 75 77 // Skip publishing and use provided record ID 76 78 collection.markCardLinkAsPublished(card.cardId, publishedRecordId); ··· 96 98 } 97 99 98 100 // Mark the card link as published in the collection 99 - collection.markCardLinkAsPublished(card.cardId, publishLinkResult.value); 101 + collection.markCardLinkAsPublished( 102 + card.cardId, 103 + publishLinkResult.value, 104 + ); 100 105 } 101 106 102 107 // Save the updated collection
+9 -4
src/modules/cards/tests/utils/InMemoryAtUriResolutionService.ts
··· 64 64 } 65 65 66 66 async resolveCollectionLinkId( 67 - atUri: string 68 - ): Promise<Result<{collectionId: CollectionId, cardId: CardId} | null>> { 67 + atUri: string, 68 + ): Promise<Result<{ collectionId: CollectionId; cardId: CardId } | null>> { 69 69 const result = await this.resolveAtUri(atUri); 70 70 71 71 if (result.isErr()) { 72 72 return err(result.error); 73 73 } 74 74 75 - if (!result.value || result.value.type !== AtUriResourceType.COLLECTION_LINK) { 75 + if ( 76 + !result.value || 77 + result.value.type !== AtUriResourceType.COLLECTION_LINK 78 + ) { 76 79 return ok(null); 77 80 } 78 81 79 - return ok(result.value.id as {collectionId: CollectionId, cardId: CardId}); 82 + return ok( 83 + result.value.id as { collectionId: CollectionId; cardId: CardId }, 84 + ); 80 85 } 81 86 }