refactor(usage): streamline usage invalidation event emissions

- Replaced direct event emissions with a dedicated emitUsageInvalidated function across BackupUseCase, FileUseCases, and FolderUseCases for better consistency and maintainability.
- Updated the UsageInvalidatedEvent structure to use an interface, enhancing type safety.
- Adjusted event handler to utilize a deduplication strategy for usage events, improving performance and reliability.

Author: jzunigax2

src/modules/backups/backup.usecase.ts

@@ -22,7 +22,7 @@ import { type DevicePlatform } from './device.domain';
 import { type UpdateDeviceAndFolderDto } from './dto/update-device-and-folder.dto';
 import { SequelizeFolderRepository } from '../folder/folder.repository';
 import { EventEmitter2 } from '@nestjs/event-emitter';
-import { UsageInvalidatedEvent } from '../usage-queue/events/usage-invalidated.event';
+import { emitUsageInvalidated } from '../usage-queue/events/usage-invalidated.event';
 
 @Injectable()
 export class BackupUseCase {
@@ -80,9 +80,11 @@ export class BackupUseCase {
       id: deviceId,
     });
 
-    this.eventEmitter.emit(
-      'usage.backup.device_deleted',
-      new UsageInvalidatedEvent(user.uuid, user.id, 'backup.device.delete'),
+    emitUsageInvalidated(
+      this.eventEmitter,
+      user.uuid,
+      user.id,
+      'backup.device.delete',
     );
 
     return result;
@@ -479,9 +481,11 @@ export class BackupUseCase {
       backupId,
     );
 
-    this.eventEmitter.emit(
-      'usage.backup.deleted',
-      new UsageInvalidatedEvent(user.uuid, user.id, 'backup.delete'),
+    emitUsageInvalidated(
+      this.eventEmitter,
+      user.uuid,
+      user.id,
+      'backup.delete',
     );
 
     return result;

src/modules/file/actions/create-file-version.action.ts

@@ -6,7 +6,7 @@ import { SequelizeFileRepository } from '../file.repository';
 import { SequelizeFileVersionRepository } from '../file-version.repository';
 import { FileVersionStatus } from '../file-version.domain';
 import { FeatureLimitService } from '../../feature-limit/feature-limit.service';
-import { UsageInvalidatedEvent } from '../../usage-queue/events/usage-invalidated.event';
+import { emitUsageInvalidated } from '../../usage-queue/events/usage-invalidated.event';
 import { Time } from '../../../lib/time';
 
 @Injectable()
@@ -44,9 +44,11 @@ export class CreateFileVersionAction {
       }),
     ]);
 
-    this.eventEmitter.emit(
-      'usage.file.version_created',
-      new UsageInvalidatedEvent(user.uuid, user.id, 'file.version.create'),
+    emitUsageInvalidated(
+      this.eventEmitter,
+      user.uuid,
+      user.id,
+      'file.version.create',
     );
   }
 

src/modules/file/actions/delete-file-version.action.ts

@@ -9,7 +9,7 @@ import { SequelizeFileVersionRepository } from '../file-version.repository';
 import { SequelizeFileRepository } from '../file.repository';
 import { type User } from '../../user/user.domain';
 import { FileVersionStatus } from '../file-version.domain';
-import { UsageInvalidatedEvent } from '../../usage-queue/events/usage-invalidated.event';
+import { emitUsageInvalidated } from '../../usage-queue/events/usage-invalidated.event';
 
 @Injectable()
 export class DeleteFileVersionAction {
@@ -49,9 +49,11 @@ export class DeleteFileVersionAction {
       FileVersionStatus.DELETED,
     );
 
-    this.eventEmitter.emit(
-      'usage.file.version_deleted',
-      new UsageInvalidatedEvent(user.uuid, user.id, 'file.version.delete'),
+    emitUsageInvalidated(
+      this.eventEmitter,
+      user.uuid,
+      user.id,
+      'file.version.delete',
     );
   }
 }

src/modules/file/file.usecase.ts

@@ -55,7 +55,7 @@ import { RedisService } from '../../externals/redis/redis.service';
 import { type Usage } from '../usage/usage.domain';
 import { CacheManagerService } from '../cache-manager/cache-manager.service';
 import { EventEmitter2 } from '@nestjs/event-emitter';
-import { UsageInvalidatedEvent } from '../usage-queue/events/usage-invalidated.event';
+import { emitUsageInvalidated } from '../usage-queue/events/usage-invalidated.event';
 import { PaymentRequiredException } from '../feature-limit/exceptions/payment-required.exception';
 import {
   DeleteFileVersionAction,
@@ -216,10 +216,7 @@ export class FileUseCases {
 
     await this.fileRepository.deleteFilesByUser(user, [file]);
 
-    this.eventEmitter.emit(
-      'usage.file.deleted',
-      new UsageInvalidatedEvent(user.uuid, user.id, 'file.delete'),
-    );
+    emitUsageInvalidated(this.eventEmitter, user.uuid, user.id, 'file.delete');
 
     return { id, uuid };
   }
@@ -368,10 +365,7 @@ export class FileUseCases {
       creationTime: newFileDto.creationTime || newFileDto.date || new Date(),
     });
 
-    this.eventEmitter.emit(
-      'usage.file.created',
-      new UsageInvalidatedEvent(user.uuid, user.id, 'file.create'),
-    );
+    emitUsageInvalidated(this.eventEmitter, user.uuid, user.id, 'file.create');
 
     if (!hadFilesBeforeUpload) {
       const isUserFreeTier = tier?.label === PLAN_FREE_INDIVIDUAL_TIER_LABEL;
@@ -858,10 +852,7 @@ export class FileUseCases {
       ),
     ]);
 
-    this.eventEmitter.emit(
-      'usage.file.trashed',
-      new UsageInvalidatedEvent(user.uuid, user.id, 'file.trash'),
-    );
+    emitUsageInvalidated(this.eventEmitter, user.uuid, user.id, 'file.trash');
   }
 
   async getEncryptionKeyFromFile(
@@ -988,10 +979,7 @@ export class FileUseCases {
       }
     }
 
-    this.eventEmitter.emit(
-      'usage.file.replaced',
-      new UsageInvalidatedEvent(user.uuid, user.id, 'file.replace'),
-    );
+    emitUsageInvalidated(this.eventEmitter, user.uuid, user.id, 'file.replace');
 
     return {
       ...file.toJSON(),
@@ -1037,9 +1025,11 @@ export class FileUseCases {
     );
 
     if (deleted > 0) {
-      this.eventEmitter.emit(
-        'usage.file.batch_deleted',
-        new UsageInvalidatedEvent(user.uuid, user.id, 'file.batch_delete'),
+      emitUsageInvalidated(
+        this.eventEmitter,
+        user.uuid,
+        user.id,
+        'file.batch_delete',
       );
     }
 

src/modules/folder/folder.usecase.ts

@@ -36,7 +36,7 @@ import { type FolderModel } from './folder.model';
 import { type MoveFolderDto } from './dto/move-folder.dto';
 import { FeatureLimitService } from '../feature-limit/feature-limit.service';
 import { EventEmitter2 } from '@nestjs/event-emitter';
-import { UsageInvalidatedEvent } from '../usage-queue/events/usage-invalidated.event';
+import { emitUsageInvalidated } from '../usage-queue/events/usage-invalidated.event';
 
 const invalidName = /[\\/]|^\s*$/;
 
@@ -529,10 +529,7 @@ export class FolderUseCases {
       ),
     ]);
 
-    this.eventEmitter.emit(
-      'usage.folder.trashed',
-      new UsageInvalidatedEvent(user.uuid, user.id, 'folder.trash'),
-    );
+    emitUsageInvalidated(this.eventEmitter, user.uuid, user.id, 'folder.trash');
   }
 
   async getFoldersByParentId(

src/modules/usage-queue/events/usage-invalidated.event.ts

@@ -1,7 +1,18 @@
-export class UsageInvalidatedEvent {
-  constructor(
-    public readonly userUuid: string,
-    public readonly userId: number,
-    public readonly source: string,
-  ) {}
+import { type EventEmitter2 } from '@nestjs/event-emitter';
+
+export interface UsageInvalidatedEvent {
+  userUuid: string;
+  userId: number;
+  source: string;
+}
+
+export const USAGE_INVALIDATED_EVENT = 'usage.invalidated';
+
+export function emitUsageInvalidated(
+  eventEmitter: EventEmitter2,
+  userUuid: string,
+  userId: number,
+  source: string,
+): void {
+  eventEmitter.emit(USAGE_INVALIDATED_EVENT, { userUuid, userId, source });
 }

src/modules/usage-queue/handlers/usage-event.handler.ts

@@ -2,9 +2,14 @@ import { Injectable, Logger } from '@nestjs/common';
 import { OnEvent } from '@nestjs/event-emitter';
 import { InjectQueue } from '@nestjs/bullmq';
 import { type Queue } from 'bullmq';
-import { type UsageInvalidatedEvent } from '../events/usage-invalidated.event';
+import {
+  type UsageInvalidatedEvent,
+  USAGE_INVALIDATED_EVENT,
+} from '../events/usage-invalidated.event';
 import { USAGE_QUEUE_NAME } from '../usage-queue.constants';
 
+const DEDUPLICATION_WINDOW_MS = 3000;
+
 @Injectable()
 export class UsageEventHandler {
   private readonly logger = new Logger(UsageEventHandler.name);
@@ -13,7 +18,7 @@ export class UsageEventHandler {
     @InjectQueue(USAGE_QUEUE_NAME) private readonly usageQueue: Queue,
   ) {}
 
-  @OnEvent('usage.**')
+  @OnEvent(USAGE_INVALIDATED_EVENT)
   async handleUsageInvalidated(event: UsageInvalidatedEvent) {
     try {
       await this.usageQueue.add(
@@ -24,8 +29,12 @@ export class UsageEventHandler {
           source: event.source,
         },
         {
-          jobId: event.userUuid,
-          delay: 3000,
+          deduplication: {
+            id: event.userUuid,
+            ttl: DEDUPLICATION_WINDOW_MS,
+            extend: true,
+          },
+          delay: DEDUPLICATION_WINDOW_MS,
         },
       );
     } catch (error) {

src/modules/user/user.usecase.spec.ts

@@ -87,6 +87,7 @@ import { SharingInvite } from '../sharing/sharing.domain';
 import { aes } from '@internxt/lib';
 import { WorkspacesUsecases } from '../workspaces/workspaces.usecase';
 import { PaymentsService } from '../../externals/payments/payments.service';
+import { EventEmitter2 } from '@nestjs/event-emitter';
 import * as jwtLibrary from '../../lib/jwt';
 import { JsonWebTokenError } from 'jsonwebtoken';
 import { type LegacyRecoverAccountDto } from './dto/legacy-recover-account.dto';
@@ -141,6 +142,7 @@ describe('User use cases', () => {
   let referralsRepository: SequelizeReferralRepository;
   let userReferralsRepository: SequelizeUserReferralsRepository;
   let notificationService: NotificationService;
+  let eventEmitter: EventEmitter2;
 
   const user = User.build({
     id: 1,
@@ -246,6 +248,7 @@ describe('User use cases', () => {
     );
     notificationService =
       moduleRef.get<NotificationService>(NotificationService);
+    eventEmitter = moduleRef.get<EventEmitter2>(EventEmitter2);
   });
 
   describe('Resetting a user', () => {
@@ -2266,17 +2269,14 @@ describe('User use cases', () => {
       expect(result).toEqual(cachedUsage);
     });
 
-    it('When cache does not have user usage data, then it should get data from database and cache it', async () => {
+    it('When cache does not have user usage data, then it should compute from database and emit event for cache fill', async () => {
       const driveUsage = 2048;
       const backupUsage = 1024;
       const totalUsage = driveUsage + backupUsage;
       jest.spyOn(cacheManagerService, 'getUserUsage').mockResolvedValue(null);
       jest
         .spyOn(fileUseCases, 'getUserUsedStorage')
         .mockResolvedValue(driveUsage);
-      jest
-        .spyOn(cacheManagerService, 'setUserUsage')
-        .mockResolvedValue(undefined);
       jest
         .spyOn(backupUseCases, 'sumExistentBackupSizes')
         .mockResolvedValue(backupUsage);
@@ -2288,10 +2288,13 @@ describe('User use cases', () => {
       expect(backupUseCases.sumExistentBackupSizes).toHaveBeenCalledWith(
         user.id,
       );
-      expect(cacheManagerService.setUserUsage).toHaveBeenCalledWith(
-        user.uuid,
-        driveUsage,
-        backupUsage,
+      expect(eventEmitter.emit).toHaveBeenCalledWith(
+        'usage.cache_miss',
+        expect.objectContaining({
+          userUuid: user.uuid,
+          userId: user.id,
+          source: 'cache_miss',
+        }),
       );
       expect(result).toEqual({
         drive: driveUsage,

src/modules/user/user.usecase.ts

@@ -12,6 +12,8 @@ import {
   forwardRef,
 } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
+import { EventEmitter2 } from '@nestjs/event-emitter';
+import { emitUsageInvalidated } from '../usage-queue/events/usage-invalidated.event';
 import { v4, validate } from 'uuid';
 import { generateMnemonic } from 'bip39';
 import * as speakeasy from 'speakeasy';
@@ -167,6 +169,7 @@ export class UserUseCases {
     private readonly backupUseCases: BackupUseCase,
     private readonly cacheManager: CacheManagerService,
     private readonly asymmetricEncryptionService: AsymmetricEncryptionService,
+    private readonly eventEmitter: EventEmitter2,
   ) {}
 
   async getCachedAvatar(user: User): Promise<string | null> {
@@ -1890,7 +1893,9 @@ export class UserUseCases {
       this.backupUseCases.sumExistentBackupSizes(user.id),
     ]);
 
-    await this.cacheManager.setUserUsage(user.uuid, driveUsage, backupUsage);
+    // Let the queue processor be the sole cache writer to avoid race
+    // conditions where this read-path write overwrites a fresher value.
+    emitUsageInvalidated(this.eventEmitter, user.uuid, user.id, 'cache_miss');
 
     return {
       drive: driveUsage,