nandi.uk
This repository has no description
1import { Result, ok, err } from '../../../../shared/core/Result';
2import { CardAddedToLibraryEvent } from '../../../cards/domain/events/CardAddedToLibraryEvent';
3import { CardAddedToCollectionEvent } from '../../../cards/domain/events/CardAddedToCollectionEvent';
4import {
5 AddActivityToFeedUseCase,
6 AddCardCollectedActivityDTO,
7} from '../useCases/commands/AddActivityToFeedUseCase';
8import { ActivityTypeEnum } from '../../domain/value-objects/ActivityType';
9
10interface PendingCardActivity {
11 cardId: string;
12 actorId: string;
13 collectionIds: string[];
14 timestamp: Date;
15 hasLibraryEvent: boolean;
16 hasCollectionEvents: boolean;
17}
18
19export 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}