hrms-bangkok/hrms-api-org
7
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

rollback code handler_org

Browse source
This commit is contained in:
Warunee Tamkoo 2026-05-01 17:08:53 +07:00
parent cba5991097
commit f4798bddde
2 changed files with 1320 additions and 1515 deletions
Download patch file Download diff file Expand all files Collapse all files

143
src/services/PositionService.ts
View file

@ -11,7 +11,6 @@ import { PosMasterHistory } from "../entities/PosMasterHistory";
import { Position } from "../entities/Position"; import { Position } from "../entities/Position";
import { ProfileEducation } from "../entities/ProfileEducation"; import { ProfileEducation } from "../entities/ProfileEducation";
import { RequestWithUser } from "../middlewares/user"; import { RequestWithUser } from "../middlewares/user";
import { chunkArray } from "../interfaces/utils";
export async function CreatePosMasterHistoryOfficer( export async function CreatePosMasterHistoryOfficer(
posMasterId: string, posMasterId: string,
@ -418,145 +417,3 @@ export async function BatchSavePosMasterHistoryOfficer(
return false; return false;
} }
} }
export interface BatchHistoryOperation {
posMasterId: string;
posMasterData: PosMaster;
orgRevisionId: string;
lastUpdateUserId: string;
lastUpdateFullName: string;
}
export async function BatchUpdatePosMasters(
manager: any,
updates: { id: string; current_holderId: string | null; lastUpdateUserId: string; lastUpdateFullName: string; lastUpdatedAt: Date }[]
): Promise<void> {
if (updates.length === 0) return;
const CHUNK_SIZE = 5000;
const chunks = chunkArray(updates, CHUNK_SIZE);
for (const chunk of chunks) {
// Create a temporary table for this batch
const tempTableName = `temp_posmaster_update_${Date.now()}_${Math.random().toString(36).substring(7)}`;
try {
// Create temporary table
await manager.query(`
CREATE TEMPORARY TABLE ${tempTableName} (
id CHAR(36) PRIMARY KEY,
current_holderId CHAR(36) NULL,
lastUpdateUserId CHAR(36) NOT NULL,
lastUpdateFullName VARCHAR(255) NOT NULL,
lastUpdatedAt DATETIME NOT NULL
) ENGINE=InnoDB
`);
// Build insert query with proper parameter count
const insertParams: any[] = [];
const valuePlaceholders: string[] = [];
for (const u of chunk) {
valuePlaceholders.push('(?, ?, ?, ?, ?)');
insertParams.push(u.id, u.current_holderId, u.lastUpdateUserId, u.lastUpdateFullName, u.lastUpdatedAt);
}
// Bulk insert into temporary table
await manager.query(`
INSERT INTO ${tempTableName} (id, current_holderId, lastUpdateUserId, lastUpdateFullName, lastUpdatedAt)
VALUES ${valuePlaceholders.join(',')}
`, insertParams);
// Update using JOIN with temporary table (very fast - single query per chunk)
await manager.query(`
UPDATE posMaster p
INNER JOIN ${tempTableName} t ON p.id = t.id
SET p.current_holderId = t.current_holderId,
p.next_holderId = NULL,
p.lastUpdateUserId = t.lastUpdateUserId,
p.lastUpdateFullName = t.lastUpdateFullName,
p.lastUpdatedAt = t.lastUpdatedAt
`);
} finally {
// Drop temporary table
await manager.query(`DROP TEMPORARY TABLE IF EXISTS ${tempTableName}`).catch(() => {});
}
}
}
export async function BatchCreatePosMasterHistoryOfficer(
manager: any,
operations: BatchHistoryOperation[]
): Promise<void> {
if (operations.length === 0) return;
const repoHistory = manager.getRepository(PosMasterHistory);
const repoOrgRevision = manager.getRepository(OrgRevision);
const _null: any = null;
const orgRevisionIds = [...new Set(operations.map(op => op.orgRevisionId))];
const revisions = await repoOrgRevision.findBy({
id: In(orgRevisionIds),
orgRevisionIsCurrent: true,
orgRevisionIsDraft: false,
});
const currentRevisionIds = new Set(revisions.map((r: any) => r.id));
const historyRecords: PosMasterHistory[] = [];
for (const op of operations) {
const pm = op.posMasterData;
const checkCurrentRevision = currentRevisionIds.has(pm.orgRevisionId);
const h = new PosMasterHistory();
h.ancestorDNA = pm.ancestorDNA ?? _null;
if (checkCurrentRevision) {
h.prefix = pm.current_holder?.prefix ?? _null;
h.firstName = pm.current_holder?.firstName ?? _null;
h.lastName = pm.current_holder?.lastName ?? _null;
h.profileId = pm.current_holder?.id ?? _null;
} else {
h.prefix = pm.next_holder?.prefix ?? _null;
h.firstName = pm.next_holder?.firstName ?? _null;
h.lastName = pm.next_holder?.lastName ?? _null;
}
const selectedPosition = pm.positions?.find((p: any) => p.positionIsSelected === true) ?? null;
h.position = selectedPosition?.positionName ?? _null;
h.posType = selectedPosition?.posType?.posTypeName ?? _null;
h.posLevel = selectedPosition?.posLevel?.posLevelName ?? _null;
h.posExecutive = selectedPosition?.posExecutive?.posExecutiveName ?? _null;
h.rootDnaId = pm.orgRoot?.ancestorDNA ?? _null;
h.child1DnaId = pm.orgChild1?.ancestorDNA ?? _null;
h.child2DnaId = pm.orgChild2?.ancestorDNA ?? _null;
h.child3DnaId = pm.orgChild3?.ancestorDNA ?? _null;
h.child4DnaId = pm.orgChild4?.ancestorDNA ?? _null;
h.posMasterNoPrefix = pm.posMasterNoPrefix ?? _null;
h.posMasterNo = pm.posMasterNo ?? _null;
h.posMasterNoSuffix = pm.posMasterNoSuffix ?? _null;
h.shortName = [
pm.orgChild4?.orgChild4ShortName,
pm.orgChild3?.orgChild3ShortName,
pm.orgChild2?.orgChild2ShortName,
pm.orgChild1?.orgChild1ShortName,
pm.orgRoot?.orgRootShortName,
].find((s: any) => typeof s === "string" && s.trim().length > 0) ?? _null;
h.createdUserId = op.lastUpdateUserId;
h.createdFullName = op.lastUpdateFullName;
h.lastUpdateUserId = op.lastUpdateUserId;
h.lastUpdateFullName = op.lastUpdateFullName;
h.createdAt = new Date();
h.lastUpdatedAt = new Date();
historyRecords.push(h);
}
const CHUNK_SIZE = 500;
const chunks = chunkArray(historyRecords, CHUNK_SIZE);
for (const chunk of chunks) {
await repoHistory.save(chunk);
}
}

358
src/services/rabbitmq.ts
View file

@ -1,4 +1,4 @@
import * as amqp from "amqplib"; import amqp from "amqplib";
import { AppDataSource } from "../database/data-source"; import { AppDataSource } from "../database/data-source";
import { Command } from "../entities/Command"; import { Command } from "../entities/Command";
import { chunkArray, commandTypePath } from "../interfaces/utils"; import { chunkArray, commandTypePath } from "../interfaces/utils";
@ -24,12 +24,7 @@ import { In, Not } from "typeorm";
import { PosMasterAct } from "../entities/PosMasterAct"; import { PosMasterAct } from "../entities/PosMasterAct";
import { PermissionOrg } from "../entities/PermissionOrg"; import { PermissionOrg } from "../entities/PermissionOrg";
import { sendWebSocket } from "./webSocket"; import { sendWebSocket } from "./webSocket";
import { import { CreatePosMasterHistoryOfficer } from "./PositionService";
CreatePosMasterHistoryOfficer,
BatchUpdatePosMasters,
BatchCreatePosMasterHistoryOfficer,
BatchHistoryOperation,
} from "./PositionService";
import { PayloadSendNoti } from "../interfaces/utils"; import { PayloadSendNoti } from "../interfaces/utils";
import { PermissionProfile } from "../entities/PermissionProfile"; import { PermissionProfile } from "../entities/PermissionProfile";
@ -97,6 +92,8 @@ export async function init() {
// createConsumer(queue2, channel, handler2); // createConsumer(queue2, channel, handler2);
} }
let retries = 0;
function createConsumer( //----> consumer function createConsumer( //----> consumer
queue: string, queue: string,
channel: amqp.Channel, channel: amqp.Channel,
@ -106,15 +103,13 @@ function createConsumer( //----> consumer
queue, queue,
async (msg) => { async (msg) => {
if (!msg) return; if (!msg) return;
try { if ((await handler(msg)) || retries++ >= 3) {
await handler(msg); retries = 0;
console.log("[AMQ] Process Consumer success"); console.log("[AMQ] Process Consumer success");
} catch (error) {
console.log("[AMQ] Process Consumer failed");
} finally {
// Always acknowledge - no retries
return channel.ack(msg); return channel.ack(msg);
} }
console.log("[AMQ] Process Consumer failed");
return await new Promise((resolve) => setTimeout(() => resolve(channel.nack(msg)), 3000));
}, },
{ noAck: false }, { noAck: false },
); );
@ -410,7 +405,7 @@ async function handler_command_noti(msg: amqp.ConsumeMessage): Promise<boolean>
try { try {
let profilesNotiRequest: Promise<any> | undefined; let profilesNotiRequest: Promise<any> | undefined;
if (!["C-PM-10"].includes(command.commandType.code)) { if (!(["C-PM-10"].includes(command.commandType.code))) {
profilesNotiRequest = new CallAPI() profilesNotiRequest = new CallAPI()
.PostData( .PostData(
{ headers: { authorization: token } }, { headers: { authorization: token } },
@ -487,7 +482,8 @@ async function handler_command_noti(msg: amqp.ConsumeMessage): Promise<boolean>
/*เฉพาะคำสั่ง C-PM-10 ให้ตัด profilesNotiRequest ที่ส่ง noti ครั้งแรกออก*/ /*เฉพาะคำสั่ง C-PM-10 ให้ตัด profilesNotiRequest ที่ส่ง noti ครั้งแรกออก*/
if (["C-PM-10"].includes(command.commandType.code)) { if (["C-PM-10"].includes(command.commandType.code)) {
await Promise.all([profilesSendRequest]); await Promise.all([profilesSendRequest]);
} else { }
else {
await Promise.all([profilesNotiRequest!, profilesSendRequest]); await Promise.all([profilesNotiRequest!, profilesSendRequest]);
} }
@ -501,30 +497,26 @@ async function handler_command_noti(msg: amqp.ConsumeMessage): Promise<boolean>
async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> { async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
//----> condition before process consume //----> condition before process consume
console.time("[AMQ] handler_org_total"); console.time('[AMQ] handler_org_total');
const startTime = Date.now(); const startTime = Date.now();
console.log(`[AMQ] handler_org START at ${new Date(startTime).toISOString()}`); 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()); 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; const { id, status, lastUpdateUserId, lastUpdateFullName, lastUpdatedAt } = data;
console.log(`[AMQ] Received message - revisionId: ${id}, status: ${status}`); console.log(`[AMQ] Received message - revisionId: ${id}, status: ${status}`);
@ -539,7 +531,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
).catch(console.error); ).catch(console.error);
} }
console.time("[AMQ] query_revisions"); console.time('[AMQ] query_revisions');
const orgRevisionPublish = await repoOrgRevision const orgRevisionPublish = await repoOrgRevision
.createQueryBuilder("orgRevision") .createQueryBuilder("orgRevision")
.where("orgRevision.orgRevisionIsDraft = false") .where("orgRevision.orgRevisionIsDraft = false")
@ -551,19 +543,13 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
.where("orgRevision.orgRevisionIsDraft = true") .where("orgRevision.orgRevisionIsDraft = true")
.andWhere("orgRevision.orgRevisionIsCurrent = false") .andWhere("orgRevision.orgRevisionIsCurrent = false")
.getOne(); .getOne();
console.timeEnd("[AMQ] query_revisions"); console.timeEnd('[AMQ] query_revisions');
console.log( console.log(`[AMQ] orgRevisionPublish found: ${orgRevisionPublish ? orgRevisionPublish.id : 'null'}`);
`[AMQ] orgRevisionPublish found: ${orgRevisionPublish ? orgRevisionPublish.id : "null"}`, console.log(`[AMQ] orgRevisionDraft found: ${orgRevisionDraft ? orgRevisionDraft.id : 'null'}`);
);
console.log(
`[AMQ] orgRevisionDraft found: ${orgRevisionDraft ? orgRevisionDraft.id : "null"}`,
);
// Validate: ต้องมี orgRevisionPublish เสมอสำหรับการเผยแพร่ // Validate: ต้องมี orgRevisionPublish เสมอสำหรับการเผยแพร่
if (!orgRevisionPublish) { if (!orgRevisionPublish) {
console.error( console.error('[AMQ] Cannot publish: No current org revision found (isDraft=false, isCurrent=true)');
"[AMQ] Cannot publish: No current org revision found (isDraft=false, isCurrent=true)",
);
if (user) { if (user) {
sendWebSocket( sendWebSocket(
"send-publish-org", "send-publish-org",
@ -579,9 +565,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
// Validate: ต้องมี orgRevisionDraft ที่จะเผยแพร่ // Validate: ต้องมี orgRevisionDraft ที่จะเผยแพร่
if (!orgRevisionDraft) { if (!orgRevisionDraft) {
console.error( console.error('[AMQ] Cannot publish: No draft org revision found (isDraft=true, isCurrent=false)');
"[AMQ] Cannot publish: No draft org revision found (isDraft=true, isCurrent=false)",
);
if (user) { if (user) {
sendWebSocket( sendWebSocket(
"send-publish-org", "send-publish-org",
@ -598,15 +582,9 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
// NOTE: ย้ายการอัปเดตสถานะไปไว้หลังจากทำงานเสร็จทั้งหมด // NOTE: ย้ายการอัปเดตสถานะไปไว้หลังจากทำงานเสร็จทั้งหมด
// เพื่อป้องกันกรณี timeout/retry ทำให้สถานะเพี้ยน (ทุก row เป็น false,false) // เพื่อป้องกันกรณี timeout/retry ทำให้สถานะเพี้ยน (ทุก row เป็น false,false)
console.time("[AMQ] query_posMaster"); try {
const POS_MASTER_PAGE_SIZE = 2000; console.time('[AMQ] query_posMaster');
let totalPosMastersProcessed = 0; const posMaster = await repoPosmaster.find({
let hasMoreRecords = true;
let skip = 0;
const posMaster: PosMaster[] = [];
while (hasMoreRecords) {
const posMasterPage = await repoPosmaster.find({
where: { orgRevisionId: id }, where: { orgRevisionId: id },
relations: [ relations: [
"orgRoot", "orgRoot",
@ -619,27 +597,16 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
"positions.posType", "positions.posType",
"positions.posExecutive", "positions.posExecutive",
], ],
order: { id: "ASC" },
skip: skip,
take: POS_MASTER_PAGE_SIZE,
}); });
console.timeEnd('[AMQ] query_posMaster');
posMaster.push(...posMasterPage);
totalPosMastersProcessed += posMasterPage.length;
hasMoreRecords = posMasterPage.length === POS_MASTER_PAGE_SIZE;
skip += POS_MASTER_PAGE_SIZE;
console.log(`[AMQ] Loaded posMaster page: ${totalPosMastersProcessed} records`);
}
console.timeEnd("[AMQ] query_posMaster");
console.log(`[AMQ] posMaster count: ${posMaster.length}`); console.log(`[AMQ] posMaster count: ${posMaster.length}`);
console.time("[AMQ] query_old_data"); console.time('[AMQ] query_old_data');
const oldPosMasters = await repoPosmaster.find({ const oldPosMasters = await repoPosmaster.find({
where: { where: {
orgRevisionId: orgRevisionPublish.id, orgRevisionId: orgRevisionPublish.id,
}, },
select: ["id", "current_holderId", "ancestorDNA"], select: ['id', 'current_holderId', 'ancestorDNA']
}); });
// Task #2160 ดึง posMasterAssign ของ revision เดิม // Task #2160 ดึง posMasterAssign ของ revision เดิม
@ -651,11 +618,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] oldPosMasters count: ${oldPosMasters.length}`);
console.log(`[AMQ] oldposMasterAssigns count: ${oldposMasterAssigns.length}`); console.log(`[AMQ] oldposMasterAssigns count: ${oldposMasterAssigns.length}`);
console.time("[AMQ] build_assignMap"); console.time('[AMQ] build_assignMap');
// สร้าง assignMap เอาไว้เก็บ posMasterAssign.ancestorDNA ของ revision เดิม // สร้าง assignMap เอาไว้เก็บ posMasterAssign.ancestorDNA ของ revision เดิม
const assignMap = new Map<string, PosMasterAssignDTO[]>(); const assignMap = new Map<string, PosMasterAssignDTO[]>();
for (const posmasterAssign of oldposMasterAssigns) { for (const posmasterAssign of oldposMasterAssigns) {
@ -666,12 +633,12 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
assignMap.get(dna)!.push({ assignMap.get(dna)!.push({
id: posmasterAssign.id, id: posmasterAssign.id,
posMasterId: posmasterAssign.posMasterId, 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 // ดึง posMasterAct ของ revision เดิม xxx
const oldposMasterAct = await posMasterActRepository.find({ const oldposMasterAct = await posMasterActRepository.find({
relations: ["posMaster", "posMasterChild"], relations: ["posMaster", "posMasterChild"],
@ -681,16 +648,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}`); console.log(`[AMQ] oldposMasterAct count: ${oldposMasterAct.length}`);
type ActKey = string; // `${parentDNA}|${childDNA}` type ActKey = string; // `${parentDNA}|${childDNA}`
console.time("[AMQ] build_maps"); console.time('[AMQ] build_maps');
const posMasterActMap = new Map<ActKey, PosMasterAct[]>(); const posMasterActMap = new Map<ActKey, PosMasterAct[]>();
for (const act of oldposMasterAct) { for (const act of oldposMasterAct) {
const parentDNA = act.posMaster?.ancestorDNA?.trim() ?? ""; const parentDNA = act.posMaster?.ancestorDNA?.trim() ?? '';
const childDNA = act.posMasterChild?.ancestorDNA?.trim() ?? ""; const childDNA = act.posMasterChild?.ancestorDNA?.trim() ?? '';
const key = `${parentDNA}|${childDNA}`; const key = `${parentDNA}|${childDNA}`;
if (!posMasterActMap.has(key)) { if (!posMasterActMap.has(key)) {
@ -701,7 +668,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
const posMasterIdMap = new Map<string, string>(); const posMasterIdMap = new Map<string, string>();
for (const pm of posMaster) { 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>(); const oldPosMasterMap = new Map<string, PosMaster>();
@ -711,25 +678,25 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
oldPosMasterMap.set(dna, oldPm); oldPosMasterMap.set(dna, oldPm);
} }
} }
console.timeEnd("[AMQ] build_maps"); console.timeEnd('[AMQ] build_maps');
const _null: any = null; const _null: any = null;
// ===== BATCH PROCESSING: เตรียมข้อมูลก่อน loop ===== // ===== BATCH PROCESSING: เตรียมข้อมูลก่อน loop =====
console.time("[AMQ] prepare_batch_data"); console.time('[AMQ] prepare_batch_data');
// 1. รวบรวม profileIds ทั้งหมดที่ต้องอัพเดท // 1. รวบรวม profileIds ทั้งหมดที่ต้องอัพเดท
const profileIds = posMaster const profileIds = posMaster
.filter((item) => item.next_holderId != null) .filter(item => item.next_holderId != null)
.map((item) => item.next_holderId!) .map(item => item.next_holderId!)
.filter((id) => id != null && id !== ""); .filter(id => id != null && id !== "");
// 2. Batch load profiles ทั้งหมดในครั้งเดียว (แก้ปัญหา N+1 Query) // 2. Batch load profiles ทั้งหมดในครั้งเดียว (แก้ปัญหา N+1 Query)
const profilesMap = new Map<string, Profile>(); const profilesMap = new Map<string, Profile>();
if (profileIds.length > 0) { if (profileIds.length > 0) {
const profiles = await repoProfile.findBy({ 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}`); console.log(`[AMQ] profiles to update: ${profilesMap.size}`);
@ -757,7 +724,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
lastUpdatedAt: lastUpdatedAt, lastUpdatedAt: lastUpdatedAt,
lastUpdateFullName: lastUpdateFullName, lastUpdateFullName: lastUpdateFullName,
lastUpdateUserId: lastUpdateUserId, lastUpdateUserId: lastUpdateUserId,
}), })
); );
posMasterAssignsToSave.push(...newAssigns); posMasterAssignsToSave.push(...newAssigns);
} }
@ -808,74 +775,56 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
historyCreateIds.push(item.id); historyCreateIds.push(item.id);
} }
} }
console.timeEnd("[AMQ] prepare_batch_data"); console.timeEnd('[AMQ] prepare_batch_data');
console.log( 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: ${historyCreateIds.length}`,
);
// ===== BATCH EXECUTION: save ทีละ batch ===== // ===== BATCH EXECUTION: save ทีละ batch =====
// 4. Batch save posMasterAssign (chunk 500) // 4. Batch save posMasterAssign (chunk 500)
console.time("[AMQ] batch_save_posMasterAssign"); console.time('[AMQ] batch_save_posMasterAssign');
if (posMasterAssignsToSave.length > 0) { if (posMasterAssignsToSave.length > 0) {
const chunks = chunkArray(posMasterAssignsToSave, 500); const chunks = chunkArray(posMasterAssignsToSave, 500);
for (const chunk of chunks) { for (const chunk of chunks) {
await posMasterAssignRepository.save(chunk); await posMasterAssignRepository.save(chunk);
} }
} }
console.timeEnd("[AMQ] batch_save_posMasterAssign"); console.timeEnd('[AMQ] batch_save_posMasterAssign');
// 5. Batch save profiles (chunk 200) // 5. Batch save profiles (chunk 200)
console.time("[AMQ] batch_save_profiles"); console.time('[AMQ] batch_save_profiles');
if (profilesToSave.length > 0) { if (profilesToSave.length > 0) {
const chunks = chunkArray(profilesToSave, 200); const chunks = chunkArray(profilesToSave, 200);
for (const chunk of chunks) { for (const chunk of chunks) {
await repoProfile.save(chunk); await repoProfile.save(chunk);
} }
} }
console.timeEnd("[AMQ] batch_save_profiles"); console.timeEnd('[AMQ] batch_save_profiles');
// 6. Batch update posMasters // 6. Batch update posMasters
console.time("[AMQ] batch_update_posMasters"); console.time('[AMQ] batch_update_posMasters');
for (const update of posMasterUpdates) {
const posMasterUpdatesForBatch = posMasterUpdates.map((u: any) => ({ await repoPosmaster.update(update.id, {
id: u.id, current_holderId: update.current_holderId,
current_holderId: u.current_holderId ?? null, next_holderId: null,
lastUpdateUserId, lastUpdateUserId,
lastUpdateFullName, lastUpdateFullName,
lastUpdatedAt, lastUpdatedAt,
}));
await BatchUpdatePosMasters(AppDataSource.manager, posMasterUpdatesForBatch);
console.timeEnd("[AMQ] batch_update_posMasters");
// 7. Batch create history
console.time("[AMQ] batch_create_history");
const historyOperations: BatchHistoryOperation[] = [];
for (const id of historyCreateIds) {
const pm = posMaster.find((p) => p.id === id);
if (pm) {
historyOperations.push({
posMasterId: id,
posMasterData: pm,
orgRevisionId: pm.orgRevisionId,
lastUpdateUserId,
lastUpdateFullName,
}); });
} }
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);
} }
console.timeEnd('[AMQ] batch_create_history');
await BatchCreatePosMasterHistoryOfficer(AppDataSource.manager, historyOperations);
console.timeEnd("[AMQ] batch_create_history");
// Clone oldposMasterAct // Clone oldposMasterAct
console.time("[AMQ] clone_oldposMasterAct"); console.time('[AMQ] clone_oldposMasterAct');
for (const act of oldposMasterAct) { for (const act of oldposMasterAct) {
const parentDNA = act.posMaster?.ancestorDNA?.trim()?.toLowerCase() ?? ""; const parentDNA = act.posMaster?.ancestorDNA?.trim()?.toLowerCase() ?? '';
const childDNA = act.posMasterChild?.ancestorDNA?.trim()?.toLowerCase() ?? ""; const childDNA = act.posMasterChild?.ancestorDNA?.trim()?.toLowerCase() ?? '';
const newParentId = posMasterIdMap.get(parentDNA); const newParentId = posMasterIdMap.get(parentDNA);
const newChildId = posMasterIdMap.get(childDNA); const newChildId = posMasterIdMap.get(childDNA);
@ -898,16 +847,16 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
await posMasterActRepository.save(newAct); await posMasterActRepository.save(newAct);
} }
console.timeEnd("[AMQ] clone_oldposMasterAct"); console.timeEnd('[AMQ] clone_oldposMasterAct');
if (orgRevisionPublish != null && orgRevisionDraft != null) { if (orgRevisionPublish != null && orgRevisionDraft != null) {
console.time("[AMQ] clone_org_structure"); console.time('[AMQ] clone_org_structure');
//new main revision //new main revision
const before = null; const before = null;
//ทุก orgRoot และ orgChild ข้างล่างนี้จะเป็นตัวเก่าที่ไม่ได้เป็น current revision //ทุก orgRoot และ orgChild ข้างล่างนี้จะเป็นตัวเก่าที่ไม่ได้เป็น current revision
//cone tree //cone tree
console.time("[AMQ] query_old_org_structure"); console.time('[AMQ] query_old_org_structure');
//หา dna tree //หา dna tree
const orgRoot = await orgRootRepository.find({ const orgRoot = await orgRootRepository.find({
where: { orgRevisionId: orgRevisionPublish.id }, where: { orgRevisionId: orgRevisionPublish.id },
@ -928,27 +877,27 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
const orgChild4 = await child4Repository.find({ const orgChild4 = await child4Repository.find({
where: { orgRevisionId: orgRevisionPublish.id }, where: { orgRevisionId: orgRevisionPublish.id },
}); });
console.timeEnd("[AMQ] query_old_org_structure"); console.timeEnd('[AMQ] query_old_org_structure');
console.log( console.log(`[AMQ] Old structure - orgRoot: ${orgRoot.length}, orgChild1: ${orgChild1.length}, orgChild2: ${orgChild2.length}, orgChild3: ${orgChild3.length}, orgChild4: ${orgChild4.length}`);
`[AMQ] Old structure - orgRoot: ${orgRoot.length}, orgChild1: ${orgChild1.length}, orgChild2: ${orgChild2.length}, orgChild3: ${orgChild3.length}, orgChild4: ${orgChild4.length}`,
);
// Task #2172 ดึง orgRoot ของ revision ใหม่ // Task #2172 ดึง orgRoot ของ revision ใหม่
const newRoots = await orgRootRepository.find({ const newRoots = await orgRootRepository.find({
where: { orgRevisionId: orgRevisionDraft.id }, where: { orgRevisionId: orgRevisionDraft.id },
}); });
// สร้าง newRootMap เอาไว้เก็บ orgRoot.ancestorDNA ของ revision ใหม่ // สร้าง newRootMap เอาไว้เก็บ orgRoot.ancestorDNA ของ revision ใหม่
const newRootMap = new Map(newRoots.map((r) => [r.ancestorDNA, r.id])); const newRootMap = new Map(
newRoots.map(r => [r.ancestorDNA, r.id])
);
console.time("[AMQ] clone_permissionProfiles"); console.time('[AMQ] clone_permissionProfiles');
// ดึง permissionProfiles ของ revision เดิม // ดึง permissionProfiles ของ revision เดิม
const oldPermissionProfiles = await permissionProfilesRepository.find({ const oldPermissionProfiles = await permissionProfilesRepository.find({
relations: ["orgRootTree"], relations: ["orgRootTree"],
where: { where: {
orgRootTree: { orgRootTree: {
orgRevisionId: orgRevisionPublish.id, orgRevisionId: orgRevisionPublish.id,
}, }
}, }
}); });
const inserts: any[] = []; const inserts: any[] = [];
for (const permiss of oldPermissionProfiles) { for (const permiss of oldPermissionProfiles) {
@ -973,15 +922,15 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
if (inserts.length > 0) { if (inserts.length > 0) {
await permissionProfilesRepository.insert(inserts); await permissionProfilesRepository.insert(inserts);
} }
console.timeEnd("[AMQ] clone_permissionProfiles"); console.timeEnd('[AMQ] clone_permissionProfiles');
//หา dna posmaster ถ้าไม่มีให้เอาตัวเองเป็น dna //หา dna posmaster ถ้าไม่มีให้เอาตัวเองเป็น dna
console.time("[AMQ] query_employeePosMaster"); console.time('[AMQ] query_employeePosMaster');
const orgemployeePosMaster = await repoEmployeePosmaster.find({ const orgemployeePosMaster = await repoEmployeePosmaster.find({
where: { orgRevisionId: orgRevisionPublish.id }, where: { orgRevisionId: orgRevisionPublish.id },
relations: ["positions"], relations: ["positions"],
}); });
console.timeEnd("[AMQ] query_employeePosMaster"); console.timeEnd('[AMQ] query_employeePosMaster');
console.log(`[AMQ] orgemployeePosMaster count: ${orgemployeePosMaster.length}`); console.log(`[AMQ] orgemployeePosMaster count: ${orgemployeePosMaster.length}`);
let _orgemployeePosMaster: EmployeePosMaster[]; let _orgemployeePosMaster: EmployeePosMaster[];
@ -1006,16 +955,14 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
_orgemployeePosMaster = orgemployeePosMaster.map((x) => ({ _orgemployeePosMaster = orgemployeePosMaster.map((x) => ({
...x, ...x,
current_holderId: current_holderId:
x.current_holderId && validProfileIds.has(x.current_holderId) x.current_holderId && validProfileIds.has(x.current_holderId) ? x.current_holderId : null,
? x.current_holderId
: null,
ancestorDNA: ancestorDNA:
!x.ancestorDNA || x.ancestorDNA === "00000000-0000-0000-0000-000000000000" !x.ancestorDNA || x.ancestorDNA === "00000000-0000-0000-0000-000000000000"
? x.id ? x.id
: x.ancestorDNA, : x.ancestorDNA,
})); }));
console.time("[AMQ] insert_employeePosMaster"); console.time('[AMQ] insert_employeePosMaster');
await repoEmployeePosmaster await repoEmployeePosmaster
.createQueryBuilder() .createQueryBuilder()
.insert() .insert()
@ -1026,16 +973,16 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
overwrite: ["ancestorDNA"], overwrite: ["ancestorDNA"],
}) })
.execute(); .execute();
console.timeEnd("[AMQ] insert_employeePosMaster"); console.timeEnd('[AMQ] insert_employeePosMaster');
// } // }
//หา dna posmaster ถ้าไม่มีให้เอาตัวเองเป็น dna //หา dna posmaster ถ้าไม่มีให้เอาตัวเองเป็น dna
console.time("[AMQ] query_employeeTempPosMaster"); console.time('[AMQ] query_employeeTempPosMaster');
const orgemployeeTempPosMaster = await repoEmployeeTempPosmaster.find({ const orgemployeeTempPosMaster = await repoEmployeeTempPosmaster.find({
where: { orgRevisionId: orgRevisionPublish.id }, where: { orgRevisionId: orgRevisionPublish.id },
relations: ["positions"], relations: ["positions"],
}); });
console.timeEnd("[AMQ] query_employeeTempPosMaster"); console.timeEnd('[AMQ] query_employeeTempPosMaster');
console.log(`[AMQ] orgemployeeTempPosMaster count: ${orgemployeeTempPosMaster.length}`); console.log(`[AMQ] orgemployeeTempPosMaster count: ${orgemployeeTempPosMaster.length}`);
let _orgemployeeTempPosMaster: EmployeeTempPosMaster[]; let _orgemployeeTempPosMaster: EmployeeTempPosMaster[];
@ -1066,7 +1013,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
// } // }
//create org - forEach orgRoot (WARNING: async forEach without await) //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)`); console.log(`[AMQ] Starting forEach orgRoot loop (${orgRoot.length} items)`);
let processedOrgRoot = 0; let processedOrgRoot = 0;
orgRoot.forEach(async (x: any) => { orgRoot.forEach(async (x: any) => {
@ -1097,12 +1044,12 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
// requestBody.typeDraft.toUpperCase() == "ORG_POSITION_PERSON_ROLE" // requestBody.typeDraft.toUpperCase() == "ORG_POSITION_PERSON_ROLE"
// ) { // ) {
//create employeePosmaster //create employeePosmaster
const filteredEmployeePosMaster = _orgemployeePosMaster.filter( const filteredEmployeePosMaster = _orgemployeePosMaster
(x: EmployeePosMaster) => x.orgRootId == dataId && x.orgChild1Id == null, .filter((x: EmployeePosMaster) => x.orgRootId == dataId && x.orgChild1Id == null);
);
await Promise.all( await Promise.all(
filteredEmployeePosMaster.map(async (item: any) => { filteredEmployeePosMaster
.map(async (item: any) => {
delete item.id; delete item.id;
const employeePosMaster = Object.assign(new EmployeePosMaster(), item); const employeePosMaster = Object.assign(new EmployeePosMaster(), item);
employeePosMaster.positions = []; employeePosMaster.positions = [];
@ -1134,7 +1081,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosMaster.lastUpdatedAt = new Date(); employeePosMaster.lastUpdatedAt = new Date();
await repoEmployeePosmaster.save(employeePosMaster); await repoEmployeePosmaster.save(employeePosMaster);
for (const pos of item.positions) { //create employeePosition
item.positions.map(async (pos: any) => {
delete pos.id; delete pos.id;
const employeePosition: EmployeePosition = Object.assign( const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(), new EmployeePosition(),
@ -1154,7 +1102,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator"; employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date(); employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition); await employeePositionRepository.save(employeePosition);
} });
}), }),
); );
//create employeeTempPosmaster //create employeeTempPosmaster
@ -1193,7 +1141,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeeTempPosMaster.lastUpdatedAt = new Date(); employeeTempPosMaster.lastUpdatedAt = new Date();
await repoEmployeeTempPosmaster.save(employeeTempPosMaster); await repoEmployeeTempPosmaster.save(employeeTempPosMaster);
for (const pos of item.positions) { //create employeePosition
item.positions.map(async (pos: any) => {
delete pos.id; delete pos.id;
const employeePosition: EmployeePosition = Object.assign( const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(), new EmployeePosition(),
@ -1213,7 +1162,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator"; employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date(); employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition); await employeePositionRepository.save(employeePosition);
} });
}), }),
); );
// } // }
@ -1234,8 +1183,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
x.ancestorDNA === "00000000-0000-0000-0000-000000000000" x.ancestorDNA === "00000000-0000-0000-0000-000000000000"
) { ) {
return ( return (
i.ancestorDNA === null || i.ancestorDNA === null || i.ancestorDNA === "00000000-0000-0000-0000-000000000000"
i.ancestorDNA === "00000000-0000-0000-0000-000000000000"
); );
} }
return i.ancestorDNA === x.ancestorDNA; return i.ancestorDNA === x.ancestorDNA;
@ -1250,9 +1198,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
//create employeePosmaster //create employeePosmaster
await Promise.all( await Promise.all(
_orgemployeePosMaster _orgemployeePosMaster
.filter( .filter((x: EmployeePosMaster) => x.orgChild1Id == data1Id && x.orgChild2Id == null)
(x: EmployeePosMaster) => x.orgChild1Id == data1Id && x.orgChild2Id == null,
)
.map(async (item: any) => { .map(async (item: any) => {
delete item.id; delete item.id;
// console.log("[in case Child1] orgChild1Id == data1Id"); // console.log("[in case Child1] orgChild1Id == data1Id");
@ -1287,7 +1233,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosMaster.lastUpdatedAt = new Date(); employeePosMaster.lastUpdatedAt = new Date();
await repoEmployeePosmaster.save(employeePosMaster); await repoEmployeePosmaster.save(employeePosMaster);
for (const pos of item.positions) { //create employeePosition
item.positions.map(async (pos: any) => {
delete pos.id; delete pos.id;
const employeePosition: EmployeePosition = Object.assign( const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(), new EmployeePosition(),
@ -1307,7 +1254,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator"; employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date(); employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition); await employeePositionRepository.save(employeePosition);
} });
}), }),
); );
// create employeeTempPosmaster // create employeeTempPosmaster
@ -1349,7 +1296,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeeTempPosMaster.lastUpdatedAt = new Date(); employeeTempPosMaster.lastUpdatedAt = new Date();
await repoEmployeeTempPosmaster.save(employeeTempPosMaster); await repoEmployeeTempPosmaster.save(employeeTempPosMaster);
for (const pos of item.positions) { //create employeePosition
item.positions.map(async (pos: any) => {
delete pos.id; delete pos.id;
const employeePosition: EmployeePosition = Object.assign( const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(), new EmployeePosition(),
@ -1369,7 +1317,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator"; employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date(); employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition); await employeePositionRepository.save(employeePosition);
} });
}), }),
); );
// } // }
@ -1444,7 +1392,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosMaster.lastUpdatedAt = new Date(); employeePosMaster.lastUpdatedAt = new Date();
await repoEmployeePosmaster.save(employeePosMaster); await repoEmployeePosmaster.save(employeePosMaster);
for (const pos of item.positions) { //create employeePosition
item.positions.map(async (pos: any) => {
delete pos.id; delete pos.id;
const employeePosition: EmployeePosition = Object.assign( const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(), new EmployeePosition(),
@ -1464,7 +1413,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator"; employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date(); employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition); await employeePositionRepository.save(employeePosition);
} });
}), }),
); );
// create employeeTempPosmaster // create employeeTempPosmaster
@ -1511,7 +1460,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeeTempPosMaster.lastUpdatedAt = new Date(); employeeTempPosMaster.lastUpdatedAt = new Date();
await repoEmployeeTempPosmaster.save(employeeTempPosMaster); await repoEmployeeTempPosmaster.save(employeeTempPosMaster);
for (const pos of item.positions) { //create employeePosition
item.positions.map(async (pos: any) => {
delete pos.id; delete pos.id;
const employeePosition: EmployeePosition = Object.assign( const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(), new EmployeePosition(),
@ -1531,7 +1481,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator"; employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date(); employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition); await employeePositionRepository.save(employeePosition);
} });
}), }),
); );
// } // }
@ -1608,7 +1558,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosMaster.lastUpdatedAt = new Date(); employeePosMaster.lastUpdatedAt = new Date();
await repoEmployeePosmaster.save(employeePosMaster); await repoEmployeePosmaster.save(employeePosMaster);
for (const pos of item.positions) { //create employeePosition
item.positions.map(async (pos: any) => {
delete pos.id; delete pos.id;
const employeePosition: EmployeePosition = Object.assign( const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(), new EmployeePosition(),
@ -1628,7 +1579,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator"; employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date(); employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition); await employeePositionRepository.save(employeePosition);
} });
}), }),
); );
// create employeeTempPosmaster // create employeeTempPosmaster
@ -1676,7 +1627,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeeTempPosMaster.lastUpdatedAt = new Date(); employeeTempPosMaster.lastUpdatedAt = new Date();
await repoEmployeeTempPosmaster.save(employeeTempPosMaster); await repoEmployeeTempPosmaster.save(employeeTempPosMaster);
for (const pos of item.positions) { //create employeePosition
item.positions.map(async (pos: any) => {
delete pos.id; delete pos.id;
const employeePosition: EmployeePosition = Object.assign( const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(), new EmployeePosition(),
@ -1696,7 +1648,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator"; employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date(); employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition); await employeePositionRepository.save(employeePosition);
} });
}), }),
); );
// } // }
@ -1774,7 +1726,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosMaster.lastUpdatedAt = new Date(); employeePosMaster.lastUpdatedAt = new Date();
await repoEmployeePosmaster.save(employeePosMaster); await repoEmployeePosmaster.save(employeePosMaster);
for (const pos of item.positions) { //create employeePosition
item.positions.map(async (pos: any) => {
delete pos.id; delete pos.id;
const employeePosition: EmployeePosition = Object.assign( const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(), new EmployeePosition(),
@ -1794,7 +1747,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator"; employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date(); employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition); await employeePositionRepository.save(employeePosition);
} });
}), }),
); );
//create employeeTempPosmaster //create employeeTempPosmaster
@ -1840,7 +1793,8 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeeTempPosMaster.lastUpdatedAt = new Date(); employeeTempPosMaster.lastUpdatedAt = new Date();
await repoEmployeeTempPosmaster.save(employeeTempPosMaster); await repoEmployeeTempPosmaster.save(employeeTempPosMaster);
for (const pos of item.positions) { //create employeePosition
item.positions.map(async (pos: any) => {
delete pos.id; delete pos.id;
const employeePosition: EmployeePosition = Object.assign( const employeePosition: EmployeePosition = Object.assign(
new EmployeePosition(), new EmployeePosition(),
@ -1860,7 +1814,7 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
employeePosition.lastUpdateFullName = "System Administrator"; employeePosition.lastUpdateFullName = "System Administrator";
employeePosition.lastUpdatedAt = new Date(); employeePosition.lastUpdatedAt = new Date();
await employeePositionRepository.save(employeePosition); await employeePositionRepository.save(employeePosition);
} });
}), }),
); );
// } // }
@ -1933,10 +1887,10 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
{ userId: user?.sub }, { userId: user?.sub },
).catch(console.error); ).catch(console.error);
} }
console.timeEnd("[AMQ] clone_org_structure"); console.timeEnd('[AMQ] clone_org_structure');
// อัปเดตสถานะ orgRevision หลังจากทำงานเสร็จทั้งหมด // อัปเดตสถานะ orgRevision หลังจากทำงานเสร็จทั้งหมด
console.time("[AMQ] save_revision_status"); console.time('[AMQ] save_revision_status');
orgRevisionPublish.orgRevisionIsDraft = false; orgRevisionPublish.orgRevisionIsDraft = false;
orgRevisionPublish.orgRevisionIsCurrent = false; orgRevisionPublish.orgRevisionIsCurrent = false;
await repoOrgRevision.save(orgRevisionPublish); await repoOrgRevision.save(orgRevisionPublish);
@ -1944,29 +1898,26 @@ async function handler_org(msg: amqp.ConsumeMessage): Promise<boolean> {
orgRevisionDraft.orgRevisionIsCurrent = true; orgRevisionDraft.orgRevisionIsCurrent = true;
orgRevisionDraft.orgRevisionIsDraft = false; orgRevisionDraft.orgRevisionIsDraft = false;
await repoOrgRevision.save(orgRevisionDraft); 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.log(`[AMQ] handler_org SUCCESS - Total time: ${Date.now() - startTime}ms`);
console.timeEnd("[AMQ] handler_org_total"); console.timeEnd('[AMQ] handler_org_total');
return true; return true;
}); // ✅ END TRANSACTION - All operations succeeded, data is committed
} catch (error) { } catch (error) {
// ✅ TRANSACTION AUTOMATICALLY ROLLED BACK - No data was saved
const totalTime = Date.now() - startTime; const totalTime = Date.now() - startTime;
console.error(`[AMQ] handler_org ERROR after ${totalTime}ms:`, error); console.error(`[AMQ] handler_org ERROR after ${totalTime}ms:`, error);
console.error("[AMQ] Transaction rolled back - all changes were undone");
if (user) { if (user) {
sendWebSocket( sendWebSocket(
"send-publish-org", "send-publish-org",
{ {
success: false, success: false,
message: `เผยแพร่โครงสร้างหน่วยงานไม่สำเร็จ: ${error instanceof Error ? error.message : String(error)}`, message: `ระบบทำการเผยแพร่โครงสร้างหน่วยงานไม่สำเร็จ`,
}, },
{ userId: user?.sub }, { userId: user?.sub },
).catch(console.error); ).catch(console.error);
} }
console.timeEnd("[AMQ] handler_org_total"); console.timeEnd('[AMQ] handler_org_total');
throw error; // ✅ Re-throw to be caught by createConsumer's try-catch return false;
} }
} }
@ -2640,8 +2591,7 @@ async function handler_org_draft(msg: amqp.ConsumeMessage): Promise<boolean> {
}); });
await posMasterAssignRepository.delete({ posMasterId: In(_posMasters.map((x) => x.id)) }); 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: In(_posMasters.map((x) => x.id)) }); //ใช้ posMasterId ของ revision: draft *แต่ยังไม่เจอช็อดไหนที่ใช้โครงสร้างแบบร่างในรักษาการแทน
await posMasterActRepository.delete({ await posMasterActRepository.delete({ //ใช้ posMasterId ของ revision: draft *แต่ยังไม่เจอช็อดไหนที่ใช้โครงสร้างแบบร่างในรักษาการแทน
//ใช้ posMasterId ของ revision: draft *แต่ยังไม่เจอช็อดไหนที่ใช้โครงสร้างแบบร่างในรักษาการแทน
posMasterChildId: In(_posMasters.map((x) => x.id)), posMasterChildId: In(_posMasters.map((x) => x.id)),
}); });
// await posMasterRepository.remove(_posMasters); // await posMasterRepository.remove(_posMasters);
@ -2669,26 +2619,24 @@ async function handler_org_draft(msg: amqp.ConsumeMessage): Promise<boolean> {
await child2Repository.delete({ orgRevisionId: In(_orgRevisions.map((x) => x.id)) }); await child2Repository.delete({ orgRevisionId: In(_orgRevisions.map((x) => x.id)) });
await child1Repository.delete({ orgRevisionId: In(_orgRevisions.map((x) => x.id)) }); await child1Repository.delete({ orgRevisionId: In(_orgRevisions.map((x) => x.id)) });
// Task #2160 อัพเดทหน้าที่จัดการโครงสร้างแบบร่าง // Task #2160 อัพเดทหน้าที่จัดการโครงสร้างแบบร่าง
if ( if (["ORG", "ORG_POSITION", "ORG_POSITION_PERSON", "ORG_POSITION_ROLE", "ORG_POSITION_PERSON_ROLE"].includes(requestBody.typeDraft?.toUpperCase())) {
[
"ORG",
"ORG_POSITION",
"ORG_POSITION_PERSON",
"ORG_POSITION_ROLE",
"ORG_POSITION_PERSON_ROLE",
].includes(requestBody.typeDraft?.toUpperCase())
) {
const _newRoots = await orgRootRepository.find({ 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) { for (const oldRoot of _roots) {
const newRootId = newRootMap.get(oldRoot.ancestorDNA); const newRootId = newRootMap.get(oldRoot.ancestorDNA);
if (!newRootId) continue; if (!newRootId) continue;
// อัพเดท orgRootId ที่อยู่ภายใต้ orgRevision แบบร่างเดิมเป็นของ orgRevision แบบร่างใหม่ // อัพเดท orgRootId ที่อยู่ภายใต้ orgRevision แบบร่างเดิมเป็นของ orgRevision แบบร่างใหม่
await permissionOrgRepository.update({ orgRootId: oldRoot.id }, { orgRootId: newRootId }); await permissionOrgRepository.update(
{ orgRootId: oldRoot.id },
{ orgRootId: newRootId }
);
} }
} else { }
else {
await permissionOrgRepository.delete({ await permissionOrgRepository.delete({
orgRootId: In(_roots.map((x) => x.id)), orgRootId: In(_roots.map((x) => x.id)),
}); });