nandi.uk
This repository has no description
update early testers list
author
Wesley Finck
date
(Feb 9, 2026, 11:59 AM -0800)
commit
dd5ee44f
dd5ee44fef7162391bb1c420d4898cefe9cb1fd2
parent
d76a026b
d76a026bec5e92d2a0798211f59e7e3faf0279f6
+1982
+663
.agent/logs/20260305_following_feed_application_layer.md
+663
.agent/logs/20260305_following_feed_application_layer.md
···
1
+
# Following Feed: Application Layer Implementation Guide
2
+
3
+
## Architectural Decision: Single Use Case with Fan-out
4
+
5
+
### Decision
6
+
7
+
**Fan-out logic will be coordinated inside `AddActivityToFeedUseCase`**, not in a separate use case.
8
+
9
+
## Application Layer Architecture
10
+
11
+
### Overview
12
+
13
+
```
14
+
Event: CARD_ADDED_TO_LIBRARY / CARD_ADDED_TO_COLLECTION
15
+
↓
16
+
Event Handler (unchanged interface)
17
+
↓
18
+
AddActivityToFeedUseCase.execute()
19
+
├─ 1. Validate input (existing)
20
+
├─ 2. Fetch card metadata (existing)
21
+
├─ 3. Create activity via FeedService
22
+
│ └─ [Distributed lock + dedup check + insert/update]
23
+
├─ 4. Get followers of actor (user) ← NEW
24
+
├─ 5. Get followers of collections (if any) ← NEW
25
+
├─ 6. Combine & deduplicate follower IDs ← NEW
26
+
└─ 7. Fan-out to followers ← NEW
27
+
└─ [Idempotent bulk insert via ON CONFLICT DO NOTHING]
28
+
```
29
+
30
+
### Responsibilities
31
+
32
+
**Use Case (`AddActivityToFeedUseCase`):**
33
+
34
+
- Orchestrates the complete flow: validation → activity creation → fan-out
35
+
- Coordinates calls to multiple repositories
36
+
- Implements business logic
37
+
- Handles errors with appropriate strategies
38
+
39
+
---
40
+
41
+
## Repository Interfaces
42
+
43
+
### New Repository: `IFollowsRepository`
44
+
45
+
**Purpose:** Query following relationships to determine fan-out targets.
46
+
47
+
**Note:** This repository is **read-only** for the purpose of feed fan-out. Follow/unfollow operations (write) are out of scope for this implementation.
48
+
49
+
```typescript
50
+
export interface Follow {
51
+
followerId: string; // DID of user who is following
52
+
targetId: string; // User DID or Collection UUID being followed
53
+
targetType: 'USER' | 'COLLECTION';
54
+
}
55
+
56
+
export interface IFollowsRepository {
57
+
/**
58
+
* Get all followers of a specific user.
59
+
*
60
+
* @param targetId - User DID being followed
61
+
* @param targetType - Must be 'USER' for this method
62
+
* @returns Array of Follow records (can be empty if no followers)
63
+
*
64
+
* Example:
65
+
* getFollowers('did:plc:alice123', 'USER')
66
+
* → [{ followerId: 'did:plc:bob456', targetId: 'did:plc:alice123', targetType: 'USER' }]
67
+
*/
68
+
getFollowers(targetId: string, targetType: 'USER'): Promise<Result<Follow[]>>;
69
+
70
+
/**
71
+
* Get all followers of multiple collections (combined, deduplicated).
72
+
*
73
+
* @param collectionIds - Array of collection UUIDs
74
+
* @returns Array of Follow records (can be empty)
75
+
*
76
+
* Example:
77
+
* getFollowersOfCollections(['uuid-1', 'uuid-2'])
78
+
* → [
79
+
* { followerId: 'did:plc:bob456', targetId: 'uuid-1', targetType: 'COLLECTION' },
80
+
* { followerId: 'did:plc:carol789', targetId: 'uuid-2', targetType: 'COLLECTION' }
81
+
* ]
82
+
*
83
+
* Notes:
84
+
* - Returns empty array if collectionIds is empty
85
+
* - Results may include duplicates if a user follows multiple input collections
86
+
* (deduplication happens at use case level)
87
+
*/
88
+
getFollowersOfCollections(collectionIds: string[]): Promise<Result<Follow[]>>;
89
+
}
90
+
```
91
+
92
+
### Updated Repository: `IFeedRepository`
93
+
94
+
**Add these methods to the existing interface:**
95
+
96
+
```typescript
97
+
export interface IFeedRepository {
98
+
// ... existing methods (addActivity, updateActivity, etc.) ...
99
+
100
+
/**
101
+
* Fan-out an activity to multiple followers' following feeds.
102
+
*
103
+
* @param activityId - Activity to distribute
104
+
* @param followerIds - User DIDs to receive this activity (deduplicated by caller)
105
+
* @param createdAt - Activity timestamp (denormalized for sorting)
106
+
* @returns Success or error
107
+
*
108
+
* Idempotency guarantee:
109
+
* - Uses ON CONFLICT DO NOTHING on primary key (user_id, activity_id)
110
+
* - Safe to call multiple times with same inputs
111
+
* - Retries are silent (no error on duplicate)
112
+
*
113
+
* Performance:
114
+
* - Bulk insert operation (single query)
115
+
* - Returns immediately if followerIds is empty (no-op)
116
+
*
117
+
* Example:
118
+
* fanOutActivityToFollowers(
119
+
* activityId,
120
+
* ['did:plc:bob456', 'did:plc:carol789'],
121
+
* new Date('2026-03-05T10:00:00Z')
122
+
* )
123
+
* → Inserts 2 rows into following_feed_items
124
+
*/
125
+
fanOutActivityToFollowers(
126
+
activityId: ActivityId,
127
+
followerIds: string[],
128
+
createdAt: Date,
129
+
): Promise<Result<void>>;
130
+
131
+
/**
132
+
* Get a user's following feed (paginated).
133
+
*
134
+
* @param userId - User DID whose feed to fetch
135
+
* @param options - Pagination, filters (urlType, source, beforeActivityId)
136
+
* @returns Paginated feed activities
137
+
*
138
+
* Query pattern:
139
+
* - Filters by user_id on following_feed_items
140
+
* - JOINs to feed_activities for full activity data
141
+
* - Supports same filters as global feed (urlType, source)
142
+
* - Cursor-based pagination via beforeActivityId
143
+
*
144
+
* Example:
145
+
* getFollowingFeed('did:plc:bob456', { page: 1, limit: 20, urlType: 'article' })
146
+
* → Returns up to 20 activities from Bob's following feed
147
+
*/
148
+
getFollowingFeed(
149
+
userId: string,
150
+
options: FeedQueryOptions,
151
+
): Promise<Result<PaginatedFeedResult>>;
152
+
}
153
+
```
154
+
155
+
**Interface Contract:**
156
+
157
+
- `fanOutActivityToFollowers`:
158
+
- **Idempotent** - Safe to retry with same inputs
159
+
- **Bulk operation** - Single query for all follower IDs
160
+
- **No-op if empty** - Returns `ok(undefined)` immediately if `followerIds.length === 0`
161
+
- **Denormalized timestamp** - Stores `createdAt` for efficient sorting (no JOIN needed)
162
+
163
+
- `getFollowingFeed`:
164
+
- **Same query options as global feed** - Reuses `FeedQueryOptions` type
165
+
- **Same result format** - Returns `PaginatedFeedResult` (consistent with other feed queries)
166
+
- **Efficient JOIN** - Single query with proper index usage
167
+
168
+
---
169
+
170
+
## Use Case Orchestration
171
+
172
+
### Updated `AddActivityToFeedUseCase`
173
+
174
+
#### Constructor Dependencies
175
+
176
+
```typescript
177
+
export class AddActivityToFeedUseCase implements UseCase<...> {
178
+
constructor(
179
+
private feedService: FeedService, // Existing
180
+
private cardRepository: ICardRepository, // Existing
181
+
private followsRepository: IFollowsRepository, // NEW - for fetching followers
182
+
private feedRepository: IFeedRepository, // NEW - for fan-out operation
183
+
) {}
184
+
}
185
+
```
186
+
187
+
#### Execution Flow
188
+
189
+
```typescript
190
+
async execute(request: AddActivityToFeedDTO): Promise<Result<...>> {
191
+
try {
192
+
// ========================================
193
+
// PHASE 1: VALIDATION (Existing)
194
+
// ========================================
195
+
196
+
// Validate actorId, cardId, collectionIds
197
+
// (existing code, unchanged)
198
+
199
+
// ========================================
200
+
// PHASE 2: FETCH CARD METADATA (Existing)
201
+
// ========================================
202
+
203
+
// Fetch card to get urlType and source
204
+
// (existing code, unchanged)
205
+
206
+
// ========================================
207
+
// PHASE 3: CREATE ACTIVITY (Existing)
208
+
// ========================================
209
+
210
+
// Call FeedService with distributed locking
211
+
const activityResult = await this.feedService.addCardCollectedActivity(
212
+
actorId,
213
+
cardId,
214
+
collectionIds,
215
+
urlType,
216
+
source,
217
+
createdAt,
218
+
);
219
+
220
+
if (activityResult.isErr()) {
221
+
return err(new ValidationError(activityResult.error.message));
222
+
}
223
+
224
+
const activity = activityResult.value;
225
+
226
+
// ========================================
227
+
// PHASE 4: GET FOLLOWERS (NEW)
228
+
// ========================================
229
+
230
+
// 4a. Get followers of the actor (user who created activity)
231
+
const userFollowersResult = await this.followsRepository.getFollowers(
232
+
actorId.value,
233
+
'USER'
234
+
);
235
+
236
+
const userFollowers = userFollowersResult.isOk()
237
+
? userFollowersResult.value.map(f => f.followerId)
238
+
: [];
239
+
240
+
// 4b. Get followers of collections (if any)
241
+
let collectionFollowers: string[] = [];
242
+
if (collectionIds && collectionIds.length > 0) {
243
+
const collectionIdStrings = collectionIds.map(id => id.getStringValue());
244
+
const collectionFollowersResult =
245
+
await this.followsRepository.getFollowersOfCollections(collectionIdStrings);
246
+
247
+
collectionFollowers = collectionFollowersResult.isOk()
248
+
? collectionFollowersResult.value.map(f => f.followerId)
249
+
: [];
250
+
}
251
+
252
+
// 4c. Combine and deduplicate follower IDs
253
+
const allFollowerIds = new Set([...userFollowers, ...collectionFollowers]);
254
+
255
+
// ========================================
256
+
// PHASE 5: FAN-OUT (NEW)
257
+
// ========================================
258
+
259
+
if (allFollowerIds.size > 0) {
260
+
const fanOutResult = await this.feedRepository.fanOutActivityToFollowers(
261
+
activity.activityId,
262
+
Array.from(allFollowerIds),
263
+
activity.createdAt,
264
+
);
265
+
266
+
// Error handling: Log but don't fail the use case
267
+
// Activity already exists in global feed
268
+
// Event retries will eventually distribute it
269
+
if (fanOutResult.isErr()) {
270
+
console.error('Fan-out failed (will retry on event retry):', fanOutResult.error);
271
+
// Note: We do NOT return err here - see Error Handling section
272
+
}
273
+
}
274
+
275
+
// ========================================
276
+
// RETURN SUCCESS
277
+
// ========================================
278
+
279
+
return ok({
280
+
activityId: activity.activityId.getStringValue(),
281
+
});
282
+
283
+
} catch (error) {
284
+
return err(AppError.UnexpectedError.create(error));
285
+
}
286
+
}
287
+
```
288
+
289
+
### Orchestration Principles
290
+
291
+
1. **Sequential execution** - Each phase depends on previous phase results
292
+
2. **Graceful degradation** - Follower lookup errors don't fail the use case (returns empty array)
293
+
3. **Deduplication at use case level** - Combine user + collection followers using `Set`
294
+
4. **Skip empty fan-out** - Don't call repository if no followers
295
+
5. **Business logic lives here** - Future conditional fan-out rules (e.g., "don't fan-out Margin library activities") implemented in this orchestration layer
296
+
297
+
---
298
+
299
+
## Integration with Event Handlers
300
+
301
+
### No Changes Required
302
+
303
+
Event handlers remain unchanged:
304
+
305
+
```typescript
306
+
export class CardAddedToLibraryEventHandler
307
+
implements IEventHandler<CardAddedToLibraryEvent>
308
+
{
309
+
constructor(private addActivityToFeedUseCase: AddActivityToFeedUseCase) {}
310
+
311
+
async handle(event: CardAddedToLibraryEvent): Promise<Result<void>> {
312
+
const result = await this.addActivityToFeedUseCase.execute({
313
+
type: ActivityTypeEnum.CARD_COLLECTED,
314
+
actorId: event.curatorId.value,
315
+
cardId: event.cardId.getStringValue(),
316
+
collectionIds: undefined,
317
+
createdAt: event.addedAt,
318
+
});
319
+
320
+
if (result.isErr()) {
321
+
console.error('Failed to add library activity to feed:', result.error);
322
+
return err(result.error);
323
+
}
324
+
325
+
return ok(undefined);
326
+
}
327
+
}
328
+
```
329
+
330
+
## Dependency Injection Updates
331
+
332
+
### Factory Changes Required
333
+
334
+
Update the factory/composition root to inject new dependencies:
335
+
336
+
```typescript
337
+
// UseCaseFactory or similar
338
+
export class UseCaseFactory {
339
+
createAddActivityToFeedUseCase(): AddActivityToFeedUseCase {
340
+
return new AddActivityToFeedUseCase(
341
+
this.feedService, // Existing
342
+
this.cardRepository, // Existing
343
+
this.repositoryFactory.followsRepository, // NEW
344
+
this.repositoryFactory.feedRepository, // NEW (already exists, just inject)
345
+
);
346
+
}
347
+
}
348
+
```
349
+
350
+
```typescript
351
+
// RepositoryFactory
352
+
export class RepositoryFactory {
353
+
// Existing
354
+
get feedRepository(): IFeedRepository { ... }
355
+
get cardRepository(): ICardRepository { ... }
356
+
357
+
// NEW
358
+
get followsRepository(): IFollowsRepository {
359
+
return this._followsRepository ??= new DrizzleFollowsRepository(this.db);
360
+
}
361
+
}
362
+
```
363
+
364
+
---
365
+
366
+
## Database Schema
367
+
368
+
### New Tables
369
+
370
+
```sql
371
+
-- Following relationships (polymorphic - supports users AND collections)
372
+
CREATE TABLE follows (
373
+
follower_id TEXT NOT NULL, -- DID of user who is following
374
+
target_id TEXT NOT NULL, -- ID of what's being followed (user DID or collection UUID)
375
+
target_type TEXT NOT NULL, -- 'USER' or 'COLLECTION'
376
+
created_at TIMESTAMP NOT NULL DEFAULT NOW(),
377
+
PRIMARY KEY (follower_id, target_id, target_type)
378
+
);
379
+
380
+
CREATE INDEX idx_follows_follower ON follows(follower_id);
381
+
CREATE INDEX idx_follows_target ON follows(target_id, target_type);
382
+
383
+
-- Following feed items (fan-out target)
384
+
CREATE TABLE following_feed_items (
385
+
user_id TEXT NOT NULL, -- DID of feed owner
386
+
activity_id UUID NOT NULL REFERENCES feed_activities(id) ON DELETE CASCADE,
387
+
created_at TIMESTAMP NOT NULL, -- Denormalized from activity for sorting
388
+
PRIMARY KEY (user_id, activity_id)
389
+
);
390
+
391
+
CREATE INDEX idx_following_feed_user_time ON following_feed_items(user_id, created_at DESC);
392
+
```
393
+
394
+
### New Repository: IFollowsRepository
395
+
396
+
**Note:** This repository only defines methods needed for feed fan-out. Follow/unfollow functionality is out of scope for this document.
397
+
398
+
```typescript
399
+
export interface Follow {
400
+
followerId: string;
401
+
targetId: string;
402
+
targetType: 'USER' | 'COLLECTION';
403
+
}
404
+
405
+
export interface IFollowsRepository {
406
+
// Get all followers of a specific user
407
+
getFollowers(targetId: string, targetType: 'USER'): Promise<Result<Follow[]>>;
408
+
409
+
// Get all followers of multiple collections (combined)
410
+
getFollowersOfCollections(collectionIds: string[]): Promise<Result<Follow[]>>;
411
+
}
412
+
```
413
+
414
+
---
415
+
416
+
### New Repository Implementation: DrizzleFollowsRepository
417
+
418
+
```typescript
419
+
export class DrizzleFollowsRepository implements IFollowsRepository {
420
+
constructor(private db: PostgresJsDatabase) {}
421
+
422
+
async getFollowers(
423
+
targetId: string,
424
+
targetType: 'USER',
425
+
): Promise<Result<Follow[]>> {
426
+
try {
427
+
const results = await this.db
428
+
.select()
429
+
.from(follows)
430
+
.where(
431
+
and(
432
+
eq(follows.target_id, targetId),
433
+
eq(follows.target_type, targetType),
434
+
),
435
+
);
436
+
437
+
return ok(
438
+
results.map((r) => ({
439
+
followerId: r.follower_id,
440
+
targetId: r.target_id,
441
+
targetType: r.target_type as 'USER' | 'COLLECTION',
442
+
})),
443
+
);
444
+
} catch (error) {
445
+
return err(error as Error);
446
+
}
447
+
}
448
+
449
+
async getFollowersOfCollections(
450
+
collectionIds: string[],
451
+
): Promise<Result<Follow[]>> {
452
+
try {
453
+
if (collectionIds.length === 0) {
454
+
return ok([]);
455
+
}
456
+
457
+
const results = await this.db
458
+
.select()
459
+
.from(follows)
460
+
.where(
461
+
and(
462
+
sql`${follows.target_id} = ANY(${collectionIds}::text[])`,
463
+
eq(follows.target_type, 'COLLECTION'),
464
+
),
465
+
);
466
+
467
+
return ok(
468
+
results.map((r) => ({
469
+
followerId: r.follower_id,
470
+
targetId: r.target_id,
471
+
targetType: r.target_type as 'USER' | 'COLLECTION',
472
+
})),
473
+
);
474
+
} catch (error) {
475
+
return err(error as Error);
476
+
}
477
+
}
478
+
}
479
+
```
480
+
481
+
---
482
+
483
+
### Updated FeedRepository: Add Fan-out Method
484
+
485
+
```typescript
486
+
// Add to IFeedRepository interface
487
+
export interface IFeedRepository {
488
+
// ... existing methods ...
489
+
490
+
fanOutActivityToFollowers(
491
+
activityId: ActivityId,
492
+
followerIds: string[],
493
+
createdAt: Date,
494
+
): Promise<Result<void>>;
495
+
}
496
+
497
+
// DrizzleFeedRepository implementation
498
+
async fanOutActivityToFollowers(
499
+
activityId: ActivityId,
500
+
followerIds: string[],
501
+
createdAt: Date,
502
+
): Promise<Result<void>> {
503
+
try {
504
+
if (followerIds.length === 0) {
505
+
return ok(undefined);
506
+
}
507
+
508
+
const values = followerIds.map(userId => ({
509
+
user_id: userId,
510
+
activity_id: activityId.getStringValue(),
511
+
created_at: createdAt,
512
+
}));
513
+
514
+
await this.db
515
+
.insert(followingFeedItems)
516
+
.values(values)
517
+
.onConflictDoNothing(); // Idempotent (handles retries)
518
+
519
+
return ok(undefined);
520
+
} catch (error) {
521
+
return err(error as Error);
522
+
}
523
+
}
524
+
```
525
+
526
+
---
527
+
528
+
### Idempotency Handling
529
+
530
+
Both activity creation and fan-out use `ON CONFLICT DO NOTHING`:
531
+
532
+
```typescript
533
+
// FeedService.addCardCollectedActivity() uses distributed locking (ILockService) to prevent
534
+
// race conditions during check-and-insert. Lock key: feed:activity:{actorId}:{cardId}
535
+
// This ensures that concurrent events for the same card+actor are serialized.
536
+
537
+
// Additionally, activity lookup uses findRecentCardCollectedActivity (2-minute window)
538
+
539
+
// Fan-out is idempotent
540
+
await this.db.insert(followingFeedItems).values(values).onConflictDoNothing(); // Primary key (user_id, activity_id) prevents duplicates
541
+
```
542
+
543
+
**Result:** If event handler retries:
544
+
545
+
1. FeedService tries to acquire lock → either:
546
+
- Gets lock immediately (first attempt), or
547
+
- Waits for lock, then finds existing activity (created by previous attempt) → returns it
548
+
2. Activity creation finds existing activity via findRecentCardCollectedActivity → returns it
549
+
3. Fan-out silently skips duplicate rows (ON CONFLICT DO NOTHING) → succeeds
550
+
4. Use case completes successfully with idempotent behavior
551
+
552
+
---
553
+
554
+
## Query Pattern: Getting Following Feed
555
+
556
+
```typescript
557
+
async getFollowingFeed(
558
+
userId: string,
559
+
options: FeedQueryOptions,
560
+
): Promise<Result<PaginatedFeedResult>> {
561
+
try {
562
+
const { page, limit, beforeActivityId } = options;
563
+
const offset = (page - 1) * limit;
564
+
565
+
// Build where conditions
566
+
const whereConditions = [
567
+
eq(followingFeedItems.user_id, userId)
568
+
];
569
+
570
+
if (options.urlType) {
571
+
whereConditions.push(eq(feedActivities.urlType, options.urlType));
572
+
}
573
+
574
+
if (options.source) {
575
+
if (options.source === ActivitySource.SEMBLE) {
576
+
whereConditions.push(sql`${feedActivities.source} IS NULL`);
577
+
} else {
578
+
whereConditions.push(eq(feedActivities.source, options.source));
579
+
}
580
+
}
581
+
582
+
// Cursor-based pagination
583
+
if (beforeActivityId) {
584
+
const beforeActivity = await this.db
585
+
.select({ createdAt: followingFeedItems.created_at })
586
+
.from(followingFeedItems)
587
+
.where(
588
+
and(
589
+
eq(followingFeedItems.user_id, userId),
590
+
eq(followingFeedItems.activity_id, beforeActivityId.getStringValue())
591
+
)
592
+
)
593
+
.limit(1);
594
+
595
+
if (beforeActivity.length > 0) {
596
+
whereConditions.push(
597
+
lt(followingFeedItems.created_at, beforeActivity[0].createdAt)
598
+
);
599
+
}
600
+
}
601
+
602
+
// Main query with JOIN
603
+
const activitiesResult = await this.db
604
+
.select({
605
+
id: feedActivities.id,
606
+
actorId: feedActivities.actorId,
607
+
cardId: feedActivities.cardId,
608
+
type: feedActivities.type,
609
+
metadata: feedActivities.metadata,
610
+
urlType: feedActivities.urlType,
611
+
source: feedActivities.source,
612
+
createdAt: followingFeedItems.created_at, // Use denormalized timestamp
613
+
})
614
+
.from(followingFeedItems)
615
+
.innerJoin(
616
+
feedActivities,
617
+
eq(feedActivities.id, followingFeedItems.activity_id)
618
+
)
619
+
.where(and(...whereConditions))
620
+
.orderBy(
621
+
desc(followingFeedItems.created_at),
622
+
desc(followingFeedItems.activity_id)
623
+
)
624
+
.limit(limit)
625
+
.offset(offset);
626
+
627
+
// Count total (with same filters)
628
+
const totalCountResult = await this.db
629
+
.select({ count: count() })
630
+
.from(followingFeedItems)
631
+
.innerJoin(
632
+
feedActivities,
633
+
eq(feedActivities.id, followingFeedItems.activity_id)
634
+
)
635
+
.where(and(...whereConditions));
636
+
637
+
const totalCount = totalCountResult[0]?.count || 0;
638
+
639
+
// Map to domain objects (same as existing feeds)
640
+
const activities = await this.mapToDomainActivities(activitiesResult);
641
+
642
+
const hasMore = offset + activities.length < totalCount;
643
+
const nextCursor = hasMore && activities.length > 0
644
+
? activities[activities.length - 1].activityId
645
+
: undefined;
646
+
647
+
return ok({
648
+
activities,
649
+
totalCount,
650
+
hasMore,
651
+
nextCursor,
652
+
});
653
+
} catch (error) {
654
+
return err(error as Error);
655
+
}
656
+
}
657
+
```
658
+
659
+
**Index Usage:**
660
+
661
+
- `idx_following_feed_user_time` handles WHERE + ORDER BY efficiently
662
+
- JOIN to `feed_activities` uses primary key (fast)
663
+
- Additional filters (urlType, source) use existing indexes on `feed_activities`
+801
.agent/logs/20260305_following_feed_spec.md
+801
.agent/logs/20260305_following_feed_spec.md
···
1
+
# Following Feed Implementation Specification
2
+
3
+
**Date:** 2026-03-05
4
+
**Status:** Planning
5
+
**Architecture:** Application-level fan-out
6
+
7
+
**Scope:** This document covers how to add activities to following feeds and how to fetch them. Follow/unfollow functionality is **out of scope** and will be handled separately.
8
+
9
+
---
10
+
11
+
## Overview
12
+
13
+
The following feed extends the existing event-driven feed activity system. Activities are already created asynchronously via `CARD_ADDED_TO_LIBRARY` and `CARD_ADDED_TO_COLLECTION` events. We will perform **application-level synchronous fan-out** after activity creation.
14
+
15
+
**Key Design:** Event handlers call `AddActivityToFeedUseCase` directly (no saga). The use case coordinates:
16
+
17
+
1. Activity creation (with distributed locking via `ILockService` to prevent race conditions)
18
+
2. Following feed fan-out (fetching followers and inserting into following feed)
19
+
20
+
**Following Targets:** Users can follow both **users** AND **collections**, so fan-out must consider:
21
+
22
+
- Followers of the actor (user who created the activity)
23
+
- Followers of the collections (if the activity includes collections)
24
+
25
+
---
26
+
27
+
## Database Schema
28
+
29
+
### New Tables
30
+
31
+
```sql
32
+
-- Following relationships (polymorphic - supports users AND collections)
33
+
CREATE TABLE follows (
34
+
follower_id TEXT NOT NULL, -- DID of user who is following
35
+
target_id TEXT NOT NULL, -- ID of what's being followed (user DID or collection UUID)
36
+
target_type TEXT NOT NULL, -- 'USER' or 'COLLECTION'
37
+
created_at TIMESTAMP NOT NULL DEFAULT NOW(),
38
+
PRIMARY KEY (follower_id, target_id, target_type)
39
+
);
40
+
41
+
CREATE INDEX idx_follows_follower ON follows(follower_id);
42
+
CREATE INDEX idx_follows_target ON follows(target_id, target_type);
43
+
44
+
-- Following feed items (fan-out target)
45
+
CREATE TABLE following_feed_items (
46
+
user_id TEXT NOT NULL, -- DID of feed owner
47
+
activity_id UUID NOT NULL REFERENCES feed_activities(id) ON DELETE CASCADE,
48
+
created_at TIMESTAMP NOT NULL, -- Denormalized from activity for sorting
49
+
PRIMARY KEY (user_id, activity_id)
50
+
);
51
+
52
+
CREATE INDEX idx_following_feed_user_time ON following_feed_items(user_id, created_at DESC);
53
+
```
54
+
55
+
### Why This Schema?
56
+
57
+
1. **`follows` table (Polymorphic):**
58
+
- Stores who follows what (users OR collections)
59
+
- `target_type` distinguishes between following a user vs a collection
60
+
- Single table keeps queries simple (one lookup for all followers)
61
+
- Alternative: separate `user_follows` and `collection_follows` tables (more normalized, but requires 2 queries)
62
+
63
+
2. **`following_feed_items` table:**
64
+
- Stores the fan-out results
65
+
- Each row = "this activity should appear in this user's following feed"
66
+
- Same structure regardless of whether activity came from following a user or collection
67
+
68
+
3. **Denormalized `created_at`:**
69
+
- Avoids JOIN to `feed_activities` for sorting queries
70
+
- Small storage cost for massive read performance gain
71
+
72
+
4. **ON DELETE CASCADE:**
73
+
- When an activity is deleted, automatically remove from all following feeds
74
+
75
+
---
76
+
77
+
## Flow from Event to Following Feed
78
+
79
+
### Current Flow (After Saga Removal)
80
+
81
+
```
82
+
Event: CARD_ADDED_TO_LIBRARY
83
+
↓
84
+
CardAddedToLibraryEventHandler
85
+
↓
86
+
AddActivityToFeedUseCase.execute()
87
+
↓
88
+
FeedService.addCardCollectedActivity()
89
+
├─ [ACQUIRE DISTRIBUTED LOCK via ILockService]
90
+
├─ [CHECK for recent activity]
91
+
├─ [INSERT or UPDATE feed_activities]
92
+
└─ [RELEASE DISTRIBUTED LOCK]
93
+
↓
94
+
DrizzleFeedRepository.addActivity()
95
+
↓
96
+
INSERT INTO feed_activities (...)
97
+
```
98
+
99
+
### New Flow (With Following Feed)
100
+
101
+
```
102
+
Event: CARD_ADDED_TO_LIBRARY
103
+
↓
104
+
CardAddedToLibraryEventHandler
105
+
↓
106
+
AddActivityToFeedUseCase.execute()
107
+
├─ FeedService.addCardCollectedActivity()
108
+
│ ├─ [ACQUIRE DISTRIBUTED LOCK]
109
+
│ ├─ [CHECK for recent activity]
110
+
│ └─ [INSERT or UPDATE feed_activities]
111
+
│
112
+
├─ FollowsRepository.getFollowers(actorId, 'USER')
113
+
│
114
+
├─ FollowsRepository.getFollowersOfCollections(collectionIds)
115
+
│
116
+
├─ Combine & deduplicate follower lists
117
+
│
118
+
└─ FeedRepository.fanOutActivityToFollowers()
119
+
└─ INSERT INTO following_feed_items (user_id, activity_id, created_at)
120
+
VALUES (...follower_ids..., activity_id, created_at)
121
+
ON CONFLICT DO NOTHING
122
+
```
123
+
124
+
**Key Points:**
125
+
126
+
- ✅ Activity creation is protected by distributed lock (prevents race conditions)
127
+
- ✅ No additional async layer needed (already in event handler)
128
+
- ✅ Fan-out happens at application level after activity creation
129
+
- ✅ If either activity creation or fan-out fails, event handler retries entire use case
130
+
- ✅ Both operations are idempotent (ON CONFLICT DO NOTHING) for retry safety
131
+
132
+
---
133
+
134
+
## Application-Level Fan-out Design
135
+
136
+
Fan-out happens in `AddActivityToFeedUseCase.execute()` by orchestrating multiple repository calls.
137
+
138
+
```typescript
139
+
// AddActivityToFeedUseCase.execute()
140
+
async execute(request: AddActivityToFeedDTO): Promise<Result<...>> {
141
+
// ... validation (existing code) ...
142
+
143
+
// 1. Create activity WITHOUT fan-out
144
+
const activityResult = await this.feedService.addCardCollectedActivity(
145
+
actorId,
146
+
cardId,
147
+
collectionIds,
148
+
urlType,
149
+
source,
150
+
createdAt,
151
+
);
152
+
153
+
if (activityResult.isErr()) {
154
+
return err(new ValidationError(activityResult.error.message));
155
+
}
156
+
157
+
const activity = activityResult.value;
158
+
159
+
// 2. Get followers of the actor (user)
160
+
const userFollowersResult = await this.followsRepository.getFollowers(
161
+
actorId.value,
162
+
'USER'
163
+
);
164
+
165
+
const userFollowers = userFollowersResult.isOk()
166
+
? userFollowersResult.value
167
+
: [];
168
+
169
+
// 3. Get followers of collections (if any)
170
+
let collectionFollowers: string[] = [];
171
+
if (collectionIds && collectionIds.length > 0) {
172
+
const collectionIdStrings = collectionIds.map(id => id.getStringValue());
173
+
const collectionFollowersResult =
174
+
await this.followsRepository.getFollowersOfCollections(collectionIdStrings);
175
+
176
+
collectionFollowers = collectionFollowersResult.isOk()
177
+
? collectionFollowersResult.value
178
+
: [];
179
+
}
180
+
181
+
// 4. Combine and dedupe
182
+
const allFollowers = new Set([...userFollowers, ...collectionFollowers]);
183
+
184
+
// 5. Fan-out to all followers
185
+
if (allFollowers.size > 0) {
186
+
await this.feedRepository.fanOutActivityToFollowers(
187
+
activity.activityId,
188
+
Array.from(allFollowers),
189
+
activity.createdAt,
190
+
);
191
+
}
192
+
193
+
return ok({
194
+
activityId: activity.activityId.getStringValue(),
195
+
});
196
+
}
197
+
```
198
+
199
+
**Why Application-Level:**
200
+
201
+
- ✅ Proper separation of concerns (use case orchestrates, repositories handle data)
202
+
- ✅ Full context available for business decisions (can add conditional logic based on source, type, etc.)
203
+
- ✅ Easier to test (can mock each repository call independently)
204
+
- ✅ Repositories stay simple (data access only, no business logic)
205
+
206
+
**Handling Non-Atomicity:**
207
+
208
+
We mitigate this with:
209
+
210
+
1. **Idempotent operations:** Both activity creation and fan-out use `ON CONFLICT DO NOTHING`
211
+
2. **Event retries:** If fan-out fails, BullMQ retries the entire use case
212
+
3. **Distributed locking:** FeedService uses ILockService to prevent race conditions during activity creation
213
+
214
+
**Result:** If event handler retries:
215
+
216
+
1. Activity creation finds existing activity via distributed lock + 2-minute window check → returns it
217
+
2. Fan-out silently skips duplicate rows (ON CONFLICT DO NOTHING) → succeeds
218
+
3. Use case completes successfully with idempotent behavior
219
+
220
+
---
221
+
222
+
## Application-Level Implementation
223
+
224
+
### Updated Use Case: AddActivityToFeedUseCase
225
+
226
+
```typescript
227
+
export class AddActivityToFeedUseCase implements UseCase<...> {
228
+
constructor(
229
+
private feedService: FeedService,
230
+
private cardRepository: ICardRepository,
231
+
private followsRepository: IFollowsRepository, // NEW
232
+
private feedRepository: IFeedRepository, // NEW (for fan-out)
233
+
) {}
234
+
235
+
async execute(request: AddActivityToFeedDTO): Promise<Result<...>> {
236
+
try {
237
+
// ... existing validation (actorId, cardId, collectionIds) ...
238
+
239
+
// ... existing card fetch logic ...
240
+
241
+
// 1. Create activity in global feed (no fan-out yet)
242
+
// Note: FeedService internally uses distributed locking (via ILockService) during check-and-insert
243
+
// to prevent race conditions when multiple events arrive for the same card+actor
244
+
const activityResult = await this.feedService.addCardCollectedActivity(
245
+
actorId,
246
+
cardId,
247
+
collectionIds,
248
+
urlType,
249
+
source,
250
+
createdAt,
251
+
);
252
+
253
+
if (activityResult.isErr()) {
254
+
return err(new ValidationError(activityResult.error.message));
255
+
}
256
+
257
+
const activity = activityResult.value;
258
+
259
+
// 2. Get followers of the actor (user who created activity)
260
+
const userFollowersResult = await this.followsRepository.getFollowers(
261
+
actorId.value,
262
+
'USER'
263
+
);
264
+
265
+
const userFollowers = userFollowersResult.isOk()
266
+
? userFollowersResult.value.map(f => f.followerId)
267
+
: [];
268
+
269
+
// 3. Get followers of collections (if any)
270
+
let collectionFollowers: string[] = [];
271
+
if (collectionIds && collectionIds.length > 0) {
272
+
const collectionIdStrings = collectionIds.map(id => id.getStringValue());
273
+
const collectionFollowersResult =
274
+
await this.followsRepository.getFollowersOfCollections(collectionIdStrings);
275
+
276
+
collectionFollowers = collectionFollowersResult.isOk()
277
+
? collectionFollowersResult.value.map(f => f.followerId)
278
+
: [];
279
+
}
280
+
281
+
// 4. Combine and deduplicate
282
+
const allFollowerIds = new Set([...userFollowers, ...collectionFollowers]);
283
+
284
+
// 5. Fan-out to all followers
285
+
if (allFollowerIds.size > 0) {
286
+
const fanOutResult = await this.feedRepository.fanOutActivityToFollowers(
287
+
activity.activityId,
288
+
Array.from(allFollowerIds),
289
+
activity.createdAt,
290
+
);
291
+
292
+
if (fanOutResult.isErr()) {
293
+
// Log error - event handler will retry entire use case if critical
294
+
// Distributed lock in FeedService ensures no duplicate activities
295
+
console.error('Fan-out failed:', fanOutResult.error);
296
+
}
297
+
}
298
+
299
+
return ok({
300
+
activityId: activity.activityId.getStringValue(),
301
+
});
302
+
} catch (error) {
303
+
return err(AppError.UnexpectedError.create(error));
304
+
}
305
+
}
306
+
}
307
+
```
308
+
309
+
---
310
+
311
+
### New Repository: IFollowsRepository
312
+
313
+
**Note:** This repository only defines methods needed for feed fan-out. Follow/unfollow functionality is out of scope for this document.
314
+
315
+
```typescript
316
+
export interface Follow {
317
+
followerId: string;
318
+
targetId: string;
319
+
targetType: 'USER' | 'COLLECTION';
320
+
}
321
+
322
+
export interface IFollowsRepository {
323
+
// Get all followers of a specific user
324
+
getFollowers(targetId: string, targetType: 'USER'): Promise<Result<Follow[]>>;
325
+
326
+
// Get all followers of multiple collections (combined)
327
+
getFollowersOfCollections(collectionIds: string[]): Promise<Result<Follow[]>>;
328
+
}
329
+
```
330
+
331
+
---
332
+
333
+
### New Repository Implementation: DrizzleFollowsRepository
334
+
335
+
```typescript
336
+
export class DrizzleFollowsRepository implements IFollowsRepository {
337
+
constructor(private db: PostgresJsDatabase) {}
338
+
339
+
async getFollowers(
340
+
targetId: string,
341
+
targetType: 'USER',
342
+
): Promise<Result<Follow[]>> {
343
+
try {
344
+
const results = await this.db
345
+
.select()
346
+
.from(follows)
347
+
.where(
348
+
and(
349
+
eq(follows.target_id, targetId),
350
+
eq(follows.target_type, targetType),
351
+
),
352
+
);
353
+
354
+
return ok(
355
+
results.map((r) => ({
356
+
followerId: r.follower_id,
357
+
targetId: r.target_id,
358
+
targetType: r.target_type as 'USER' | 'COLLECTION',
359
+
})),
360
+
);
361
+
} catch (error) {
362
+
return err(error as Error);
363
+
}
364
+
}
365
+
366
+
async getFollowersOfCollections(
367
+
collectionIds: string[],
368
+
): Promise<Result<Follow[]>> {
369
+
try {
370
+
if (collectionIds.length === 0) {
371
+
return ok([]);
372
+
}
373
+
374
+
const results = await this.db
375
+
.select()
376
+
.from(follows)
377
+
.where(
378
+
and(
379
+
sql`${follows.target_id} = ANY(${collectionIds}::text[])`,
380
+
eq(follows.target_type, 'COLLECTION'),
381
+
),
382
+
);
383
+
384
+
return ok(
385
+
results.map((r) => ({
386
+
followerId: r.follower_id,
387
+
targetId: r.target_id,
388
+
targetType: r.target_type as 'USER' | 'COLLECTION',
389
+
})),
390
+
);
391
+
} catch (error) {
392
+
return err(error as Error);
393
+
}
394
+
}
395
+
}
396
+
```
397
+
398
+
---
399
+
400
+
### Updated FeedRepository: Add Fan-out Method
401
+
402
+
```typescript
403
+
// Add to IFeedRepository interface
404
+
export interface IFeedRepository {
405
+
// ... existing methods ...
406
+
407
+
fanOutActivityToFollowers(
408
+
activityId: ActivityId,
409
+
followerIds: string[],
410
+
createdAt: Date,
411
+
): Promise<Result<void>>;
412
+
}
413
+
414
+
// DrizzleFeedRepository implementation
415
+
async fanOutActivityToFollowers(
416
+
activityId: ActivityId,
417
+
followerIds: string[],
418
+
createdAt: Date,
419
+
): Promise<Result<void>> {
420
+
try {
421
+
if (followerIds.length === 0) {
422
+
return ok(undefined);
423
+
}
424
+
425
+
const values = followerIds.map(userId => ({
426
+
user_id: userId,
427
+
activity_id: activityId.getStringValue(),
428
+
created_at: createdAt,
429
+
}));
430
+
431
+
await this.db
432
+
.insert(followingFeedItems)
433
+
.values(values)
434
+
.onConflictDoNothing(); // Idempotent (handles retries)
435
+
436
+
return ok(undefined);
437
+
} catch (error) {
438
+
return err(error as Error);
439
+
}
440
+
}
441
+
```
442
+
443
+
---
444
+
445
+
### Idempotency Handling
446
+
447
+
Both activity creation and fan-out use `ON CONFLICT DO NOTHING`:
448
+
449
+
```typescript
450
+
// FeedService.addCardCollectedActivity() uses distributed locking (ILockService) to prevent
451
+
// race conditions during check-and-insert. Lock key: feed:activity:{actorId}:{cardId}
452
+
// This ensures that concurrent events for the same card+actor are serialized.
453
+
454
+
// Additionally, activity lookup uses findRecentCardCollectedActivity (2-minute window)
455
+
456
+
// Fan-out is idempotent
457
+
await this.db.insert(followingFeedItems).values(values).onConflictDoNothing(); // Primary key (user_id, activity_id) prevents duplicates
458
+
```
459
+
460
+
**Result:** If event handler retries:
461
+
462
+
1. FeedService tries to acquire lock → either:
463
+
- Gets lock immediately (first attempt), or
464
+
- Waits for lock, then finds existing activity (created by previous attempt) → returns it
465
+
2. Activity creation finds existing activity via findRecentCardCollectedActivity → returns it
466
+
3. Fan-out silently skips duplicate rows (ON CONFLICT DO NOTHING) → succeeds
467
+
4. Use case completes successfully with idempotent behavior
468
+
469
+
---
470
+
471
+
## Query Pattern: Getting Following Feed
472
+
473
+
```typescript
474
+
async getFollowingFeed(
475
+
userId: string,
476
+
options: FeedQueryOptions,
477
+
): Promise<Result<PaginatedFeedResult>> {
478
+
try {
479
+
const { page, limit, beforeActivityId } = options;
480
+
const offset = (page - 1) * limit;
481
+
482
+
// Build where conditions
483
+
const whereConditions = [
484
+
eq(followingFeedItems.user_id, userId)
485
+
];
486
+
487
+
if (options.urlType) {
488
+
whereConditions.push(eq(feedActivities.urlType, options.urlType));
489
+
}
490
+
491
+
if (options.source) {
492
+
if (options.source === ActivitySource.SEMBLE) {
493
+
whereConditions.push(sql`${feedActivities.source} IS NULL`);
494
+
} else {
495
+
whereConditions.push(eq(feedActivities.source, options.source));
496
+
}
497
+
}
498
+
499
+
// Cursor-based pagination
500
+
if (beforeActivityId) {
501
+
const beforeActivity = await this.db
502
+
.select({ createdAt: followingFeedItems.created_at })
503
+
.from(followingFeedItems)
504
+
.where(
505
+
and(
506
+
eq(followingFeedItems.user_id, userId),
507
+
eq(followingFeedItems.activity_id, beforeActivityId.getStringValue())
508
+
)
509
+
)
510
+
.limit(1);
511
+
512
+
if (beforeActivity.length > 0) {
513
+
whereConditions.push(
514
+
lt(followingFeedItems.created_at, beforeActivity[0].createdAt)
515
+
);
516
+
}
517
+
}
518
+
519
+
// Main query with JOIN
520
+
const activitiesResult = await this.db
521
+
.select({
522
+
id: feedActivities.id,
523
+
actorId: feedActivities.actorId,
524
+
cardId: feedActivities.cardId,
525
+
type: feedActivities.type,
526
+
metadata: feedActivities.metadata,
527
+
urlType: feedActivities.urlType,
528
+
source: feedActivities.source,
529
+
createdAt: followingFeedItems.created_at, // Use denormalized timestamp
530
+
})
531
+
.from(followingFeedItems)
532
+
.innerJoin(
533
+
feedActivities,
534
+
eq(feedActivities.id, followingFeedItems.activity_id)
535
+
)
536
+
.where(and(...whereConditions))
537
+
.orderBy(
538
+
desc(followingFeedItems.created_at),
539
+
desc(followingFeedItems.activity_id)
540
+
)
541
+
.limit(limit)
542
+
.offset(offset);
543
+
544
+
// Count total (with same filters)
545
+
const totalCountResult = await this.db
546
+
.select({ count: count() })
547
+
.from(followingFeedItems)
548
+
.innerJoin(
549
+
feedActivities,
550
+
eq(feedActivities.id, followingFeedItems.activity_id)
551
+
)
552
+
.where(and(...whereConditions));
553
+
554
+
const totalCount = totalCountResult[0]?.count || 0;
555
+
556
+
// Map to domain objects (same as existing feeds)
557
+
const activities = await this.mapToDomainActivities(activitiesResult);
558
+
559
+
const hasMore = offset + activities.length < totalCount;
560
+
const nextCursor = hasMore && activities.length > 0
561
+
? activities[activities.length - 1].activityId
562
+
: undefined;
563
+
564
+
return ok({
565
+
activities,
566
+
totalCount,
567
+
hasMore,
568
+
nextCursor,
569
+
});
570
+
} catch (error) {
571
+
return err(error as Error);
572
+
}
573
+
}
574
+
```
575
+
576
+
**Index Usage:**
577
+
578
+
- `idx_following_feed_user_time` handles WHERE + ORDER BY efficiently
579
+
- JOIN to `feed_activities` uses primary key (fast)
580
+
- Additional filters (urlType, source) use existing indexes on `feed_activities`
581
+
582
+
---
583
+
584
+
## Performance Analysis
585
+
586
+
### Write Performance (Fan-out)
587
+
588
+
**Scenario:** User with 1,000 followers posts an activity
589
+
590
+
| Approach | Database Operations | Estimated Time |
591
+
| ------------------------ | ---------------------------------------- | -------------- |
592
+
| **Option A (2 queries)** | 1 INSERT + 1 SELECT + 1 INSERT (1k rows) | ~30-50ms |
593
+
| **Option B (CTE)** | 1 combined query | ~20-40ms |
594
+
595
+
**Bottleneck:** Not the DB, but finding follower IDs. With proper index (`idx_follows_following`), this is O(1) lookup.
596
+
597
+
**Scaling:**
598
+
599
+
- 100 followers: ~10ms
600
+
- 1,000 followers: ~40ms
601
+
- 10,000 followers: ~300ms (still acceptable for async event handler)
602
+
- 100,000 followers: ~3s (need optimization - see below)
603
+
604
+
### Read Performance (Query Following Feed)
605
+
606
+
**Scenario:** User queries their following feed (page 1, 20 items)
607
+
608
+
| Operation | Strategy | Estimated Time |
609
+
| ---------------------- | ------------------------------------------------ | -------------- |
610
+
| **Filter by user** | `idx_following_feed_user_time` (index-only scan) | ~1ms |
611
+
| **Sort by time** | Already in index order | 0ms |
612
+
| **JOIN to activities** | Primary key lookup (20 items) | ~1ms |
613
+
| **Total** | | **~2-5ms** |
614
+
615
+
**Comparison to other feeds:**
616
+
617
+
- Global feed: ~2-5ms (same)
618
+
- Gems feed: ~5-10ms (JSONB filtering slower)
619
+
- Following feed: ~2-5ms ✅
620
+
621
+
---
622
+
623
+
## Edge Cases & Considerations
624
+
625
+
### 1. **What if user has no followers?**
626
+
627
+
```sql
628
+
SELECT follower_id FROM follows WHERE target_id = ? AND target_type = 'USER'
629
+
-- Returns empty result set
630
+
-- INSERT following_feed_items skipped (no rows)
631
+
```
632
+
633
+
✅ Works fine, no fan-out occurs.
634
+
635
+
### 2. **What if fan-out operation fails?**
636
+
637
+
If fan-out fails (e.g., DB connection lost):
638
+
639
+
1. Activity is already created in global feed
640
+
2. Event handler throws error
641
+
3. BullMQ retries the event
642
+
4. Activity creation finds existing activity (via distributed lock + 2-minute window)
643
+
5. Fan-out retries and succeeds (idempotent with ON CONFLICT DO NOTHING)
644
+
645
+
✅ Eventual consistency with retries.
646
+
647
+
### 3. **Duplicate activities in following feed?**
648
+
649
+
Can't happen:
650
+
651
+
- Primary key `(user_id, activity_id)` prevents duplicates
652
+
- Fan-out uses `ON CONFLICT DO NOTHING` for idempotency
653
+
654
+
```typescript
655
+
await tx.insert(followingFeedItems).values(fanOutValues).onConflictDoNothing();
656
+
```
657
+
658
+
---
659
+
660
+
## Optimizations for High Follower Counts
661
+
662
+
### When user has > 10,000 followers:
663
+
664
+
**Problem:** Fan-out takes > 1 second, blocks event handler
665
+
666
+
**Solution: Hybrid approach**
667
+
668
+
```typescript
669
+
const SYNC_FANOUT_THRESHOLD = 5000;
670
+
671
+
async addActivity(activity: FeedActivity): Promise<Result<void>> {
672
+
await this.db.transaction(async (tx) => {
673
+
// 1. Always insert activity
674
+
await tx.insert(feedActivities).values(...);
675
+
676
+
// 2. Count followers
677
+
const followerCount = await tx
678
+
.select({ count: count() })
679
+
.from(follows)
680
+
.where(eq(follows.following_id, dto.actorId));
681
+
682
+
if (followerCount[0].count < SYNC_FANOUT_THRESHOLD) {
683
+
// 3a. Small follower count: fan-out synchronously
684
+
await tx.execute(sql`INSERT INTO following_feed_items ...`);
685
+
} else {
686
+
// 3b. Large follower count: queue background job
687
+
await this.queueFanoutJob(dto.id, dto.actorId);
688
+
}
689
+
});
690
+
}
691
+
```
692
+
693
+
**For medium scale (< 100k users):** Sync approach is fine. This optimization can wait.
694
+
695
+
---
696
+
697
+
## Summary: Implementation Changes Required
698
+
699
+
### Database Changes
700
+
701
+
**New Tables (2):**
702
+
703
+
1. **`follows`** - Polymorphic table for following users AND collections
704
+
- `follower_id` (TEXT) - User doing the following
705
+
- `target_id` (TEXT) - User DID or Collection UUID being followed
706
+
- `target_type` (TEXT) - 'USER' or 'COLLECTION'
707
+
- Primary key: `(follower_id, target_id, target_type)`
708
+
709
+
2. **`following_feed_items`** - Fan-out target table
710
+
- `user_id` (TEXT) - Feed owner
711
+
- `activity_id` (UUID) - Reference to feed_activities
712
+
- `created_at` (TIMESTAMP) - Denormalized for sorting
713
+
- Primary key: `(user_id, activity_id)`
714
+
715
+
### Application Layer Changes
716
+
717
+
**New Repository (1):**
718
+
719
+
- **`IFollowsRepository` / `DrizzleFollowsRepository`**
720
+
- `getFollowers(targetId, targetType)` - Get followers of a user
721
+
- `getFollowersOfCollections(collectionIds[])` - Get followers of multiple collections
722
+
723
+
**Updated Repository (1):**
724
+
725
+
- **`IFeedRepository` / `DrizzleFeedRepository`**
726
+
- Add: `fanOutActivityToFollowers(activityId, followerIds[], createdAt)` - Insert into following_feed_items
727
+
- Add: `getFollowingFeed(userId, options)` - Query following feed (similar to getGemsFeed)
728
+
729
+
**Updated Use Case (1):**
730
+
731
+
- **`AddActivityToFeedUseCase`**
732
+
- Add `followsRepository` to constructor
733
+
- Add `feedRepository` to constructor (for fan-out)
734
+
- After creating activity:
735
+
1. Get followers of actor (user)
736
+
2. Get followers of collections (if any)
737
+
3. Combine and dedupe
738
+
4. Call `fanOutActivityToFollowers()`
739
+
740
+
### Key Design Principles
741
+
742
+
1. ✅ **Application-level fan-out** - Use case orchestrates, repositories handle data
743
+
2. ✅ **Distributed locking** - FeedService uses ILockService to prevent race conditions during activity creation
744
+
3. ✅ **Idempotency** - Both activity creation and fan-out use `ON CONFLICT DO NOTHING` for retry safety
745
+
4. ✅ **Polymorphic follows** - Single table supports both user and collection follows
746
+
5. ✅ **Eventual consistency** - Event retries handle failures gracefully
747
+
748
+
---
749
+
750
+
## Implementation Steps
751
+
752
+
### Phase 1: Database Schema
753
+
754
+
1. **Create migration** for `follows` and `following_feed_items` tables
755
+
2. **Update** `createTestSchema.ts` with new tables and indexes
756
+
3. **Generate migration** with `npm run db:generate`
757
+
4. **Apply migration** to development database
758
+
759
+
### Phase 2: Repositories
760
+
761
+
1. **Create** `IFollowsRepository` interface with `getFollowers()` and `getFollowersOfCollections()`
762
+
2. **Implement** `DrizzleFollowsRepository`
763
+
3. **Update** `IFeedRepository` with `fanOutActivityToFollowers()` and `getFollowingFeed()`
764
+
4. **Implement** new methods in `DrizzleFeedRepository`
765
+
5. **Add** repository instances to `RepositoryFactory`
766
+
767
+
### Phase 3: Use Case Updates
768
+
769
+
1. **Update** `AddActivityToFeedUseCase` constructor to include `followsRepository` and `feedRepository`
770
+
2. **Add** fan-out logic after activity creation (as shown in implementation section)
771
+
3. **Update** `UseCaseFactory` to inject new dependencies
772
+
773
+
### Phase 4: Feed Query API
774
+
775
+
1. **Create** `GetFollowingFeedUseCase` (similar to existing feed use cases)
776
+
2. **Add** HTTP route and controller for following feed query
777
+
778
+
### Phase 5: Testing
779
+
780
+
1. **Unit tests** for `DrizzleFollowsRepository`
781
+
2. **Unit tests** for fan-out logic in `AddActivityToFeedUseCase`
782
+
3. **Integration tests** for complete flow (event → activity → fan-out)
783
+
4. **Performance tests** with various follower counts (100, 1k, 10k)
784
+
5. **Test retry scenarios** (ensure idempotency works)
785
+
6. **Run** `npm run build:check` to verify no type errors
786
+
787
+
---
788
+
789
+
## Open Questions
790
+
791
+
1. **Following feed filters:** Should following feed support same filters as global feed?
792
+
- `urlType` filter (show only articles, videos, etc.)
793
+
- `source` filter (show only Semble, Margin, etc.)
794
+
795
+
2. **Conditional fan-out:** Should certain activities NOT be fanned out?
796
+
- Example: "Don't fan-out Margin activities to collection followers" (only to user followers)
797
+
- Application-level approach makes this easy to implement
798
+
799
+
3. **Performance optimization threshold:** At what follower count should we implement async fan-out?
800
+
- Current plan: Sync fan-out for all users
801
+
- Future consideration: Hybrid approach for users with > 10k followers
+515
.agent/logs/20260305_removing_feed_saga.md
+515
.agent/logs/20260305_removing_feed_saga.md
···
1
+
# Removing SAGA Pattern from Feed Event Processing
2
+
3
+
**Date**: 2026-03-05
4
+
**Status**: Design Document
5
+
**Author**: System Design
6
+
7
+
## Executive Summary
8
+
9
+
This document outlines the plan to remove the `CardCollectionSaga` pattern from feed event processing. The saga adds unnecessary complexity for handling `CardAddedToLibrary` and `CardAddedToCollection` events. We will replace it with direct event handling using distributed locking at the service layer to prevent race conditions.
10
+
11
+
---
12
+
13
+
## 1. Current State
14
+
15
+
### Event Flow (With SAGA)
16
+
17
+
```
18
+
CardAddedToLibraryEvent → CardAddedToLibraryEventHandler
19
+
↓
20
+
CardCollectionSaga (3-second aggregation)
21
+
↓
22
+
AddActivityToFeedUseCase
23
+
↓
24
+
FeedService (2-minute deduplication)
25
+
↓
26
+
DrizzleFeedRepository
27
+
28
+
CardAddedToCollectionEvent → CardAddedToCollectionEventHandler
29
+
↓
30
+
CardCollectionSaga (3-second aggregation)
31
+
↓
32
+
AddActivityToFeedUseCase
33
+
↓
34
+
FeedService (2-minute deduplication)
35
+
↓
36
+
DrizzleFeedRepository
37
+
```
38
+
39
+
### What CardCollectionSaga Does
40
+
41
+
**File**: `src/modules/feeds/application/sagas/CardCollectionSaga.ts`
42
+
43
+
1. **Short-term event aggregation (3 seconds)**: Merges events that happen nearly simultaneously
44
+
2. **Distributed locking via Redis**: Prevents race conditions when multiple workers process events
45
+
- Lock key pattern: `saga:feed:lock:{cardId}-{actorId}`
46
+
- State key pattern: `saga:feed:pending:{cardId}-{actorId}`
47
+
- Lock TTL: 3-8 seconds (aggregation window + 5 seconds)
48
+
- Retry: 15 attempts with exponential backoff (100ms base, capped at 2s)
49
+
3. **Timestamp preservation**: Tracks earliest `addedAt` timestamp from all events
50
+
4. **Deferred execution**: Batches events before hitting the database
51
+
52
+
### What FeedService Does
53
+
54
+
**File**: `src/modules/feeds/domain/services/FeedService.ts`
55
+
56
+
1. **Medium-term deduplication (2 minutes)**: Checks for recent activities via `findRecentCardCollectedActivity`
57
+
2. **Collection merging**: Updates existing activities with new collection IDs
58
+
3. **Activity creation**: Creates new activities when no recent one exists
59
+
60
+
### Why It Was Built
61
+
62
+
The saga was designed to handle the case where:
63
+
64
+
- A user clicks "save" on a card
65
+
- System generates `CardAddedToLibraryEvent`
66
+
- System also generates `CardAddedToCollectionEvent` (if added to collections)
67
+
- Both events arrive nearly simultaneously at different workers
68
+
- Without coordination, this could create 2 separate activities instead of 1 merged activity
69
+
70
+
---
71
+
72
+
## 2. Why Remove It
73
+
74
+
### Reasons for Removal
75
+
76
+
1. **Unnecessary Complexity**: Two layers of deduplication (3-second saga + 2-minute service) is overkill
77
+
2. **Cognitive Overhead**: Understanding saga pattern adds mental load for future developers
78
+
3. **Inflexibility**: Eventually we want to handle events differently:
79
+
- Following feed (only from followed users/collections)
80
+
- Global feed (all activities)
81
+
- Different aggregation rules per feed type
82
+
4. **Redundancy**: FeedService already handles the core use case via 2-minute window
83
+
5. **Maintenance Burden**: More code to maintain, debug, and test
84
+
6. **Over-Engineering**: The 3-second aggregation is a micro-optimization that adds macro-complexity
85
+
86
+
### What We're Keeping
87
+
88
+
- FeedService's 2-minute deduplication window (good enough)
89
+
- Time-based activity merging at insert level
90
+
- Clean separation between event handling and business logic
91
+
92
+
---
93
+
94
+
## 3. New Architecture
95
+
96
+
### Simplified Event Flow (Without SAGA)
97
+
98
+
```
99
+
CardAddedToLibraryEvent → CardAddedToLibraryEventHandler
100
+
↓
101
+
AddActivityToFeedUseCase
102
+
↓
103
+
FeedService (with locking + 2-minute deduplication)
104
+
↓
105
+
DrizzleFeedRepository
106
+
107
+
CardAddedToCollectionEvent → CardAddedToCollectionEventHandler
108
+
↓
109
+
AddActivityToFeedUseCase
110
+
↓
111
+
FeedService (with locking + 2-minute deduplication)
112
+
↓
113
+
DrizzleFeedRepository
114
+
```
115
+
116
+
### Key Changes
117
+
118
+
1. **Event handlers call use case directly** (no saga intermediary)
119
+
2. **Locking moves to FeedService layer** (during check-and-insert)
120
+
3. **Each event processed independently** (simpler to reason about)
121
+
4. **Future flexibility** for different feed types
122
+
123
+
---
124
+
125
+
## 4. Locking Strategy
126
+
127
+
### Why We Need Locking
128
+
129
+
Without locking, race conditions can occur:
130
+
131
+
```
132
+
Timeline without locking:
133
+
T0: Worker A receives CardAddedToLibraryEvent
134
+
T1: Worker B receives CardAddedToCollectionEvent (same card+actor)
135
+
T2: Worker A queries: findRecentCardCollectedActivity → finds nothing
136
+
T3: Worker B queries: findRecentCardCollectedActivity → finds nothing
137
+
T4: Worker A inserts new activity
138
+
T5: Worker B inserts new activity (DUPLICATE!)
139
+
```
140
+
141
+
With locking:
142
+
143
+
```
144
+
Timeline with locking:
145
+
T0: Worker A receives CardAddedToLibraryEvent
146
+
T1: Worker B receives CardAddedToCollectionEvent
147
+
T2: Worker A acquires lock for "card123-actor456"
148
+
T3: Worker B tries to acquire same lock → WAITS
149
+
T4: Worker A queries → finds nothing → inserts activity → releases lock
150
+
T5: Worker B acquires lock
151
+
T6: Worker B queries → finds Worker A's activity → merges collections → releases lock
152
+
```
153
+
154
+
### Implementation: Use Existing RedisLockService
155
+
156
+
**Files**:
157
+
158
+
- Interface: `src/shared/infrastructure/locking/ILockService.ts`
159
+
- Implementation: `src/shared/infrastructure/locking/RedisLockService.ts`
160
+
161
+
The codebase already has a production-ready distributed locking service using **Redlock** algorithm.
162
+
163
+
**Characteristics**:
164
+
165
+
- Uses `redlock` npm package (industry standard)
166
+
- 3 retry attempts with exponential backoff (200ms base + 200ms jitter)
167
+
- Configurable lock TTL
168
+
- Handles container shutdown gracefully (SIGTERM)
169
+
170
+
### Lock Configuration
171
+
172
+
```typescript
173
+
// Lock key pattern
174
+
const lockKey = `feed:activity:${actorId}:${cardId}`;
175
+
176
+
// Lock TTL: 10 seconds (generous for database operation)
177
+
const lockTTL = 10000; // milliseconds
178
+
179
+
// Lock during FeedService.addCardCollectedActivity()
180
+
await lockService.withLock(lockKey, lockTTL, async () => {
181
+
// Check for recent activity
182
+
// Insert or update activity
183
+
});
184
+
```
185
+
186
+
### Handling Lock Failures
187
+
188
+
If lock acquisition fails after retries:
189
+
190
+
- Log warning (for monitoring)
191
+
- Proceed with operation anyway (accept small risk of duplicate)
192
+
- Rationale: Better to process event than drop it
193
+
194
+
**Why this is acceptable**:
195
+
196
+
- Lock failures are rare with Redlock
197
+
- 2-minute deduplication window catches most duplicates anyway
198
+
- Better to have occasional duplicate than lose user actions
199
+
200
+
---
201
+
202
+
## 5. Implementation Steps
203
+
204
+
### Step 1: Modify Event Handlers
205
+
206
+
**Files to Modify**:
207
+
208
+
- `src/modules/feeds/application/eventHandlers/CardAddedToLibraryEventHandler.ts`
209
+
- `src/modules/feeds/application/eventHandlers/CardAddedToCollectionEventHandler.ts`
210
+
211
+
**Changes**:
212
+
213
+
```typescript
214
+
// BEFORE (with saga)
215
+
export class CardAddedToLibraryEventHandler
216
+
implements IEventHandler<CardAddedToLibraryEvent>
217
+
{
218
+
constructor(private cardCollectionSaga: CardCollectionSaga) {}
219
+
220
+
async handle(event: CardAddedToLibraryEvent): Promise<Result<void>> {
221
+
return this.cardCollectionSaga.handleCardEvent(event);
222
+
}
223
+
}
224
+
225
+
// AFTER (direct to use case)
226
+
export class CardAddedToLibraryEventHandler
227
+
implements IEventHandler<CardAddedToLibraryEvent>
228
+
{
229
+
constructor(private addActivityToFeedUseCase: AddActivityToFeedUseCase) {}
230
+
231
+
async handle(event: CardAddedToLibraryEvent): Promise<Result<void>> {
232
+
const result = await this.addActivityToFeedUseCase.execute({
233
+
type: ActivityTypeEnum.CARD_COLLECTED,
234
+
actorId: event.curatorId.value,
235
+
cardId: event.cardId.getStringValue(),
236
+
collectionIds: undefined, // Library event has no collections
237
+
createdAt: event.addedAt,
238
+
});
239
+
240
+
if (result.isErr()) {
241
+
console.error('Failed to add library activity to feed:', result.error);
242
+
return err(result.error);
243
+
}
244
+
245
+
return ok(undefined);
246
+
}
247
+
}
248
+
```
249
+
250
+
### Step 2: Add Locking to FeedService
251
+
252
+
**File to Modify**: `src/modules/feeds/domain/services/FeedService.ts`
253
+
254
+
**Changes**:
255
+
256
+
```typescript
257
+
export class FeedService implements DomainService {
258
+
constructor(
259
+
private feedRepository: IFeedRepository,
260
+
private lockService: ILockService, // ADD THIS
261
+
) {}
262
+
263
+
async addCardCollectedActivity(
264
+
actorId: CuratorId,
265
+
cardId: CardId,
266
+
collectionIds?: CollectionId[],
267
+
urlType?: UrlType,
268
+
source?: string,
269
+
createdAt?: Date,
270
+
): Promise<Result<FeedActivity, FeedServiceError>> {
271
+
const lockKey = `feed:activity:${actorId.value}:${cardId.getStringValue()}`;
272
+
const lockTTL = 10000; // 10 seconds
273
+
274
+
try {
275
+
// Acquire distributed lock
276
+
return await this.lockService.withLock(lockKey, lockTTL, async () => {
277
+
// Existing logic: check for recent activity, insert or update
278
+
const recentActivityResult =
279
+
await this.feedRepository.findRecentCardCollectedActivity(
280
+
actorId,
281
+
cardId,
282
+
2, // 2 minutes
283
+
);
284
+
285
+
if (recentActivityResult.isErr()) {
286
+
return err(
287
+
new FeedServiceError(
288
+
`Failed to check for recent activity: ${recentActivityResult.error.message}`,
289
+
),
290
+
);
291
+
}
292
+
293
+
const recentActivity = recentActivityResult.value;
294
+
295
+
if (recentActivity && collectionIds && collectionIds.length > 0) {
296
+
// Update existing activity by merging collections
297
+
recentActivity.mergeCollections(collectionIds);
298
+
299
+
const updateResult =
300
+
await this.feedRepository.updateActivity(recentActivity);
301
+
302
+
if (updateResult.isErr()) {
303
+
return err(
304
+
new FeedServiceError(
305
+
`Failed to update activity: ${updateResult.error.message}`,
306
+
),
307
+
);
308
+
}
309
+
310
+
return ok(recentActivity);
311
+
} else if (recentActivity) {
312
+
// Recent activity exists but no new collections to add, return existing
313
+
return ok(recentActivity);
314
+
}
315
+
316
+
// No recent activity found, create new one
317
+
const activityResult = FeedActivity.createCardCollected(
318
+
actorId,
319
+
cardId,
320
+
collectionIds,
321
+
urlType,
322
+
source,
323
+
createdAt,
324
+
);
325
+
326
+
if (activityResult.isErr()) {
327
+
return err(new FeedServiceError(activityResult.error.message));
328
+
}
329
+
330
+
const activity = activityResult.value;
331
+
const saveResult = await this.feedRepository.addActivity(activity);
332
+
333
+
if (saveResult.isErr()) {
334
+
return err(
335
+
new FeedServiceError(
336
+
`Failed to save activity: ${saveResult.error.message}`,
337
+
),
338
+
);
339
+
}
340
+
341
+
return ok(activity);
342
+
});
343
+
} catch (error) {
344
+
// Lock acquisition failed after retries
345
+
console.warn(
346
+
`Failed to acquire lock for ${lockKey}, proceeding without lock`,
347
+
error,
348
+
);
349
+
350
+
// Proceed with operation anyway (fallback behavior)
351
+
// ... (duplicate the lock body here, or extract to a method)
352
+
}
353
+
}
354
+
}
355
+
```
356
+
357
+
### Step 3: Update FeedWorkerProcess
358
+
359
+
**File to Modify**: `src/shared/infrastructure/processes/FeedWorkerProcess.ts`
360
+
361
+
**Changes**:
362
+
363
+
```typescript
364
+
// BEFORE
365
+
protected async registerHandlers(
366
+
subscriber: IEventSubscriber,
367
+
services: WorkerServices,
368
+
repositories: Repositories,
369
+
): Promise<void> {
370
+
const useCases = UseCaseFactory.createForWorker(repositories, services);
371
+
372
+
// Create saga with proper use case dependency and state store from services
373
+
const cardCollectionSaga = new CardCollectionSaga(
374
+
useCases.addActivityToFeedUseCase,
375
+
services.sagaStateStore,
376
+
);
377
+
378
+
const cardAddedToLibraryHandler = new CardAddedToLibraryEventHandler(
379
+
cardCollectionSaga,
380
+
);
381
+
const cardAddedToCollectionHandler = new CardAddedToCollectionEventHandler(
382
+
cardCollectionSaga,
383
+
);
384
+
385
+
await subscriber.subscribe(
386
+
EventNames.CARD_ADDED_TO_LIBRARY,
387
+
cardAddedToLibraryHandler,
388
+
);
389
+
390
+
await subscriber.subscribe(
391
+
EventNames.CARD_ADDED_TO_COLLECTION,
392
+
cardAddedToCollectionHandler,
393
+
);
394
+
}
395
+
396
+
// AFTER
397
+
protected async registerHandlers(
398
+
subscriber: IEventSubscriber,
399
+
services: WorkerServices,
400
+
repositories: Repositories,
401
+
): Promise<void> {
402
+
const useCases = UseCaseFactory.createForWorker(repositories, services);
403
+
404
+
const cardAddedToLibraryHandler = new CardAddedToLibraryEventHandler(
405
+
useCases.addActivityToFeedUseCase,
406
+
);
407
+
const cardAddedToCollectionHandler = new CardAddedToCollectionEventHandler(
408
+
useCases.addActivityToFeedUseCase,
409
+
);
410
+
411
+
await subscriber.subscribe(
412
+
EventNames.CARD_ADDED_TO_LIBRARY,
413
+
cardAddedToLibraryHandler,
414
+
);
415
+
416
+
await subscriber.subscribe(
417
+
EventNames.CARD_ADDED_TO_COLLECTION,
418
+
cardAddedToCollectionHandler,
419
+
);
420
+
}
421
+
```
422
+
423
+
### Step 4: Update ServiceFactory
424
+
425
+
**File to Modify**: `src/shared/infrastructure/http/factories/ServiceFactory.ts`
426
+
427
+
**Changes**:
428
+
429
+
- Inject `lockService` into `FeedService` constructor
430
+
- Lock service already created in `createForWorker()`
431
+
432
+
```typescript
433
+
// In createForWorker() method
434
+
const feedService = new FeedService(
435
+
feedRepository,
436
+
services.lockService, // ADD THIS
437
+
);
438
+
```
439
+
440
+
### Step 5: Remove Saga Files
441
+
442
+
**Files to Remove**:
443
+
444
+
- `src/modules/feeds/application/sagas/CardCollectionSaga.ts`
445
+
446
+
**Files to Keep** (still used by CardNotificationSaga):
447
+
448
+
- `src/modules/feeds/application/sagas/ISagaStateStore.ts`
449
+
- `src/modules/feeds/infrastructure/RedisSagaStateStore.ts`
450
+
- `src/modules/feeds/infrastructure/InMemorySagaStateStore.ts`
451
+
452
+
### Step 6: Update Tests
453
+
454
+
**Files to Update**:
455
+
456
+
- Event handler tests
457
+
- FeedService tests (add lock service mock)
458
+
- Integration tests for feed worker
459
+
460
+
---
461
+
462
+
## 6. Files Summary
463
+
464
+
### Files to Modify
465
+
466
+
| File | Changes |
467
+
| -------------------------------------- | ------------------------------------------------------ |
468
+
| `CardAddedToLibraryEventHandler.ts` | Inject `AddActivityToFeedUseCase`, call directly |
469
+
| `CardAddedToCollectionEventHandler.ts` | Inject `AddActivityToFeedUseCase`, call directly |
470
+
| `FeedService.ts` | Add `ILockService` dependency, wrap operations in lock |
471
+
| `FeedWorkerProcess.ts` | Remove saga instantiation, pass use case to handlers |
472
+
| `ServiceFactory.ts` | Inject lock service into FeedService |
473
+
474
+
### Files to Remove
475
+
476
+
| File | Reason |
477
+
| ----------------------- | ---------------- |
478
+
| `CardCollectionSaga.ts` | No longer needed |
479
+
480
+
### Files to Keep (Used by Other Features)
481
+
482
+
| File | Reason |
483
+
| --------------------------- | ---------------------------- |
484
+
| `ISagaStateStore.ts` | Used by CardNotificationSaga |
485
+
| `RedisSagaStateStore.ts` | Used by CardNotificationSaga |
486
+
| `InMemorySagaStateStore.ts` | Used by CardNotificationSaga |
487
+
488
+
---
489
+
490
+
## 11. Conclusion
491
+
492
+
Removing `CardCollectionSaga` simplifies the architecture while maintaining correctness and performance. The existing FeedService deduplication logic, combined with distributed locking, provides sufficient protection against race conditions. This change enables future flexibility for implementing different feed types without the constraints of the saga pattern.
493
+
494
+
**Key Takeaway**: Keep it simple. The 2-minute deduplication window + distributed locking is good enough. The 3-second saga aggregation was premature optimization.
495
+
496
+
---
497
+
498
+
## Appendix: Code References
499
+
500
+
### Current Implementation
501
+
502
+
- **CardCollectionSaga**: `src/modules/feeds/application/sagas/CardCollectionSaga.ts:21-236`
503
+
- **FeedService deduplication**: `src/modules/feeds/domain/services/FeedService.ts:29-66`
504
+
- **DrizzleFeedRepository.findRecentCardCollectedActivity**: `src/modules/feeds/infrastructure/repositories/DrizzleFeedRepository.ts:362-408`
505
+
506
+
### Locking Infrastructure
507
+
508
+
- **ILockService**: `src/shared/infrastructure/locking/ILockService.ts`
509
+
- **RedisLockService**: `src/shared/infrastructure/locking/RedisLockService.ts`
510
+
- **Redlock configuration**: Uses `redlock` npm package with 3 retries, 200ms delays
511
+
512
+
### Worker Configuration
513
+
514
+
- **FeedWorkerProcess**: `src/shared/infrastructure/processes/FeedWorkerProcess.ts:18-66`
515
+
- **ServiceFactory**: `src/shared/infrastructure/http/factories/ServiceFactory.ts`