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';
9import { ISagaStateStore } from './ISagaStateStore';
10
11interface PendingCardActivity {
12 cardId: string;
13 actorId: string;
14 collectionIds: string[];
15 timestamp: Date;
16 hasLibraryEvent: boolean;
17 hasCollectionEvents: boolean;
18}
19
20export class CardCollectionSaga {
21 private readonly AGGREGATION_WINDOW_MS = 3000;
22 private readonly REDIS_KEY_PREFIX = 'saga:feed';
23
24 constructor(
25 private addActivityToFeedUseCase: AddActivityToFeedUseCase,
26 private stateStore: ISagaStateStore,
27 ) {}
28
29 async handleCardEvent(
30 event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
31 ): Promise<Result<void>> {
32 const aggregationKey = this.createKey(event);
33
34 // Retry lock acquisition with longer delays and more attempts for high concurrency
35 const maxRetries = 15; // Increased from 5
36 const baseDelay = 100; // Increased from 50ms
37 const maxDelay = 2000; // Cap the maximum delay
38
39 for (let attempt = 0; attempt < maxRetries; attempt++) {
40 const lockAcquired = await this.acquireLock(aggregationKey);
41
42 if (lockAcquired) {
43 try {
44 const existing = await this.getPendingActivity(aggregationKey);
45
46 if (existing && this.isWithinWindow(existing)) {
47 this.mergeActivity(existing, event);
48 await this.setPendingActivity(aggregationKey, existing);
49 } else {
50 const newActivity = this.createNewPendingActivity(event);
51 await this.setPendingActivity(aggregationKey, newActivity);
52 await this.scheduleFlush(aggregationKey);
53 }
54
55 return ok(undefined);
56 } finally {
57 await this.releaseLock(aggregationKey);
58 }
59 }
60
61 // Lock not acquired, wait and retry
62 if (attempt < maxRetries - 1) {
63 // Use exponential backoff with jitter and cap at maxDelay
64 const exponentialDelay = baseDelay * Math.pow(1.5, attempt);
65 const jitter = Math.random() * 50; // Add randomness to prevent thundering herd
66 const delay = Math.min(exponentialDelay + jitter, maxDelay);
67
68 console.log(
69 `Lock acquisition failed for ${aggregationKey}, retrying in ${Math.round(delay)}ms (attempt ${attempt + 1}/${maxRetries})`,
70 );
71 await new Promise((resolve) => setTimeout(resolve, delay));
72 }
73 }
74
75 // All retries failed - this shouldn't happen often
76 console.warn(
77 `Failed to acquire lock after ${maxRetries} attempts for ${aggregationKey}`,
78 );
79 return ok(undefined);
80 }
81
82 // Key helpers
83 private getPendingKey(aggregationKey: string): string {
84 return `${this.REDIS_KEY_PREFIX}:pending:${aggregationKey}`;
85 }
86
87 private getLockKey(aggregationKey: string): string {
88 return `${this.REDIS_KEY_PREFIX}:lock:${aggregationKey}`;
89 }
90
91 // State management
92 private async getPendingActivity(
93 aggregationKey: string,
94 ): Promise<PendingCardActivity | null> {
95 const data = await this.stateStore.get(this.getPendingKey(aggregationKey));
96 if (!data) return null;
97
98 const parsed = JSON.parse(data);
99 // Convert timestamp string back to Date object
100 parsed.timestamp = new Date(parsed.timestamp);
101 return parsed;
102 }
103
104 private async setPendingActivity(
105 aggregationKey: string,
106 activity: PendingCardActivity,
107 ): Promise<void> {
108 const key = this.getPendingKey(aggregationKey);
109 const ttlSeconds = Math.ceil(this.AGGREGATION_WINDOW_MS / 1000) + 5;
110 await this.stateStore.setex(key, ttlSeconds, JSON.stringify(activity));
111 }
112
113 private async deletePendingActivity(aggregationKey: string): Promise<void> {
114 await this.stateStore.del(this.getPendingKey(aggregationKey));
115 }
116
117 // Distributed locking
118 private async acquireLock(aggregationKey: string): Promise<boolean> {
119 const lockKey = this.getLockKey(aggregationKey);
120 // Reduced lock TTL to allow faster recovery in high concurrency scenarios
121 const lockTtl = Math.ceil(this.AGGREGATION_WINDOW_MS / 1000) + 5; // Reduced from +10 to +5
122 const result = await this.stateStore.set(lockKey, '1', 'EX', lockTtl, 'NX');
123 return result === 'OK';
124 }
125
126 private async releaseLock(aggregationKey: string): Promise<void> {
127 await this.stateStore.del(this.getLockKey(aggregationKey));
128 }
129
130 private createKey(
131 event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
132 ): string {
133 const cardId = event.cardId.getStringValue();
134 const actorId = this.getActorId(event);
135 return `${cardId}-${actorId}`;
136 }
137
138 private getActorId(
139 event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
140 ): string {
141 if ('curatorId' in event) {
142 return event.curatorId.value; // CardAddedToLibraryEvent
143 } else {
144 return event.addedBy.value; // CardAddedToCollectionEvent
145 }
146 }
147
148 private isWithinWindow(pending: PendingCardActivity): boolean {
149 const now = new Date();
150 const timeDiff = now.getTime() - pending.timestamp.getTime();
151 return timeDiff <= this.AGGREGATION_WINDOW_MS;
152 }
153
154 private createNewPendingActivity(
155 event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
156 ): PendingCardActivity {
157 const cardId = event.cardId.getStringValue();
158 const actorId = this.getActorId(event);
159
160 const pending: PendingCardActivity = {
161 cardId,
162 actorId,
163 collectionIds: [],
164 timestamp: new Date(),
165 hasLibraryEvent: false,
166 hasCollectionEvents: false,
167 };
168
169 this.mergeActivity(pending, event);
170 return pending;
171 }
172
173 private mergeActivity(
174 existing: PendingCardActivity,
175 event: CardAddedToLibraryEvent | CardAddedToCollectionEvent,
176 ): void {
177 if ('curatorId' in event) {
178 // CardAddedToLibraryEvent
179 existing.hasLibraryEvent = true;
180 } else {
181 // CardAddedToCollectionEvent
182 existing.hasCollectionEvents = true;
183 const collectionId = event.collectionId.getStringValue();
184 if (!existing.collectionIds.includes(collectionId)) {
185 existing.collectionIds.push(collectionId);
186 }
187 }
188 }
189
190 private async scheduleFlush(aggregationKey: string): Promise<void> {
191 setTimeout(async () => {
192 await this.flushActivity(aggregationKey);
193 }, this.AGGREGATION_WINDOW_MS);
194 }
195
196 private async flushActivity(aggregationKey: string): Promise<void> {
197 const lockAcquired = await this.acquireLock(aggregationKey);
198 if (!lockAcquired) return;
199
200 try {
201 const pending = await this.getPendingActivity(aggregationKey);
202 if (!pending) return;
203
204 const request: AddCardCollectedActivityDTO = {
205 type: ActivityTypeEnum.CARD_COLLECTED,
206 actorId: pending.actorId,
207 cardId: pending.cardId,
208 collectionIds:
209 pending.collectionIds.length > 0 ? pending.collectionIds : undefined,
210 };
211
212 await this.addActivityToFeedUseCase.execute(request);
213 } finally {
214 await this.deletePendingActivity(aggregationKey);
215 await this.releaseLock(aggregationKey);
216 }
217 }
218}