[ExecuteSalaryCurrentService] ครอบ transaction + respone success/fail count

src/controllers/CommandController.ts

@@ -106,7 +106,7 @@ import { reOrderCommandRecivesAndDelete } from "../services/CommandService";
 import { RetirementService } from "../services/RetirementService";
 import { ExecuteOfficerProfileService } from "../services/ExecuteOfficerProfileService";
 import { ExecuteSalaryService } from "../services/ExecuteSalaryService";
-import { ExecuteSalaryCurrentService } from "../services/ExecuteSalaryCurrentService";
+import { ExecuteSalaryCurrentService, ExecuteSalaryResult } from "../services/ExecuteSalaryCurrentService";
 import { ExecuteSalaryEmployeeCurrentService } from "../services/ExecuteSalaryEmployeeCurrentService";
 import { ExecuteSalaryLeaveService } from "../services/ExecuteSalaryLeaveService";
 import { ExecuteSalaryEmployeeLeaveService } from "../services/ExecuteSalaryEmployeeLeaveService";
@@ -3702,11 +3702,14 @@ export class CommandController extends Controller {
       }[];
     },
   ) {
-    await new ExecuteSalaryCurrentService().executeSalaryCurrent(body.data, {
+    const result: ExecuteSalaryResult = await new ExecuteSalaryCurrentService().executeSalaryCurrent(
+      body.data,
+      {
         user: { sub: req.user.sub, name: req.user.name },
         req,
-    });
+      },
-    return new HttpSuccess();
+    );
+    return new HttpSuccess(result);
   }
 
   @Post("excexute/salary-employee-current")

src/services/ExecuteSalaryCurrentService.ts

@@ -1,4 +1,4 @@
-import { Double } from "typeorm";
+import { Double, EntityManager } from "typeorm";
 import { AppDataSource } from "../database/data-source";
 import HttpError from "../interfaces/http-error";
 import HttpStatusCode from "../interfaces/http-status";
@@ -60,6 +60,16 @@ export interface SalaryCurrentExecutionContext {
   req?: any;
 }
 
+/**
+ * ผลลัพธ์การประมวลผล batch — แต่ละคนทำงานแบบ independent
+ * คนที่ fail จะ rollback เฉพาะตัว (per-item transaction) ไม่กระทบคนอื่น
+ */
+export interface ExecuteSalaryResult {
+  successCount: number;
+  failureCount: number;
+  failures: { profileId: string; reason: string }[];
+}
+
 /**
  * Service สำหรับสร้าง ProfileSalary ของข้าราชการ + อัปเดตตำแหน่งปัจจุบัน (เปลี่ยนตำแหน่ง)
  *
@@ -69,28 +79,29 @@ export interface SalaryCurrentExecutionContext {
  * - consumer ใน rabbitmq handler เรียกผ่าน service นี้โดยตรง (Linear Flow)
  *
  * Behavior ทั้งหมด preserve จาก CommandController.newSalaryAndUpdateCurrent ต้นฉบับ
+ *
+ * Batch semantics: ประมวลผลทุกคนแบบ sequential (ทีละคน) แต่ละคนครอบด้วย
+ * transaction ของตัวเอง เพื่อกัน race condition เมื่อหลายคนใน batch อ้างอิง
+ * posMaster/position ตัวเดียวกัน — คนที่ throw จะ rollback เฉพาะตัว ไม่กระทบคนอื่น
+ * ผลลัพธ์รายงานเป็น success/failure count + รายชื่อคนที่ fail
  */
 export class ExecuteSalaryCurrentService {
   private commandRepository = AppDataSource.getRepository(Command);
   private profileRepository = AppDataSource.getRepository(Profile);
-  private salaryRepo = AppDataSource.getRepository(ProfileSalary);
-  private salaryHistoryRepo = AppDataSource.getRepository(ProfileSalaryHistory);
-  private posMasterRepository = AppDataSource.getRepository(PosMaster);
-  private positionRepository = AppDataSource.getRepository(Position);
   private orgRootRepository = AppDataSource.getRepository(OrgRoot);
 
   /**
-   * ประมวลผลสร้าง ProfileSalary + อัปเดตตำแหน่งปัจจุบันของข้าราชการ
+   * ประมวลผลสร้าง ProfileSalary + อัปเดตตำแหน่งปัจจุบันของข้าราชการทั้ง batch
+   *
+   * @returns สรุปผล success/failure ต่อคน
    */
   async executeSalaryCurrent(
     data: SalaryCurrentItem[],
     ctx: SalaryCurrentExecutionContext,
-  ): Promise<void> {
+  ): Promise<ExecuteSalaryResult> {
     console.log("[ExecuteSalaryCurrentService] Starting executeSalaryCurrent");
     console.log("[ExecuteSalaryCurrentService] Request body count:", data?.length);
 
-    const req = ctx.req;
-
     // ─────────────────────────────────────────────────────────────
     // Normalize date fields (ผ่าน handler จะได้ string → ต้องแปลงเป็น Date)
     // ─────────────────────────────────────────────────────────────
@@ -149,14 +160,69 @@ export class ExecuteSalaryCurrentService {
             .orgRootShortName ?? "";
       }
     }
-    await Promise.all(
+
-      data.map(async (item) => {
+    // ─────────────────────────────────────────────────────────────
-        const profile: any = await this.profileRepository.findOneBy({ id: item.profileId });
+    // Per-item transaction: แต่ละคนมี transaction ของตัวเอง (sequential)
+    // ประมวลทีละคนเพื่อกัน race condition เมื่อหลายคนใน batch อ้างอิง
+    // posMaster/position ตัวเดียวกัน คนที่ throw จะ rollback เฉพาะตัว (manager)
+    // และไม่กระทบคนอื่นใน batch
+    // ─────────────────────────────────────────────────────────────
+    const failures: ExecuteSalaryResult["failures"] = [];
+    let successCount = 0;
+    for (const item of data ?? []) {
+      try {
+        await AppDataSource.transaction(async (manager) => {
+          await this.processOne(item, ctx, manager, _posNumCodeSit, _posNumCodeSitAbb);
+        });
+        successCount++;
+      } catch (err) {
+        const reason =
+          err instanceof HttpError
+            ? err.message
+            : err instanceof Error
+              ? err.message
+              : "unexpected error";
+        console.error(
+          `[ExecuteSalaryCurrentService] Failed profileId=${item.profileId}: ${reason}`,
+          err,
+        );
+        failures.push({ profileId: item.profileId ?? "unknown", reason });
+      }
+    }
+
+    console.log(
+      `[ExecuteSalaryCurrentService] executeSalaryCurrent completed — success: ${successCount}, failure: ${failures.length}`,
+    );
+
+    return { successCount, failureCount: failures.length, failures };
+  }
+
+  /**
+   * ประมวลผล 1 คน ภายใน transaction เดียว (manager)
+   * ทุก save ใช้ manager.getRepository(...) เพื่อให้อยู่ใน transaction เดียวกัน
+   * ถ้า throw ระหว่างทาง → rollback ทั้งหมดของคนนี้ (กัน partial commit)
+   */
+  private async processOne(
+    item: SalaryCurrentItem,
+    ctx: SalaryCurrentExecutionContext,
+    manager: EntityManager,
+    _posNumCodeSit: string,
+    _posNumCodeSitAbb: string,
+  ): Promise<void> {
+    const req = ctx.req;
+
+    const profileRepository = manager.getRepository(Profile);
+    const salaryRepo = manager.getRepository(ProfileSalary);
+    const salaryHistoryRepo = manager.getRepository(ProfileSalaryHistory);
+    const posMasterRepository = manager.getRepository(PosMaster);
+    const positionRepository = manager.getRepository(Position);
+
+    const profile: any = await profileRepository.findOneBy({ id: item.profileId });
     if (!profile) {
       throw new HttpError(HttpStatusCode.NOT_FOUND, "ไม่พบข้อมูลทะเบียนประวัตินี้");
     }
     let _null: any = null;
-        const dest_item = await this.salaryRepo.findOne({
+    const dest_item = await salaryRepo.findOne({
       where: { profileId: item.profileId },
       order: { order: "DESC" },
     });
@@ -177,14 +243,14 @@ export class ExecuteSalaryCurrentService {
     Object.assign(dataSalary, { ...item, ...meta });
     const history = new ProfileSalaryHistory();
     Object.assign(history, { ...dataSalary, id: undefined });
-        await this.salaryRepo.save(dataSalary, { data: req });
+    await salaryRepo.save(dataSalary, { data: req });
     setLogDataDiff(req, { before, after: dataSalary });
     history.commandId = item.commandId ?? _null;
     history.profileSalaryId = dataSalary.id;
-        await this.salaryHistoryRepo.save(history, { data: req });
+    await salaryHistoryRepo.save(history, { data: req });
 
     // STEP 1: หา posMaster ที่จะใช้งานตาม id ที่ส่งมา
-        let posMaster = await this.posMasterRepository.findOne({
+    let posMaster = await posMasterRepository.findOne({
       where: { id: item.posmasterId },
       relations: {
         orgRevision: true,
@@ -203,7 +269,7 @@ export class ExecuteSalaryCurrentService {
 
     // ถ้าไม่อยู่ในโครงสร้างปัจจุบัน ให้หาตัวใหม่จาก ancestorDNA
     if (!isCurrent && posMaster?.ancestorDNA) {
-          posMaster = await this.posMasterRepository.findOne({
+      posMaster = await posMasterRepository.findOne({
         where: {
           ancestorDNA: posMaster.ancestorDNA,
           orgRevision: {
@@ -229,7 +295,7 @@ export class ExecuteSalaryCurrentService {
       throw new HttpError(HttpStatusCode.NOT_FOUND, "ไม่พบข้อมูลตำแหน่งนี้");
     }
 
-        const posMasterOld = await this.posMasterRepository.findOne({
+    const posMasterOld = await posMasterRepository.findOne({
       where: {
         current_holderId: item.profileId,
         orgRevisionId: posMaster.orgRevisionId,
@@ -240,7 +306,7 @@ export class ExecuteSalaryCurrentService {
       posMasterOld.lastUpdatedAt = new Date();
     }
 
-        const positionOld = await this.positionRepository.findOne({
+    const positionOld = await positionRepository.findOne({
       where: {
         posMasterId: posMasterOld?.id,
         positionIsSelected: true,
@@ -255,10 +321,10 @@ export class ExecuteSalaryCurrentService {
       });
 
       positionOld.positionIsSelected = false;
-          await this.positionRepository.save(positionOld);
+      await positionRepository.save(positionOld);
     }
 
-        const checkPosition = await this.positionRepository.find({
+    const checkPosition = await positionRepository.find({
       where: {
         posMasterId: posMaster!.id, // ใช้ posMaster ตัวใหม่ (ที่อาจจะเปลี่ยนจาก ancestorDNA)
         positionIsSelected: true,
@@ -282,7 +348,7 @@ export class ExecuteSalaryCurrentService {
           positionIsSelected: false,
         };
       });
-          await this.positionRepository.save(clearPosition);
+      await positionRepository.save(clearPosition);
     }
 
     posMaster.current_holderId = item.profileId;
@@ -290,10 +356,11 @@ export class ExecuteSalaryCurrentService {
     // posMaster.conditionReason = _null;
     // posMaster.isCondition = false;
     if (posMasterOld != null) {
-          await this.posMasterRepository.save(posMasterOld);
+      await posMasterRepository.save(posMasterOld);
-          await CreatePosMasterHistoryOfficer(posMasterOld.id, req);
+      // ส่ง manager เข้าไปเพื่อให้อยู่ใน transaction เดียวกัน
+      await CreatePosMasterHistoryOfficer(posMasterOld.id, req, null, null, manager);
     }
-        await this.posMasterRepository.save(posMaster);
+    await posMasterRepository.save(posMaster);
 
     // STEP 2: กำหนด position ใหม่
     // Match position ตามลำดับ priority:
@@ -312,7 +379,7 @@ export class ExecuteSalaryCurrentService {
     // CONDITION 1: เช็คจาก positionId ตรง
     // ═══════════════════════════════════════════════════════════
     if (item.positionId) {
-          const positionById = await this.positionRepository.findOne({
+      const positionById = await positionRepository.findOne({
         where: {
           id: item.positionId,
           posMasterId: posMaster.id, // ต้องอยู่ใน posMaster ที่ถูกต้อง
@@ -351,7 +418,7 @@ export class ExecuteSalaryCurrentService {
         whereCondition.positionArea = item.positionArea;
       }
 
-          const positionBy7Fields = await this.positionRepository.findOne({
+      const positionBy7Fields = await positionRepository.findOne({
         where: whereCondition,
         relations: ["posExecutive"],
         order: { orderNo: "ASC" },
@@ -366,7 +433,7 @@ export class ExecuteSalaryCurrentService {
     // CONDITION 3: Match 3 ฟิลด์ (ถ้า Condition 2 ไม่ match)
     // ═══════════════════════════════════════════════════════════
     if (!positionNew && item.positionName && posTypeId && posLevelId) {
-          const positionBy3Fields = await this.positionRepository.findOne({
+      const positionBy3Fields = await positionRepository.findOne({
         where: {
           posMasterId: posMaster.id,
           positionName: item.positionName,
@@ -386,7 +453,7 @@ export class ExecuteSalaryCurrentService {
     // // FALLBACK: ถ้าทั้ง 3 ไม่ match ให้เลือก position แรกใน posMaster
     // // ═══════════════════════════════════════════════════════════
     // if (!positionNew) {
-        //   const fallbackPositions = await this.positionRepository.find({
+    //   const fallbackPositions = await positionRepository.find({
     //     where: {
     //       posMasterId: posMaster.id,
     //     },
@@ -419,13 +486,10 @@ export class ExecuteSalaryCurrentService {
       }
       profile.amount = item.amount ?? null;
       profile.amountSpecial = item.amountSpecial ?? null;
-          await this.profileRepository.save(profile);
+      await profileRepository.save(profile);
-          await this.positionRepository.save(positionNew);
+      await positionRepository.save(positionNew);
     }
-        await CreatePosMasterHistoryOfficer(posMaster.id, req);
+    // ส่ง manager เข้าไปเพื่อให้อยู่ใน transaction เดียวกัน
-      }),
+    await CreatePosMasterHistoryOfficer(posMaster.id, req, null, null, manager);
-    );
-
-    console.log("[ExecuteSalaryCurrentService] executeSalaryCurrent completed successfully");
   }
 }

src/services/rabbitmq.ts

@@ -388,8 +388,16 @@ async function handler(msg: amqp.ConsumeMessage): Promise<boolean> {
           await new ExecuteOfficerProfileService().executeCreateOfficerProfile(resultData, ctx);
           console.log(`[AMQ] Processed ${resultData.length} profiles via ExecuteOfficerProfileService`);
         } else if (isSalaryCurrent) {
-          await new ExecuteSalaryCurrentService().executeSalaryCurrent(resultData, ctx);
+          const salaryResult = await new ExecuteSalaryCurrentService().executeSalaryCurrent(
-          console.log(`[AMQ] Processed ${resultData.length} profiles via ExecuteSalaryCurrentService`);
+            resultData,
+            ctx,
+          );
+          console.log(
+            `[AMQ] Processed via ExecuteSalaryCurrentService — success: ${salaryResult.successCount}, failure: ${salaryResult.failureCount}`,
+          );
+          for (const f of salaryResult.failures) {
+            console.error(`[AMQ] ExecuteSalaryCurrentService failed profileId=${f.profileId}: ${f.reason}`);
+          }
         } else if (isSalaryEmployeeCurrent) {
           await new ExecuteSalaryEmployeeCurrentService().executeSalaryEmployeeCurrent(resultData, ctx);
           console.log(`[AMQ] Processed ${resultData.length} profiles via ExecuteSalaryEmployeeCurrentService`);