refactor: update user usage handling and query structure

Author: kastov

src/modules/users/builders/bulk-update-user-used-traffic/bulk-update-user-used-traffic.builder.ts

@@ -10,26 +10,27 @@ export class BulkUpdateUserUsedTrafficBuilder {
 
     public getQuery(userUsageList: { u: string; b: string; n: string }[]): Prisma.Sql {
         const query = `
-            WITH updated_users AS ( UPDATE "users" AS u
-                SET
-                    "used_traffic_bytes"          = u."used_traffic_bytes" + data."inc_used",
-                    "lifetime_used_traffic_bytes" = u."lifetime_used_traffic_bytes" + data."inc_used",
-                    "online_at"                   = NOW(),
-                    "first_connected_at"          = COALESCE(u."first_connected_at", NOW()),
-                    "updated_at"                  = NOW(),
-                    "last_connected_node_uuid"   = data."last_connected_node_uuid"
-            FROM (
-                VALUES ${userUsageList.map((usageHistory) => `(${usageHistory.b}::bigint, '${usageHistory.u}'::uuid, '${usageHistory.n}'::uuid)`).join(',')}
-                ) AS data("inc_used", "uuid", "last_connected_node_uuid")
-            WHERE data."uuid" = u."uuid"
-            RETURNING
-                u."uuid",
-                (u."first_connected_at" = u."online_at") AS "isFirstConnection"
-            )
-            SELECT uuid
-            FROM updated_users
-            WHERE "isFirstConnection";
-        `;
+        WITH sorted_data AS (
+            SELECT * FROM (
+                VALUES ${userUsageList.map((h) => `(${h.b}, ${h.u}, '${h.n}'::uuid)`).join(',')}
+            ) AS data("inc_used", "t_id", "last_connected_node_uuid")
+            ORDER BY "t_id"
+        ),
+        updated_users AS (
+            UPDATE "users" AS u
+            SET
+                "used_traffic_bytes" = u."used_traffic_bytes" + sorted_data."inc_used",
+                "lifetime_used_traffic_bytes" = u."lifetime_used_traffic_bytes" + sorted_data."inc_used",
+                "online_at" = NOW(),
+                "first_connected_at" = COALESCE(u."first_connected_at", NOW()),
+                "updated_at" = NOW(),
+                "last_connected_node_uuid" = sorted_data."last_connected_node_uuid"
+            FROM sorted_data
+            WHERE sorted_data."t_id" = u."t_id"
+            RETURNING u."uuid", (u."first_connected_at" = u."online_at") AS "isFirstConnection"
+        )
+        SELECT uuid FROM updated_users WHERE "isFirstConnection";
+    `;
         return Prisma.raw(query);
     }
 }

src/modules/users/queries/get-uuid-by-username/get-uuid-by-username.handler.ts

@@ -1,20 +1,17 @@
 import { IQueryHandler, QueryHandler } from '@nestjs/cqrs';
 import { Logger } from '@nestjs/common';
 
-import { ICommandResponse } from '@common/types/command-response.type';
 import { ERRORS } from '@libs/contracts/constants';
 
 import { GetUuidByUsernameQuery } from './get-uuid-by-username.query';
 import { UsersRepository } from '../../repositories/users.repository';
 
 @QueryHandler(GetUuidByUsernameQuery)
-export class GetUuidByUsernameHandler
-    implements IQueryHandler<GetUuidByUsernameQuery, ICommandResponse<string | null>>
-{
+export class GetUuidByUsernameHandler implements IQueryHandler<GetUuidByUsernameQuery> {
     private readonly logger = new Logger(GetUuidByUsernameHandler.name);
     constructor(private readonly usersRepository: UsersRepository) {}
 
-    async execute(query: GetUuidByUsernameQuery): Promise<ICommandResponse<string | null>> {
+    async execute(query: GetUuidByUsernameQuery) {
         try {
             const user = await this.usersRepository.getUserUuidByUsername(query.username);
 

src/modules/users/queries/get-uuid-by-username/get-uuid-by-username.query.ts

@@ -2,7 +2,12 @@ import { Query } from '@nestjs/cqrs';
 
 import { ICommandResponse } from '@common/types/command-response.type';
 
-export class GetUuidByUsernameQuery extends Query<ICommandResponse<string | null>> {
+export class GetUuidByUsernameQuery extends Query<
+    ICommandResponse<{
+        uuid: string;
+        tId: bigint;
+    } | null>
+> {
     constructor(public readonly username: string) {
         super();
     }

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

@@ -1214,18 +1214,21 @@ export class UsersRepository implements ICrud<BaseUserEntity> {
         ).as('activeInternalSquads');
     }
 
-    public async getUserUuidByUsername(username: string): Promise<string | null> {
+    public async getUserUuidByUsername(
+        username: string,
+    ): Promise<{ uuid: string; tId: bigint } | null> {
         const result = await this.qb.kysely
             .selectFrom('users')
-            .select('uuid')
+            .select(['uuid'])
+            .select(sql.ref<bigint>('t_id').as('tId'))
             .where('username', '=', username)
             .executeTakeFirst();
 
         if (!result) {
             return null;
         }
 
-        return result.uuid;
+        return { uuid: result.uuid, tId: result.tId };
     }
 
     public async findNotConnectedUsers(startDate: Date, endDate: Date): Promise<UserEntity[]> {

src/queue/record-user-usage/record-user-usage.processor.ts

@@ -24,7 +24,7 @@ import { RecordUserUsageJobNames } from './enums';
 import { QueueNames } from '../queue.enum';
 
 @Processor(QueueNames.recordUserUsage, {
-    concurrency: 100,
+    concurrency: 50,
 })
 export class RecordUserUsageQueueProcessor extends WorkerHost {
     private readonly logger = new Logger(RecordUserUsageQueueProcessor.name);
@@ -109,20 +109,20 @@ export class RecordUserUsageQueueProcessor extends WorkerHost {
 
                     const totalBytes = xrayUser.downlink + xrayUser.uplink;
 
-                    const userUuid = userResponse.response;
+                    const { uuid, tId } = userResponse.response;
 
                     allUsageRecords.push(
                         new NodesUserUsageHistoryEntity({
                             nodeUuid,
-                            userUuid,
+                            userUuid: uuid,
                             totalBytes: BigInt(totalBytes),
                             uploadBytes: BigInt(xrayUser.uplink),
                             downloadBytes: BigInt(xrayUser.downlink),
                         }),
                     );
 
                     userUsageList.push({
-                        u: userUuid,
+                        u: tId.toString(),
                         b: this.multiplyConsumption(consumptionMultiplier, totalBytes).toString(),
                         n: nodeUuid,
                     });

src/queue/update-users-usage/update-users-usage.processor.ts

@@ -14,7 +14,7 @@ import { UpdateUsersUsageJobNames } from './enums';
 import { QueueNames } from '../queue.enum';
 
 @Processor(QueueNames.updateUsersUsage, {
-    concurrency: 1,
+    concurrency: 10,
 })
 export class UpdateUsersUsageQueueProcessor extends WorkerHost {
     private readonly logger = new Logger(UpdateUsersUsageQueueProcessor.name);

src/queue/update-users-usage/update-users-usage.service.ts

@@ -34,7 +34,7 @@ export class UpdateUsersUsageQueueService
         await this.checkConnection();
     }
 
-    public async updateUserUsage(payload: Array<{ u: string; b: string; n: string }>) {
+    public async updateUserUsage(payload: { u: string; b: string; n: string }[]) {
         const chunks = this.chunks(payload, 1500);
         for await (const chunk of chunks) {
             await this.addJob(UpdateUsersUsageJobNames.UpdateUsersUsage, chunk, {