refactor: update user commands and repository methods to use tId

Author: kastov

src/modules/users/builders/trigger-threshold-notifications-builder/trigger-threshold-notifications-builder.builder.ts

@@ -15,7 +15,7 @@ export class TriggerThresholdNotificationsBuilder {
       WITH thresholds(pct) AS (VALUES ${pctValues}),
       candidates AS (
         SELECT
-          u."uuid",
+          u."t_id",
           MIN(u."created_at") AS created_at_for_order,
           MAX(t.pct)          AS new_threshold
         FROM "users" u
@@ -26,16 +26,16 @@ export class TriggerThresholdNotificationsBuilder {
          AND u."traffic_limit_bytes" > 0
          AND ut."used_traffic_bytes" >= (u."traffic_limit_bytes" * t.pct / 100)
          AND u."last_triggered_threshold" < t.pct
-        GROUP BY u."uuid"
+        GROUP BY u."t_id"
         ORDER BY created_at_for_order
         LIMIT 5000
       )
       UPDATE "users" AS u
       SET "last_triggered_threshold" = c.new_threshold,
           "updated_at"               = NOW()
       FROM candidates c
-      WHERE u."uuid" = c."uuid"
-      RETURNING u."uuid";
+      WHERE u."t_id" = c."t_id"
+      RETURNING u."t_id" AS "tId";
     `;
     }
 }

src/modules/users/commands/trigger-threshold-notification/trigger-threshold-notification.handler.ts

@@ -13,7 +13,7 @@ import { UsersRepository } from '../../repositories/users.repository';
 @CommandHandler(TriggerThresholdNotificationCommand)
 export class TriggerThresholdNotificationHandler
     implements
-        ICommandHandler<TriggerThresholdNotificationCommand, ICommandResponse<{ uuid: string }[]>>
+        ICommandHandler<TriggerThresholdNotificationCommand, ICommandResponse<{ tId: bigint }[]>>
 {
     public readonly logger = new Logger(TriggerThresholdNotificationHandler.name);
 
@@ -25,7 +25,7 @@ export class TriggerThresholdNotificationHandler
     })
     async execute(
         command: TriggerThresholdNotificationCommand,
-    ): Promise<ICommandResponse<{ uuid: string }[]>> {
+    ): Promise<ICommandResponse<{ tId: bigint }[]>> {
         try {
             const result = await this.usersRepository.triggerThresholdNotifications(
                 command.percentages,

src/modules/users/commands/update-exceeded-users/update-exceeded-users.handler.ts

@@ -13,7 +13,7 @@ import { UsersRepository } from '../../repositories/users.repository';
 @CommandHandler(UpdateExceededTrafficUsersCommand)
 export class UpdateExceededTrafficUsersHandler
     implements
-        ICommandHandler<UpdateExceededTrafficUsersCommand, ICommandResponse<{ uuid: string }[]>>
+        ICommandHandler<UpdateExceededTrafficUsersCommand, ICommandResponse<{ tId: bigint }[]>>
 {
     public readonly logger = new Logger(UpdateExceededTrafficUsersHandler.name);
 
@@ -23,7 +23,7 @@ export class UpdateExceededTrafficUsersHandler
         maxWait: 20_000,
         timeout: 120_000,
     })
-    async execute(): Promise<ICommandResponse<{ uuid: string }[]>> {
+    async execute(): Promise<ICommandResponse<{ tId: bigint }[]>> {
         try {
             const result = await this.usersRepository.updateExceededTrafficUsers();
 

src/modules/users/commands/update-expired-users/update-expired-users.handler.ts

@@ -12,7 +12,7 @@ import { UsersRepository } from '../../repositories/users.repository';
 
 @CommandHandler(UpdateExpiredUsersCommand)
 export class UpdateExpiredUsersHandler
-    implements ICommandHandler<UpdateExpiredUsersCommand, ICommandResponse<{ uuid: string }[]>>
+    implements ICommandHandler<UpdateExpiredUsersCommand, ICommandResponse<{ tId: bigint }[]>>
 {
     public readonly logger = new Logger(UpdateExpiredUsersHandler.name);
 
@@ -22,7 +22,7 @@ export class UpdateExpiredUsersHandler
         maxWait: 20_000,
         timeout: 120_000,
     })
-    async execute(): Promise<ICommandResponse<{ uuid: string }[]>> {
+    async execute(): Promise<ICommandResponse<{ tId: bigint }[]>> {
         try {
             const result = await this.usersRepository.updateExpiredUsers();
 

src/modules/users/repositories/users.repository.ts

@@ -83,9 +83,9 @@ export class UsersRepository {
         return result;
     }
 
-    public async triggerThresholdNotifications(percentages: number[]): Promise<{ uuid: string }[]> {
+    public async triggerThresholdNotifications(percentages: number[]): Promise<{ tId: bigint }[]> {
         const { query } = new TriggerThresholdNotificationsBuilder(percentages);
-        return await this.prisma.tx.$queryRaw<{ uuid: string }[]>(query);
+        return await this.prisma.tx.$queryRaw<{ tId: bigint }[]>(query);
     }
 
     public async updateStatusAndTrafficAndResetAt(
@@ -121,7 +121,7 @@ export class UsersRepository {
             .execute();
     }
 
-    public async updateExceededTrafficUsers(): Promise<{ uuid: string }[]> {
+    public async updateExceededTrafficUsers(): Promise<{ tId: bigint }[]> {
         const result = await this.qb.kysely
             .updateTable('users')
             .set({ status: USERS_STATUS.LIMITED })
@@ -130,7 +130,7 @@ export class UsersRepository {
             .where('users.status', '=', USERS_STATUS.ACTIVE)
             .where('users.trafficLimitBytes', '!=', 0n)
             .whereRef('userTraffic.usedTrafficBytes', '>=', 'users.trafficLimitBytes')
-            .returning(['users.uuid'])
+            .returning(['users.tId'])
             .execute();
 
         return result;
@@ -148,11 +148,11 @@ export class UsersRepository {
         return result.map((value) => new UserEntity(value));
     }
 
-    public async updateExpiredUsers(): Promise<{ uuid: string }[]> {
+    public async updateExpiredUsers(): Promise<{ tId: bigint }[]> {
         // UPDATE "public"."users" SET "status" = $1, "updated_at" = $2 WHERE ("public"."users"."status" IN ($3,$4) AND "public"."users"."expire_at" < $5) RETURNING "public"."users"."uuid"
         const result = await this.prisma.tx.users.updateManyAndReturn({
             select: {
-                uuid: true,
+                tId: true,
             },
             where: {
                 AND: [

src/queue/user-jobs/user-jobs.processor.ts

@@ -87,12 +87,12 @@ export class UserJobsQueueProcessor extends WorkerHost {
                 `Job ${job.name} Found ${usersResponse.response.length} expired users.`,
             );
 
-            await pMap(usersResponse.response, async (userUuid) => {
+            await pMap(usersResponse.response, async (userIterator) => {
                 try {
                     const userResponse = await this.queryBus.execute(
                         new GetUserByUniqueFieldQuery(
                             {
-                                uuid: userUuid.uuid,
+                                tId: userIterator.tId,
                             },
                             {
                                 activeInternalSquads: true,
@@ -161,11 +161,11 @@ export class UserJobsQueueProcessor extends WorkerHost {
                 `Job ${job.name} has found ${usersResponse.response.length} exceeded traffic usage users.`,
             );
 
-            await pMap(usersResponse.response, async (userUuid) => {
+            await pMap(usersResponse.response, async (userIterator) => {
                 try {
                     const userResponse = await this.queryBus.execute(
                         new GetUserByUniqueFieldQuery({
-                            uuid: userUuid.uuid,
+                            tId: userIterator.tId,
                         }),
                     );
 
@@ -246,11 +246,11 @@ export class UserJobsQueueProcessor extends WorkerHost {
 
                 await pMap(
                     usersResponse.response,
-                    async (userUuid) => {
+                    async (userIterator) => {
                         try {
                             const userResponse = await this.queryBus.execute(
                                 new GetUserByUniqueFieldQuery({
-                                    uuid: userUuid.uuid,
+                                    tId: userIterator.tId,
                                 }),
                             );
 
@@ -275,7 +275,7 @@ export class UserJobsQueueProcessor extends WorkerHost {
                             );
                         }
                     },
-                    { concurrency: 100 },
+                    { concurrency: 80 },
                 );
 
                 continue;
@@ -358,26 +358,26 @@ export class UserJobsQueueProcessor extends WorkerHost {
         }
     }
 
-    private async updateExpiredUsers(): Promise<ICommandResponse<{ uuid: string }[]>> {
+    private async updateExpiredUsers(): Promise<ICommandResponse<{ tId: bigint }[]>> {
         return this.commandBus.execute<
             UpdateExpiredUsersCommand,
-            ICommandResponse<{ uuid: string }[]>
+            ICommandResponse<{ tId: bigint }[]>
         >(new UpdateExpiredUsersCommand());
     }
 
-    private async updateExceededTrafficUsers(): Promise<ICommandResponse<{ uuid: string }[]>> {
+    private async updateExceededTrafficUsers(): Promise<ICommandResponse<{ tId: bigint }[]>> {
         return this.commandBus.execute<
             UpdateExceededTrafficUsersCommand,
-            ICommandResponse<{ uuid: string }[]>
+            ICommandResponse<{ tId: bigint }[]>
         >(new UpdateExceededTrafficUsersCommand());
     }
 
     private async triggerThresholdNotifications(
         percentages: number[],
-    ): Promise<ICommandResponse<{ uuid: string }[]>> {
+    ): Promise<ICommandResponse<{ tId: bigint }[]>> {
         return this.commandBus.execute<
             TriggerThresholdNotificationCommand,
-            ICommandResponse<{ uuid: string }[]>
+            ICommandResponse<{ tId: bigint }[]>
         >(new TriggerThresholdNotificationCommand(percentages));
     }
 }