nandi.uk
This repository has no description
remove CardCollectionSaga from feed subdomain
author
Wesley Finck
date
(Feb 5, 2026, 5:45 PM -0800)
commit
bfad8d65
bfad8d65bc6ebca5f76ffd72a4234aec7bbf2e5b
parent
dd4cff82
dd4cff82c3cc8b83f6a72b589ea2e53fc0294acd
+274
-582
+6
-260
src/modules/cards/tests/integration/BullMQEventSystem.integration.test.ts
+6
-260
src/modules/cards/tests/integration/BullMQEventSystem.integration.test.ts
···
12
12
import { EventNames } from '../../../../shared/infrastructure/events/EventConfig';
13
13
import { Queue } from 'bullmq';
14
14
import { QueueNames } from 'src/shared/infrastructure/events/QueueConfig';
15
-
import { CardCollectionSaga } from '../../../feeds/application/sagas/CardCollectionSaga';
16
-
import { RedisSagaStateStore } from '../../../feeds/infrastructure/RedisSagaStateStore';
17
15
18
16
describe('BullMQ Event System Integration', () => {
19
17
let redisContainer: StartedRedisContainer;
···
298
296
}, 10000);
299
297
});
300
298
301
-
describe('Redis-Based Saga Integration', () => {
302
-
it('should handle distributed saga state across multiple workers', async () => {
303
-
// Arrange - Create two saga instances (simulating multiple workers)
304
-
const mockUseCase = {
305
-
execute: jest
306
-
.fn()
307
-
.mockResolvedValue(ok({ activityId: 'test-activity' })),
308
-
} as any;
309
-
310
-
const stateStore = new RedisSagaStateStore(redis);
311
-
const saga1 = new CardCollectionSaga(mockUseCase, stateStore);
312
-
const saga2 = new CardCollectionSaga(mockUseCase, stateStore);
313
-
314
-
// Create test events for same card/user (should be aggregated)
315
-
const cardId = CardId.createFromString('saga-test-card').unwrap();
316
-
const curatorId = CuratorId.create('did:plc:sagatest').unwrap();
317
-
318
-
const libraryEvent = CardAddedToLibraryEvent.create(
319
-
cardId,
320
-
curatorId,
321
-
new Date(),
322
-
).unwrap();
323
-
const collectionEvent = CardAddedToCollectionEvent.create(
324
-
cardId,
325
-
CollectionId.createFromString('test-collection').unwrap(),
326
-
curatorId,
327
-
new Date(),
328
-
).unwrap();
329
-
330
-
// Act - Process events with different saga instances
331
-
const result1 = await saga1.handleCardEvent(libraryEvent);
332
-
const result2 = await saga2.handleCardEvent(collectionEvent);
333
-
334
-
// Assert - Both operations succeeded
335
-
expect(result1.isOk()).toBe(true);
336
-
expect(result2.isOk()).toBe(true);
337
-
338
-
// Wait for aggregation window
339
-
await new Promise((resolve) => setTimeout(resolve, 3500));
340
-
341
-
// Assert - Only one aggregated activity was created
342
-
expect(mockUseCase.execute).toHaveBeenCalledTimes(1);
343
-
344
-
const call = mockUseCase.execute.mock.calls[0][0];
345
-
expect(call.cardId).toBe(cardId.getStringValue());
346
-
expect(call.actorId).toBe(curatorId.value);
347
-
expect(call.collectionIds).toContain('test-collection');
348
-
}, 15000);
349
-
350
-
it('should handle concurrent lock contention with retry mechanism', async () => {
351
-
// Arrange - Create multiple saga instances
352
-
const mockUseCase = {
353
-
execute: jest
354
-
.fn()
355
-
.mockResolvedValue(ok({ activityId: 'test-activity' })),
356
-
} as any;
357
-
358
-
const stateStore = new RedisSagaStateStore(redis);
359
-
const saga1 = new CardCollectionSaga(mockUseCase, stateStore);
360
-
const saga2 = new CardCollectionSaga(mockUseCase, stateStore);
361
-
const saga3 = new CardCollectionSaga(mockUseCase, stateStore);
362
-
363
-
// Create events for same card/user (will compete for same lock)
364
-
const cardId = CardId.createFromString('concurrent-test-card').unwrap();
365
-
const curatorId = CuratorId.create('did:plc:concurrentuser').unwrap();
366
-
const collectionId1 =
367
-
CollectionId.createFromString('collection-1').unwrap();
368
-
const collectionId2 =
369
-
CollectionId.createFromString('collection-2').unwrap();
370
-
371
-
const events = [
372
-
CardAddedToLibraryEvent.create(cardId, curatorId, new Date()).unwrap(),
373
-
CardAddedToCollectionEvent.create(
374
-
cardId,
375
-
collectionId1,
376
-
curatorId,
377
-
new Date(),
378
-
).unwrap(),
379
-
CardAddedToCollectionEvent.create(
380
-
cardId,
381
-
collectionId2,
382
-
curatorId,
383
-
new Date(),
384
-
).unwrap(),
385
-
];
386
-
387
-
// Act - Process all events concurrently (not sequentially)
388
-
const results = await Promise.all([
389
-
saga1.handleCardEvent(events[0]!),
390
-
saga2.handleCardEvent(events[1]!),
391
-
saga3.handleCardEvent(events[2]!),
392
-
]);
393
-
394
-
// Assert - All should succeed (no events dropped due to lock contention)
395
-
results.forEach((result) => expect(result.isOk()).toBe(true));
396
-
397
-
// Wait for aggregation window
398
-
await new Promise((resolve) => setTimeout(resolve, 3500));
399
-
400
-
// Assert - Should create single activity with all collections
401
-
expect(mockUseCase.execute).toHaveBeenCalledTimes(1);
402
-
const call = mockUseCase.execute.mock.calls[0][0];
403
-
expect(call.cardId).toBe(cardId.getStringValue());
404
-
expect(call.actorId).toBe(curatorId.value);
405
-
expect(call.collectionIds).toHaveLength(2);
406
-
expect(call.collectionIds).toContain('collection-1');
407
-
expect(call.collectionIds).toContain('collection-2');
408
-
}, 20000);
409
-
410
-
it('should handle high concurrency with many simultaneous events', async () => {
411
-
// Arrange - Create many saga instances
412
-
const mockUseCase = {
413
-
execute: jest
414
-
.fn()
415
-
.mockResolvedValue(ok({ activityId: 'test-activity' })),
416
-
} as any;
417
-
418
-
const stateStore = new RedisSagaStateStore(redis);
419
-
const cardId = CardId.createFromString('high-concurrency-card').unwrap();
420
-
const curatorId = CuratorId.create('did:plc:highconcurrency').unwrap();
421
-
422
-
// Create 10 collection events that will all compete for the same lock
423
-
const events = Array.from({ length: 10 }, (_, i) => {
424
-
const saga = new CardCollectionSaga(mockUseCase, stateStore);
425
-
const collectionId = CollectionId.createFromString(
426
-
`collection-${i}`,
427
-
).unwrap();
428
-
const event = CardAddedToCollectionEvent.create(
429
-
cardId,
430
-
collectionId,
431
-
curatorId,
432
-
new Date(),
433
-
).unwrap();
434
-
return { saga, event };
435
-
});
436
-
437
-
// Add one library event
438
-
const librarySaga = new CardCollectionSaga(mockUseCase, stateStore);
439
-
const libraryEvent = CardAddedToLibraryEvent.create(
440
-
cardId,
441
-
curatorId,
442
-
new Date(),
443
-
).unwrap();
444
-
445
-
// Act - Process all events concurrently
446
-
const allPromises = [
447
-
librarySaga.handleCardEvent(libraryEvent),
448
-
...events.map(({ saga, event }) => saga.handleCardEvent(event)),
449
-
];
450
-
451
-
const results = await Promise.all(allPromises);
452
-
453
-
// Assert - All should succeed
454
-
results.forEach((result) => expect(result.isOk()).toBe(true));
455
-
456
-
// Wait longer for aggregation window and retry processing
457
-
await new Promise((resolve) => setTimeout(resolve, 8000)); // Increased from 3500ms
458
-
459
-
// Assert - Should create single activity with all 10 collections
460
-
expect(mockUseCase.execute).toHaveBeenCalledTimes(1);
461
-
const call = mockUseCase.execute.mock.calls[0][0];
462
-
expect(call.cardId).toBe(cardId.getStringValue());
463
-
expect(call.actorId).toBe(curatorId.value);
464
-
expect(call.collectionIds).toHaveLength(10);
465
-
466
-
// Verify all collection IDs are present
467
-
for (let i = 0; i < 10; i++) {
468
-
expect(call.collectionIds).toContain(`collection-${i}`);
469
-
}
470
-
}, 35000); // Increased timeout from 25000ms
471
-
472
-
it('should recover when lock expires due to timeout', async () => {
473
-
// Arrange
474
-
const mockUseCase = {
475
-
execute: jest
476
-
.fn()
477
-
.mockResolvedValue(ok({ activityId: 'test-activity' })),
478
-
} as any;
479
-
480
-
const stateStore = new RedisSagaStateStore(redis);
481
-
const cardId = CardId.createFromString('timeout-test-card').unwrap();
482
-
const curatorId = CuratorId.create('did:plc:timeoutuser').unwrap();
483
-
484
-
// Manually acquire lock with short TTL to simulate crashed worker
485
-
const lockKey = 'saga:feed:lock:timeout-test-card-did:plc:timeoutuser';
486
-
await stateStore.set(lockKey, '1', 'EX', 2, 'NX'); // 2 second TTL (increased from 1)
487
-
488
-
// Create saga and event
489
-
const saga = new CardCollectionSaga(mockUseCase, stateStore);
490
-
const event = CardAddedToLibraryEvent.create(
491
-
cardId,
492
-
curatorId,
493
-
new Date(),
494
-
).unwrap();
495
-
496
-
// Act - Try to process event (should initially be blocked by lock)
497
-
// But should succeed after lock expires and retry mechanism kicks in
498
-
const result = await saga.handleCardEvent(event);
499
-
500
-
// Assert - Should succeed after lock expires
501
-
expect(result.isOk()).toBe(true);
502
-
503
-
// Wait longer for aggregation window
504
-
await new Promise((resolve) => setTimeout(resolve, 5000)); // Increased from 3500ms
505
-
506
-
// Assert - Activity should be created
507
-
expect(mockUseCase.execute).toHaveBeenCalledTimes(1);
508
-
}, 20000); // Increased timeout from 15000ms
509
-
510
-
it('should handle retry mechanism under lock contention', async () => {
511
-
// Arrange
512
-
const mockUseCase = {
513
-
execute: jest
514
-
.fn()
515
-
.mockResolvedValue(ok({ activityId: 'test-activity' })),
516
-
} as any;
517
-
518
-
const stateStore = new RedisSagaStateStore(redis);
519
-
const cardId = CardId.createFromString('retry-test-card').unwrap();
520
-
const curatorId = CuratorId.create('did:plc:retryuser').unwrap();
521
-
522
-
// Create a saga that will hold the lock for a while
523
-
const lockHoldingSaga = new CardCollectionSaga(mockUseCase, stateStore);
524
-
const retryingSaga = new CardCollectionSaga(mockUseCase, stateStore);
525
-
526
-
const event1 = CardAddedToLibraryEvent.create(
527
-
cardId,
528
-
curatorId,
529
-
new Date(),
530
-
).unwrap();
531
-
const event2 = CardAddedToCollectionEvent.create(
532
-
cardId,
533
-
CollectionId.createFromString('retry-collection').unwrap(),
534
-
curatorId,
535
-
new Date(),
536
-
).unwrap();
537
-
538
-
// Act - Start first saga (will acquire lock)
539
-
const promise1 = lockHoldingSaga.handleCardEvent(event1);
540
-
541
-
// Immediately start second saga (will need to retry)
542
-
const promise2 = retryingSaga.handleCardEvent(event2);
543
-
544
-
const results = await Promise.all([promise1, promise2]);
545
-
546
-
// Assert - Both should succeed
547
-
expect(results[0].isOk()).toBe(true);
548
-
expect(results[1].isOk()).toBe(true);
549
-
550
-
// Wait for aggregation window
551
-
await new Promise((resolve) => setTimeout(resolve, 3500));
552
-
553
-
// Assert - Should create single aggregated activity
554
-
expect(mockUseCase.execute).toHaveBeenCalledTimes(1);
555
-
const call = mockUseCase.execute.mock.calls[0][0];
556
-
expect(call.collectionIds).toContain('retry-collection');
557
-
}, 20000);
558
-
});
299
+
// NOTE: Redis-Based Saga Integration tests have been removed as CardCollectionSaga
300
+
// has been replaced with direct event handling using distributed locks at the service layer.
301
+
// The saga pattern was removed to simplify the architecture while maintaining correctness
302
+
// through FeedService's 2-minute deduplication window combined with distributed locking.
303
+
//
304
+
// See: .agent/logs/20260305_removing_feed_saga.md for details on the architectural change.
559
305
560
306
describe('Multi-Queue Event Routing', () => {
561
307
it('should route events to multiple queues', async () => {
+18
-4
src/modules/feeds/application/eventHandlers/CardAddedToCollectionEventHandler.ts
+18
-4
src/modules/feeds/application/eventHandlers/CardAddedToCollectionEventHandler.ts
···
1
1
import { CardAddedToCollectionEvent } from '../../../cards/domain/events/CardAddedToCollectionEvent';
2
2
import { IEventHandler } from '../../../../shared/application/events/IEventSubscriber';
3
-
import { Result } from '../../../../shared/core/Result';
4
-
import { CardCollectionSaga } from '../sagas/CardCollectionSaga';
3
+
import { Result, ok, err } from '../../../../shared/core/Result';
4
+
import { AddActivityToFeedUseCase } from '../useCases/commands/AddActivityToFeedUseCase';
5
+
import { ActivityTypeEnum } from '../../../feeds/domain/value-objects/ActivityType';
5
6
6
7
export class CardAddedToCollectionEventHandler
7
8
implements IEventHandler<CardAddedToCollectionEvent>
8
9
{
9
-
constructor(private cardCollectionSaga: CardCollectionSaga) {}
10
+
constructor(private addActivityToFeedUseCase: AddActivityToFeedUseCase) {}
10
11
11
12
async handle(event: CardAddedToCollectionEvent): Promise<Result<void>> {
12
-
return this.cardCollectionSaga.handleCardEvent(event);
13
+
const result = await this.addActivityToFeedUseCase.execute({
14
+
type: ActivityTypeEnum.CARD_COLLECTED,
15
+
actorId: event.addedBy.value,
16
+
cardId: event.cardId.getStringValue(),
17
+
collectionIds: [event.collectionId.getStringValue()],
18
+
createdAt: event.addedAt,
19
+
});
20
+
21
+
if (result.isErr()) {
22
+
console.error('Failed to add collection activity to feed:', result.error);
23
+
return err(result.error);
24
+
}
25
+
26
+
return ok(undefined);
13
27
}
14
28
}
+18
-4
src/modules/feeds/application/eventHandlers/CardAddedToLibraryEventHandler.ts
+18
-4
src/modules/feeds/application/eventHandlers/CardAddedToLibraryEventHandler.ts
···
1
1
import { CardAddedToLibraryEvent } from '../../../cards/domain/events/CardAddedToLibraryEvent';
2
2
import { IEventHandler } from '../../../../shared/application/events/IEventSubscriber';
3
-
import { Result } from '../../../../shared/core/Result';
4
-
import { CardCollectionSaga } from '../sagas/CardCollectionSaga';
3
+
import { Result, ok, err } from '../../../../shared/core/Result';
4
+
import { AddActivityToFeedUseCase } from '../useCases/commands/AddActivityToFeedUseCase';
5
+
import { ActivityTypeEnum } from '../../../feeds/domain/value-objects/ActivityType';
5
6
6
7
export class CardAddedToLibraryEventHandler
7
8
implements IEventHandler<CardAddedToLibraryEvent>
8
9
{
9
-
constructor(private cardCollectionSaga: CardCollectionSaga) {}
10
+
constructor(private addActivityToFeedUseCase: AddActivityToFeedUseCase) {}
10
11
11
12
async handle(event: CardAddedToLibraryEvent): Promise<Result<void>> {
12
-
return this.cardCollectionSaga.handleCardEvent(event);
13
+
const result = await this.addActivityToFeedUseCase.execute({
14
+
type: ActivityTypeEnum.CARD_COLLECTED,
15
+
actorId: event.curatorId.value,
16
+
cardId: event.cardId.getStringValue(),
17
+
collectionIds: undefined, // Library event has no collections
18
+
createdAt: event.addedAt,
19
+
});
20
+
21
+
if (result.isErr()) {
22
+
console.error('Failed to add library activity to feed:', result.error);
23
+
return err(result.error);
24
+
}
25
+
26
+
return ok(undefined);
13
27
}
14
28
}
-236
src/modules/feeds/application/sagas/CardCollectionSaga.ts
-236
src/modules/feeds/application/sagas/CardCollectionSaga.ts
···
1
-
import { Result, ok, err } from '../../../../shared/core/Result';
2
-
import { CardAddedToLibraryEvent } from '../../../cards/domain/events/CardAddedToLibraryEvent';
3
-
import { CardAddedToCollectionEvent } from '../../../cards/domain/events/CardAddedToCollectionEvent';
4
-
import {
5
-
AddActivityToFeedUseCase,
6
-
AddCardCollectedActivityDTO,
7
-
} from '../useCases/commands/AddActivityToFeedUseCase';
8
-
import { ActivityTypeEnum } from '../../domain/value-objects/ActivityType';
9
-
import { ISagaStateStore } from './ISagaStateStore';
10
-
11
-
interface PendingCardActivity {
12
-
cardId: string;
13
-
actorId: string;
14
-
collectionIds: string[];
15
-
timestamp: Date; // Timestamp for the aggregation window
16
-
eventTimestamps: Date[]; // Track all addedAt timestamps from events
17
-
hasLibraryEvent: boolean;
18
-
hasCollectionEvents: boolean;
19
-
}
20
-
21
-
export class CardCollectionSaga {
22
-
private readonly AGGREGATION_WINDOW_MS = 3000;
23
-
private readonly REDIS_KEY_PREFIX = 'saga:feed';
24
-
25
-
constructor(
26
-
private addActivityToFeedUseCase: AddActivityToFeedUseCase,
27
-
private stateStore: ISagaStateStore,
28
-
) {}
29
-
30
-
async handleCardEvent(
31
-
event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
32
-
): Promise<Result<void>> {
33
-
const aggregationKey = this.createKey(event);
34
-
35
-
// Retry lock acquisition with longer delays and more attempts for high concurrency
36
-
const maxRetries = 15; // Increased from 5
37
-
const baseDelay = 100; // Increased from 50ms
38
-
const maxDelay = 2000; // Cap the maximum delay
39
-
40
-
for (let attempt = 0; attempt < maxRetries; attempt++) {
41
-
const lockAcquired = await this.acquireLock(aggregationKey);
42
-
43
-
if (lockAcquired) {
44
-
try {
45
-
const existing = await this.getPendingActivity(aggregationKey);
46
-
47
-
if (existing && this.isWithinWindow(existing)) {
48
-
this.mergeActivity(existing, event);
49
-
await this.setPendingActivity(aggregationKey, existing);
50
-
} else {
51
-
const newActivity = this.createNewPendingActivity(event);
52
-
await this.setPendingActivity(aggregationKey, newActivity);
53
-
await this.scheduleFlush(aggregationKey);
54
-
}
55
-
56
-
return ok(undefined);
57
-
} finally {
58
-
await this.releaseLock(aggregationKey);
59
-
}
60
-
}
61
-
62
-
// Lock not acquired, wait and retry
63
-
if (attempt < maxRetries - 1) {
64
-
// Use exponential backoff with jitter and cap at maxDelay
65
-
const exponentialDelay = baseDelay * Math.pow(1.5, attempt);
66
-
const jitter = Math.random() * 50; // Add randomness to prevent thundering herd
67
-
const delay = Math.min(exponentialDelay + jitter, maxDelay);
68
-
69
-
console.log(
70
-
`Lock acquisition failed for ${aggregationKey}, retrying in ${Math.round(delay)}ms (attempt ${attempt + 1}/${maxRetries})`,
71
-
);
72
-
await new Promise((resolve) => setTimeout(resolve, delay));
73
-
}
74
-
}
75
-
76
-
// All retries failed - this shouldn't happen often
77
-
console.warn(
78
-
`Failed to acquire lock after ${maxRetries} attempts for ${aggregationKey}`,
79
-
);
80
-
return ok(undefined);
81
-
}
82
-
83
-
// Key helpers
84
-
private getPendingKey(aggregationKey: string): string {
85
-
return `${this.REDIS_KEY_PREFIX}:pending:${aggregationKey}`;
86
-
}
87
-
88
-
private getLockKey(aggregationKey: string): string {
89
-
return `${this.REDIS_KEY_PREFIX}:lock:${aggregationKey}`;
90
-
}
91
-
92
-
// State management
93
-
private async getPendingActivity(
94
-
aggregationKey: string,
95
-
): Promise<PendingCardActivity | null> {
96
-
const data = await this.stateStore.get(this.getPendingKey(aggregationKey));
97
-
if (!data) return null;
98
-
99
-
const parsed = JSON.parse(data);
100
-
// Convert timestamp strings back to Date objects
101
-
parsed.timestamp = new Date(parsed.timestamp);
102
-
parsed.eventTimestamps = parsed.eventTimestamps.map(
103
-
(ts: string) => new Date(ts),
104
-
);
105
-
return parsed;
106
-
}
107
-
108
-
private async setPendingActivity(
109
-
aggregationKey: string,
110
-
activity: PendingCardActivity,
111
-
): Promise<void> {
112
-
const key = this.getPendingKey(aggregationKey);
113
-
const ttlSeconds = Math.ceil(this.AGGREGATION_WINDOW_MS / 1000) + 5;
114
-
await this.stateStore.setex(key, ttlSeconds, JSON.stringify(activity));
115
-
}
116
-
117
-
private async deletePendingActivity(aggregationKey: string): Promise<void> {
118
-
await this.stateStore.del(this.getPendingKey(aggregationKey));
119
-
}
120
-
121
-
// Distributed locking
122
-
private async acquireLock(aggregationKey: string): Promise<boolean> {
123
-
const lockKey = this.getLockKey(aggregationKey);
124
-
// Reduced lock TTL to allow faster recovery in high concurrency scenarios
125
-
const lockTtl = Math.ceil(this.AGGREGATION_WINDOW_MS / 1000) + 5; // Reduced from +10 to +5
126
-
const result = await this.stateStore.set(lockKey, '1', 'EX', lockTtl, 'NX');
127
-
return result === 'OK';
128
-
}
129
-
130
-
private async releaseLock(aggregationKey: string): Promise<void> {
131
-
await this.stateStore.del(this.getLockKey(aggregationKey));
132
-
}
133
-
134
-
private createKey(
135
-
event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
136
-
): string {
137
-
const cardId = event.cardId.getStringValue();
138
-
const actorId = this.getActorId(event);
139
-
return `${cardId}-${actorId}`;
140
-
}
141
-
142
-
private getActorId(
143
-
event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
144
-
): string {
145
-
if ('curatorId' in event) {
146
-
return event.curatorId.value; // CardAddedToLibraryEvent
147
-
} else {
148
-
return event.addedBy.value; // CardAddedToCollectionEvent
149
-
}
150
-
}
151
-
152
-
private isWithinWindow(pending: PendingCardActivity): boolean {
153
-
const now = new Date();
154
-
const timeDiff = now.getTime() - pending.timestamp.getTime();
155
-
return timeDiff <= this.AGGREGATION_WINDOW_MS;
156
-
}
157
-
158
-
private createNewPendingActivity(
159
-
event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
160
-
): PendingCardActivity {
161
-
const cardId = event.cardId.getStringValue();
162
-
const actorId = this.getActorId(event);
163
-
164
-
const pending: PendingCardActivity = {
165
-
cardId,
166
-
actorId,
167
-
collectionIds: [],
168
-
timestamp: new Date(),
169
-
eventTimestamps: [],
170
-
hasLibraryEvent: false,
171
-
hasCollectionEvents: false,
172
-
};
173
-
174
-
this.mergeActivity(pending, event);
175
-
return pending;
176
-
}
177
-
178
-
private mergeActivity(
179
-
existing: PendingCardActivity,
180
-
event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
181
-
): void {
182
-
// Track the addedAt timestamp from the event
183
-
existing.eventTimestamps.push(event.addedAt);
184
-
185
-
if ('curatorId' in event) {
186
-
// CardAddedToLibraryEvent
187
-
existing.hasLibraryEvent = true;
188
-
} else {
189
-
// CardAddedToCollectionEvent
190
-
existing.hasCollectionEvents = true;
191
-
const collectionId = event.collectionId.getStringValue();
192
-
if (!existing.collectionIds.includes(collectionId)) {
193
-
existing.collectionIds.push(collectionId);
194
-
}
195
-
}
196
-
}
197
-
198
-
private async scheduleFlush(aggregationKey: string): Promise<void> {
199
-
setTimeout(async () => {
200
-
await this.flushActivity(aggregationKey);
201
-
}, this.AGGREGATION_WINDOW_MS);
202
-
}
203
-
204
-
private async flushActivity(aggregationKey: string): Promise<void> {
205
-
const lockAcquired = await this.acquireLock(aggregationKey);
206
-
if (!lockAcquired) return;
207
-
208
-
try {
209
-
const pending = await this.getPendingActivity(aggregationKey);
210
-
if (!pending) return;
211
-
212
-
// Calculate the earliest timestamp from all events
213
-
// If no timestamps (shouldn't happen), use current time as fallback
214
-
const earliestTimestamp =
215
-
pending.eventTimestamps.length > 0
216
-
? pending.eventTimestamps.reduce((earliest, current) =>
217
-
current < earliest ? current : earliest,
218
-
)
219
-
: undefined;
220
-
221
-
const request: AddCardCollectedActivityDTO = {
222
-
type: ActivityTypeEnum.CARD_COLLECTED,
223
-
actorId: pending.actorId,
224
-
cardId: pending.cardId,
225
-
collectionIds:
226
-
pending.collectionIds.length > 0 ? pending.collectionIds : undefined,
227
-
createdAt: earliestTimestamp,
228
-
};
229
-
230
-
await this.addActivityToFeedUseCase.execute(request);
231
-
} finally {
232
-
await this.deletePendingActivity(aggregationKey);
233
-
await this.releaseLock(aggregationKey);
234
-
}
235
-
}
236
-
}
+71
-56
src/modules/feeds/domain/services/FeedService.ts
+71
-56
src/modules/feeds/domain/services/FeedService.ts
···
6
6
import { CardId } from '../../../cards/domain/value-objects/CardId';
7
7
import { CollectionId } from '../../../cards/domain/value-objects/CollectionId';
8
8
import { UrlType } from '../../../cards/domain/value-objects/UrlType';
9
+
import { IDistributedLockService } from '../../../../shared/infrastructure/locking/IDistributedLockService';
9
10
10
11
export class FeedServiceError extends Error {
11
12
constructor(message: string) {
···
15
16
}
16
17
17
18
export class FeedService implements DomainService {
18
-
constructor(private feedRepository: IFeedRepository) {}
19
+
constructor(
20
+
private feedRepository: IFeedRepository,
21
+
private distributedLockService: IDistributedLockService,
22
+
) {}
19
23
20
24
async addCardCollectedActivity(
21
25
actorId: CuratorId,
···
25
29
source?: string,
26
30
createdAt?: Date,
27
31
): Promise<Result<FeedActivity, FeedServiceError>> {
32
+
const lockKey = `feed:activity:${actorId.value}:${cardId.getStringValue()}`;
33
+
const lockTTL = 10000; // 10 seconds
34
+
28
35
try {
29
-
// Check for recent duplicate activity (within 2 minutes)
30
-
const recentActivityResult =
31
-
await this.feedRepository.findRecentCardCollectedActivity(
32
-
actorId,
33
-
cardId,
34
-
2, // 2 minutes
35
-
);
36
+
// Acquire distributed lock to prevent race conditions when multiple workers
37
+
// process events for the same card+actor combination
38
+
return await this.distributedLockService.withLock(
39
+
lockKey,
40
+
lockTTL,
41
+
async () => {
42
+
// Check for recent duplicate activity (within 2 minutes)
43
+
const recentActivityResult =
44
+
await this.feedRepository.findRecentCardCollectedActivity(
45
+
actorId,
46
+
cardId,
47
+
2, // 2 minutes
48
+
);
36
49
37
-
if (recentActivityResult.isErr()) {
38
-
return err(
39
-
new FeedServiceError(
40
-
`Failed to check for recent activity: ${recentActivityResult.error.message}`,
41
-
),
42
-
);
43
-
}
50
+
if (recentActivityResult.isErr()) {
51
+
return err(
52
+
new FeedServiceError(
53
+
`Failed to check for recent activity: ${recentActivityResult.error.message}`,
54
+
),
55
+
);
56
+
}
44
57
45
-
const recentActivity = recentActivityResult.value;
58
+
const recentActivity = recentActivityResult.value;
46
59
47
-
if (recentActivity && collectionIds && collectionIds.length > 0) {
48
-
// Update existing activity by merging collections
49
-
recentActivity.mergeCollections(collectionIds);
60
+
if (recentActivity && collectionIds && collectionIds.length > 0) {
61
+
// Update existing activity by merging collections
62
+
recentActivity.mergeCollections(collectionIds);
50
63
51
-
const updateResult =
52
-
await this.feedRepository.updateActivity(recentActivity);
64
+
const updateResult =
65
+
await this.feedRepository.updateActivity(recentActivity);
53
66
54
-
if (updateResult.isErr()) {
55
-
return err(
56
-
new FeedServiceError(
57
-
`Failed to update activity: ${updateResult.error.message}`,
58
-
),
59
-
);
60
-
}
67
+
if (updateResult.isErr()) {
68
+
return err(
69
+
new FeedServiceError(
70
+
`Failed to update activity: ${updateResult.error.message}`,
71
+
),
72
+
);
73
+
}
61
74
62
-
return ok(recentActivity);
63
-
} else if (recentActivity) {
64
-
// Recent activity exists but no new collections to add, return existing
65
-
return ok(recentActivity);
66
-
}
75
+
return ok(recentActivity);
76
+
} else if (recentActivity) {
77
+
// Recent activity exists but no new collections to add, return existing
78
+
return ok(recentActivity);
79
+
}
67
80
68
-
// No recent activity found, create new one
69
-
const activityResult = FeedActivity.createCardCollected(
70
-
actorId,
71
-
cardId,
72
-
collectionIds,
73
-
urlType,
74
-
source,
75
-
createdAt,
76
-
);
81
+
// No recent activity found, create new one
82
+
const activityResult = FeedActivity.createCardCollected(
83
+
actorId,
84
+
cardId,
85
+
collectionIds,
86
+
urlType,
87
+
source,
88
+
createdAt,
89
+
);
77
90
78
-
if (activityResult.isErr()) {
79
-
return err(new FeedServiceError(activityResult.error.message));
80
-
}
91
+
if (activityResult.isErr()) {
92
+
return err(new FeedServiceError(activityResult.error.message));
93
+
}
81
94
82
-
const activity = activityResult.value;
83
-
const saveResult = await this.feedRepository.addActivity(activity);
95
+
const activity = activityResult.value;
96
+
const saveResult = await this.feedRepository.addActivity(activity);
84
97
85
-
if (saveResult.isErr()) {
86
-
return err(
87
-
new FeedServiceError(
88
-
`Failed to save activity: ${saveResult.error.message}`,
89
-
),
90
-
);
91
-
}
98
+
if (saveResult.isErr()) {
99
+
return err(
100
+
new FeedServiceError(
101
+
`Failed to save activity: ${saveResult.error.message}`,
102
+
),
103
+
);
104
+
}
92
105
93
-
return ok(activity);
106
+
return ok(activity);
107
+
},
108
+
);
94
109
} catch (error) {
95
110
return err(
96
111
new FeedServiceError(
97
-
`Unexpected error: ${error instanceof Error ? error.message : 'Unknown error'}`,
112
+
`Failed to acquire lock or process activity: ${error instanceof Error ? error.message : 'Unknown error'}`,
98
113
),
99
114
);
100
115
}