nandi.uk
This repository has no description
feat: Implement CardCollectionSaga for event aggregation and deduplication
This commit introduces a Saga pattern to handle card collection events, providing:
- Event aggregation within a 5-second window
- Merging library and collection events for the same card
- Coordinated activity creation through a centralized saga
- Improved event handling with minimal changes to existing architecture
Changes:
- Added CardCollectionSaga to coordinate event processing
- Updated event handlers to use the new saga
- Implemented event merging and deduplication logic
- Added timer-based event flushing mechanism
Co-authored-by: aider (anthropic/claude-sonnet-4-20250514) <aider@aider.chat>
author
Wesley Finck
co-author
aider (anthropic/claude-sonnet-4-20250514)
date
(Aug 1, 2025, 9:49 AM -0500)
commit
cf27f325
cf27f32587f8dbf1b473ed616950398d136e6222
parent
8d8b908a
8d8b908a2815572d17d1ffa4fef2d9fd6a40d61d
+179
-73
+4
-36
src/modules/feeds/application/eventHandlers/CardAddedToCollectionEventHandler.ts
+4
-36
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, ok, err } from '../../../../shared/core/Result';
4
-
import {
5
-
AddActivityToFeedUseCase,
6
-
AddCardCollectedActivityDTO,
7
-
} from '../useCases/commands/AddActivityToFeedUseCase';
8
-
import { ActivityTypeEnum } from '../../domain/value-objects/ActivityType';
3
+
import { Result } from '../../../../shared/core/Result';
4
+
import { CardCollectionSaga } from '../sagas/CardCollectionSaga';
9
5
10
6
export class CardAddedToCollectionEventHandler
11
7
implements IEventHandler<CardAddedToCollectionEvent>
12
8
{
13
-
constructor(private addActivityToFeedUseCase: AddActivityToFeedUseCase) {}
9
+
constructor(private cardCollectionSaga: CardCollectionSaga) {}
14
10
15
11
async handle(event: CardAddedToCollectionEvent): Promise<Result<void>> {
16
-
try {
17
-
const request: AddCardCollectedActivityDTO = {
18
-
type: ActivityTypeEnum.CARD_COLLECTED,
19
-
actorId: event.addedBy.value,
20
-
cardId: event.cardId.getStringValue(),
21
-
collectionIds: [event.collectionId.getStringValue()],
22
-
// TODO: Fetch card metadata (title, URL) from card repository
23
-
cardTitle: undefined,
24
-
cardUrl: undefined,
25
-
};
26
-
27
-
const result = await this.addActivityToFeedUseCase.execute(request);
28
-
29
-
if (result.isErr()) {
30
-
console.error(
31
-
'[FEEDS] Failed to add card-added-to-collection activity:',
32
-
result.error,
33
-
);
34
-
return err(new Error(result.error.message));
35
-
}
36
-
37
-
return ok(undefined);
38
-
} catch (error) {
39
-
console.error(
40
-
'[FEEDS] Unexpected error handling CardAddedToCollectionEvent:',
41
-
error,
42
-
);
43
-
return err(error as Error);
44
-
}
12
+
return this.cardCollectionSaga.handleCardEvent(event);
45
13
}
46
14
}
+4
-37
src/modules/feeds/application/eventHandlers/CardAddedToLibraryEventHandler.ts
+4
-37
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, ok, err } from '../../../../shared/core/Result';
4
-
import {
5
-
AddActivityToFeedUseCase,
6
-
AddCardCollectedActivityDTO,
7
-
} from '../useCases/commands/AddActivityToFeedUseCase';
8
-
import { ActivityTypeEnum } from '../../domain/value-objects/ActivityType';
3
+
import { Result } from '../../../../shared/core/Result';
4
+
import { CardCollectionSaga } from '../sagas/CardCollectionSaga';
9
5
10
6
export class CardAddedToLibraryEventHandler
11
7
implements IEventHandler<CardAddedToLibraryEvent>
12
8
{
13
-
constructor(private addActivityToFeedUseCase: AddActivityToFeedUseCase) {}
9
+
constructor(private cardCollectionSaga: CardCollectionSaga) {}
14
10
15
11
async handle(event: CardAddedToLibraryEvent): Promise<Result<void>> {
16
-
try {
17
-
const request: AddCardCollectedActivityDTO = {
18
-
type: ActivityTypeEnum.CARD_COLLECTED,
19
-
actorId: event.curatorId.value,
20
-
cardId: event.cardId.getStringValue(),
21
-
// No collection IDs for library-only additions
22
-
collectionIds: undefined,
23
-
};
24
-
25
-
const result = await this.addActivityToFeedUseCase.execute(request);
26
-
27
-
if (result.isErr()) {
28
-
console.error(
29
-
'[FEEDS] Error processing CardAddedToLibraryEvent:',
30
-
result.error,
31
-
);
32
-
return err(new Error(result.error.message));
33
-
}
34
-
35
-
console.log(
36
-
`[FEEDS] Successfully processed CardAddedToLibraryEvent for card ${event.cardId.getStringValue()}, created activity ${result.value.activityId}`,
37
-
);
38
-
return ok(undefined);
39
-
} catch (error) {
40
-
console.error(
41
-
'[FEEDS] Unexpected error handling CardAddedToLibraryEvent:',
42
-
error,
43
-
);
44
-
return err(error as Error);
45
-
}
12
+
return this.cardCollectionSaga.handleCardEvent(event);
46
13
}
47
14
}
+171
src/modules/feeds/application/sagas/CardCollectionSaga.ts
+171
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
+
10
+
interface PendingCardActivity {
11
+
cardId: string;
12
+
actorId: string;
13
+
collectionIds: string[];
14
+
timestamp: Date;
15
+
hasLibraryEvent: boolean;
16
+
hasCollectionEvents: boolean;
17
+
}
18
+
19
+
export class CardCollectionSaga {
20
+
private pendingActivities = new Map<string, PendingCardActivity>();
21
+
private flushTimers = new Map<string, NodeJS.Timeout>();
22
+
private readonly AGGREGATION_WINDOW_MS = 3000;
23
+
24
+
constructor(private addActivityToFeedUseCase: AddActivityToFeedUseCase) {}
25
+
26
+
async handleCardEvent(
27
+
event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
28
+
): Promise<Result<void>> {
29
+
try {
30
+
const aggregationKey = this.createKey(event);
31
+
const existing = this.pendingActivities.get(aggregationKey);
32
+
33
+
if (existing && this.isWithinWindow(existing)) {
34
+
// Merge with existing activity
35
+
this.mergeActivity(existing, event);
36
+
// Reset the timer
37
+
this.rescheduleFlush(aggregationKey);
38
+
} else {
39
+
// Create new pending activity
40
+
this.createPendingActivity(aggregationKey, event);
41
+
this.scheduleFlush(aggregationKey);
42
+
}
43
+
44
+
return ok(undefined);
45
+
} catch (error) {
46
+
console.error('[SAGA] Error handling card event:', error);
47
+
return err(error as Error);
48
+
}
49
+
}
50
+
51
+
private createKey(
52
+
event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
53
+
): string {
54
+
const cardId = event.cardId.getStringValue();
55
+
const actorId = this.getActorId(event);
56
+
return `${cardId}-${actorId}`;
57
+
}
58
+
59
+
private getActorId(
60
+
event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
61
+
): string {
62
+
if ('curatorId' in event) {
63
+
return event.curatorId.value; // CardAddedToLibraryEvent
64
+
} else {
65
+
return event.addedBy.value; // CardAddedToCollectionEvent
66
+
}
67
+
}
68
+
69
+
private isWithinWindow(pending: PendingCardActivity): boolean {
70
+
const now = new Date();
71
+
const timeDiff = now.getTime() - pending.timestamp.getTime();
72
+
return timeDiff <= this.AGGREGATION_WINDOW_MS;
73
+
}
74
+
75
+
private createPendingActivity(
76
+
key: string,
77
+
event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
78
+
): void {
79
+
const cardId = event.cardId.getStringValue();
80
+
const actorId = this.getActorId(event);
81
+
82
+
const pending: PendingCardActivity = {
83
+
cardId,
84
+
actorId,
85
+
collectionIds: [],
86
+
timestamp: new Date(),
87
+
hasLibraryEvent: false,
88
+
hasCollectionEvents: false,
89
+
};
90
+
91
+
this.mergeActivity(pending, event);
92
+
this.pendingActivities.set(key, pending);
93
+
}
94
+
95
+
private mergeActivity(
96
+
existing: PendingCardActivity,
97
+
event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
98
+
): void {
99
+
if ('curatorId' in event) {
100
+
// CardAddedToLibraryEvent
101
+
existing.hasLibraryEvent = true;
102
+
} else {
103
+
// CardAddedToCollectionEvent
104
+
existing.hasCollectionEvents = true;
105
+
const collectionId = event.collectionId.getStringValue();
106
+
if (!existing.collectionIds.includes(collectionId)) {
107
+
existing.collectionIds.push(collectionId);
108
+
}
109
+
}
110
+
}
111
+
112
+
private scheduleFlush(key: string): void {
113
+
const timer = setTimeout(() => {
114
+
this.flushActivity(key);
115
+
}, this.AGGREGATION_WINDOW_MS);
116
+
117
+
this.flushTimers.set(key, timer);
118
+
}
119
+
120
+
private rescheduleFlush(key: string): void {
121
+
// Clear existing timer
122
+
const existingTimer = this.flushTimers.get(key);
123
+
if (existingTimer) {
124
+
clearTimeout(existingTimer);
125
+
}
126
+
127
+
// Schedule new timer
128
+
this.scheduleFlush(key);
129
+
}
130
+
131
+
private async flushActivity(key: string): Promise<void> {
132
+
const pending = this.pendingActivities.get(key);
133
+
if (!pending) return;
134
+
135
+
try {
136
+
// Create the aggregated activity
137
+
const request: AddCardCollectedActivityDTO = {
138
+
type: ActivityTypeEnum.CARD_COLLECTED,
139
+
actorId: pending.actorId,
140
+
cardId: pending.cardId,
141
+
collectionIds:
142
+
pending.collectionIds.length > 0 ? pending.collectionIds : undefined,
143
+
};
144
+
145
+
const result = await this.addActivityToFeedUseCase.execute(request);
146
+
147
+
if (result.isErr()) {
148
+
console.error(
149
+
'[SAGA] Failed to create aggregated activity:',
150
+
result.error,
151
+
);
152
+
} else {
153
+
console.log(
154
+
`[SAGA] Successfully created aggregated activity ${result.value.activityId} for card ${pending.cardId}`,
155
+
);
156
+
}
157
+
} catch (error) {
158
+
console.error('[SAGA] Error flushing activity:', error);
159
+
} finally {
160
+
// Clean up
161
+
this.pendingActivities.delete(key);
162
+
this.flushTimers.delete(key);
163
+
}
164
+
}
165
+
166
+
// For testing or graceful shutdown
167
+
public async flushAll(): Promise<void> {
168
+
const keys = Array.from(this.pendingActivities.keys());
169
+
await Promise.all(keys.map((key) => this.flushActivity(key)));
170
+
}
171
+
}