From b5c75379ffd44ffe6ce6002e5a9a1337bb3600ce Mon Sep 17 00:00:00 2001 From: waruneeauy Date: Tue, 5 May 2026 15:59:39 +0700 Subject: [PATCH 1/5] fixed error and not retry --- src/services/rabbitmq.ts | 361 +++++++++++++++++++++++++++++---------- 1 file changed, 273 insertions(+), 88 deletions(-) diff --git a/src/services/rabbitmq.ts b/src/services/rabbitmq.ts index 42c386e0..097473f8 100644 --- a/src/services/rabbitmq.ts +++ b/src/services/rabbitmq.ts @@ -24,9 +24,27 @@ import { In, Not } from "typeorm"; import { PosMasterAct } from "../entities/PosMasterAct"; import { PermissionOrg } from "../entities/PermissionOrg"; import { sendWebSocket } from "./webSocket"; -import { CreatePosMasterHistoryOfficer } from "./PositionService"; import { PayloadSendNoti } from "../interfaces/utils"; import { PermissionProfile } from "../entities/PermissionProfile"; +import { PosMasterHistory } from "../entities/PosMasterHistory"; + +let reconnectTimer: ReturnType | null = null; + +function scheduleReconnect() { + if (reconnectTimer) { + return; + } + + reconnectTimer = setTimeout(async () => { + reconnectTimer = null; + try { + await init(); + } catch (error) { + console.error("[AMQ] Reconnect failed:", error); + scheduleReconnect(); + } + }, 1000); +} export let sendToQueue: (payload: any) => void; export let sendToQueueOrg: (payload: any) => void; @@ -55,10 +73,28 @@ export async function init() { console.log(connection ? "[AMQ] Connection success" : "[AMQ] Connection failed"); + connection.on("error", (error) => { + console.error("[AMQ] Connection error:", error); + }); + + connection.on("close", () => { + console.error("[AMQ] Connection closed. Scheduling reconnect..."); + scheduleReconnect(); + }); + const channel = await connection.createChannel(); //----> (1.4) create Channel console.log(channel ? "[AMQ] Create channel success" : "[AMQ] Create channel failed"); + channel.on("error", (error) => { + console.error("[AMQ] Channel error:", error); + }); + + channel.on("close", () => { + console.error("[AMQ] Channel closed. Scheduling reconnect..."); + scheduleReconnect(); + }); + channel.assertQueue(queue, { durable: true }), //----> (1.5) assert queue and set durable (if "true" save to disk on RabbitMQ) channel.assertQueue(queue_org, { durable: true }), channel.assertQueue(queue_org_draft, { durable: true }), @@ -97,18 +133,26 @@ function createConsumer( //----> consumer channel: amqp.Channel, handler: (msg: amqp.ConsumeMessage) => Promise | boolean, ) { - let retries = 0; channel.consume( queue, async (msg) => { if (!msg) return; - if ((await handler(msg)) || retries++ >= 3) { - retries = 0; - console.log("[AMQ] Process Consumer success"); + try { + if (await handler(msg)) { + console.log("[AMQ] Process Consumer success"); + return channel.ack(msg); + } + console.error(`[AMQ] Process Consumer failed on queue ${queue}, acknowledging without retry`); return channel.ack(msg); + } catch (error) { + console.error(`[AMQ] Consumer processing error on queue ${queue}:`, error); + try { + console.error(`[AMQ] Acknowledging failed message on queue ${queue} without retry`); + channel.ack(msg); + } catch (channelError) { + console.error(`[AMQ] Failed to ack/nack message on queue ${queue}:`, channelError); + } } - console.log("[AMQ] Process Consumer failed"); - return await new Promise((resolve) => setTimeout(() => resolve(channel.nack(msg)), 3000)); }, { noAck: false }, ); @@ -740,7 +784,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { // 3. เตรียม arrays สำหรับ batch operations const profilesToSave: Profile[] = []; const posMasterAssignsToSave: PosMasterAssign[] = []; - const historyCreateIds: string[] = []; + const historyRowsToSave: Partial[] = []; const posMasterUpdates: { id: string; current_holderId: string | null | undefined }[] = []; // ===== LOOP: เก็บข้อมูลทั้งหมด ===== @@ -809,12 +853,54 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { const isHolderChanged = oldHolderId !== newHolderId; if (isHolderChanged) { - historyCreateIds.push(item.id); + const nextHolderProfile = + item.next_holderId != null && item.next_holderId !== "" + ? profilesMap.get(item.next_holderId) + : null; + const selectedPosition = + item.positions.length > 0 + ? item.positions.find((position) => position.positionIsSelected === true) ?? null + : null; + const shortName = + [ + item.orgChild4?.orgChild4ShortName, + item.orgChild3?.orgChild3ShortName, + item.orgChild2?.orgChild2ShortName, + item.orgChild1?.orgChild1ShortName, + item.orgRoot?.orgRootShortName, + ].find((name) => typeof name === "string" && name.trim().length > 0) ?? _null; + + historyRowsToSave.push({ + ancestorDNA: item.ancestorDNA ? item.ancestorDNA : _null, + prefix: nextHolderProfile?.prefix || _null, + firstName: nextHolderProfile?.firstName || _null, + lastName: nextHolderProfile?.lastName || _null, + shortName, + posMasterNoPrefix: item.posMasterNoPrefix ?? _null, + posMasterNo: item.posMasterNo ?? _null, + posMasterNoSuffix: item.posMasterNoSuffix ?? _null, + position: selectedPosition?.positionName ?? _null, + posType: selectedPosition?.posType?.posTypeName ?? _null, + posLevel: selectedPosition?.posLevel?.posLevelName ?? _null, + posExecutive: selectedPosition?.posExecutive?.posExecutiveName ?? _null, + profileId: _null, + rootDnaId: item.orgRoot?.ancestorDNA || _null, + child1DnaId: item.orgChild1?.ancestorDNA || _null, + child2DnaId: item.orgChild2?.ancestorDNA || _null, + child3DnaId: item.orgChild3?.ancestorDNA || _null, + child4DnaId: item.orgChild4?.ancestorDNA || _null, + createdUserId: "", + createdFullName: "system", + lastUpdateUserId: "", + lastUpdateFullName: "system", + createdAt: new Date(), + lastUpdatedAt: new Date(), + }); } } console.timeEnd("[AMQ] prepare_batch_data"); console.log( - `[AMQ] Prepared - posMasterAssignsToSave: ${posMasterAssignsToSave.length}, profilesToSave: ${profilesToSave.length}, posMasterUpdates: ${posMasterUpdates.length}, historyCreateIds: ${historyCreateIds.length}`, + `[AMQ] Prepared - posMasterAssignsToSave: ${posMasterAssignsToSave.length}, profilesToSave: ${profilesToSave.length}, posMasterUpdates: ${posMasterUpdates.length}, historyCreateIds: ${historyRowsToSave.length}`, ); // ===== BATCH EXECUTION: save ทีละ batch ===== @@ -835,6 +921,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { const child2Repository = manager.getRepository(OrgChild2); const child3Repository = manager.getRepository(OrgChild3); const child4Repository = manager.getRepository(OrgChild4); + const posMasterHistoryRepository = manager.getRepository(PosMasterHistory); const targetOrgRevision = await repoOrgRevision .createQueryBuilder("orgRevision") @@ -903,21 +990,45 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { // 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, - }); + if (posMasterUpdates.length > 0) { + const chunks = chunkArray(posMasterUpdates, 500); + const posMasterTableName = repoPosmaster.metadata.tableName; + for (const chunk of chunks as typeof posMasterUpdates[]) { + const caseClauses = chunk.map(() => "WHEN ? THEN ?").join(" "); + const wherePlaceholders = chunk.map(() => "?").join(", "); + const params = chunk.flatMap((update: (typeof posMasterUpdates)[number]) => [ + update.id, + update.current_holderId ?? null, + ]); + + params.push( + lastUpdateUserId, + lastUpdateFullName, + lastUpdatedAt, + ...chunk.map((update: (typeof posMasterUpdates)[number]) => update.id), + ); + + await manager.query( + `UPDATE \`${posMasterTableName}\` + SET current_holderId = CASE id ${caseClauses} END, + next_holderId = NULL, + lastUpdateUserId = ?, + lastUpdateFullName = ?, + lastUpdatedAt = ? + WHERE id IN (${wherePlaceholders})`, + params, + ); + } } console.timeEnd("[AMQ] batch_update_posMasters"); // 7. Batch create history console.time("[AMQ] batch_create_history"); - for (const id of historyCreateIds) { - await CreatePosMasterHistoryOfficer(id, null, undefined, undefined, manager); + if (historyRowsToSave.length > 0) { + const chunks = chunkArray(historyRowsToSave, 500); + for (const chunk of chunks) { + await posMasterHistoryRepository.save(posMasterHistoryRepository.create(chunk)); + } } console.timeEnd("[AMQ] batch_create_history"); @@ -958,25 +1069,23 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { //cone tree console.time("[AMQ] query_old_org_structure"); //หา dna tree - const orgRoot = await orgRootRepository.find({ - where: { orgRevisionId: orgRevisionPublish.id }, - }); - - const orgChild1 = await child1Repository.find({ - where: { orgRevisionId: orgRevisionPublish.id }, - }); - - const orgChild2 = await child2Repository.find({ - where: { orgRevisionId: orgRevisionPublish.id }, - }); - - const orgChild3 = await child3Repository.find({ - where: { orgRevisionId: orgRevisionPublish.id }, - }); - - const orgChild4 = await child4Repository.find({ - where: { orgRevisionId: orgRevisionPublish.id }, - }); + const [orgRoot, orgChild1, orgChild2, orgChild3, orgChild4] = await Promise.all([ + orgRootRepository.find({ + where: { orgRevisionId: orgRevisionPublish.id }, + }), + child1Repository.find({ + where: { orgRevisionId: orgRevisionPublish.id }, + }), + child2Repository.find({ + where: { orgRevisionId: orgRevisionPublish.id }, + }), + child3Repository.find({ + where: { orgRevisionId: orgRevisionPublish.id }, + }), + child4Repository.find({ + where: { orgRevisionId: orgRevisionPublish.id }, + }), + ]); console.timeEnd("[AMQ] query_old_org_structure"); console.log( `[AMQ] Old structure - orgRoot: ${orgRoot.length}, orgChild1: ${orgChild1.length}, orgChild2: ${orgChild2.length}, orgChild3: ${orgChild3.length}, orgChild4: ${orgChild4.length}`, @@ -1106,6 +1215,78 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { ? x.id : x.ancestorDNA, })); + + const groupByParentId = ( + items: T[], + getParentId: (item: T) => string | null | undefined, + ) => { + const grouped = new Map(); + for (const item of items) { + const parentId = getParentId(item); + if (!parentId) { + continue; + } + const current = grouped.get(parentId); + if (current) { + current.push(item); + } else { + grouped.set(parentId, [item]); + } + } + return grouped; + }; + + const buildEmployeeNodeKey = (item: { + orgRootId?: string | null; + orgChild1Id?: string | null; + orgChild2Id?: string | null; + orgChild3Id?: string | null; + orgChild4Id?: string | null; + }) => { + if (item.orgChild4Id) return `child4:${item.orgChild4Id}`; + if (item.orgChild3Id && item.orgChild4Id == null) return `child3:${item.orgChild3Id}`; + if (item.orgChild2Id && item.orgChild3Id == null) return `child2:${item.orgChild2Id}`; + if (item.orgChild1Id && item.orgChild2Id == null) return `child1:${item.orgChild1Id}`; + if (item.orgRootId && item.orgChild1Id == null) return `root:${item.orgRootId}`; + return null; + }; + + const groupByEmployeeNode = < + T extends { + orgRootId?: string | null; + orgChild1Id?: string | null; + orgChild2Id?: string | null; + orgChild3Id?: string | null; + orgChild4Id?: string | null; + }, + >( + items: T[], + ) => { + const grouped = new Map(); + for (const item of items) { + const key = buildEmployeeNodeKey(item); + if (!key) { + continue; + } + const current = grouped.get(key); + if (current) { + current.push(item); + } else { + grouped.set(key, [item]); + } + } + return grouped; + }; + + const employeePosMasterByNode = groupByEmployeeNode(_orgemployeePosMaster); + const employeeTempPosMasterByNode = groupByEmployeeNode(_orgemployeeTempPosMaster); + const getNodeKey = (level: "root" | "child1" | "child2" | "child3" | "child4", id: string) => + `${level}:${id}`; + const orgChild1ByRoot = groupByParentId(orgChild1, (item) => item.orgRootId); + const orgChild2ByChild1 = groupByParentId(orgChild2, (item) => item.orgChild1Id); + const orgChild3ByChild2 = groupByParentId(orgChild3, (item) => item.orgChild2Id); + const orgChild4ByChild3 = groupByParentId(orgChild4, (item) => item.orgChild3Id); + await repoEmployeeTempPosmaster .createQueryBuilder() .insert() @@ -1124,9 +1305,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { const dataId = x.id; const matchedOrgRoot = findMatchedNodeByAncestorDNA(orgRootCurrent, x); - const filteredEmployeePosMaster = _orgemployeePosMaster.filter( - (x: EmployeePosMaster) => x.orgRootId == dataId && x.orgChild1Id == null, - ); + const filteredEmployeePosMaster = employeePosMasterByNode.get(getNodeKey("root", dataId)) ?? []; await Promise.all( filteredEmployeePosMaster.map(async (item: any) => { @@ -1164,8 +1343,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { ); await Promise.all( - _orgemployeeTempPosMaster - .filter((x: EmployeeTempPosMaster) => x.orgRootId == dataId && x.orgChild1Id == null) + (employeeTempPosMasterByNode.get(getNodeKey("root", dataId)) ?? []) .map(async (item: any) => { delete item.id; const employeeTempPosMaster = Object.assign(new EmployeeTempPosMaster(), item); @@ -1200,12 +1378,11 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { }), ); - for (const x of orgChild1.filter((item: OrgChild1) => item.orgRootId == dataId)) { + for (const x of orgChild1ByRoot.get(dataId) ?? []) { const data1Id = x.id; const matchedOrgChild1 = findMatchedNodeByAncestorDNA(orgChild1Current, x); await Promise.all( - _orgemployeePosMaster - .filter((x: EmployeePosMaster) => x.orgChild1Id == data1Id && x.orgChild2Id == null) + (employeePosMasterByNode.get(getNodeKey("child1", data1Id)) ?? []) .map(async (item: any) => { delete item.id; const employeePosMaster = Object.assign(new EmployeePosMaster(), item); @@ -1242,10 +1419,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { ); await Promise.all( - _orgemployeeTempPosMaster - .filter( - (x: EmployeeTempPosMaster) => x.orgChild1Id == data1Id && x.orgChild2Id == null, - ) + (employeeTempPosMasterByNode.get(getNodeKey("child1", data1Id)) ?? []) .map(async (item: any) => { delete item.id; const employeeTempPosMaster = Object.assign(new EmployeeTempPosMaster(), item); @@ -1281,12 +1455,11 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { }), ); - for (const x of orgChild2.filter((item: OrgChild2) => item.orgChild1Id == data1Id)) { + for (const x of orgChild2ByChild1.get(data1Id) ?? []) { const data2Id = x.id; const matchedOrgChild2 = findMatchedNodeByAncestorDNA(orgChild2Current, x); await Promise.all( - _orgemployeePosMaster - .filter((x: EmployeePosMaster) => x.orgChild2Id == data2Id && x.orgChild3Id == null) + (employeePosMasterByNode.get(getNodeKey("child2", data2Id)) ?? []) .map(async (item: any) => { delete item.id; const employeePosMaster = Object.assign(new EmployeePosMaster(), item); @@ -1324,10 +1497,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { ); await Promise.all( - _orgemployeeTempPosMaster - .filter( - (x: EmployeeTempPosMaster) => x.orgChild2Id == data2Id && x.orgChild3Id == null, - ) + (employeeTempPosMasterByNode.get(getNodeKey("child2", data2Id)) ?? []) .map(async (item: any) => { delete item.id; const employeeTempPosMaster = Object.assign(new EmployeeTempPosMaster(), item); @@ -1364,14 +1534,11 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { }), ); - for (const x of orgChild3.filter((item: OrgChild3) => item.orgChild2Id == data2Id)) { + for (const x of orgChild3ByChild2.get(data2Id) ?? []) { const data3Id = x.id; const matchedOrgChild3 = findMatchedNodeByAncestorDNA(orgChild3Current, x); await Promise.all( - _orgemployeePosMaster - .filter( - (x: EmployeePosMaster) => x.orgChild3Id == data3Id && x.orgChild4Id == null, - ) + (employeePosMasterByNode.get(getNodeKey("child3", data3Id)) ?? []) .map(async (item: any) => { delete item.id; const employeePosMaster = Object.assign(new EmployeePosMaster(), item); @@ -1410,10 +1577,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { ); await Promise.all( - _orgemployeeTempPosMaster - .filter( - (x: EmployeeTempPosMaster) => x.orgChild3Id == data3Id && x.orgChild4Id == null, - ) + (employeeTempPosMasterByNode.get(getNodeKey("child3", data3Id)) ?? []) .map(async (item: any) => { delete item.id; const employeeTempPosMaster = Object.assign(new EmployeeTempPosMaster(), item); @@ -1451,12 +1615,11 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { }), ); - for (const x of orgChild4.filter((item: OrgChild4) => item.orgChild3Id == data3Id)) { + for (const x of orgChild4ByChild3.get(data3Id) ?? []) { const data4Id = x.id; const matchedOrgChild4 = findMatchedNodeByAncestorDNA(orgChild4Current, x); await Promise.all( - _orgemployeePosMaster - .filter((x: EmployeePosMaster) => x.orgChild4Id == data4Id) + (employeePosMasterByNode.get(getNodeKey("child4", data4Id)) ?? []) .map(async (item: any) => { delete item.id; const employeePosMaster = Object.assign(new EmployeePosMaster(), item); @@ -1496,8 +1659,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { ); await Promise.all( - _orgemployeeTempPosMaster - .filter((x: EmployeeTempPosMaster) => x.orgChild4Id == data4Id) + (employeeTempPosMasterByNode.get(getNodeKey("child4", data4Id)) ?? []) .map(async (item: any) => { delete item.id; const employeeTempPosMaster = Object.assign( @@ -1544,22 +1706,42 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { } } - const employeePosMaster = await repoEmployeePosmaster.find({ - where: { orgRevisionId: orgRevisionDraft.id }, - relations: ["positions", "positions.posLevel", "positions.posType"], - }); + const [employeePosMaster, employeeTempPosMaster] = await Promise.all([ + repoEmployeePosmaster.find({ + where: { orgRevisionId: orgRevisionDraft.id }, + relations: ["positions", "positions.posLevel", "positions.posType"], + }), + repoEmployeeTempPosmaster.find({ + where: { orgRevisionId: orgRevisionDraft.id }, + relations: ["positions", "positions.posLevel", "positions.posType"], + }), + ]); + const profileEmployeeIds = Array.from( + new Set( + [...employeePosMaster, ...employeeTempPosMaster] + .map((item) => item.next_holderId) + .filter((profileId): profileId is string => !!profileId), + ), + ); + const profileEmployeeMap = new Map(); + if (profileEmployeeIds.length > 0) { + const profiles = await repoProfileEmployee.findBy({ + id: In(profileEmployeeIds), + }); + profiles.forEach((profile) => profileEmployeeMap.set(profile.id, profile)); + } + const updatedProfileEmployeeIds = new Set(); + for (const item of employeePosMaster) { if (item.next_holderId != null) { - const profile = await repoProfileEmployee.findOne({ - where: { id: item.next_holderId == null ? "" : item.next_holderId }, - }); + const profile = profileEmployeeMap.get(item.next_holderId); const position = await item.positions.find((x) => x.positionIsSelected == true); const _null: any = null; if (profile != null) { profile.posLevelId = position?.posLevelId ?? _null; profile.posTypeId = position?.posTypeId ?? _null; profile.position = position?.positionName ?? _null; - await repoProfileEmployee.save(profile); + updatedProfileEmployeeIds.add(profile.id); } } item.lastUpdateUserId = lastUpdateUserId; @@ -1567,22 +1749,16 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { item.lastUpdatedAt = lastUpdatedAt; await repoEmployeePosmaster.save(item); } - const employeeTempPosMaster = await repoEmployeeTempPosmaster.find({ - where: { orgRevisionId: orgRevisionDraft.id }, - relations: ["positions", "positions.posLevel", "positions.posType"], - }); for (const item of employeeTempPosMaster) { if (item.next_holderId != null) { - const profile = await repoProfileEmployee.findOne({ - where: { id: item.next_holderId == null ? "" : item.next_holderId }, - }); + const profile = profileEmployeeMap.get(item.next_holderId); const position = await item.positions.find((x) => x.positionIsSelected == true); const _null: any = null; if (profile != null) { profile.posLevelId = position?.posLevelId ?? _null; profile.posTypeId = position?.posTypeId ?? _null; profile.position = position?.positionName ?? _null; - await repoProfileEmployee.save(profile); + updatedProfileEmployeeIds.add(profile.id); } } item.lastUpdateUserId = lastUpdateUserId; @@ -1590,6 +1766,15 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { item.lastUpdatedAt = lastUpdatedAt; await repoEmployeeTempPosmaster.save(item); } + if (updatedProfileEmployeeIds.size > 0) { + const profilesToSave = Array.from(updatedProfileEmployeeIds) + .map((profileId) => profileEmployeeMap.get(profileId)) + .filter((profile): profile is ProfileEmployee => !!profile); + const chunks = chunkArray(profilesToSave, 200); + for (const chunk of chunks) { + await repoProfileEmployee.save(chunk); + } + } console.timeEnd("[AMQ] clone_org_structure"); console.time("[AMQ] save_revision_status"); From e7e4e2075b71d7b65f271605b2b35456ee23c6b1 Mon Sep 17 00:00:00 2001 From: waruneeauy Date: Tue, 5 May 2026 16:25:06 +0700 Subject: [PATCH 2/5] =?UTF-8?q?1.=20=E0=B8=A3=E0=B8=A7=E0=B8=A1=20query=5F?= =?UTF-8?q?employeePosMaster=20=E0=B8=81=E0=B8=B1=E0=B8=9A=20query=5Femplo?= =?UTF-8?q?yeeTempPosMaster=20=E0=B9=83=E0=B8=AB=E0=B9=89=E0=B8=94?= =?UTF-8?q?=E0=B8=B6=E0=B8=87=E0=B9=81=E0=B8=9A=E0=B8=9A=E0=B8=82=E0=B8=99?= =?UTF-8?q?=E0=B8=B2=E0=B8=99=E0=B8=94=E0=B9=89=E0=B8=A7=E0=B8=A2=20Promis?= =?UTF-8?q?e.all=202.=20=E0=B8=95=E0=B8=B1=E0=B8=94=20full-table=20scan=20?= =?UTF-8?q?=E0=B8=82=E0=B8=AD=E0=B8=87=20ProfileEmployee=20=E0=B8=AD?= =?UTF-8?q?=E0=B8=AD=E0=B8=81=20=E0=B9=82=E0=B8=94=E0=B8=A2=E0=B9=80?= =?UTF-8?q?=E0=B8=9B=E0=B8=A5=E0=B8=B5=E0=B9=88=E0=B8=A2=E0=B8=99=E0=B8=88?= =?UTF-8?q?=E0=B8=B2=E0=B8=81=20find({=20select:=20["id"]=20})=20=E0=B8=97?= =?UTF-8?q?=E0=B8=B1=E0=B9=89=E0=B8=87=E0=B8=95=E0=B8=B2=E0=B8=A3=E0=B8=B2?= =?UTF-8?q?=E0=B8=87=20=E0=B8=A1=E0=B8=B2=E0=B9=80=E0=B8=9B=E0=B9=87?= =?UTF-8?q?=E0=B8=99=20query=20=E0=B9=80=E0=B8=89=E0=B8=9E=E0=B8=B2?= =?UTF-8?q?=E0=B8=B0=20current=5FholderId=20=E0=B8=97=E0=B8=B5=E0=B9=88?= =?UTF-8?q?=E0=B8=AD=E0=B9=89=E0=B8=B2=E0=B8=87=E0=B8=96=E0=B8=B6=E0=B8=87?= =?UTF-8?q?=E0=B8=88=E0=B8=A3=E0=B8=B4=E0=B8=87=E0=B9=83=E0=B8=99=E0=B8=8A?= =?UTF-8?q?=E0=B8=B8=E0=B8=94=E0=B8=82=E0=B9=89=E0=B8=AD=E0=B8=A1=E0=B8=B9?= =?UTF-8?q?=E0=B8=A5=20publish=203.=20=E0=B9=80=E0=B8=81=E0=B9=87=E0=B8=9A?= =?UTF-8?q?=20normalization=20=E0=B8=82=E0=B8=AD=E0=B8=87=20=5Forgemployee?= =?UTF-8?q?PosMaster=20=E0=B9=81=E0=B8=A5=E0=B8=B0=20=5ForgemployeeTempPos?= =?UTF-8?q?Master=20=E0=B9=84=E0=B8=A7=E0=B9=89=E0=B8=AB=E0=B8=A5=E0=B8=B1?= =?UTF-8?q?=E0=B8=87=20query=20=E0=B8=8A=E0=B8=B8=E0=B8=94=E0=B9=80?= =?UTF-8?q?=E0=B8=94=E0=B8=B5=E0=B8=A2=E0=B8=A7=E0=B8=81=E0=B8=B1=E0=B8=99?= =?UTF-8?q?=20=E0=B8=97=E0=B8=B3=E0=B9=83=E0=B8=AB=E0=B9=89=20block=20?= =?UTF-8?q?=E0=B8=99=E0=B8=B5=E0=B9=89=E0=B8=81=E0=B8=A3=E0=B8=B0=E0=B8=8A?= =?UTF-8?q?=E0=B8=B1=E0=B8=9A=E0=B8=82=E0=B8=B6=E0=B9=89=E0=B8=99=E0=B9=81?= =?UTF-8?q?=E0=B8=A5=E0=B8=B0=E0=B8=A5=E0=B8=94=20read=20cost=20=E0=B8=97?= =?UTF-8?q?=E0=B8=B5=E0=B9=88=E0=B9=84=E0=B8=A1=E0=B9=88=E0=B8=88=E0=B8=B3?= =?UTF-8?q?=E0=B9=80=E0=B8=9B=E0=B9=87=E0=B8=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/services/rabbitmq.ts | 64 ++++++++++++++++++++++------------------ 1 file changed, 35 insertions(+), 29 deletions(-) diff --git a/src/services/rabbitmq.ts b/src/services/rabbitmq.ts index 097473f8..da93d32f 100644 --- a/src/services/rabbitmq.ts +++ b/src/services/rabbitmq.ts @@ -1161,21 +1161,38 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { } console.timeEnd("[AMQ] clone_permissionProfiles"); - //หา dna posmaster ถ้าไม่มีให้เอาตัวเองเป็น dna - console.time("[AMQ] query_employeePosMaster"); - const orgemployeePosMaster = await repoEmployeePosmaster.find({ - where: { orgRevisionId: orgRevisionPublish.id }, - relations: ["positions"], - }); - console.timeEnd("[AMQ] query_employeePosMaster"); + console.time("[AMQ] query_employee_org_structures"); + const [orgemployeePosMaster, orgemployeeTempPosMaster] = await Promise.all([ + repoEmployeePosmaster.find({ + where: { orgRevisionId: orgRevisionPublish.id }, + relations: ["positions"], + }), + repoEmployeeTempPosmaster.find({ + where: { orgRevisionId: orgRevisionPublish.id }, + relations: ["positions"], + }), + ]); + console.timeEnd("[AMQ] query_employee_org_structures"); console.log(`[AMQ] orgemployeePosMaster count: ${orgemployeePosMaster.length}`); + console.log(`[AMQ] orgemployeeTempPosMaster count: ${orgemployeeTempPosMaster.length}`); - let _orgemployeePosMaster: EmployeePosMaster[]; - const validProfileIds = new Set( - (await repoProfileEmployee.find({ select: ["id"] })).map((p) => p.id), + const currentHolderIds = Array.from( + new Set( + orgemployeePosMaster + .map((item) => item.current_holderId) + .filter((holderId): holderId is string => !!holderId), + ), ); + const validProfileIds = new Set(); + if (currentHolderIds.length > 0) { + const profiles = await repoProfileEmployee.find({ + select: ["id"], + where: { id: In(currentHolderIds) }, + }); + profiles.forEach((profile) => validProfileIds.add(profile.id)); + } - _orgemployeePosMaster = orgemployeePosMaster.map((x) => ({ + const _orgemployeePosMaster: EmployeePosMaster[] = orgemployeePosMaster.map((x) => ({ ...x, current_holderId: x.current_holderId && validProfileIds.has(x.current_holderId) ? x.current_holderId : null, @@ -1184,6 +1201,13 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { ? x.id : x.ancestorDNA, })); + const _orgemployeeTempPosMaster: EmployeeTempPosMaster[] = orgemployeeTempPosMaster.map((x) => ({ + ...x, + ancestorDNA: + x.ancestorDNA == null || x.ancestorDNA == "00000000-0000-0000-0000-000000000000" + ? x.id + : x.ancestorDNA, + })); console.time("[AMQ] insert_employeePosMaster"); await repoEmployeePosmaster @@ -1198,24 +1222,6 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { .execute(); console.timeEnd("[AMQ] insert_employeePosMaster"); - //หา dna posmaster ถ้าไม่มีให้เอาตัวเองเป็น dna - console.time("[AMQ] query_employeeTempPosMaster"); - const orgemployeeTempPosMaster = await repoEmployeeTempPosmaster.find({ - where: { orgRevisionId: orgRevisionPublish.id }, - relations: ["positions"], - }); - console.timeEnd("[AMQ] query_employeeTempPosMaster"); - console.log(`[AMQ] orgemployeeTempPosMaster count: ${orgemployeeTempPosMaster.length}`); - - let _orgemployeeTempPosMaster: EmployeeTempPosMaster[]; - _orgemployeeTempPosMaster = orgemployeeTempPosMaster.map((x) => ({ - ...x, - ancestorDNA: - x.ancestorDNA == null || x.ancestorDNA == "00000000-0000-0000-0000-000000000000" - ? x.id - : x.ancestorDNA, - })); - const groupByParentId = ( items: T[], getParentId: (item: T) => string | null | undefined, From 750947f34fb6cb62f8e98f7207e1682c8027f1d0 Mon Sep 17 00:00:00 2001 From: waruneeauy Date: Tue, 5 May 2026 16:38:54 +0700 Subject: [PATCH 3/5] =?UTF-8?q?1.=20=E0=B9=80=E0=B8=9E=E0=B8=B4=E0=B9=88?= =?UTF-8?q?=E0=B8=A1=20helper=20=E0=B8=AA=E0=B8=B3=E0=B8=AB=E0=B8=A3?= =?UTF-8?q?=E0=B8=B1=E0=B8=9A=20build=20clone=20rows=20=E0=B8=88=E0=B8=B2?= =?UTF-8?q?=E0=B8=81=20metadata=20=E0=B8=82=E0=B8=AD=E0=B8=87=20repository?= =?UTF-8?q?=20=E0=B9=81=E0=B8=A5=E0=B9=89=E0=B8=A7=20pre-generate=20UUID?= =?UTF-8?q?=20=E0=B9=83=E0=B8=AB=E0=B9=89=20parent=20=E0=B9=81=E0=B8=A5?= =?UTF-8?q?=E0=B8=B0=20child=20=E0=B8=A5=E0=B9=88=E0=B8=A7=E0=B8=87?= =?UTF-8?q?=E0=B8=AB=E0=B8=99=E0=B9=89=E0=B8=B2=202.=20=E0=B9=80=E0=B8=9B?= =?UTF-8?q?=E0=B8=A5=E0=B8=B5=E0=B9=88=E0=B8=A2=E0=B8=99=20inner=20clone?= =?UTF-8?q?=20flow=20=E0=B9=80=E0=B8=9B=E0=B9=87=E0=B8=99=20cloneEmployeeN?= =?UTF-8?q?odeBatch(...)=20=E0=B8=97=E0=B8=B5=E0=B9=88=E0=B8=97=E0=B8=B3?= =?UTF-8?q?=E0=B8=87=E0=B8=B2=E0=B8=99=E0=B9=80=E0=B8=9B=E0=B9=87=E0=B8=99?= =?UTF-8?q?=E0=B8=8A=E0=B8=B8=E0=B8=94=20=E0=B9=81=E0=B8=97=E0=B8=99?= =?UTF-8?q?=E0=B8=81=E0=B8=B2=E0=B8=A3=20save()=20parent=20=E0=B9=81?= =?UTF-8?q?=E0=B8=A5=E0=B9=89=E0=B8=A7=20save()=20children=20=E0=B8=97?= =?UTF-8?q?=E0=B8=B5=E0=B8=A5=E0=B8=B0=E0=B8=A3=E0=B8=B2=E0=B8=A2=E0=B8=81?= =?UTF-8?q?=E0=B8=B2=E0=B8=A3=203.=20=E0=B9=83=E0=B8=8A=E0=B9=89=20insertI?= =?UTF-8?q?nChunks(...)=20=E0=B8=AA=E0=B8=B3=E0=B8=AB=E0=B8=A3=E0=B8=B1?= =?UTF-8?q?=E0=B8=9A=20batch=20insert=20=E0=B8=82=E0=B8=AD=E0=B8=87=20pare?= =?UTF-8?q?nt=20rows=20=E0=B9=81=E0=B8=A5=E0=B8=B0=20EmployeePosition=20ro?= =?UTF-8?q?ws=204.=20=E0=B9=83=E0=B8=8A=E0=B9=89=20helper=20=E0=B9=80?= =?UTF-8?q?=E0=B8=94=E0=B8=B5=E0=B8=A2=E0=B8=A7=E0=B8=81=E0=B8=B1=E0=B8=99?= =?UTF-8?q?=E0=B8=8B=E0=B9=89=E0=B8=B3=E0=B8=97=E0=B8=B8=E0=B8=81=E0=B8=A3?= =?UTF-8?q?=E0=B8=B0=E0=B8=94=E0=B8=B1=E0=B8=9A=E0=B8=82=E0=B8=AD=E0=B8=87?= =?UTF-8?q?=20tree=20(root,=20child1,=20child2,=20child3,=20child4)=20?= =?UTF-8?q?=E0=B9=80=E0=B8=9E=E0=B8=B7=E0=B9=88=E0=B8=AD=E0=B8=A5=E0=B8=94?= =?UTF-8?q?=20code=20duplication=20=E0=B9=81=E0=B8=A5=E0=B8=B0=E0=B8=84?= =?UTF-8?q?=E0=B8=87=20mapping=20=E0=B8=82=E0=B8=AD=E0=B8=87=20destination?= =?UTF-8?q?=20org=20ids=20=E0=B8=95=E0=B8=B2=E0=B8=A1=20logic=20=E0=B9=80?= =?UTF-8?q?=E0=B8=94=E0=B8=B4=E0=B8=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/services/rabbitmq.ts | 543 +++++++++++++-------------------------- 1 file changed, 180 insertions(+), 363 deletions(-) diff --git a/src/services/rabbitmq.ts b/src/services/rabbitmq.ts index da93d32f..073042c9 100644 --- a/src/services/rabbitmq.ts +++ b/src/services/rabbitmq.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "crypto"; import amqp from "amqplib"; import { AppDataSource } from "../database/data-source"; import { Command } from "../entities/Command"; @@ -20,7 +21,7 @@ import { OrgChild4 } from "../entities/OrgChild4"; import { OrgRoot } from "../entities/OrgRoot"; import { PosMasterAssign, PosMasterAssignDTO } from "../entities/PosMasterAssign"; import { Position } from "../entities/Position"; -import { In, Not } from "typeorm"; +import { In, Not, Repository } from "typeorm"; import { PosMasterAct } from "../entities/PosMasterAct"; import { PermissionOrg } from "../entities/PermissionOrg"; import { sendWebSocket } from "./webSocket"; @@ -1292,6 +1293,98 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { const orgChild2ByChild1 = groupByParentId(orgChild2, (item) => item.orgChild1Id); const orgChild3ByChild2 = groupByParentId(orgChild3, (item) => item.orgChild2Id); const orgChild4ByChild3 = groupByParentId(orgChild4, (item) => item.orgChild3Id); + type OrgDestinationIds = { + orgRootId?: string | null; + orgChild1Id?: string | null; + orgChild2Id?: string | null; + orgChild3Id?: string | null; + orgChild4Id?: string | null; + }; + type CloneEmployeeSource = { + positions: EmployeePosition[]; + } & (EmployeePosMaster | EmployeeTempPosMaster); + const buildAuditFields = (timestamp: Date) => ({ + createdUserId: "", + createdFullName: "System Administrator", + createdAt: timestamp, + lastUpdateUserId: "", + lastUpdateFullName: "System Administrator", + lastUpdatedAt: timestamp, + }); + const buildColumnData = (repository: Repository, source: T): Partial => { + const row = {} as Partial; + const target = row as Record; + const sourceRecord = source as Record; + for (const column of repository.metadata.columns) { + target[column.propertyName] = sourceRecord[column.propertyName]; + } + return row; + }; + const insertInChunks = async ( + repository: Repository, + rows: Partial[], + chunkSize: number, + ) => { + if (rows.length === 0) { + return; + } + for (const chunk of chunkArray(rows, chunkSize) as Partial[][]) { + await repository.insert(chunk as Parameters["insert"]>[0]); + } + }; + const buildEmployeeCloneBatch = ( + items: T[], + parentRepository: Repository, + positionParentKey: "posMasterId" | "posMasterTempId", + targetIds: OrgDestinationIds, + ) => { + const parentRows: Partial[] = []; + const positionRows: Partial[] = []; + + for (const item of items) { + const parentId = randomUUID(); + const parentTimestamp = new Date(); + parentRows.push({ + ...buildColumnData(parentRepository, item), + id: parentId, + orgRevisionId: orgRevisionDraft.id, + ...targetIds, + ...buildAuditFields(parentTimestamp), + }); + + for (const position of item.positions ?? []) { + const positionTimestamp = new Date(); + positionRows.push({ + ...buildColumnData(employeePositionRepository, position), + id: randomUUID(), + posMasterId: positionParentKey === "posMasterId" ? parentId : undefined, + posMasterTempId: + positionParentKey === "posMasterTempId" ? parentId : undefined, + ...buildAuditFields(positionTimestamp), + }); + } + } + + return { parentRows, positionRows }; + }; + const cloneEmployeeNodeBatch = async ( + items: T[], + parentRepository: Repository, + positionParentKey: "posMasterId" | "posMasterTempId", + targetIds: OrgDestinationIds, + ) => { + if (items.length === 0) { + return; + } + const { parentRows, positionRows } = buildEmployeeCloneBatch( + items, + parentRepository, + positionParentKey, + targetIds, + ); + await insertInChunks(parentRepository, parentRows, 200); + await insertInChunks(employeePositionRepository, positionRows, 500); + }; await repoEmployeeTempPosmaster .createQueryBuilder() @@ -1313,398 +1406,122 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { const filteredEmployeePosMaster = employeePosMasterByNode.get(getNodeKey("root", dataId)) ?? []; - await Promise.all( - filteredEmployeePosMaster.map(async (item: any) => { - delete item.id; - const employeePosMaster = Object.assign(new EmployeePosMaster(), item); - employeePosMaster.positions = []; - employeePosMaster.orgRevisionId = orgRevisionDraft.id; - employeePosMaster.orgRootId = matchedOrgRoot?.id ?? null; - employeePosMaster.createdUserId = ""; - employeePosMaster.createdFullName = "System Administrator"; - employeePosMaster.createdAt = new Date(); - employeePosMaster.lastUpdateUserId = ""; - employeePosMaster.lastUpdateFullName = "System Administrator"; - employeePosMaster.lastUpdatedAt = new Date(); - await repoEmployeePosmaster.save(employeePosMaster); - - await Promise.all( - item.positions.map(async (pos: any) => { - delete pos.id; - const employeePosition: EmployeePosition = Object.assign( - new EmployeePosition(), - pos, - ); - employeePosition.posMasterId = employeePosMaster.id; - employeePosition.createdUserId = ""; - employeePosition.createdFullName = "System Administrator"; - employeePosition.createdAt = new Date(); - employeePosition.lastUpdateUserId = ""; - employeePosition.lastUpdateFullName = "System Administrator"; - employeePosition.lastUpdatedAt = new Date(); - await employeePositionRepository.save(employeePosition); - }), - ); - }), + await cloneEmployeeNodeBatch( + filteredEmployeePosMaster, + repoEmployeePosmaster, + "posMasterId", + { orgRootId: matchedOrgRoot?.id ?? null }, ); - await Promise.all( - (employeeTempPosMasterByNode.get(getNodeKey("root", dataId)) ?? []) - .map(async (item: any) => { - delete item.id; - const employeeTempPosMaster = Object.assign(new EmployeeTempPosMaster(), item); - employeeTempPosMaster.positions = []; - employeeTempPosMaster.orgRevisionId = orgRevisionDraft.id; - employeeTempPosMaster.orgRootId = matchedOrgRoot?.id ?? null; - employeeTempPosMaster.createdUserId = ""; - employeeTempPosMaster.createdFullName = "System Administrator"; - employeeTempPosMaster.createdAt = new Date(); - employeeTempPosMaster.lastUpdateUserId = ""; - employeeTempPosMaster.lastUpdateFullName = "System Administrator"; - employeeTempPosMaster.lastUpdatedAt = new Date(); - await repoEmployeeTempPosmaster.save(employeeTempPosMaster); - - await Promise.all( - item.positions.map(async (pos: any) => { - delete pos.id; - const employeePosition: EmployeePosition = Object.assign( - new EmployeePosition(), - pos, - ); - employeePosition.posMasterTempId = employeeTempPosMaster.id; - employeePosition.createdUserId = ""; - employeePosition.createdFullName = "System Administrator"; - employeePosition.createdAt = new Date(); - employeePosition.lastUpdateUserId = ""; - employeePosition.lastUpdateFullName = "System Administrator"; - employeePosition.lastUpdatedAt = new Date(); - await employeePositionRepository.save(employeePosition); - }), - ); - }), + await cloneEmployeeNodeBatch( + employeeTempPosMasterByNode.get(getNodeKey("root", dataId)) ?? [], + repoEmployeeTempPosmaster, + "posMasterTempId", + { orgRootId: matchedOrgRoot?.id ?? null }, ); for (const x of orgChild1ByRoot.get(dataId) ?? []) { const data1Id = x.id; const matchedOrgChild1 = findMatchedNodeByAncestorDNA(orgChild1Current, x); - await Promise.all( - (employeePosMasterByNode.get(getNodeKey("child1", data1Id)) ?? []) - .map(async (item: any) => { - delete item.id; - const employeePosMaster = Object.assign(new EmployeePosMaster(), item); - employeePosMaster.positions = []; - employeePosMaster.orgRevisionId = orgRevisionDraft.id; - employeePosMaster.orgRootId = matchedOrgRoot?.id ?? null; - employeePosMaster.orgChild1Id = matchedOrgChild1?.id ?? null; - employeePosMaster.createdUserId = ""; - employeePosMaster.createdFullName = "System Administrator"; - employeePosMaster.createdAt = new Date(); - employeePosMaster.lastUpdateUserId = ""; - employeePosMaster.lastUpdateFullName = "System Administrator"; - employeePosMaster.lastUpdatedAt = new Date(); - await repoEmployeePosmaster.save(employeePosMaster); - - await Promise.all( - item.positions.map(async (pos: any) => { - delete pos.id; - const employeePosition: EmployeePosition = Object.assign( - new EmployeePosition(), - pos, - ); - employeePosition.posMasterId = employeePosMaster.id; - employeePosition.createdUserId = ""; - employeePosition.createdFullName = "System Administrator"; - employeePosition.createdAt = new Date(); - employeePosition.lastUpdateUserId = ""; - employeePosition.lastUpdateFullName = "System Administrator"; - employeePosition.lastUpdatedAt = new Date(); - await employeePositionRepository.save(employeePosition); - }), - ); - }), + await cloneEmployeeNodeBatch( + employeePosMasterByNode.get(getNodeKey("child1", data1Id)) ?? [], + repoEmployeePosmaster, + "posMasterId", + { + orgRootId: matchedOrgRoot?.id ?? null, + orgChild1Id: matchedOrgChild1?.id ?? null, + }, ); - await Promise.all( - (employeeTempPosMasterByNode.get(getNodeKey("child1", data1Id)) ?? []) - .map(async (item: any) => { - delete item.id; - const employeeTempPosMaster = Object.assign(new EmployeeTempPosMaster(), item); - employeeTempPosMaster.positions = []; - employeeTempPosMaster.orgRevisionId = orgRevisionDraft.id; - employeeTempPosMaster.orgRootId = matchedOrgRoot?.id ?? null; - employeeTempPosMaster.orgChild1Id = matchedOrgChild1?.id ?? null; - employeeTempPosMaster.createdUserId = ""; - employeeTempPosMaster.createdFullName = "System Administrator"; - employeeTempPosMaster.createdAt = new Date(); - employeeTempPosMaster.lastUpdateUserId = ""; - employeeTempPosMaster.lastUpdateFullName = "System Administrator"; - employeeTempPosMaster.lastUpdatedAt = new Date(); - await repoEmployeeTempPosmaster.save(employeeTempPosMaster); - - await Promise.all( - item.positions.map(async (pos: any) => { - delete pos.id; - const employeePosition: EmployeePosition = Object.assign( - new EmployeePosition(), - pos, - ); - employeePosition.posMasterTempId = employeeTempPosMaster.id; - employeePosition.createdUserId = ""; - employeePosition.createdFullName = "System Administrator"; - employeePosition.createdAt = new Date(); - employeePosition.lastUpdateUserId = ""; - employeePosition.lastUpdateFullName = "System Administrator"; - employeePosition.lastUpdatedAt = new Date(); - await employeePositionRepository.save(employeePosition); - }), - ); - }), + await cloneEmployeeNodeBatch( + employeeTempPosMasterByNode.get(getNodeKey("child1", data1Id)) ?? [], + repoEmployeeTempPosmaster, + "posMasterTempId", + { + orgRootId: matchedOrgRoot?.id ?? null, + orgChild1Id: matchedOrgChild1?.id ?? null, + }, ); for (const x of orgChild2ByChild1.get(data1Id) ?? []) { const data2Id = x.id; const matchedOrgChild2 = findMatchedNodeByAncestorDNA(orgChild2Current, x); - await Promise.all( - (employeePosMasterByNode.get(getNodeKey("child2", data2Id)) ?? []) - .map(async (item: any) => { - delete item.id; - const employeePosMaster = Object.assign(new EmployeePosMaster(), item); - employeePosMaster.positions = []; - employeePosMaster.orgRevisionId = orgRevisionDraft.id; - employeePosMaster.orgRootId = matchedOrgRoot?.id ?? null; - employeePosMaster.orgChild1Id = matchedOrgChild1?.id ?? null; - employeePosMaster.orgChild2Id = matchedOrgChild2?.id ?? null; - employeePosMaster.createdUserId = ""; - employeePosMaster.createdFullName = "System Administrator"; - employeePosMaster.createdAt = new Date(); - employeePosMaster.lastUpdateUserId = ""; - employeePosMaster.lastUpdateFullName = "System Administrator"; - employeePosMaster.lastUpdatedAt = new Date(); - await repoEmployeePosmaster.save(employeePosMaster); - - await Promise.all( - item.positions.map(async (pos: any) => { - delete pos.id; - const employeePosition: EmployeePosition = Object.assign( - new EmployeePosition(), - pos, - ); - employeePosition.posMasterId = employeePosMaster.id; - employeePosition.createdUserId = ""; - employeePosition.createdFullName = "System Administrator"; - employeePosition.createdAt = new Date(); - employeePosition.lastUpdateUserId = ""; - employeePosition.lastUpdateFullName = "System Administrator"; - employeePosition.lastUpdatedAt = new Date(); - await employeePositionRepository.save(employeePosition); - }), - ); - }), + await cloneEmployeeNodeBatch( + employeePosMasterByNode.get(getNodeKey("child2", data2Id)) ?? [], + repoEmployeePosmaster, + "posMasterId", + { + orgRootId: matchedOrgRoot?.id ?? null, + orgChild1Id: matchedOrgChild1?.id ?? null, + orgChild2Id: matchedOrgChild2?.id ?? null, + }, ); - await Promise.all( - (employeeTempPosMasterByNode.get(getNodeKey("child2", data2Id)) ?? []) - .map(async (item: any) => { - delete item.id; - const employeeTempPosMaster = Object.assign(new EmployeeTempPosMaster(), item); - employeeTempPosMaster.positions = []; - employeeTempPosMaster.orgRevisionId = orgRevisionDraft.id; - employeeTempPosMaster.orgRootId = dataId; - employeeTempPosMaster.orgChild1Id = data1Id; - employeeTempPosMaster.orgChild2Id = data2Id; - employeeTempPosMaster.createdUserId = ""; - employeeTempPosMaster.createdFullName = "System Administrator"; - employeeTempPosMaster.createdAt = new Date(); - employeeTempPosMaster.lastUpdateUserId = ""; - employeeTempPosMaster.lastUpdateFullName = "System Administrator"; - employeeTempPosMaster.lastUpdatedAt = new Date(); - await repoEmployeeTempPosmaster.save(employeeTempPosMaster); - - await Promise.all( - item.positions.map(async (pos: any) => { - delete pos.id; - const employeePosition: EmployeePosition = Object.assign( - new EmployeePosition(), - pos, - ); - employeePosition.posMasterTempId = employeeTempPosMaster.id; - employeePosition.createdUserId = ""; - employeePosition.createdFullName = "System Administrator"; - employeePosition.createdAt = new Date(); - employeePosition.lastUpdateUserId = ""; - employeePosition.lastUpdateFullName = "System Administrator"; - employeePosition.lastUpdatedAt = new Date(); - await employeePositionRepository.save(employeePosition); - }), - ); - }), + await cloneEmployeeNodeBatch( + employeeTempPosMasterByNode.get(getNodeKey("child2", data2Id)) ?? [], + repoEmployeeTempPosmaster, + "posMasterTempId", + { + orgRootId: dataId, + orgChild1Id: data1Id, + orgChild2Id: data2Id, + }, ); for (const x of orgChild3ByChild2.get(data2Id) ?? []) { const data3Id = x.id; const matchedOrgChild3 = findMatchedNodeByAncestorDNA(orgChild3Current, x); - await Promise.all( - (employeePosMasterByNode.get(getNodeKey("child3", data3Id)) ?? []) - .map(async (item: any) => { - delete item.id; - const employeePosMaster = Object.assign(new EmployeePosMaster(), item); - employeePosMaster.positions = []; - employeePosMaster.orgRevisionId = orgRevisionDraft.id; - employeePosMaster.orgRootId = matchedOrgRoot?.id ?? null; - employeePosMaster.orgChild1Id = matchedOrgChild1?.id ?? null; - employeePosMaster.orgChild2Id = matchedOrgChild2?.id ?? null; - employeePosMaster.orgChild3Id = matchedOrgChild3?.id ?? null; - employeePosMaster.createdUserId = ""; - employeePosMaster.createdFullName = "System Administrator"; - employeePosMaster.createdAt = new Date(); - employeePosMaster.lastUpdateUserId = ""; - employeePosMaster.lastUpdateFullName = "System Administrator"; - employeePosMaster.lastUpdatedAt = new Date(); - await repoEmployeePosmaster.save(employeePosMaster); - - await Promise.all( - item.positions.map(async (pos: any) => { - delete pos.id; - const employeePosition: EmployeePosition = Object.assign( - new EmployeePosition(), - pos, - ); - employeePosition.posMasterId = employeePosMaster.id; - employeePosition.createdUserId = ""; - employeePosition.createdFullName = "System Administrator"; - employeePosition.createdAt = new Date(); - employeePosition.lastUpdateUserId = ""; - employeePosition.lastUpdateFullName = "System Administrator"; - employeePosition.lastUpdatedAt = new Date(); - await employeePositionRepository.save(employeePosition); - }), - ); - }), + await cloneEmployeeNodeBatch( + employeePosMasterByNode.get(getNodeKey("child3", data3Id)) ?? [], + repoEmployeePosmaster, + "posMasterId", + { + orgRootId: matchedOrgRoot?.id ?? null, + orgChild1Id: matchedOrgChild1?.id ?? null, + orgChild2Id: matchedOrgChild2?.id ?? null, + orgChild3Id: matchedOrgChild3?.id ?? null, + }, ); - await Promise.all( - (employeeTempPosMasterByNode.get(getNodeKey("child3", data3Id)) ?? []) - .map(async (item: any) => { - delete item.id; - const employeeTempPosMaster = Object.assign(new EmployeeTempPosMaster(), item); - employeeTempPosMaster.positions = []; - employeeTempPosMaster.orgRevisionId = orgRevisionDraft.id; - employeeTempPosMaster.orgRootId = matchedOrgRoot?.id ?? null; - employeeTempPosMaster.orgChild1Id = matchedOrgChild1?.id ?? null; - employeeTempPosMaster.orgChild2Id = matchedOrgChild2?.id ?? null; - employeeTempPosMaster.orgChild3Id = matchedOrgChild3?.id ?? null; - employeeTempPosMaster.createdUserId = ""; - employeeTempPosMaster.createdFullName = "System Administrator"; - employeeTempPosMaster.createdAt = new Date(); - employeeTempPosMaster.lastUpdateUserId = ""; - employeeTempPosMaster.lastUpdateFullName = "System Administrator"; - employeeTempPosMaster.lastUpdatedAt = new Date(); - await repoEmployeeTempPosmaster.save(employeeTempPosMaster); - - await Promise.all( - item.positions.map(async (pos: any) => { - delete pos.id; - const employeePosition: EmployeePosition = Object.assign( - new EmployeePosition(), - pos, - ); - employeePosition.posMasterTempId = employeeTempPosMaster.id; - employeePosition.createdUserId = ""; - employeePosition.createdFullName = "System Administrator"; - employeePosition.createdAt = new Date(); - employeePosition.lastUpdateUserId = ""; - employeePosition.lastUpdateFullName = "System Administrator"; - employeePosition.lastUpdatedAt = new Date(); - await employeePositionRepository.save(employeePosition); - }), - ); - }), + await cloneEmployeeNodeBatch( + employeeTempPosMasterByNode.get(getNodeKey("child3", data3Id)) ?? [], + repoEmployeeTempPosmaster, + "posMasterTempId", + { + orgRootId: matchedOrgRoot?.id ?? null, + orgChild1Id: matchedOrgChild1?.id ?? null, + orgChild2Id: matchedOrgChild2?.id ?? null, + orgChild3Id: matchedOrgChild3?.id ?? null, + }, ); for (const x of orgChild4ByChild3.get(data3Id) ?? []) { const data4Id = x.id; const matchedOrgChild4 = findMatchedNodeByAncestorDNA(orgChild4Current, x); - await Promise.all( - (employeePosMasterByNode.get(getNodeKey("child4", data4Id)) ?? []) - .map(async (item: any) => { - delete item.id; - const employeePosMaster = Object.assign(new EmployeePosMaster(), item); - employeePosMaster.positions = []; - employeePosMaster.orgRevisionId = orgRevisionDraft.id; - employeePosMaster.orgRootId = matchedOrgRoot?.id ?? null; - employeePosMaster.orgChild1Id = matchedOrgChild1?.id ?? null; - employeePosMaster.orgChild2Id = matchedOrgChild2?.id ?? null; - employeePosMaster.orgChild3Id = matchedOrgChild3?.id ?? null; - employeePosMaster.orgChild4Id = matchedOrgChild4?.id ?? null; - employeePosMaster.createdUserId = ""; - employeePosMaster.createdFullName = "System Administrator"; - employeePosMaster.createdAt = new Date(); - employeePosMaster.lastUpdateUserId = ""; - employeePosMaster.lastUpdateFullName = "System Administrator"; - employeePosMaster.lastUpdatedAt = new Date(); - await repoEmployeePosmaster.save(employeePosMaster); - - await Promise.all( - item.positions.map(async (pos: any) => { - delete pos.id; - const employeePosition: EmployeePosition = Object.assign( - new EmployeePosition(), - pos, - ); - employeePosition.posMasterId = employeePosMaster.id; - employeePosition.createdUserId = ""; - employeePosition.createdFullName = "System Administrator"; - employeePosition.createdAt = new Date(); - employeePosition.lastUpdateUserId = ""; - employeePosition.lastUpdateFullName = "System Administrator"; - employeePosition.lastUpdatedAt = new Date(); - await employeePositionRepository.save(employeePosition); - }), - ); - }), + await cloneEmployeeNodeBatch( + employeePosMasterByNode.get(getNodeKey("child4", data4Id)) ?? [], + repoEmployeePosmaster, + "posMasterId", + { + orgRootId: matchedOrgRoot?.id ?? null, + orgChild1Id: matchedOrgChild1?.id ?? null, + orgChild2Id: matchedOrgChild2?.id ?? null, + orgChild3Id: matchedOrgChild3?.id ?? null, + orgChild4Id: matchedOrgChild4?.id ?? null, + }, ); - await Promise.all( - (employeeTempPosMasterByNode.get(getNodeKey("child4", data4Id)) ?? []) - .map(async (item: any) => { - delete item.id; - const employeeTempPosMaster = Object.assign( - new EmployeeTempPosMaster(), - item, - ); - employeeTempPosMaster.positions = []; - employeeTempPosMaster.orgRevisionId = orgRevisionDraft.id; - employeeTempPosMaster.orgRootId = matchedOrgRoot?.id ?? null; - employeeTempPosMaster.orgChild1Id = matchedOrgChild1?.id ?? null; - employeeTempPosMaster.orgChild2Id = matchedOrgChild2?.id ?? null; - employeeTempPosMaster.orgChild3Id = matchedOrgChild3?.id ?? null; - employeeTempPosMaster.orgChild4Id = matchedOrgChild4?.id ?? null; - employeeTempPosMaster.createdUserId = ""; - employeeTempPosMaster.createdFullName = "System Administrator"; - employeeTempPosMaster.createdAt = new Date(); - employeeTempPosMaster.lastUpdateUserId = ""; - employeeTempPosMaster.lastUpdateFullName = "System Administrator"; - employeeTempPosMaster.lastUpdatedAt = new Date(); - await repoEmployeeTempPosmaster.save(employeeTempPosMaster); - - await Promise.all( - item.positions.map(async (pos: any) => { - delete pos.id; - const employeePosition: EmployeePosition = Object.assign( - new EmployeePosition(), - pos, - ); - employeePosition.posMasterTempId = employeeTempPosMaster.id; - employeePosition.createdUserId = ""; - employeePosition.createdFullName = "System Administrator"; - employeePosition.createdAt = new Date(); - employeePosition.lastUpdateUserId = ""; - employeePosition.lastUpdateFullName = "System Administrator"; - employeePosition.lastUpdatedAt = new Date(); - await employeePositionRepository.save(employeePosition); - }), - ); - }), + await cloneEmployeeNodeBatch( + employeeTempPosMasterByNode.get(getNodeKey("child4", data4Id)) ?? [], + repoEmployeeTempPosmaster, + "posMasterTempId", + { + orgRootId: matchedOrgRoot?.id ?? null, + orgChild1Id: matchedOrgChild1?.id ?? null, + orgChild2Id: matchedOrgChild2?.id ?? null, + orgChild3Id: matchedOrgChild3?.id ?? null, + orgChild4Id: matchedOrgChild4?.id ?? null, + }, ); } } From 93d4857ea102a6a3b5b0d4bac3693dcc0aa2f698 Mon Sep 17 00:00:00 2001 From: harid Date: Tue, 5 May 2026 16:43:47 +0700 Subject: [PATCH 4/5] =?UTF-8?q?cronjob=20=E0=B8=AA=E0=B9=88=E0=B8=87?= =?UTF-8?q?=E0=B8=82=E0=B9=89=E0=B8=AD=E0=B8=A1=E0=B8=B9=E0=B8=A5=E0=B8=9C?= =?UTF-8?q?=E0=B8=B9=E0=B9=89=E0=B9=80=E0=B8=81=E0=B8=A9=E0=B8=B5=E0=B8=A2?= =?UTF-8?q?=E0=B8=93=E0=B9=84=E0=B8=9B=E0=B9=83=E0=B8=AB=E0=B9=89=E0=B8=A3?= =?UTF-8?q?=E0=B8=B0=E0=B8=9A=E0=B8=9A=E0=B8=9E=E0=B9=89=E0=B8=99=E0=B8=A3?= =?UTF-8?q?=E0=B8=B2=E0=B8=8A=E0=B8=81=E0=B8=B2=E0=B8=A3=20#2330?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/app.ts | 12 ++ src/controllers/CommandController.ts | 19 +++- src/controllers/ExRetirementController.ts | 23 ++-- src/services/RetirementService.ts | 133 ++++++++++++++++++++++ 4 files changed, 175 insertions(+), 12 deletions(-) create mode 100644 src/services/RetirementService.ts diff --git a/src/app.ts b/src/app.ts index 06f76548..c2cbe5f7 100644 --- a/src/app.ts +++ b/src/app.ts @@ -19,6 +19,7 @@ import { ScriptProfileOrgController } from "./controllers/ScriptProfileOrgContro import { DateSerializer } from "./interfaces/date-serializer"; import { initWebSocket } from "./services/webSocket"; +import { RetirementService } from "./services/RetirementService"; async function main() { await AppDataSource.initialize(); @@ -114,6 +115,17 @@ async function main() { } }); + // Cron job for posting retirement data to Exprofile - every day at 04:30:00 on the 1st of October + const cronTime_PostRetire = "0 30 4 1 10 *"; + cron.schedule(cronTime_PostRetire, async () => { + try { + const retirementService = new RetirementService(); + await retirementService.cronjobPostRetireToExprofile(); + } catch (error) { + console.error("[Cronjob] Error executing cronjobPostRetireToExprofile:", error); + } + }); + // app.listen(APP_PORT, APP_HOST, () => console.log(`Listening on: http://localhost:${APP_PORT}`)); const server = app.listen( APP_PORT, diff --git a/src/controllers/CommandController.ts b/src/controllers/CommandController.ts index f12df5be..f7e68083 100644 --- a/src/controllers/CommandController.ts +++ b/src/controllers/CommandController.ts @@ -104,6 +104,7 @@ import { PostRetireToExprofile } from "./ExRetirementController"; import { LeaveType } from "../entities/LeaveType"; import { KeycloakAttributeService } from "../services/KeycloakAttributeService"; import { reOrderCommandRecivesAndDelete } from "../services/CommandService"; +import { RetirementService } from "../services/RetirementService"; @Route("api/v1/org/command") @Tags("Command") @Security("bearerAuth") @@ -1608,8 +1609,7 @@ export class CommandController extends Controller { return new HttpSuccess(); } - // @Get("XXX") - async cronjobUpdateRetirementStatus(/*@Request() request: RequestWithUser*/) { + async cronjobUpdateRetirementStatus() { const adminToken = (await getToken()) ?? ""; const today = new Date(); today.setUTCHours(0, 0, 0, 0); @@ -1887,6 +1887,21 @@ export class CommandController extends Controller { return new HttpSuccess(); } + /** + * API ทดสอบ cronjobPostRetireToExprofile + * @summary ทดสอบส่งข้อมูลผู้เกษียณไปยังระบบพ้นราชการ (Exprofile) + */ + @Get("cronjob/cronjobPostRetireToExprofile") + async runCronjobPostRetireToExprofile() { + try { + const retirementService = new RetirementService(); + const result = await retirementService.cronjobPostRetireToExprofile(); + return new HttpSuccess(result); + } catch (error: any) { + throw new HttpError(HttpStatusCode.INTERNAL_SERVER_ERROR, error.message || "เกิดข้อผิดพลาด"); + } + } + /** * API รายละเอียดรายการคำสั่ง tab4 คำสั่ง * diff --git a/src/controllers/ExRetirementController.ts b/src/controllers/ExRetirementController.ts index c8ffe5da..6720f19d 100644 --- a/src/controllers/ExRetirementController.ts +++ b/src/controllers/ExRetirementController.ts @@ -237,16 +237,19 @@ export async function PostRetireToExprofile( continue; } - addLogSequence(request, { - action: "request", - status: "error", - description: "unconnected to exprofile api", - request: { - method: "POST", - url: API_URL_BANGKOK + "/importData", - response: JSON.stringify(error), - }, - }); + // เช็ค request ก่อนเรียก addLogSequence (สำหรับ cronjob ที่ส่ง null) + if (request) { + addLogSequence(request, { + action: "request", + status: "error", + description: "unconnected to exprofile api", + request: { + method: "POST", + url: API_URL_BANGKOK + "/importData", + response: JSON.stringify(error), + }, + }); + } throw new HttpError(HttpStatusCode.INTERNAL_SERVER_ERROR, "ไม่สามารถติดต่อ API ได้"); } diff --git a/src/services/RetirementService.ts b/src/services/RetirementService.ts new file mode 100644 index 00000000..1997c2e3 --- /dev/null +++ b/src/services/RetirementService.ts @@ -0,0 +1,133 @@ +import { AppDataSource } from "../database/data-source"; +import { Profile } from "../entities/Profile"; +import { PostRetireToExprofile } from "../controllers/ExRetirementController"; +import { Between, MoreThanOrEqual } from "typeorm"; + +const BATCH_SIZE = 100; +const CONCURRENT_PER_BATCH = 10; // ส่ง parallel ทีละ 10 คนในแต่ละ batch + +export class RetirementService { + private profileRepository = AppDataSource.getRepository(Profile); + + /** + * Cronjob สำหรับส่งข้อมูลผู้เกษียณไปยังระบบพ้นราชการ (Exprofile) + * ทำงานเวลา 04:30:00 ของทุกวันที่ 1 ตุลาคม + * + * รายละเอียด: + * - Query profiles ที่ leaveDate = วันที่ 1 ตุลาคมของปีนั้น และ leaveType = "RETIRE" + * - Batch ทีละ 100 records + * - Concurrent ทีละ 10 คน (parallel) ในแต่ละ batch + * - ถ้า fail ให้ log error แล้วทำคนต่อไป + */ + async cronjobPostRetireToExprofile(): Promise<{ + success: number; + failed: number; + failedProfiles: Array<{ id: string; name: string; error: string }>; + }> { + const result = { + success: 0, + failed: 0, + failedProfiles: [] as Array<{ id: string; name: string; error: string }>, + }; + + try { + + // หาวันที่ 1 ตุลาคมของปีปัจจุบัน + const now = new Date(); + const currentYear = now.getFullYear(); + + // สร้างวันที่ 1 ตุลาคมของปีปัจจุบัน (เวลา 00:00:00) + const startDate = new Date(currentYear, 9, 1, 0, 0, 0); // Month 9 = October (0-indexed) + const endDate = new Date(currentYear, 9, 1, 23, 59, 59); + + // Query profiles ที่ leaveDate อยู่ในวันที่ 1 ตุลาคม และ leaveType = "RETIRE" + const profiles = await this.profileRepository.find({ + where: [ + { leaveDate: Between(startDate, endDate), leaveType: "RETIRE" as any }, + { leaveDate: MoreThanOrEqual(startDate), leaveType: "RETIRE" as any }, + ], + relations: ["posLevel", "posType"], + }); + + // Filter เอาเฉพาะวันที่ 1 ตุลาคมเท่านั้น + const filteredProfiles = profiles.filter(p => { + if (!p.leaveDate) return false; + const leaveDate = new Date(p.leaveDate); + return ( + leaveDate.getFullYear() === currentYear && + leaveDate.getMonth() === 9 && // October + leaveDate.getDate() === 1 + ); + }); + + if (filteredProfiles.length === 0) { + return result; + } + + // แบ่ง batch ทีละ 100 records + for (let i = 0; i < filteredProfiles.length; i += BATCH_SIZE) { + const batch = filteredProfiles.slice(i, i + BATCH_SIZE); + + // แบ่งเป็น chunk เล็กๆ ทีละ CONCURRENT_PER_BATCH เพื่อส่ง parallel + for (let j = 0; j < batch.length; j += CONCURRENT_PER_BATCH) { + const chunk = batch.slice(j, j + CONCURRENT_PER_BATCH); + + // ส่ง parallel ในแต่ละ chunk + await Promise.all( + chunk.map(async (profile) => { + try { + await this.postSingleProfileToExprofile(profile); + result.success++; + } catch (error: any) { + result.failed++; + const errorInfo = { + id: profile.id, + name: `${profile.prefix}${profile.firstName} ${profile.lastName}`, + error: error.message || String(error), + }; + result.failedProfiles.push(errorInfo); + } + }) + ); + } + } + + } catch (error: any) { + throw error; + } + + return result; + } + + /** + * ส่งข้อมูล profile ไปยัง Exprofile + */ + private async postSingleProfileToExprofile(profile: Profile): Promise { + if (!profile.leaveDate) { + return; + } + + if (!profile.citizenId) { + return; + } + + const retireYear = profile.leaveDate.getFullYear(); + const retireDate = new Date(profile.leaveDate); + + // ส่งไปยัง Exprofile + PostRetireToExprofile( + null, + profile.citizenId, + profile.prefix || "", + profile.firstName || "", + profile.lastName || "", + retireYear.toString(), + profile.position || "", + profile.posType?.posTypeName || "", + profile.posLevel?.posLevelName || "", + retireDate, + profile.org || "", + profile.leaveReason || "เกษียณอายุราชการ" + ); + } +} From 6c1e4a1e427a82d908cc14c8d17d83eb72688f4c Mon Sep 17 00:00:00 2001 From: waruneeauy Date: Tue, 5 May 2026 18:11:55 +0700 Subject: [PATCH 5/5] Optimize handler_org batch writes Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/services/rabbitmq.ts | 59 ++++++++++++++++++++++++++++++---------- 1 file changed, 45 insertions(+), 14 deletions(-) diff --git a/src/services/rabbitmq.ts b/src/services/rabbitmq.ts index 073042c9..990cd5fb 100644 --- a/src/services/rabbitmq.ts +++ b/src/services/rabbitmq.ts @@ -1035,6 +1035,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { // Clone oldposMasterAct console.time("[AMQ] clone_oldposMasterAct"); + const posMasterActRowsToInsert: Partial[] = []; for (const act of oldposMasterAct) { const parentDNA = act.posMaster?.ancestorDNA?.trim()?.toLowerCase() ?? ""; const childDNA = act.posMasterChild?.ancestorDNA?.trim()?.toLowerCase() ?? ""; @@ -1046,7 +1047,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { const { id, posMaster, posMasterChild, ...fields } = act; - const newAct = { + posMasterActRowsToInsert.push({ ...fields, posMasterId: newParentId, posMasterChildId: newChildId, @@ -1056,9 +1057,13 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { lastUpdatedAt: new Date(), lastUpdateFullName: user ? user.name : "system", lastUpdateUserId: user ? user.sub : "system", - }; - - await posMasterActRepository.save(newAct); + }); + } + if (posMasterActRowsToInsert.length > 0) { + const chunks = chunkArray(posMasterActRowsToInsert, 500); + for (const chunk of chunks) { + await posMasterActRepository.insert(chunk); + } } console.timeEnd("[AMQ] clone_oldposMasterAct"); @@ -1554,11 +1559,13 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { profiles.forEach((profile) => profileEmployeeMap.set(profile.id, profile)); } const updatedProfileEmployeeIds = new Set(); + const employeePosMasterIdsToTouch: string[] = []; + const employeeTempPosMasterIdsToTouch: string[] = []; for (const item of employeePosMaster) { if (item.next_holderId != null) { const profile = profileEmployeeMap.get(item.next_holderId); - const position = await item.positions.find((x) => x.positionIsSelected == true); + const position = item.positions.find((x) => x.positionIsSelected == true); const _null: any = null; if (profile != null) { profile.posLevelId = position?.posLevelId ?? _null; @@ -1567,15 +1574,12 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { updatedProfileEmployeeIds.add(profile.id); } } - item.lastUpdateUserId = lastUpdateUserId; - item.lastUpdateFullName = lastUpdateFullName; - item.lastUpdatedAt = lastUpdatedAt; - await repoEmployeePosmaster.save(item); + employeePosMasterIdsToTouch.push(item.id); } for (const item of employeeTempPosMaster) { if (item.next_holderId != null) { const profile = profileEmployeeMap.get(item.next_holderId); - const position = await item.positions.find((x) => x.positionIsSelected == true); + const position = item.positions.find((x) => x.positionIsSelected == true); const _null: any = null; if (profile != null) { profile.posLevelId = position?.posLevelId ?? _null; @@ -1584,10 +1588,37 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise { updatedProfileEmployeeIds.add(profile.id); } } - item.lastUpdateUserId = lastUpdateUserId; - item.lastUpdateFullName = lastUpdateFullName; - item.lastUpdatedAt = lastUpdatedAt; - await repoEmployeeTempPosmaster.save(item); + employeeTempPosMasterIdsToTouch.push(item.id); + } + if (employeePosMasterIdsToTouch.length > 0) { + const chunks = chunkArray(employeePosMasterIdsToTouch, 500); + const employeePosMasterTableName = repoEmployeePosmaster.metadata.tableName; + for (const chunk of chunks) { + const wherePlaceholders = chunk.map(() => "?").join(", "); + await manager.query( + `UPDATE \`${employeePosMasterTableName}\` + SET lastUpdateUserId = ?, + lastUpdateFullName = ?, + lastUpdatedAt = ? + WHERE id IN (${wherePlaceholders})`, + [lastUpdateUserId, lastUpdateFullName, lastUpdatedAt, ...chunk], + ); + } + } + if (employeeTempPosMasterIdsToTouch.length > 0) { + const chunks = chunkArray(employeeTempPosMasterIdsToTouch, 500); + const employeeTempPosMasterTableName = repoEmployeeTempPosmaster.metadata.tableName; + for (const chunk of chunks) { + const wherePlaceholders = chunk.map(() => "?").join(", "); + await manager.query( + `UPDATE \`${employeeTempPosMasterTableName}\` + SET lastUpdateUserId = ?, + lastUpdateFullName = ?, + lastUpdatedAt = ? + WHERE id IN (${wherePlaceholders})`, + [lastUpdateUserId, lastUpdateFullName, lastUpdatedAt, ...chunk], + ); + } } if (updatedProfileEmployeeIds.size > 0) { const profilesToSave = Array.from(updatedProfileEmployeeIds)