Merge branch 'develop' of github.com:Frappet/bma-ehr-organization into develop

src/app.ts

@@ -5,11 +5,14 @@ import express from "express";
 import swaggerUi from "swagger-ui-express";
 import swaggerDocument from "./swagger.json";
 import * as cron from "node-cron";
+import { init as rabbitmqInit } from './services/rabbitmq'; 
 import error from "./middlewares/error";
 import { AppDataSource } from "./database/data-source";
 import { RegisterRoutes } from "./routes";
 import { OrganizationController } from "./controllers/OrganizationController";
 import logMiddleware from "./middlewares/logs";
+import { run } from "node:test";
+import { CommandController } from "./controllers/CommandController";
 
 async function main() {
   await AppDataSource.initialize();
@@ -43,6 +46,17 @@ async function main() {
     }
   });
 
+  const cronTime_command = "0 2 * * * *";
+  // const cronTime_command = "*/10 * * * * *";
+  cron.schedule(cronTime_command, async () => {
+    try {
+      const commandController = new CommandController();
+      await commandController.cronjobCommand();
+    } catch (error) {
+      console.error("Error executing function from controller:", error);
+    }
+  });
+
   // app.listen(APP_PORT, APP_HOST, () => console.log(`Listening on: http://localhost:${APP_PORT}`));
   app.listen(
     APP_PORT,
@@ -52,6 +66,16 @@ async function main() {
       console.log(`[APP] Swagger on: http://localhost:${APP_PORT}/api-docs`)
     ),
   );
+  async function runMessageQueue() {
+    try {
+      await rabbitmqInit();
+    } catch (e) {
+      console.log(e);
+      setTimeout(runMessageQueue, 1000);
+    }
+  }
+
+  runMessageQueue()
 }
 
 main();

src/controllers/CommandController.ts

@@ -19,7 +19,7 @@ import HttpSuccess from "../interfaces/http-success";
 import HttpStatusCode from "../interfaces/http-status";
 import HttpError from "../interfaces/http-error";
 import { Command } from "../entities/Command";
-import { Brackets, LessThan, MoreThan, Double, In, Not } from "typeorm";
+import { Brackets, LessThan, MoreThan, Double, In, Not ,Between } from "typeorm";
 import { CommandType } from "../entities/CommandType";
 import { CommandSend } from "../entities/CommandSend";
 import { Profile, CreateProfileAllFields } from "../entities/Profile";
@@ -47,6 +47,7 @@ import { EmployeePosMaster } from "../entities/EmployeePosMaster";
 import { ProfileDiscipline } from "../entities/ProfileDiscipline";
 import { ProfileDisciplineHistory } from "../entities/ProfileDisciplineHistory";
 import { PosMasterAct } from "../entities/PosMasterAct";
+import { sendToQueue } from "../services/rabbitmq";
 import { PosLevel } from "../entities/PosLevel";
 import { PosType } from "../entities/PosType";
 import { addUserRoles, createUser, getRoles } from "../keycloak";
@@ -1038,34 +1039,86 @@ export class CommandController extends Controller {
       )
     ) {
       command.status = "WAITING";
+      command.lastUpdateUserId = request.user.sub;
+      command.lastUpdateFullName = request.user.name;
+      command.lastUpdatedAt = new Date();
+      await this.commandRepository.save(command);
     } else {
       const path = this.commandTypePath(command.commandType.code);
       if (path == null) throw new HttpError(HttpStatusCode.NOT_FOUND, "ไม่พบประเภทคำสั่งนี้ในระบบ");
-
-      await new CallAPI()
-        .PostData(request, path + "/excecute", {
-          refIds: command.commandRecives
-            .filter((x) => x.refId != null)
-            .map((x) => ({
-              refId: x.refId,
-              commandAffectDate: command.commandAffectDate,
-              commandNo: command.commandNo,
-              commandYear: command.commandYear,
-              templateDoc: command.positionDetail,
-              amount: x.amount,
-              positionSalaryAmount: x.positionSalaryAmount,
-              mouthSalaryAmount: x.mouthSalaryAmount,
-            })),
-        })
-        .then(async (res) => {
-          command.status = "REPORTED";
-        })
-        .catch((e) => {});
+      const msg = {
+        data: {
+          id: command.id,
+          status: "REPORTED",
+          lastUpdateUserId: request.user.sub,
+          lastUpdateFullName: request.user.name,
+          lastUpdatedAt: new Date(),
+        },
+        user: request.user,
+        token: request.headers["authorization"],
+      };
+      sendToQueue(msg);
     }
-    command.lastUpdateUserId = request.user.sub;
-    command.lastUpdateFullName = request.user.name;
-    command.lastUpdatedAt = new Date();
-    await this.commandRepository.save(command);
+    return new HttpSuccess();
+  }
+
+  async cronjobCommand(@Request() request?: RequestWithUser) {
+    console.log(request);
+    const today = new Date();
+    today.setHours(7, 0, 0, 0); //+7 เพื่อให้ตรง local time (อาจจะต้องใช้ moment)
+    const tomorrow = new Date(today);
+    tomorrow.setDate(tomorrow.getDate() + 1);
+
+    const command = await this.commandRepository.find({
+      relations: ["commandType", "commandRecives"],
+      where:{
+        commandExcecuteDate: Between(today, tomorrow),
+        status: "WAITING" 
+      }
+    });
+    
+    const data = {
+      client_id: "gettoken",
+      client_secret: process.env.AUTH_ACCOUNT_SECRET,
+      grant_type: "password",
+      requested_token_type: "urn:ietf:params:oauth:token-type:refresh_token",
+      username: process.env.USERNAME_,
+      password: process.env.PASSWORD_,
+    };
+    let _data: any = null;
+    await Promise.all([
+      await new CallAPI()
+        .PostDataKeycloak("/realms/bma-ehr/protocol/openid-connect/token", data)
+        .then(async (x) => {
+          _data = x;
+        })
+        .catch(async (x) => {
+          throw new HttpError(HttpStatus.UNAUTHORIZED, "ชื่อผู้ใช้งานหรือรหัสผ่านไม่ถูกต้อง");
+        }),
+    ]);
+    if (_data == null) {
+      return new HttpError(HttpStatus.UNAUTHORIZED, "ชื่อผู้ใช้งานหรือรหัสผ่านไม่ถูกต้อง");
+    } 
+
+    command.forEach(async (x) => {
+    const path = this.commandTypePath(x.commandType.code);
+    if (path == null) throw new HttpError(HttpStatusCode.NOT_FOUND, "ไม่พบประเภทคำสั่งนี้ในระบบ");
+    const msg = {
+      data: {
+        id: x.id,
+        status: "REPORTED",
+        lastUpdateUserId: "system",
+        lastUpdateFullName: "system",
+        // lastUpdateUserId: _data.user.sub,
+        // lastUpdateFullName: _data.user.name,
+        lastUpdatedAt: new Date(),
+      },
+      user: _data.user,
+      token: _data.access_token,
+    };
+    sendToQueue(msg);
+    })
+    
     return new HttpSuccess();
   }
 
@@ -2704,6 +2757,7 @@ export class CommandController extends Controller {
   ) {
     return new HttpSuccess();
   }
+
   commandTypePath(commandCode: string) {
     switch (commandCode) {
       case "C-PM-01":

src/interfaces/call-api.ts

@@ -41,7 +41,7 @@ class CallAPI {
     }
   }
   //Post
-  public async PostData(request: any, @Path() path: any, sendData: any) {
+  public async PostData(request: any, @Path() path: any, sendData: any, log = true) {
     const token = "Bearer " + request.headers.authorization.replace("Bearer ", "");
     const url = process.env.API_URL + path;
     try {
@@ -52,7 +52,7 @@ class CallAPI {
           api_key: process.env.API_KEY,
         },
       });
-      addLogSequence(request, {
+      if (log) addLogSequence(request, {
         action: "request",
         status: "success",
         description: "connected",
@@ -65,7 +65,7 @@ class CallAPI {
       });
       return response.data.result;
     } catch (error) {
-      addLogSequence(request, {
+      if (log) addLogSequence(request, {
         action: "request",
         status: "error",
         description: "unconnected",
@@ -79,6 +79,7 @@ class CallAPI {
       throw error;
     }
   }
+
   //Post
   public async PostDataKeycloak(@Path() path: any, sendData: any) {
     // const token = request.headers.authorization;

src/interfaces/utils.ts

@@ -209,3 +209,93 @@ export function addLogSequence(req: RequestWithUser, data: LogSequence) {
 export function editLogSequence(req: RequestWithUser, index: number, data: LogSequence) {
   req.app.locals.logData.sequence[index] = data;
 }
+
+export function commandTypePath(commandCode: string): string | null {
+  switch (commandCode) {
+    case "C-PM-01":
+      return "/placement/recruit/report";
+    case "C-PM-02":
+      return "/placement/candidate/report";
+    case "C-PM-03":
+      return "/placement/appoint/report";
+    case "C-PM-04":
+      return "/placement/move/report";
+    case "C-PM-05":
+      return "/placement/appointment/appoint/report";
+    case "C-PM-06":
+      return "/placement/appointment/slip/report";
+    case "C-PM-07":
+      return "/placement/appointment/move/report";
+    case "C-PM-08":
+      return "/retirement/other/appoint/report";
+    case "C-PM-09":
+      return "/retirement/other/out/report";
+    case "C-PM-10":
+      return "/xxxxxx";
+    case "C-PM-11":
+      return "/probation/report/command11/officer/report";
+    case "C-PM-12":
+      return "/probation/report/command12/officer/report";
+    case "C-PM-13":
+      return "/placement/transfer/command/report";
+    case "C-PM-14":
+      return "/placement/Receive/command/report";
+    case "C-PM-15":
+      return "/placement/officer/command/report";
+    case "C-PM-16":
+      return "/placement/repatriation/command/report";
+    case "C-PM-17":
+      return "/retirement/resign/command/report";
+    case "C-PM-18":
+      return "/retirement/out/command/report";
+    case "C-PM-19":
+      return "/discipline/result/command19/report";
+    case "C-PM-20":
+      return "/discipline/result/command20/report";
+    case "C-PM-21":
+      return "/org/command/command21/employee/report";
+    case "C-PM-22":
+      return "/placement/appointment/employee-appoint/report";
+    case "C-PM-23":
+      return "/retirement/resign/employee/report";
+    case "C-PM-24":
+      return "/placement/appointment/employee-move/report";
+    case "C-PM-25":
+      return "/discipline/result/command25/report";
+    case "C-PM-26":
+      return "/discipline/result/command26/report";
+    case "C-PM-27":
+      return "/discipline/result/command27/report";
+    case "C-PM-28":
+      return "/discipline/result/command28/report";
+    case "C-PM-29":
+      return "/discipline/result/command29/report";
+    case "C-PM-30":
+      return "/discipline/result/command30/report";
+    case "C-PM-31":
+      return "/discipline/result/command31/report";
+    case "C-PM-32":
+      return "/discipline/result/command32/report";
+    case "C-PM-33":
+      return "/salary/report/command/officer/report";
+    case "C-PM-34":
+      return "/salary/report/command/officer/report";
+    case "C-PM-35":
+      return "/salary/report/command/officer/report";
+    case "C-PM-36":
+      return "/salary/report/command/employee/report";
+    case "C-PM-37":
+      return "/salary/report/command/employee/report";
+    case "C-PM-38":
+      return "/org/command/command38/officer/report";
+    case "C-PM-39":
+      return "/placement/slip/report";
+    case "C-PM-40":
+      return "/org/command/command40/officer/report";
+    case "C-PM-41":
+      return "/retirement/resign/leave-cancel/report";
+    default:
+      return null;
+  }
+}
+

src/services/rabbitmq.ts

@@ -0,0 +1,126 @@
+import amqp from "amqplib";
+import { AppDataSource } from "../database/data-source";
+import { Command } from "../entities/Command";
+import { commandTypePath } from "../interfaces/utils";
+import CallAPI from "../interfaces/call-api";
+import HttpError from "../interfaces/http-error";
+import HttpStatusCode from "../interfaces/http-status";
+import { RequestWithUser } from "../middlewares/user";
+
+export let sendToQueue: (payload: any) => void;
+
+export async function init() { //----> (1) Producer
+  if (!process.env.AMQ_URL || !process.env.AMQ_QUEUE) return;
+
+  const { AMQ_URL: url, AMQ_QUEUE: queue } = process.env; //----> (1.2) get url and queue from .env
+
+  const connection = await amqp.connect(url); //----> (1.3) set up url with amqp protocol
+
+  const channel = await connection.createChannel(); //----> (1.4) create Channel
+
+  channel.assertQueue(queue, { durable: true }); //----> (1.5) assert queue and set durable (if "true" save to disk on RabbitMQ)
+  channel.prefetch(1);
+
+  sendToQueue = (payload: any, persistent = true) => { //----> (2) sendQueue To RabbitMQ and set persistent (if "true" redo the failed queue when server run again)
+    channel.sendToQueue(queue, Buffer.from(JSON.stringify(payload)), {
+      persistent,
+    });
+  };
+
+  console.log("[AMQ] Listening for message...");
+
+  createConsumer(queue, channel, handler); //----> (3) Process Consumer
+  // createConsumer(queue1, channel, handler1);
+  // createConsumer(queue2, channel, handler2);
+}
+
+function createConsumer( //----> consumer
+  queue: string,
+  channel: amqp.Channel,
+  handler: (msg: amqp.ConsumeMessage) => Promise<boolean> | boolean,
+) {
+  channel.consume(
+    queue,
+    async (msg) => {
+      if (!msg) return;
+
+      if (await handler(msg)) return channel.ack(msg);
+
+      return await new Promise((resolve) => setTimeout(() => resolve(channel.nack(msg)), 3000));
+    },
+    { noAck: false },
+  );
+}
+
+async function handler(msg: amqp.ConsumeMessage): Promise<boolean> { //----> condition before process consumer
+  const repo = AppDataSource.getRepository(Command);
+  const { data, token, user } = JSON.parse(msg.content.toString());
+
+  const { id, status, lastUpdateUserId, lastUpdateFullName, lastUpdatedAt } = data;
+
+  const command = await repo.findOne({
+    where: { id: id },
+    relations: ["commandType", "commandRecives"],
+  });
+
+  if (!command) return true;
+
+  const path = commandTypePath(command.commandType.code);
+  if (path == null) throw new HttpError(HttpStatusCode.NOT_FOUND, "ไม่พบประเภทคำสั่งนี้ในระบบ");
+
+  return await new CallAPI()
+    .PostData(
+      {
+        headers: { authorization: token }, //time bomb
+      },
+      path + "/excecute",
+      {
+        refIds: command.commandRecives
+          .filter((x) => x.refId != null)
+          .map((x) => ({
+            refId: x.refId,
+            commandAffectDate: command.commandAffectDate,
+            commandNo: command.commandNo,
+            commandYear: command.commandYear,
+            templateDoc: command.positionDetail,
+            amount: x.amount,
+            positionSalaryAmount: x.positionSalaryAmount,
+            mouthSalaryAmount: x.mouthSalaryAmount,
+          })),
+      },
+      false,
+    )
+    .then(async (res) => {
+      console.log("[AMQ] Excecute Command Success");
+      Object.assign(command, { status, lastUpdateUserId, lastUpdateFullName, lastUpdatedAt });
+      const result = await repo.save(command).catch((e) => console.log(e));
+      if (result) return true;
+      return false;
+    })
+    .catch((e) => {
+      console.error(e);
+      return false;
+    });
+}
+
+// async function handler(msg: amqp.ConsumeMessage): Promise<boolean> { //----> condition before process consumer
+//   const repo = AppDataSource.getRepository(Command);
+
+//   const { id, status, lastUpdateUserId, lastUpdateFullName, lastUpdatedAt } = JSON.parse(
+//     msg.content.toString(),
+//   );
+
+//   const record = await repo.findOne({
+//     where: { id },
+//   });
+
+//   if (!record) return true;
+
+//   Object.assign(record, { status, lastUpdateUserId, lastUpdateFullName, lastUpdatedAt });
+
+//   const result = await repo.save(record).catch((e) => console.log(e));
+
+//   if (result) return true;
+
+//   return false;
+// }