fix handler_org and add transaction

This commit is contained in:
Warunee Tamkoo 2026-04-30 22:41:29 +07:00
parent b5e80ba1e9
commit 50bd145053

View file

@ -1,4 +1,4 @@
import amqp from "amqplib";
import * as amqp from "amqplib";
import { AppDataSource } from "../database/data-source";
import { Command } from "../entities/Command";
import { chunkArray, commandTypePath } from "../interfaces/utils";
@ -24,12 +24,7 @@ import { In, Not } from "typeorm";
import { PosMasterAct } from "../entities/PosMasterAct";
import { PermissionOrg } from "../entities/PermissionOrg";
import { sendWebSocket } from "./webSocket";
import {
CreatePosMasterHistoryOfficer,
BatchUpdatePosMasters,
BatchCreatePosMasterHistoryOfficer,
BatchHistoryOperation,
} from "./PositionService";
import { CreatePosMasterHistoryOfficer, BatchUpdatePosMasters, BatchCreatePosMasterHistoryOfficer, BatchHistoryOperation } from "./PositionService";
import { PayloadSendNoti } from "../interfaces/utils";
import { PermissionProfile } from "../entities/PermissionProfile";
@ -410,7 +405,7 @@ async function handler_command_noti(msg: amqp.ConsumeMessage): Promise<boolean>
try {
let profilesNotiRequest: Promise<any> | undefined;
if (!["C-PM-10"].includes(command.commandType.code)) {
if (!(["C-PM-10"].includes(command.commandType.code))) {
profilesNotiRequest = new CallAPI()
.PostData(
{ headers: { authorization: token } },
@ -487,7 +482,8 @@ async function handler_command_noti(msg: amqp.ConsumeMessage): Promise<boolean>
/*เฉพาะคำสั่ง C-PM-10 ให้ตัด profilesNotiRequest ที่ส่ง noti ครั้งแรกออก*/
if (["C-PM-10"].includes(command.commandType.code)) {
await Promise.all([profilesSendRequest]);
} else {
}
else {
await Promise.all([profilesNotiRequest!, profilesSendRequest]);
}
@ -501,26 +497,30 @@ async function handler_command_noti(msg: amqp.ConsumeMessage): Promise<boolean>
async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
//----> condition before process consume
console.time("[AMQ] handler_org_total");
console.time('[AMQ] handler_org_total');
const startTime = Date.now();
console.log(`[AMQ] handler_org START at ${new Date(startTime).toISOString()}`);
const repoPosmaster = AppDataSource.getRepository(PosMaster);
const posMasterAssignRepository = AppDataSource.getRepository(PosMasterAssign);
const posMasterActRepository = AppDataSource.getRepository(PosMasterAct);
const permissionProfilesRepository = AppDataSource.getRepository(PermissionProfile);
const repoEmployeePosmaster = AppDataSource.getRepository(EmployeePosMaster);
const repoEmployeeTempPosmaster = AppDataSource.getRepository(EmployeeTempPosMaster);
const repoProfile = AppDataSource.getRepository(Profile);
const repoProfileEmployee = AppDataSource.getRepository(ProfileEmployee);
const employeePositionRepository = AppDataSource.getRepository(EmployeePosition);
const repoOrgRevision = AppDataSource.getRepository(OrgRevision);
const orgRootRepository = AppDataSource.getRepository(OrgRoot);
const child1Repository = AppDataSource.getRepository(OrgChild1);
const child2Repository = AppDataSource.getRepository(OrgChild2);
const child3Repository = AppDataSource.getRepository(OrgChild3);
const child4Repository = AppDataSource.getRepository(OrgChild4);
const { data, token, user } = JSON.parse(msg.content.toString());
try {
// ✅ WRAP ALL DATABASE OPERATIONS IN TRANSACTION FOR AUTOMATIC ROLLBACK ON ERROR
return await AppDataSource.transaction(async (manager) => {
const repoPosmaster = manager.getRepository(PosMaster);
const posMasterAssignRepository = manager.getRepository(PosMasterAssign);
const posMasterActRepository = manager.getRepository(PosMasterAct);
const permissionProfilesRepository = manager.getRepository(PermissionProfile);
const repoEmployeePosmaster = manager.getRepository(EmployeePosMaster);
const repoEmployeeTempPosmaster = manager.getRepository(EmployeeTempPosMaster);
const repoProfile = manager.getRepository(Profile);
const repoProfileEmployee = manager.getRepository(ProfileEmployee);
const employeePositionRepository = manager.getRepository(EmployeePosition);
const repoOrgRevision = manager.getRepository(OrgRevision);
const orgRootRepository = manager.getRepository(OrgRoot);
const child1Repository = manager.getRepository(OrgChild1);
const child2Repository = manager.getRepository(OrgChild2);
const child3Repository = manager.getRepository(OrgChild3);
const child4Repository = manager.getRepository(OrgChild4);
const { id, status, lastUpdateUserId, lastUpdateFullName, lastUpdatedAt } = data;
console.log(`[AMQ] Received message - revisionId: ${id}, status: ${status}`);
@ -535,7 +535,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
).catch(console.error);
}
console.time("[AMQ] query_revisions");
console.time('[AMQ] query_revisions');
const orgRevisionPublish = await repoOrgRevision
.createQueryBuilder("orgRevision")
.where("orgRevision.orgRevisionIsDraft = false")
@ -547,17 +547,13 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
.where("orgRevision.orgRevisionIsDraft = true")
.andWhere("orgRevision.orgRevisionIsCurrent = false")
.getOne();
console.timeEnd("[AMQ] query_revisions");
console.log(
`[AMQ] orgRevisionPublish found: ${orgRevisionPublish ? orgRevisionPublish.id : "null"}`,
);
console.log(`[AMQ] orgRevisionDraft found: ${orgRevisionDraft ? orgRevisionDraft.id : "null"}`);
console.timeEnd('[AMQ] query_revisions');
console.log(`[AMQ] orgRevisionPublish found: ${orgRevisionPublish ? orgRevisionPublish.id : 'null'}`);
console.log(`[AMQ] orgRevisionDraft found: ${orgRevisionDraft ? orgRevisionDraft.id : 'null'}`);
// Validate: ต้องมี orgRevisionPublish เสมอสำหรับการเผยแพร่
if (!orgRevisionPublish) {
console.error(
"[AMQ] Cannot publish: No current org revision found (isDraft=false, isCurrent=true)",
);
console.error('[AMQ] Cannot publish: No current org revision found (isDraft=false, isCurrent=true)');
if (user) {
sendWebSocket(
"send-publish-org",
@ -573,9 +569,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
// Validate: ต้องมี orgRevisionDraft ที่จะเผยแพร่
if (!orgRevisionDraft) {
console.error(
"[AMQ] Cannot publish: No draft org revision found (isDraft=true, isCurrent=false)",
);
console.error('[AMQ] Cannot publish: No draft org revision found (isDraft=true, isCurrent=false)');
if (user) {
sendWebSocket(
"send-publish-org",
@ -592,8 +586,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
// NOTE: ย้ายการอัปเดตสถานะไปไว้หลังจากทำงานเสร็จทั้งหมด
// เพื่อป้องกันกรณี timeout/retry ทำให้สถานะเพี้ยน (ทุก row เป็น false,false)
try {
console.time("[AMQ] query_posMaster");
console.time('[AMQ] query_posMaster');
const POS_MASTER_PAGE_SIZE = 2000;
let totalPosMastersProcessed = 0;
let hasMoreRecords = true;
@ -614,7 +607,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
"positions.posType",
"positions.posExecutive",
],
order: { id: "ASC" },
order: { id: 'ASC' },
skip: skip,
take: POS_MASTER_PAGE_SIZE,
});
@ -626,15 +619,15 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
console.log(`[AMQ] Loaded posMaster page: ${totalPosMastersProcessed} records`);
}
console.timeEnd("[AMQ] query_posMaster");
console.timeEnd('[AMQ] query_posMaster');
console.log(`[AMQ] posMaster count: ${posMaster.length}`);
console.time("[AMQ] query_old_data");
console.time('[AMQ] query_old_data');
const oldPosMasters = await repoPosmaster.find({
where: {
orgRevisionId: orgRevisionPublish.id,
},
select: ["id", "current_holderId", "ancestorDNA"],
select: ['id', 'current_holderId', 'ancestorDNA']
});
// Task #2160 ดึง posMasterAssign ของ revision เดิม
@ -646,11 +639,11 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
},
},
});
console.timeEnd("[AMQ] query_old_data");
console.timeEnd('[AMQ] query_old_data');
console.log(`[AMQ] oldPosMasters count: ${oldPosMasters.length}`);
console.log(`[AMQ] oldposMasterAssigns count: ${oldposMasterAssigns.length}`);
console.time("[AMQ] build_assignMap");
console.time('[AMQ] build_assignMap');
// สร้าง assignMap เอาไว้เก็บ posMasterAssign.ancestorDNA ของ revision เดิม
const assignMap = new Map<string, PosMasterAssignDTO[]>();
for (const posmasterAssign of oldposMasterAssigns) {
@ -661,12 +654,12 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
assignMap.get(dna)!.push({
id: posmasterAssign.id,
posMasterId: posmasterAssign.posMasterId,
assignId: posmasterAssign.assignId,
assignId: posmasterAssign.assignId
});
}
console.timeEnd("[AMQ] build_assignMap");
console.timeEnd('[AMQ] build_assignMap');
console.time("[AMQ] query_oldposMasterAct");
console.time('[AMQ] query_oldposMasterAct');
// ดึง posMasterAct ของ revision เดิม xxx
const oldposMasterAct = await posMasterActRepository.find({
relations: ["posMaster", "posMasterChild"],
@ -676,16 +669,16 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
},
},
});
console.timeEnd("[AMQ] query_oldposMasterAct");
console.timeEnd('[AMQ] query_oldposMasterAct');
console.log(`[AMQ] oldposMasterAct count: ${oldposMasterAct.length}`);
type ActKey = string; // `${parentDNA}|${childDNA}`
console.time("[AMQ] build_maps");
console.time('[AMQ] build_maps');
const posMasterActMap = new Map<ActKey, PosMasterAct[]>();
for (const act of oldposMasterAct) {
const parentDNA = act.posMaster?.ancestorDNA?.trim() ?? "";
const childDNA = act.posMasterChild?.ancestorDNA?.trim() ?? "";
const parentDNA = act.posMaster?.ancestorDNA?.trim() ?? '';
const childDNA = act.posMasterChild?.ancestorDNA?.trim() ?? '';
const key = `${parentDNA}|${childDNA}`;
if (!posMasterActMap.has(key)) {
@ -696,7 +689,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
const posMasterIdMap = new Map<string, string>();
for (const pm of posMaster) {
posMasterIdMap.set(pm.ancestorDNA?.trim() ?? "", pm.id);
posMasterIdMap.set(pm.ancestorDNA?.trim() ?? '', pm.id);
}
const oldPosMasterMap = new Map<string, PosMaster>();
@ -706,25 +699,25 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
oldPosMasterMap.set(dna, oldPm);
}
}
console.timeEnd("[AMQ] build_maps");
console.timeEnd('[AMQ] build_maps');
const _null: any = null;
// ===== BATCH PROCESSING: เตรียมข้อมูลก่อน loop =====
console.time("[AMQ] prepare_batch_data");
console.time('[AMQ] prepare_batch_data');
// 1. รวบรวม profileIds ทั้งหมดที่ต้องอัพเดท
const profileIds = posMaster
.filter((item) => item.next_holderId != null)
.map((item) => item.next_holderId!)
.filter((id) => id != null && id !== "");
.filter(item => item.next_holderId != null)
.map(item => item.next_holderId!)
.filter(id => id != null && id !== "");
// 2. Batch load profiles ทั้งหมดในครั้งเดียว (แก้ปัญหา N+1 Query)
const profilesMap = new Map<string, Profile>();
if (profileIds.length > 0) {
const profiles = await repoProfile.findBy({
id: In(profileIds),
id: In(profileIds)
});
profiles.forEach((p) => profilesMap.set(p.id, p));
profiles.forEach(p => profilesMap.set(p.id, p));
}
console.log(`[AMQ] profiles to update: ${profilesMap.size}`);
@ -752,7 +745,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
lastUpdatedAt: lastUpdatedAt,
lastUpdateFullName: lastUpdateFullName,
lastUpdateUserId: lastUpdateUserId,
}),
})
);
posMasterAssignsToSave.push(...newAssigns);
}
@ -803,35 +796,33 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
historyCreateIds.push(item.id);
}
}
console.timeEnd("[AMQ] prepare_batch_data");
console.log(
`[AMQ] Prepared - posMasterAssignsToSave: ${posMasterAssignsToSave.length}, profilesToSave: ${profilesToSave.length}, posMasterUpdates: ${posMasterUpdates.length}, historyCreateIds: ${historyCreateIds.length}`,
);
console.timeEnd('[AMQ] prepare_batch_data');
console.log(`[AMQ] Prepared - posMasterAssignsToSave: ${posMasterAssignsToSave.length}, profilesToSave: ${profilesToSave.length}, posMasterUpdates: ${posMasterUpdates.length}, historyCreateIds: ${historyCreateIds.length}`);
// ===== BATCH EXECUTION: save ทีละ batch =====
// 4. Batch save posMasterAssign (chunk 500)
console.time("[AMQ] batch_save_posMasterAssign");
console.time('[AMQ] batch_save_posMasterAssign');
if (posMasterAssignsToSave.length > 0) {
const chunks = chunkArray(posMasterAssignsToSave, 500);
for (const chunk of chunks) {
await posMasterAssignRepository.save(chunk);
}
}
console.timeEnd("[AMQ] batch_save_posMasterAssign");
console.timeEnd('[AMQ] batch_save_posMasterAssign');
// 5. Batch save profiles (chunk 200)
console.time("[AMQ] batch_save_profiles");
console.time('[AMQ] batch_save_profiles');
if (profilesToSave.length > 0) {
const chunks = chunkArray(profilesToSave, 200);
for (const chunk of chunks) {
await repoProfile.save(chunk);
}
}
console.timeEnd("[AMQ] batch_save_profiles");
console.timeEnd('[AMQ] batch_save_profiles');
// 6. Batch update posMasters
console.time("[AMQ] batch_update_posMasters");
console.time('[AMQ] batch_update_posMasters');
const posMasterUpdatesForBatch = posMasterUpdates.map((u: any) => ({
id: u.id,
@ -841,16 +832,19 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
lastUpdatedAt,
}));
await BatchUpdatePosMasters(AppDataSource.manager, posMasterUpdatesForBatch);
await BatchUpdatePosMasters(
AppDataSource.manager,
posMasterUpdatesForBatch
);
console.timeEnd("[AMQ] batch_update_posMasters");
console.timeEnd('[AMQ] batch_update_posMasters');
// 7. Batch create history
console.time("[AMQ] batch_create_history");
console.time('[AMQ] batch_create_history');
const historyOperations: BatchHistoryOperation[] = [];
for (const id of historyCreateIds) {
const pm = posMaster.find((p) => p.id === id);
const pm = posMaster.find(p => p.id === id);
if (pm) {
historyOperations.push({
posMasterId: id,
@ -862,15 +856,18 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
}
}
await BatchCreatePosMasterHistoryOfficer(AppDataSource.manager, historyOperations);
await BatchCreatePosMasterHistoryOfficer(
AppDataSource.manager,
historyOperations
);
console.timeEnd("[AMQ] batch_create_history");
console.timeEnd('[AMQ] batch_create_history');
// Clone oldposMasterAct
console.time("[AMQ] clone_oldposMasterAct");
console.time('[AMQ] clone_oldposMasterAct');
for (const act of oldposMasterAct) {
const parentDNA = act.posMaster?.ancestorDNA?.trim()?.toLowerCase() ?? "";
const childDNA = act.posMasterChild?.ancestorDNA?.trim()?.toLowerCase() ?? "";
const parentDNA = act.posMaster?.ancestorDNA?.trim()?.toLowerCase() ?? '';
const childDNA = act.posMasterChild?.ancestorDNA?.trim()?.toLowerCase() ?? '';
const newParentId = posMasterIdMap.get(parentDNA);
const newChildId = posMasterIdMap.get(childDNA);
@ -893,16 +890,16 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
await posMasterActRepository.save(newAct);
}
console.timeEnd("[AMQ] clone_oldposMasterAct");
console.timeEnd('[AMQ] clone_oldposMasterAct');
if (orgRevisionPublish != null && orgRevisionDraft != null) {
console.time("[AMQ] clone_org_structure");
console.time('[AMQ] clone_org_structure');
//new main revision
const before = null;
//ทุก orgRoot และ orgChild ข้างล่างนี้จะเป็นตัวเก่าที่ไม่ได้เป็น current revision
//cone tree
console.time("[AMQ] query_old_org_structure");
console.time('[AMQ] query_old_org_structure');
//หา dna tree
const orgRoot = await orgRootRepository.find({
where: { orgRevisionId: orgRevisionPublish.id },
@ -923,46 +920,27 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
const orgChild4 = await 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}`,
);
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}`);
// Task #2172 ดึง orgRoot ของ revision ใหม่
const newRoots = await orgRootRepository.find({
where: { orgRevisionId: orgRevisionDraft.id },
});
// สร้าง newRootMap เอาไว้เก็บ orgRoot.ancestorDNA ของ revision ใหม่
const newRootMap = new Map(newRoots.map((r) => [r.ancestorDNA, r.id]));
// Pre-load new revision org structure data to avoid N+1 queries inside nested loops
console.time("[AMQ] query_new_org_structure");
const newOrgChild1List = await child1Repository.find({
where: { orgRevisionId: orgRevisionDraft.id },
});
const newOrgChild2List = await child2Repository.find({
where: { orgRevisionId: orgRevisionDraft.id },
});
const newOrgChild3List = await child3Repository.find({
where: { orgRevisionId: orgRevisionDraft.id },
});
const newOrgChild4List = await child4Repository.find({
where: { orgRevisionId: orgRevisionDraft.id },
});
console.timeEnd("[AMQ] query_new_org_structure");
console.log(
`[AMQ] New structure loaded - Child1: ${newOrgChild1List.length}, Child2: ${newOrgChild2List.length}, Child3: ${newOrgChild3List.length}, Child4: ${newOrgChild4List.length}`,
const newRootMap = new Map(
newRoots.map(r => [r.ancestorDNA, r.id])
);
console.time("[AMQ] clone_permissionProfiles");
console.time('[AMQ] clone_permissionProfiles');
// ดึง permissionProfiles ของ revision เดิม
const oldPermissionProfiles = await permissionProfilesRepository.find({
relations: ["orgRootTree"],
where: {
orgRootTree: {
orgRevisionId: orgRevisionPublish.id,
},
},
}
}
});
const inserts: any[] = [];
for (const permiss of oldPermissionProfiles) {
@ -987,15 +965,15 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
if (inserts.length > 0) {
await permissionProfilesRepository.insert(inserts);
}
console.timeEnd("[AMQ] clone_permissionProfiles");
console.timeEnd('[AMQ] clone_permissionProfiles');
//หา dna posmaster ถ้าไม่มีให้เอาตัวเองเป็น dna
console.time("[AMQ] query_employeePosMaster");
console.time('[AMQ] query_employeePosMaster');
const orgemployeePosMaster = await repoEmployeePosmaster.find({
where: { orgRevisionId: orgRevisionPublish.id },
relations: ["positions"],
});
console.timeEnd("[AMQ] query_employeePosMaster");
console.timeEnd('[AMQ] query_employeePosMaster');
console.log(`[AMQ] orgemployeePosMaster count: ${orgemployeePosMaster.length}`);
let _orgemployeePosMaster: EmployeePosMaster[];
@ -1027,7 +1005,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
: x.ancestorDNA,
}));
console.time("[AMQ] insert_employeePosMaster");
console.time('[AMQ] insert_employeePosMaster');
await repoEmployeePosmaster
.createQueryBuilder()
.insert()
@ -1038,16 +1016,16 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
overwrite: ["ancestorDNA"],
})
.execute();
console.timeEnd("[AMQ] insert_employeePosMaster");
console.timeEnd('[AMQ] insert_employeePosMaster');
// }
//หา dna posmaster ถ้าไม่มีให้เอาตัวเองเป็น dna
console.time("[AMQ] query_employeeTempPosMaster");
console.time('[AMQ] query_employeeTempPosMaster');
const orgemployeeTempPosMaster = await repoEmployeeTempPosmaster.find({
where: { orgRevisionId: orgRevisionPublish.id },
relations: ["positions"],
});
console.timeEnd("[AMQ] query_employeeTempPosMaster");
console.timeEnd('[AMQ] query_employeeTempPosMaster');
console.log(`[AMQ] orgemployeeTempPosMaster count: ${orgemployeeTempPosMaster.length}`);
let _orgemployeeTempPosMaster: EmployeeTempPosMaster[];
@ -1078,11 +1056,10 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
// }
//create org - forEach orgRoot (WARNING: async forEach without await)
console.time("[AMQ] forEach_orgRoot");
console.time('[AMQ] forEach_orgRoot');
console.log(`[AMQ] Starting forEach orgRoot loop (${orgRoot.length} items)`);
let processedOrgRoot = 0;
await Promise.all(
orgRoot.map(async (x: any) => {
orgRoot.forEach(async (x: any) => {
const itemStartTime = Date.now();
var dataId = x.id;
@ -1110,12 +1087,12 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
// requestBody.typeDraft.toUpperCase() == "ORG_POSITION_PERSON_ROLE"
// ) {
//create employeePosmaster
const filteredEmployeePosMaster = _orgemployeePosMaster.filter(
(x: EmployeePosMaster) => x.orgRootId == dataId && x.orgChild1Id == null,
);
const filteredEmployeePosMaster = _orgemployeePosMaster
.filter((x: EmployeePosMaster) => x.orgRootId == dataId && x.orgChild1Id == null);
await Promise.all(
filteredEmployeePosMaster.map(async (item: any) => {
filteredEmployeePosMaster
.map(async (item: any) => {
delete item.id;
const employeePosMaster = Object.assign(new EmployeePosMaster(), item);
employeePosMaster.positions = [];
@ -1147,8 +1124,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosMaster.lastUpdatedAt = new Date();
await repoEmployeePosmaster.save(employeePosMaster);
//create employeePosition
item.positions.map(async (pos: any) => {
for (const pos of item.positions) {
delete pos.id;
const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(),
@ -1168,7 +1144,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition);
});
}
}),
);
//create employeeTempPosmaster
@ -1207,8 +1184,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeeTempPosMaster.lastUpdatedAt = new Date();
await repoEmployeeTempPosmaster.save(employeeTempPosMaster);
//create employeePosition
item.positions.map(async (pos: any) => {
for (const pos of item.positions) {
delete pos.id;
const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(),
@ -1228,7 +1204,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition);
});
}
}),
);
// }
@ -1249,8 +1226,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
x.ancestorDNA === "00000000-0000-0000-0000-000000000000"
) {
return (
i.ancestorDNA === null ||
i.ancestorDNA === "00000000-0000-0000-0000-000000000000"
i.ancestorDNA === null || i.ancestorDNA === "00000000-0000-0000-0000-000000000000"
);
}
return i.ancestorDNA === x.ancestorDNA;
@ -1265,9 +1241,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
//create employeePosmaster
await Promise.all(
_orgemployeePosMaster
.filter(
(x: EmployeePosMaster) => x.orgChild1Id == data1Id && x.orgChild2Id == null,
)
.filter((x: EmployeePosMaster) => x.orgChild1Id == data1Id && x.orgChild2Id == null)
.map(async (item: any) => {
delete item.id;
// console.log("[in case Child1] orgChild1Id == data1Id");
@ -1302,8 +1276,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosMaster.lastUpdatedAt = new Date();
await repoEmployeePosmaster.save(employeePosMaster);
//create employeePosition
item.positions.map(async (pos: any) => {
for (const pos of item.positions) {
delete pos.id;
const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(),
@ -1323,7 +1296,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition);
});
}
}),
);
// create employeeTempPosmaster
@ -1365,8 +1339,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeeTempPosMaster.lastUpdatedAt = new Date();
await repoEmployeeTempPosmaster.save(employeeTempPosMaster);
//create employeePosition
item.positions.map(async (pos: any) => {
for (const pos of item.positions) {
delete pos.id;
const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(),
@ -1386,7 +1359,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition);
});
}
}),
);
// }
@ -1461,8 +1435,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosMaster.lastUpdatedAt = new Date();
await repoEmployeePosmaster.save(employeePosMaster);
//create employeePosition
item.positions.map(async (pos: any) => {
for (const pos of item.positions) {
delete pos.id;
const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(),
@ -1482,7 +1455,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition);
});
}
}),
);
// create employeeTempPosmaster
@ -1529,8 +1503,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeeTempPosMaster.lastUpdatedAt = new Date();
await repoEmployeeTempPosmaster.save(employeeTempPosMaster);
//create employeePosition
item.positions.map(async (pos: any) => {
for (const pos of item.positions) {
delete pos.id;
const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(),
@ -1550,7 +1523,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition);
});
}
}),
);
// }
@ -1627,8 +1601,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosMaster.lastUpdatedAt = new Date();
await repoEmployeePosmaster.save(employeePosMaster);
//create employeePosition
item.positions.map(async (pos: any) => {
for (const pos of item.positions) {
delete pos.id;
const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(),
@ -1648,7 +1621,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition);
});
}
}),
);
// create employeeTempPosmaster
@ -1696,8 +1670,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeeTempPosMaster.lastUpdatedAt = new Date();
await repoEmployeeTempPosmaster.save(employeeTempPosMaster);
//create employeePosition
item.positions.map(async (pos: any) => {
for (const pos of item.positions) {
delete pos.id;
const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(),
@ -1717,7 +1690,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition);
});
}
}),
);
// }
@ -1795,8 +1769,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosMaster.lastUpdatedAt = new Date();
await repoEmployeePosmaster.save(employeePosMaster);
//create employeePosition
item.positions.map(async (pos: any) => {
for (const pos of item.positions) {
delete pos.id;
const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(),
@ -1816,7 +1789,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition);
});
}
}),
);
//create employeeTempPosmaster
@ -1862,8 +1836,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeeTempPosMaster.lastUpdatedAt = new Date();
await repoEmployeeTempPosmaster.save(employeeTempPosMaster);
//create employeePosition
item.positions.map(async (pos: any) => {
for (const pos of item.positions) {
delete pos.id;
const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(),
@ -1883,7 +1856,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition);
});
}
}),
);
// }
@ -1891,8 +1865,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
});
});
});
}),
);
});
// }
const employeePosMaster = await repoEmployeePosmaster.find({
@ -1957,10 +1930,10 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
{ userId: user?.sub },
).catch(console.error);
}
console.timeEnd("[AMQ] clone_org_structure");
console.timeEnd('[AMQ] clone_org_structure');
// อัปเดตสถานะ orgRevision หลังจากทำงานเสร็จทั้งหมด
console.time("[AMQ] save_revision_status");
console.time('[AMQ] save_revision_status');
orgRevisionPublish.orgRevisionIsDraft = false;
orgRevisionPublish.orgRevisionIsCurrent = false;
await repoOrgRevision.save(orgRevisionPublish);
@ -1968,43 +1941,29 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
orgRevisionDraft.orgRevisionIsCurrent = true;
orgRevisionDraft.orgRevisionIsDraft = false;
await repoOrgRevision.save(orgRevisionDraft);
console.timeEnd("[AMQ] save_revision_status");
console.timeEnd('[AMQ] save_revision_status');
console.log(`[AMQ] handler_org SUCCESS - Total time: ${Date.now() - startTime}ms`);
console.timeEnd("[AMQ] handler_org_total");
// Memory cleanup: clear arrays that are in scope
try {
if (typeof posMaster !== "undefined" && posMaster) posMaster.length = 0;
if (typeof oldPosMasters !== "undefined" && oldPosMasters) oldPosMasters.length = 0;
if (typeof oldposMasterAssigns !== "undefined" && oldposMasterAssigns)
oldposMasterAssigns.length = 0;
if (typeof oldposMasterAct !== "undefined" && oldposMasterAct) oldposMasterAct.length = 0;
if (typeof assignMap !== "undefined" && assignMap) assignMap.clear();
if (typeof posMasterActMap !== "undefined" && posMasterActMap) posMasterActMap.clear();
if (typeof posMasterIdMap !== "undefined" && posMasterIdMap) posMasterIdMap.clear();
if (typeof oldPosMasterMap !== "undefined" && oldPosMasterMap) oldPosMasterMap.clear();
if (typeof profilesMap !== "undefined" && profilesMap) profilesMap.clear();
} catch (cleanupError) {
console.error("[AMQ] Error during memory cleanup:", cleanupError);
}
console.timeEnd('[AMQ] handler_org_total');
return true;
}); // ✅ END TRANSACTION - All operations succeeded, data is committed
} catch (error) {
// ✅ TRANSACTION AUTOMATICALLY ROLLED BACK - No data was saved
const totalTime = Date.now() - startTime;
console.error(`[AMQ] handler_org ERROR after ${totalTime}ms:`, error);
console.error('[AMQ] Transaction rolled back - all changes were undone');
if (user) {
sendWebSocket(
"send-publish-org",
{
success: false,
message: `ระบบทำการเผยแพร่โครงสร้างหน่วยงานไม่สำเร็จ`,
message: `เผยแพร่โครงสร้างหน่วยงานไม่สำเร็จ: ${error instanceof Error ? error.message : String(error)}`,
},
{ userId: user?.sub },
).catch(console.error);
}
console.timeEnd("[AMQ] handler_org_total");
return false;
console.timeEnd('[AMQ] handler_org_total');
return false; // ✅ Return false to prevent RabbitMQ retry
}
}
@ -2678,8 +2637,7 @@ async function handler_org_draft(msg: amqp.ConsumeMessage): Promise<boolean> {
});
await posMasterAssignRepository.delete({ posMasterId: In(_posMasters.map((x) => x.id)) });
await posMasterActRepository.delete({ posMasterId: In(_posMasters.map((x) => x.id)) }); //ใช้ posMasterId ของ revision: draft *แต่ยังไม่เจอช็อดไหนที่ใช้โครงสร้างแบบร่างในรักษาการแทน
await posMasterActRepository.delete({
//ใช้ posMasterId ของ revision: draft *แต่ยังไม่เจอช็อดไหนที่ใช้โครงสร้างแบบร่างในรักษาการแทน
await posMasterActRepository.delete({ //ใช้ posMasterId ของ revision: draft *แต่ยังไม่เจอช็อดไหนที่ใช้โครงสร้างแบบร่างในรักษาการแทน
posMasterChildId: In(_posMasters.map((x) => x.id)),
});
// await posMasterRepository.remove(_posMasters);
@ -2707,26 +2665,24 @@ async function handler_org_draft(msg: amqp.ConsumeMessage): Promise<boolean> {
await child2Repository.delete({ orgRevisionId: In(_orgRevisions.map((x) => x.id)) });
await child1Repository.delete({ orgRevisionId: In(_orgRevisions.map((x) => x.id)) });
// Task #2160 อัพเดทหน้าที่จัดการโครงสร้างแบบร่าง
if (
[
"ORG",
"ORG_POSITION",
"ORG_POSITION_PERSON",
"ORG_POSITION_ROLE",
"ORG_POSITION_PERSON_ROLE",
].includes(requestBody.typeDraft?.toUpperCase())
) {
if (["ORG", "ORG_POSITION", "ORG_POSITION_PERSON", "ORG_POSITION_ROLE", "ORG_POSITION_PERSON_ROLE"].includes(requestBody.typeDraft?.toUpperCase())) {
const _newRoots = await orgRootRepository.find({
where: { orgRevisionId: revision.id },
where: { orgRevisionId: revision.id }
});
const newRootMap = new Map(_newRoots.map((r) => [r.ancestorDNA, r.id]));
const newRootMap = new Map(
_newRoots.map(r => [r.ancestorDNA, r.id])
);
for (const oldRoot of _roots) {
const newRootId = newRootMap.get(oldRoot.ancestorDNA);
if (!newRootId) continue;
// อัพเดท orgRootId ที่อยู่ภายใต้ orgRevision แบบร่างเดิมเป็นของ orgRevision แบบร่างใหม่
await permissionOrgRepository.update({ orgRootId: oldRoot.id }, { orgRootId: newRootId });
await permissionOrgRepository.update(
{ orgRootId: oldRoot.id },
{ orgRootId: newRootId }
);
}
} else {
}
else {
await permissionOrgRepository.delete({
orgRootId: In(_roots.map((x) => x.id)),
});