Add migration to create CheckInJobStatuses table for RMQ task control
- Introduced a new migration that creates the CheckInJobStatuses table. - The table includes fields for tracking job statuses, timestamps, user information, and error messages. - Supports various statuses such as PENDING, PROCESSING, COMPLETED, and FAILED.
This commit is contained in:
Suphonchai Phoonsawat
parent
3532df32fd
commit
a463df5716
12 changed files with 2259 additions and 126 deletions
|
|
@ -58,6 +58,7 @@ namespace BMA.EHR.Leave.Service.Controllers
|
|||
private readonly LeaveRequestRepository _leaveRequestRepository;
|
||||
private readonly UserCalendarRepository _userCalendarRepository;
|
||||
private readonly PermissionRepository _permission;
|
||||
private readonly CheckInJobStatusRepository _checkInJobStatusRepository;
|
||||
|
||||
private readonly CommandRepository _commandRepository;
|
||||
|
||||
|
|
@ -92,6 +93,7 @@ namespace BMA.EHR.Leave.Service.Controllers
|
|||
ObjectPool<IModel> objectPool,
|
||||
PermissionRepository permission,
|
||||
NotificationRepository notificationRepository,
|
||||
CheckInJobStatusRepository checkInJobStatusRepository,
|
||||
HttpClient httpClient)
|
||||
{
|
||||
_dutyTimeRepository = dutyTimeRepository;
|
||||
|
|
@ -109,6 +111,7 @@ namespace BMA.EHR.Leave.Service.Controllers
|
|||
_commandRepository = commandRepository;
|
||||
_leaveRequestRepository = leaveRequestRepository;
|
||||
_notificationRepository = notificationRepository;
|
||||
_checkInJobStatusRepository = checkInJobStatusRepository;
|
||||
|
||||
_objectPool = objectPool;
|
||||
_permission = permission;
|
||||
|
|
@ -540,11 +543,15 @@ namespace BMA.EHR.Leave.Service.Controllers
|
|||
}
|
||||
}
|
||||
|
||||
// add task id for check in queue
|
||||
string taskId = Guid.NewGuid().ToString();
|
||||
|
||||
var checkData = new CheckTimeDtoRB
|
||||
{
|
||||
UserId = userId,
|
||||
CurrentDate = currentDate,
|
||||
CheckInId = data.CheckInId,
|
||||
TaskId = Guid.Parse(taskId),
|
||||
Lat = data.Lat,
|
||||
Lon = data.Lon,
|
||||
POI = data.POI,
|
||||
|
|
@ -564,11 +571,27 @@ namespace BMA.EHR.Leave.Service.Controllers
|
|||
var serializedObject = JsonConvert.SerializeObject(checkData);
|
||||
var body = Encoding.UTF8.GetBytes(serializedObject);
|
||||
|
||||
// add task id for check in queue
|
||||
string taskId = Guid.NewGuid().ToString();
|
||||
var properties = channel.CreateBasicProperties();
|
||||
properties.Persistent = true;
|
||||
properties.MessageId = userId.ToString("D");// ระบบลงเวลาต้องมีการเช็คสถานะใน rabbitMQ ด้วยว่ามีการรอรันอยู่ไหม ลงเวลาเข้า/ออกงาน #894
|
||||
properties.MessageId = taskId;
|
||||
|
||||
// บันทึกสถานะงานก่อนส่งไป RabbitMQ
|
||||
var jobStatus = new CheckInJobStatus
|
||||
{
|
||||
TaskId = Guid.Parse(taskId),
|
||||
KeycloakUserId = userId,
|
||||
CreatedDate = currentDate,
|
||||
Status = "PENDING",
|
||||
CheckType = data.CheckInId == null ? "CHECK_IN" : "CHECK_OUT",
|
||||
CheckInId = data.CheckInId,
|
||||
AdditionalData = JsonConvert.SerializeObject(new
|
||||
{
|
||||
IsLocation = data.IsLocation,
|
||||
LocationName = data.LocationName,
|
||||
POI = data.POI
|
||||
})
|
||||
};
|
||||
await _checkInJobStatusRepository.AddAsync(jobStatus);
|
||||
|
||||
channel.BasicPublish(exchange: "",
|
||||
routingKey: queue,
|
||||
|
|
@ -583,6 +606,78 @@ namespace BMA.EHR.Leave.Service.Controllers
|
|||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ตรวจสอบสถานะงาน check-in ด้วย Task ID
|
||||
/// </summary>
|
||||
/// <param name="taskId">Task ID ที่ได้จากการเรียก CheckInAsync</param>
|
||||
/// <returns>
|
||||
/// </returns>
|
||||
/// <response code="200">เมื่อทำรายการสำเร็จ</response>
|
||||
/// <response code="401">ไม่ได้ Login เข้าระบบ</response>
|
||||
/// <response code="404">ไม่พบข้อมูลงาน</response>
|
||||
/// <response code="500">เมื่อเกิดข้อผิดพลาดในการทำงาน</response>
|
||||
[HttpGet("job-status/{taskId:guid}")]
|
||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||
[ProducesResponseType(StatusCodes.Status401Unauthorized)]
|
||||
[ProducesResponseType(StatusCodes.Status404NotFound)]
|
||||
[ProducesResponseType(StatusCodes.Status500InternalServerError)]
|
||||
public async Task<ActionResult<ResponseObject>> GetJobStatusAsync(Guid taskId)
|
||||
{
|
||||
var jobStatus = await _checkInJobStatusRepository.GetByTaskIdAsync(taskId);
|
||||
|
||||
if (jobStatus == null)
|
||||
{
|
||||
return Error("ไม่พบข้อมูลงาน", StatusCodes.Status404NotFound);
|
||||
}
|
||||
|
||||
var result = new
|
||||
{
|
||||
taskId = jobStatus.TaskId,
|
||||
keycloakUserId = jobStatus.KeycloakUserId,
|
||||
status = jobStatus.Status,
|
||||
checkType = jobStatus.CheckType,
|
||||
checkInId = jobStatus.CheckInId,
|
||||
createdDate = jobStatus.CreatedDate,
|
||||
processingDate = jobStatus.ProcessingDate,
|
||||
completedDate = jobStatus.CompletedDate,
|
||||
errorMessage = jobStatus.ErrorMessage,
|
||||
additionalData = jobStatus.AdditionalData != null ?
|
||||
JsonConvert.DeserializeObject(jobStatus.AdditionalData) : null
|
||||
};
|
||||
|
||||
return Success(result);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ดึงรายการงานที่ยัง pending หรือ processing ของผู้ใช้
|
||||
/// </summary>
|
||||
/// <returns>
|
||||
/// </returns>
|
||||
/// <response code="200">เมื่อทำรายการสำเร็จ</response>
|
||||
/// <response code="401">ไม่ได้ Login เข้าระบบ</response>
|
||||
/// <response code="500">เมื่อเกิดข้อผิดพลาดในการทำงาน</response>
|
||||
[HttpGet("pending-jobs")]
|
||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||
[ProducesResponseType(StatusCodes.Status401Unauthorized)]
|
||||
[ProducesResponseType(StatusCodes.Status500InternalServerError)]
|
||||
public async Task<ActionResult<ResponseObject>> GetPendingJobsAsync()
|
||||
{
|
||||
var userId = UserId == null ? Guid.Empty : Guid.Parse(UserId);
|
||||
var jobs = await _checkInJobStatusRepository.GetPendingOrProcessingJobsAsync(userId);
|
||||
|
||||
var result = jobs.Select(job => new
|
||||
{
|
||||
taskId = job.TaskId,
|
||||
status = job.Status,
|
||||
checkType = job.CheckType,
|
||||
checkInId = job.CheckInId,
|
||||
createdDate = job.CreatedDate,
|
||||
processingDate = job.ProcessingDate
|
||||
}).ToList();
|
||||
|
||||
return Success(new { count = result.Count, jobs = result });
|
||||
}
|
||||
|
||||
[HttpGet("check-status")]
|
||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||
[ProducesResponseType(StatusCodes.Status401Unauthorized)]
|
||||
|
|
@ -590,61 +685,62 @@ namespace BMA.EHR.Leave.Service.Controllers
|
|||
public async Task<ActionResult<ResponseObject>> CheckInCheckStatus()
|
||||
{
|
||||
var userId = UserId == null ? Guid.Empty : Guid.Parse(UserId);
|
||||
var currentDate = DateTime.Now;
|
||||
var channel = _objectPool.Get();
|
||||
// var currentDate = DateTime.Now;
|
||||
// var channel = _objectPool.Get();
|
||||
try
|
||||
{
|
||||
var _url = _configuration["Rabbit:URL"] ?? "";
|
||||
var _queue = _configuration["Rabbit:Queue"] ?? "basic-queue";
|
||||
// var _url = _configuration["Rabbit:URL"] ?? "";
|
||||
// var _queue = _configuration["Rabbit:Queue"] ?? "basic-queue";
|
||||
|
||||
// Step 1: ตรวจสอบจำนวน message ทั้งหมดในคิว
|
||||
string queueUrl = $"{_url}{_queue}";
|
||||
var queueResponse = await _httpClient.GetAsync(queueUrl);
|
||||
if (!queueResponse.IsSuccessStatusCode)
|
||||
{
|
||||
return Error("Error accessing RabbitMQ API", (int)queueResponse.StatusCode);
|
||||
}
|
||||
// // Step 1: ตรวจสอบจำนวน message ทั้งหมดในคิว
|
||||
// string queueUrl = $"{_url}{_queue}";
|
||||
// var queueResponse = await _httpClient.GetAsync(queueUrl);
|
||||
// if (!queueResponse.IsSuccessStatusCode)
|
||||
// {
|
||||
// return Error("Error accessing RabbitMQ API", (int)queueResponse.StatusCode);
|
||||
// }
|
||||
|
||||
var queueContent = await queueResponse.Content.ReadAsStringAsync();
|
||||
var queueData = JObject.Parse(queueContent);
|
||||
int totalMessages = queueData["messages"]?.Value<int>() ?? 0;
|
||||
// var queueContent = await queueResponse.Content.ReadAsStringAsync();
|
||||
// var queueData = JObject.Parse(queueContent);
|
||||
// int totalMessages = queueData["messages"]?.Value<int>() ?? 0;
|
||||
|
||||
// Step 2: วนลูปดึง message ทีละ 100 งาน
|
||||
int batchSize = 100;
|
||||
var allMessages = new List<string>();
|
||||
int processedMessages = 0;
|
||||
// // Step 2: วนลูปดึง message ทีละ 100 งาน
|
||||
// int batchSize = 100;
|
||||
// var allMessages = new List<string>();
|
||||
// int processedMessages = 0;
|
||||
|
||||
while (processedMessages < totalMessages)
|
||||
{
|
||||
var requestBody = new StringContent(
|
||||
$"{{\"count\":{batchSize},\"requeue\":true,\"encoding\":\"auto\",\"ackmode\":\"ack_requeue_true\"}}",
|
||||
Encoding.UTF8,
|
||||
"application/json"
|
||||
);
|
||||
// while (processedMessages < totalMessages)
|
||||
// {
|
||||
// var requestBody = new StringContent(
|
||||
// $"{{\"count\":{batchSize},\"requeue\":true,\"encoding\":\"auto\",\"ackmode\":\"ack_requeue_true\"}}",
|
||||
// Encoding.UTF8,
|
||||
// "application/json"
|
||||
// );
|
||||
|
||||
string getMessagesUrl = $"{_url}{_queue}/get";
|
||||
var response = await _httpClient.PostAsync(getMessagesUrl, requestBody);
|
||||
if (!response.IsSuccessStatusCode)
|
||||
{
|
||||
return StatusCode((int)response.StatusCode, "Error retrieving messages from RabbitMQ.");
|
||||
}
|
||||
// string getMessagesUrl = $"{_url}{_queue}/get";
|
||||
// var response = await _httpClient.PostAsync(getMessagesUrl, requestBody);
|
||||
// if (!response.IsSuccessStatusCode)
|
||||
// {
|
||||
// return StatusCode((int)response.StatusCode, "Error retrieving messages from RabbitMQ.");
|
||||
// }
|
||||
|
||||
var content = await response.Content.ReadAsStringAsync();
|
||||
var messages = JArray.Parse(content);
|
||||
// var content = await response.Content.ReadAsStringAsync();
|
||||
// var messages = JArray.Parse(content);
|
||||
|
||||
if (messages.Count == 0)
|
||||
{
|
||||
break;
|
||||
}
|
||||
// if (messages.Count == 0)
|
||||
// {
|
||||
// break;
|
||||
// }
|
||||
|
||||
processedMessages += messages.Count;
|
||||
allMessages.AddRange(messages.Select(m => m["properties"].ToString()));
|
||||
}
|
||||
// processedMessages += messages.Count;
|
||||
// allMessages.AddRange(messages.Select(m => m["properties"].ToString()));
|
||||
// }
|
||||
|
||||
var jobs = await _checkInJobStatusRepository.GetPendingOrProcessingJobsAsync(userId);
|
||||
// Step 3: ค้นหา taskIds ที่อยู่ใน messages ทั้งหมด
|
||||
var foundTasks = allMessages.FirstOrDefault(x => x.Contains(userId.ToString("D")));
|
||||
//var foundTasks = allMessages.FirstOrDefault(x => x.Contains(userId.ToString("D")));
|
||||
|
||||
return Success(new { keycloakId = userId, InQueue = foundTasks != null });
|
||||
return Success(new { keycloakId = userId, InQueue = (jobs != null && jobs.Count > 0) });
|
||||
|
||||
}
|
||||
catch (Exception ex)
|
||||
|
|
@ -653,7 +749,7 @@ namespace BMA.EHR.Leave.Service.Controllers
|
|||
}
|
||||
finally
|
||||
{
|
||||
_objectPool.Return(channel);
|
||||
//_objectPool.Return(channel);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -790,21 +886,31 @@ namespace BMA.EHR.Leave.Service.Controllers
|
|||
public async Task<ActionResult<ResponseObject>> ProcessCheckInAsync([FromBody] CheckTimeDtoRB data)
|
||||
{
|
||||
var userId = data.UserId ?? Guid.Empty;
|
||||
var profile = await _userProfileRepository.GetProfileByKeycloakIdAsync(userId, data.Token);
|
||||
var taskId = data.TaskId ?? Guid.Empty;
|
||||
|
||||
if (profile == null)
|
||||
return Error(GlobalMessages.DataNotFound, StatusCodes.Status404NotFound);
|
||||
|
||||
if (data.CheckInFileName == "no-file") throw new Exception(GlobalMessages.NoFileToUpload);
|
||||
var currentDate = data.CurrentDate ?? DateTime.Now;
|
||||
|
||||
var check_status = data.CheckInId == null ? "check-in-picture" : "check-out-picture";
|
||||
|
||||
var fileName = $"{_bucketName}/{userId}/{currentDate.ToString("dd-MM-yyyy")}/{check_status}/{data.CheckInFileName}";
|
||||
using (var ms = new MemoryStream(data.CheckInFileBytes ?? new byte[0]))
|
||||
try
|
||||
{
|
||||
await _minIOService.UploadFileAsync(fileName, ms);
|
||||
}
|
||||
// อัปเดตสถานะเป็น PROCESSING
|
||||
if (taskId != Guid.Empty)
|
||||
{
|
||||
await _checkInJobStatusRepository.UpdateToProcessingAsync(taskId);
|
||||
}
|
||||
|
||||
var profile = await _userProfileRepository.GetProfileByKeycloakIdAsync(userId, data.Token);
|
||||
|
||||
if (profile == null)
|
||||
return Error(GlobalMessages.DataNotFound, StatusCodes.Status404NotFound);
|
||||
|
||||
if (data.CheckInFileName == "no-file") throw new Exception(GlobalMessages.NoFileToUpload);
|
||||
var currentDate = data.CurrentDate ?? DateTime.Now;
|
||||
|
||||
var check_status = data.CheckInId == null ? "check-in-picture" : "check-out-picture";
|
||||
|
||||
var fileName = $"{_bucketName}/{userId}/{currentDate.ToString("dd-MM-yyyy")}/{check_status}/{data.CheckInFileName}";
|
||||
using (var ms = new MemoryStream(data.CheckInFileBytes ?? new byte[0]))
|
||||
{
|
||||
await _minIOService.UploadFileAsync(fileName, ms);
|
||||
}
|
||||
|
||||
var defaultRound = await _dutyTimeRepository.GetDefaultAsync();
|
||||
if (defaultRound == null)
|
||||
|
|
@ -1058,9 +1164,32 @@ namespace BMA.EHR.Leave.Service.Controllers
|
|||
}
|
||||
|
||||
}
|
||||
|
||||
// อัปเดตสถานะเป็น COMPLETED
|
||||
if (taskId != Guid.Empty)
|
||||
{
|
||||
var additionalData = JsonConvert.SerializeObject(new
|
||||
{
|
||||
CheckInType = data.CheckInId == null ? "check-in" : "check-out",
|
||||
FileName = fileName,
|
||||
ProcessedDate = currentDate
|
||||
});
|
||||
await _checkInJobStatusRepository.UpdateToCompletedAsync(taskId, additionalData);
|
||||
}
|
||||
|
||||
var checkInType = data.CheckInId == null ? "check-in" : "check-out";
|
||||
return Success(new { user = $"{profile.FirstName} {profile.LastName}", date = currentDate, type = checkInType }); ;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// อัปเดตสถานะเป็น FAILED
|
||||
if (taskId != Guid.Empty)
|
||||
{
|
||||
await _checkInJobStatusRepository.UpdateToFailedAsync(taskId, ex.Message);
|
||||
}
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// LV1_005 - ลงเวลาเข้า-ออกงาน (USER)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue