fix performance

src/services/rabbitmq.ts

@@ -24,7 +24,7 @@ import { In, Not } from "typeorm";
 import { PosMasterAct } from "../entities/PosMasterAct";
 import { PermissionOrg } from "../entities/PermissionOrg";
 import { sendWebSocket } from "./webSocket";
-import { CreatePosMasterHistoryOfficer } from "./PositionService";
+import { CreatePosMasterHistoryOfficer, BatchUpdatePosMasters, BatchCreatePosMasterHistoryOfficer, BatchHistoryOperation } from "./PositionService";
 import { PayloadSendNoti } from "../interfaces/utils";
 import { PermissionProfile } from "../entities/PermissionProfile";
 
@@ -584,20 +584,38 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
 
   try {
     console.time('[AMQ] query_posMaster');
-    const posMaster = await repoPosmaster.find({
-      where: { orgRevisionId: id },
-      relations: [
-        "orgRoot",
-        "orgChild4",
-        "orgChild3",
-        "orgChild2",
-        "orgChild1",
-        "positions",
-        "positions.posLevel",
-        "positions.posType",
-        "positions.posExecutive",
-      ],
-    });
+    const POS_MASTER_PAGE_SIZE = 2000;
+    let totalPosMastersProcessed = 0;
+    let hasMoreRecords = true;
+    let skip = 0;
+    const posMaster: PosMaster[] = [];
+
+    while (hasMoreRecords) {
+      const posMasterPage = await repoPosmaster.find({
+        where: { orgRevisionId: id },
+        relations: [
+          "orgRoot",
+          "orgChild4",
+          "orgChild3",
+          "orgChild2",
+          "orgChild1",
+          "positions",
+          "positions.posLevel",
+          "positions.posType",
+          "positions.posExecutive",
+        ],
+        order: { id: 'ASC' },
+        skip: skip,
+        take: POS_MASTER_PAGE_SIZE,
+      });
+
+      posMaster.push(...posMasterPage);
+      totalPosMastersProcessed += posMasterPage.length;
+      hasMoreRecords = posMasterPage.length === POS_MASTER_PAGE_SIZE;
+      skip += POS_MASTER_PAGE_SIZE;
+
+      console.log(`[AMQ] Loaded posMaster page: ${totalPosMastersProcessed} records`);
+    }
     console.timeEnd('[AMQ] query_posMaster');
     console.log(`[AMQ] posMaster count: ${posMaster.length}`);
 
@@ -802,22 +820,44 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
 
     // 6. Batch update posMasters
     console.time('[AMQ] batch_update_posMasters');
-    for (const update of posMasterUpdates) {
-      await repoPosmaster.update(update.id, {
-        current_holderId: update.current_holderId,
-        next_holderId: null,
-        lastUpdateUserId,
-        lastUpdateFullName,
-        lastUpdatedAt,
-      });
-    }
+
+    const posMasterUpdatesForBatch = posMasterUpdates.map((u: any) => ({
+      id: u.id,
+      current_holderId: u.current_holderId ?? null,
+      lastUpdateUserId,
+      lastUpdateFullName,
+      lastUpdatedAt,
+    }));
+
+    await BatchUpdatePosMasters(
+      AppDataSource.manager,
+      posMasterUpdatesForBatch
+    );
+
     console.timeEnd('[AMQ] batch_update_posMasters');
 
     // 7. Batch create history
     console.time('[AMQ] batch_create_history');
+
+    const historyOperations: BatchHistoryOperation[] = [];
     for (const id of historyCreateIds) {
-      await CreatePosMasterHistoryOfficer(id, null);
+      const pm = posMaster.find(p => p.id === id);
+      if (pm) {
+        historyOperations.push({
+          posMasterId: id,
+          posMasterData: pm,
+          orgRevisionId: pm.orgRevisionId,
+          lastUpdateUserId,
+          lastUpdateFullName,
+        });
+      }
     }
+
+    await BatchCreatePosMasterHistoryOfficer(
+      AppDataSource.manager,
+      historyOperations
+    );
+
     console.timeEnd('[AMQ] batch_create_history');
 
     // Clone oldposMasterAct