diff --git a/src/modules/cards/tests/integration/BullMQEventSystem.integration.test.ts b/src/modules/cards/tests/integration/BullMQEventSystem.integration.test.ts index 5be64c09..3b04dfeb 100644 --- a/src/modules/cards/tests/integration/BullMQEventSystem.integration.test.ts +++ b/src/modules/cards/tests/integration/BullMQEventSystem.integration.test.ts @@ -350,20 +350,30 @@ describe('BullMQ Event System Integration', () => { // Create events for same card/user (will compete for same lock) const cardId = CardId.createFromString('concurrent-test-card').unwrap(); const curatorId = CuratorId.create('did:plc:concurrentuser').unwrap(); - const collectionId1 = CollectionId.createFromString('collection-1').unwrap(); - const collectionId2 = CollectionId.createFromString('collection-2').unwrap(); + const collectionId1 = + CollectionId.createFromString('collection-1').unwrap(); + const collectionId2 = + CollectionId.createFromString('collection-2').unwrap(); const events = [ CardAddedToLibraryEvent.create(cardId, curatorId).unwrap(), - CardAddedToCollectionEvent.create(cardId, collectionId1, curatorId).unwrap(), - CardAddedToCollectionEvent.create(cardId, collectionId2, curatorId).unwrap(), + CardAddedToCollectionEvent.create( + cardId, + collectionId1, + curatorId, + ).unwrap(), + CardAddedToCollectionEvent.create( + cardId, + collectionId2, + curatorId, + ).unwrap(), ]; // Act - Process all events concurrently (not sequentially) const results = await Promise.all([ - saga1.handleCardEvent(events[0]), - saga2.handleCardEvent(events[1]), - saga3.handleCardEvent(events[2]), + saga1.handleCardEvent(events[0]!), + saga2.handleCardEvent(events[1]!), + saga3.handleCardEvent(events[2]!), ]); // Assert - All should succeed (no events dropped due to lock contention) @@ -397,14 +407,23 @@ describe('BullMQ Event System Integration', () => { // Create 10 collection events that will all compete for the same lock const events = Array.from({ length: 10 }, (_, i) => { const saga = new CardCollectionSaga(mockUseCase, stateStore); - const collectionId = CollectionId.createFromString(`collection-${i}`).unwrap(); - const event = CardAddedToCollectionEvent.create(cardId, collectionId, curatorId).unwrap(); + const collectionId = CollectionId.createFromString( + `collection-${i}`, + ).unwrap(); + const event = CardAddedToCollectionEvent.create( + cardId, + collectionId, + curatorId, + ).unwrap(); return { saga, event }; }); // Add one library event const librarySaga = new CardCollectionSaga(mockUseCase, stateStore); - const libraryEvent = CardAddedToLibraryEvent.create(cardId, curatorId).unwrap(); + const libraryEvent = CardAddedToLibraryEvent.create( + cardId, + curatorId, + ).unwrap(); // Act - Process all events concurrently const allPromises = [ @@ -426,7 +445,7 @@ describe('BullMQ Event System Integration', () => { expect(call.cardId).toBe(cardId.getStringValue()); expect(call.actorId).toBe(curatorId.value); expect(call.collectionIds).toHaveLength(10); - + // Verify all collection IDs are present for (let i = 0; i < 10; i++) { expect(call.collectionIds).toContain(`collection-${i}`); @@ -492,7 +511,7 @@ describe('BullMQ Event System Integration', () => { // Act - Start first saga (will acquire lock) const promise1 = lockHoldingSaga.handleCardEvent(event1); - + // Immediately start second saga (will need to retry) const promise2 = retryingSaga.handleCardEvent(event2); diff --git a/src/modules/feeds/application/sagas/CardCollectionSaga.ts b/src/modules/feeds/application/sagas/CardCollectionSaga.ts index ebf94931..5ee00db5 100644 --- a/src/modules/feeds/application/sagas/CardCollectionSaga.ts +++ b/src/modules/feeds/application/sagas/CardCollectionSaga.ts @@ -29,7 +29,6 @@ export class CardCollectionSaga { async handleCardEvent( event: CardAddedToLibraryEvent | CardAddedToCollectionEvent, ): Promise> { - console.log('Handling card event:', event); const aggregationKey = this.createKey(event); // Retry lock acquisition with longer delays and more attempts for high concurrency @@ -45,11 +44,9 @@ export class CardCollectionSaga { const existing = await this.getPendingActivity(aggregationKey); if (existing && this.isWithinWindow(existing)) { - console.log(`Merging event into existing activity for ${aggregationKey}`); this.mergeActivity(existing, event); await this.setPendingActivity(aggregationKey, existing); } else { - console.log(`Creating new pending activity for ${aggregationKey}`); const newActivity = this.createNewPendingActivity(event); await this.setPendingActivity(aggregationKey, newActivity); await this.scheduleFlush(aggregationKey); @@ -67,8 +64,10 @@ export class CardCollectionSaga { const exponentialDelay = baseDelay * Math.pow(1.5, attempt); const jitter = Math.random() * 50; // Add randomness to prevent thundering herd const delay = Math.min(exponentialDelay + jitter, maxDelay); - - console.log(`Lock acquisition failed for ${aggregationKey}, retrying in ${Math.round(delay)}ms (attempt ${attempt + 1}/${maxRetries})`); + + console.log( + `Lock acquisition failed for ${aggregationKey}, retrying in ${Math.round(delay)}ms (attempt ${attempt + 1}/${maxRetries})`, + ); await new Promise((resolve) => setTimeout(resolve, delay)); } } diff --git a/src/workers/feed-worker.ts b/src/workers/feed-worker.ts index 7573dfb2..f160ce08 100644 --- a/src/workers/feed-worker.ts +++ b/src/workers/feed-worker.ts @@ -3,16 +3,18 @@ import { FeedWorkerProcess } from '../shared/infrastructure/processes/FeedWorker async function main() { const useInMemoryEvents = process.env.USE_IN_MEMORY_EVENTS === 'true'; - + if (useInMemoryEvents) { - console.log('Skipping feed worker startup - using in-memory events (handled by main process)'); + console.log( + 'Skipping feed worker startup - using in-memory events (handled by main process)', + ); return; } console.log('Starting dedicated feed worker process...'); const configService = new EnvironmentConfigService(); const feedWorkerProcess = new FeedWorkerProcess(configService); - + await feedWorkerProcess.start(); }