Adding
This commit is contained in:
Zeeshaun
+114
@@ -0,0 +1,114 @@
|
||||
using LCU.Api.Internal.DispatcherLib.Repositories;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services
|
||||
{
|
||||
public class CleanUpJobSettingService : ICleanUpJobSettingService
|
||||
{
|
||||
private readonly ICleanUpJobSettingRepository _cleanUpJobSettingRepository;
|
||||
private readonly IUnitOfWork _unitOfWork;
|
||||
|
||||
public CleanUpJobSettingService(IUnitOfWork unitOfWork,
|
||||
ICleanUpJobSettingRepository cleanUpJobSettingRepository)
|
||||
{
|
||||
_unitOfWork = unitOfWork;
|
||||
_cleanUpJobSettingRepository = cleanUpJobSettingRepository;
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<CleanUpJobSetting>> GetAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
return await _cleanUpJobSettingRepository.ListAsync();
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure retrieving CleanUpJobSetting.");
|
||||
throw new ServiceException("Failure retrieving CleanUpJobSetting.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<CleanUpJobSetting> GetByCleanUpJobSettingIdAsync(int cleanUpJobSettingId)
|
||||
{
|
||||
try
|
||||
{
|
||||
return await _cleanUpJobSettingRepository.FindByCleanUpJobSettingIdAsync(cleanUpJobSettingId);
|
||||
}
|
||||
catch (ArgumentNullException exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure getting dispatcher Clean Up Job Settings by Id. CleanUpJobSettingId was null");
|
||||
return null;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure retrieving dispatcher Clean Up Job Settings. CleanUpJobSettingId: {@cleanUpJobSettingId}", cleanUpJobSettingId);
|
||||
throw new ServiceException($"Failure retrieving dispatcher Clean Up Job Settings. CleanUpJobSettingId: {cleanUpJobSettingId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<CleanUpJobSetting> SaveAsync(CleanUpJobSetting request)
|
||||
{
|
||||
try
|
||||
{
|
||||
var result = await _cleanUpJobSettingRepository.AddAsync(request);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
return result;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure saving dispatcher Clean Up Job Settings. CleanUpJobSettingId: {@request.CleanUpJobSettingId}", request.CleanUpJobSettingId);
|
||||
throw new ServiceException($"Failure saving dispatcher Clean Up Job Settings. CleanUpJobSettingId: {request.CleanUpJobSettingId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<CleanUpJobSetting> UpdateAsync(CleanUpJobSetting request)
|
||||
{
|
||||
try
|
||||
{
|
||||
var existing = await GetByCleanUpJobSettingIdAsync(request.CleanUpJobSettingId);
|
||||
if(existing == null)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
existing.CompletionDaysPast = request.CompletionDaysPast;
|
||||
_cleanUpJobSettingRepository.Update(existing);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
|
||||
return existing;
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, "SQL Failure updating dispatcher Clean Up Job Settings. CleanUpJobSettingId: {@request.CleanUpJobSettingId}", request.CleanUpJobSettingId);
|
||||
throw new ServiceException($"SQL Failure updating dispatcher Clean Up Job Settings. CleanUpJobSettingId: {request.CleanUpJobSettingId}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "SQL Failure updating dispatcher Clean Up Job Settings. CleanUpJobSettingId: {@request.CleanUpJobSettingId}", request.CleanUpJobSettingId);
|
||||
throw new ServiceException($"SQL Failure updating dispatcher Clean Up Job Settings. CleanUpJobSettingId: {request.CleanUpJobSettingId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task DeleteAsync(int cleanUpJobSettingId)
|
||||
{
|
||||
try
|
||||
{
|
||||
var cleanUpJobSetting = await _cleanUpJobSettingRepository.FindByCleanUpJobSettingIdAsync(cleanUpJobSettingId);
|
||||
ArgumentNullException.ThrowIfNull(cleanUpJobSetting);
|
||||
_cleanUpJobSettingRepository.Remove(cleanUpJobSetting);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, "SQL Failure removing dispatcher Clean Up Job Settings. CleanUpJobSettingId: {@cleanUpJobSettingId}", cleanUpJobSettingId);
|
||||
throw new ServiceException($"SQL Failure removing dispatcher Clean Up Job Settings. CleanUpJobSettingId: {cleanUpJobSettingId}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure removing dispatcher Clean Up Job Settings. CleanUpJobSettingId: {@cleanUpJobSettingId}", cleanUpJobSettingId);
|
||||
throw new ServiceException($"Failure removing dispatcher Clean Up Job Settings. CleanUpJobSettingId: {cleanUpJobSettingId}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+135
@@ -0,0 +1,135 @@
|
||||
using LCU.Api.Internal.DispatcherLib.Repositories;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public class CleanUpJobTagService : ICleanUpJobTagService
|
||||
{
|
||||
private readonly ICleanUpJobTagRepository _cleanUpJobTagRepository;
|
||||
private readonly IUnitOfWork _unitOfWork;
|
||||
|
||||
public CleanUpJobTagService(ICleanUpJobTagRepository cleanUpJobRepository,
|
||||
IUnitOfWork unitOfWork)
|
||||
{
|
||||
_cleanUpJobTagRepository = cleanUpJobRepository;
|
||||
_unitOfWork = unitOfWork;
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<CleanUpJobTag>> GetAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
return await _cleanUpJobTagRepository.ListAsync();
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure retrieving CleanUpJobag.");
|
||||
throw new ServiceException("Failure retrieving CleanUpJobTag.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<CleanUpJobTag> GetByCleanUpJobTagIdAsync(int cleanUpJobTagId)
|
||||
{
|
||||
try
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(cleanUpJobTagId);
|
||||
return await _cleanUpJobTagRepository.FindByCleanUpJobTagIdAsync(cleanUpJobTagId);
|
||||
}
|
||||
catch (ArgumentNullException exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure getting dispatcher Clean Up Job Tag by Id. CleanUpJobTagId was null");
|
||||
return null;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure retrieving dispatcher Clean Up Job Tag. CleanUpJobTagId: {@cleanUpJobTagId}", cleanUpJobTagId);
|
||||
throw new ServiceException($"Failure retrieving dispatcher Clean Up Job Tag. CleanUpJobTagId: {cleanUpJobTagId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<CleanUpJobTag>> GetByCleanUpJobSettingIdAsync(int cleanUpJobSettingId)
|
||||
{
|
||||
try
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(cleanUpJobSettingId);
|
||||
return await _cleanUpJobTagRepository.FindByCleanUpJobSettingIdAsync(cleanUpJobSettingId);
|
||||
}
|
||||
catch (ArgumentNullException exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure getting dispatcher Clean Up Job Tags bySettings Id. CleanUpJobSettingId was null");
|
||||
return null;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure retrieving dispatcher Clean Up Job Tag. CleanUpJobSettingId: {@cleanUpJobSettingId}", cleanUpJobSettingId);
|
||||
throw new ServiceException($"Failure retrieving dispatcher Clean Up Job Tag. CleanUpJobSettingId: {cleanUpJobSettingId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<CleanUpJobTag> SaveAsync(CleanUpJobTag request)
|
||||
{
|
||||
try
|
||||
{
|
||||
var result = await _cleanUpJobTagRepository.AddAsync(request);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
return result;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure saving dispatcher Clean Up Job Tag. CleanUpJobTagId: {@request.CleanUpJobTagId}", request.CleanUpJobTagId);
|
||||
throw new ServiceException($"Failure saving dispatcher Clean Up Job Tag. CleanUpJobTagId: {request.CleanUpJobTagId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task DeleteAsync(int cleanUpJobTagId)
|
||||
{
|
||||
try
|
||||
{
|
||||
var cleanUpJobTag = await _cleanUpJobTagRepository.FindByCleanUpJobTagIdAsync(cleanUpJobTagId);
|
||||
ArgumentNullException.ThrowIfNull(cleanUpJobTag);
|
||||
_cleanUpJobTagRepository.Remove(cleanUpJobTag);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, "SQL Failure removing dispatcher Clean Up Job Tag. CleanUpJobTagId: {@cleanUpJobTagId}", cleanUpJobTagId);
|
||||
throw new ServiceException($"SQL Failure removing dispatcher Clean Up Job Tag. CleanUpJobTagId: {cleanUpJobTagId}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure removing dispatcher Clean Up Job Tag. CleanUpJobTagId: {@cleanUpJobTagId}", cleanUpJobTagId);
|
||||
throw new ServiceException($"Failure removing dispatcher Clean Up Job Tag. CleanUpJobTagId: {cleanUpJobTagId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task DeleteByCleanUpJobSettingIdAsync(int cleanUpJobSettingId)
|
||||
{
|
||||
try
|
||||
{
|
||||
var cleanUpJobTags = await _cleanUpJobTagRepository.FindByCleanUpJobSettingIdAsync(cleanUpJobSettingId);
|
||||
|
||||
if(!cleanUpJobTags.Any())
|
||||
{
|
||||
throw new Exception($"No CleanUpJobTags found.");
|
||||
}
|
||||
|
||||
foreach (var tag in cleanUpJobTags)
|
||||
{
|
||||
_cleanUpJobTagRepository.Remove(tag);
|
||||
}
|
||||
|
||||
await _unitOfWork.CompleteAsync();
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, "SQL Failure removing dispatcher Clean Up Job Tag. CleanUpJobSettingId: {@cleanUpJobSettingId}", cleanUpJobSettingId);
|
||||
throw new ServiceException($"SQL Failure removing dispatcher Clean Up Job Tag. CleanUpJobSettingId: {cleanUpJobSettingId}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure removing dispatcher Clean Up Job Tag. CleanUpJobSettingId: {@cleanUpJobSettingId}", cleanUpJobSettingId);
|
||||
throw new ServiceException($"Failure removing dispatcher Clean Up Job Tag. CleanUpJobSettingId: {cleanUpJobSettingId}");
|
||||
}
|
||||
}
|
||||
}
|
||||
+72
@@ -0,0 +1,72 @@
|
||||
using LCU.Api.Internal.Dispatcher.Services.QueueServices;
|
||||
using LCU.Api.Internal.MaidBotLib.Models;
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices;
|
||||
|
||||
public class BatchErroredJobsService : IBatchErroredJobsService
|
||||
{
|
||||
private readonly IQueueSendErroredJobAlertService _queueSendErroredJobAlertService;
|
||||
private readonly IJobService _jobService;
|
||||
|
||||
public BatchErroredJobsService(IJobService jobService,
|
||||
IQueueSendErroredJobAlertService queueSendErroredJobAlertService)
|
||||
{
|
||||
_jobService = jobService;
|
||||
_queueSendErroredJobAlertService = queueSendErroredJobAlertService;
|
||||
}
|
||||
|
||||
public async Task PerformBatchErroredJobs(CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
var jobs = await _jobService.GetErrorMonitoringJobs();
|
||||
if (jobs == null || !jobs.Any())
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var batchedErroredJobAlertRequest = CreateBatchedErroredJobAlerts(jobs);
|
||||
|
||||
await _queueSendErroredJobAlertService.QueueSendErroredJobAlert(batchedErroredJobAlertRequest);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
//Fatal as we need to know if we can't get out our errors
|
||||
Log.ForContext(nameof(PropertyEnum.Group),
|
||||
nameof(GroupEnum.ITDevelopment))
|
||||
.Fatal(ex, $"Failed to perform Error Monitoring.");
|
||||
}
|
||||
}
|
||||
|
||||
private static BatchedErroredJobAlertRequest CreateBatchedErroredJobAlerts(IEnumerable<Job> jobs)
|
||||
{
|
||||
var batchedErroredJobAlertRequests = new BatchedErroredJobAlertRequest()
|
||||
{
|
||||
ErroredJobAlertRequests = new List<ErroredJobAlertRequest>()
|
||||
};
|
||||
|
||||
foreach (var job in jobs)
|
||||
{
|
||||
var failedOrProcessingWorkItem = job.WorkItems
|
||||
.FirstOrDefault(wi => wi.WorkItemExecutions != null &&
|
||||
wi.WorkItemExecutions.Any(wie => wie.Status == StatusType.Failed || wie.Status == StatusType.Processing));
|
||||
|
||||
if (failedOrProcessingWorkItem == null || failedOrProcessingWorkItem.CurrentWorkItemExecution == null)
|
||||
{
|
||||
Log.Warning($"Failed or Processing workItem not found during CreateBatchedErroredJobAlerts. JobId and Name: {job.JobId}: {job.Name}");
|
||||
continue;
|
||||
}
|
||||
|
||||
batchedErroredJobAlertRequests.ErroredJobAlertRequests.Add(new ErroredJobAlertRequest
|
||||
{
|
||||
CurrentJob = job,
|
||||
CurrentWorkItem = failedOrProcessingWorkItem
|
||||
});
|
||||
}
|
||||
|
||||
return batchedErroredJobAlertRequests;
|
||||
}
|
||||
}
|
||||
+172
@@ -0,0 +1,172 @@
|
||||
using LCU.Api.Internal.Dispatcher.Configurations;
|
||||
using LCU.Api.Internal.DispatcherLib.Repositories;
|
||||
using LCU.Api.Internal.MaidBotLib.Models;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Results;
|
||||
using Newtonsoft.Json;
|
||||
using System.Threading;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices;
|
||||
|
||||
public class CleanUpJobsService : ICleanUpJobsService
|
||||
{
|
||||
private readonly IJobService _jobService;
|
||||
private readonly ICleanUpJobSettingService _cleanUpJobSettingService;
|
||||
private readonly IUnitOfWork _unitOfWork;
|
||||
private readonly ApiConfiguration _apiConfiguration;
|
||||
|
||||
public CleanUpJobsService(IJobService jobService,
|
||||
ICleanUpJobSettingService cleanUpJobSettingService,
|
||||
IUnitOfWork unitOfWork,
|
||||
ApiConfiguration apiConfiguration)
|
||||
{
|
||||
_jobService = jobService;
|
||||
_cleanUpJobSettingService = cleanUpJobSettingService;
|
||||
_unitOfWork = unitOfWork;
|
||||
_apiConfiguration = apiConfiguration;
|
||||
}
|
||||
|
||||
public async Task PerformCleanUpJobs(CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
var result = new CleanUpJobBatchResult();
|
||||
|
||||
result = await PerformSoftCleanUpJobTask(result);
|
||||
result = await PerformHardDeleteCleanUpJobTask(result);
|
||||
|
||||
await _unitOfWork.CompleteAsync();
|
||||
|
||||
if (result.IsFailure)
|
||||
{
|
||||
Log.ForContext(nameof(PropertyEnum.Group),
|
||||
nameof(GroupEnum.ITDevelopment))
|
||||
.Error("CleanUpJob Batch Result: {@CleanUpJobBatchResult}", result);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Log.ForContext(nameof(PropertyEnum.Group),
|
||||
nameof(GroupEnum.ITDevelopment))
|
||||
.Error(ex, $"Failed to perform Clean Up Jobs.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<CleanUpJobBatchResult> PerformSoftCleanUpJobTask(CleanUpJobBatchResult result)
|
||||
{
|
||||
var jobs = await _jobService.GetInactiveExcludeSoftDeletedTrackedAsync();
|
||||
|
||||
var cleanUpJobSettings = await _cleanUpJobSettingService.GetAsync();
|
||||
|
||||
foreach (var setting in cleanUpJobSettings)
|
||||
{
|
||||
if (setting.CleanUpJobTags == null || !setting.CleanUpJobTags.Any())
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
foreach (var job in jobs)
|
||||
{
|
||||
result = PerformCleanUpTaskForJob(result, job, setting);
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
public bool ValidateShouldSoftDelete(CleanUpJobSetting setting, Job job, CleanUpJobBatchResult result)
|
||||
{
|
||||
var shouldSoftDelete = ShouldSoftDeleteByTag(setting, job);
|
||||
|
||||
if (!shouldSoftDelete)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
return ShouldSoftDeleteByDate(setting, job, result);
|
||||
}
|
||||
|
||||
public bool ShouldSoftDeleteByDate(CleanUpJobSetting setting, Job job, CleanUpJobBatchResult result)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!job.WorkItems.Any() || !job.WorkItems.FirstOrDefault().WorkItemExecutions.Any())
|
||||
{
|
||||
result.AddToFailureCount($"Job would be cleaned but cannot determine if it is old enough to be cleaned due to not having any workItemExecutions. Job: {job.JobId} - {job.Name}");
|
||||
return false;
|
||||
}
|
||||
|
||||
var currentDate = DateTime.Now;
|
||||
var cutoffDate = currentDate.AddDays(setting.CompletionDaysPast);
|
||||
var jobLastCompletedTime = job.WorkItems.FirstOrDefault()
|
||||
.WorkItemExecutions.OrderByDescending(wie => wie.CompletedTimestamp).FirstOrDefault().CompletedTimestamp.Value;
|
||||
|
||||
return DateTime.Compare(cutoffDate, jobLastCompletedTime) > 0;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
var message = $"Error while attempting to compare CompletedTimestamp for SoftDeletion. Job: {job.JobId} - {job.Name}. ExeptionMessage: {ex.Message}";
|
||||
Log.Error(message);
|
||||
throw new Exception(message);
|
||||
}
|
||||
}
|
||||
|
||||
public bool ShouldSoftDeleteByTag(CleanUpJobSetting setting, Job job)
|
||||
{
|
||||
if (job.JobTags.Count != setting.CleanUpJobTags.Count)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
var orderedJobTagIds = job.JobTags.Select(t => t.TagId).OrderBy(t => t).ToList();
|
||||
var orderedCleanUpJobTagIds = setting.CleanUpJobTags.Select(t => t.TagId).OrderBy(t => t).ToList();
|
||||
|
||||
return orderedJobTagIds.SequenceEqual(orderedCleanUpJobTagIds);
|
||||
}
|
||||
|
||||
public async Task<CleanUpJobBatchResult> PerformHardDeleteCleanUpJobTask(CleanUpJobBatchResult result)
|
||||
{
|
||||
var softDeletedJobs = await _jobService.GetSoftDeletedTrackedAsync(DateTime.Now.AddDays(_apiConfiguration.CleanUpJobsDaysToDelete));
|
||||
foreach (var job in softDeletedJobs)
|
||||
{
|
||||
try
|
||||
{
|
||||
await _jobService.DeleteFromCleanupAsync(job);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Log.Error(ex.Message, "Failure to hard delete dispatcher job through Clean Up Job Task. Job: {@Job}", job);
|
||||
result.AddToFailureCount(ex.Message);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
private CleanUpJobBatchResult PerformCleanUpTaskForJob(CleanUpJobBatchResult result, Job job, CleanUpJobSetting setting)
|
||||
{
|
||||
try
|
||||
{
|
||||
var shouldSoftDelete = ValidateShouldSoftDelete(setting, job, result);
|
||||
|
||||
if (!shouldSoftDelete)
|
||||
{
|
||||
return result;
|
||||
}
|
||||
|
||||
job.SoftDeleteDate = DateTime.Now;
|
||||
|
||||
var updatedJob = _jobService.UpdateExcludingWorkItemsAsyncWithoutSave(job);
|
||||
if (updatedJob == null || updatedJob.Result.JobId != job.JobId)
|
||||
{
|
||||
throw new Exception($"Failure to setup soft delete update for job. Job:{JsonConvert.SerializeObject(job)}");
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
result.AddToFailureCount(ex.Message);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
}
|
||||
+84
@@ -0,0 +1,84 @@
|
||||
using Newtonsoft.Json;
|
||||
using LCU.Library.Internal.ApiResponse.Models;
|
||||
using LCU.Api.Internal.Dispatcher.Services.QueueServices;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using System.Collections.Generic;
|
||||
using RabbitMQ.Client;
|
||||
using LCU.Api.Internal.MaidBotLib.Models;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices;
|
||||
|
||||
public class CompletionService : ICompletionService
|
||||
{
|
||||
private readonly IJobService _jobService;
|
||||
private readonly IWorkItemService _workItemService;
|
||||
private readonly IExecutionMessageService _executionMessageService;
|
||||
private readonly IQueueJobFailureService _queueJobFailureService;
|
||||
private readonly IQueueDetermineOperationService _queueDetermineOperationService;
|
||||
|
||||
public CompletionService(IJobService jobService,
|
||||
IWorkItemService workItemService,
|
||||
IExecutionMessageService executionMessageService,
|
||||
IQueueJobFailureService queueJobFailureService,
|
||||
IQueueDetermineOperationService queueDetermineOperationService)
|
||||
{
|
||||
_jobService = jobService;
|
||||
_workItemService = workItemService;
|
||||
_executionMessageService = executionMessageService;
|
||||
_queueJobFailureService = queueJobFailureService;
|
||||
_queueDetermineOperationService = queueDetermineOperationService;
|
||||
}
|
||||
|
||||
public async Task HandleCompletionRequest(Response response)
|
||||
{
|
||||
var dispatcherReturnResponse = JsonConvert.DeserializeObject<DispatcherReturnResponse>(response.ResponsePayload);
|
||||
try
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(dispatcherReturnResponse);
|
||||
|
||||
var job = await _jobService.GetByWorkItemIdExcludingWorkItemsAsync(dispatcherReturnResponse.WorkItemId);
|
||||
|
||||
if (!response.Successful)
|
||||
{
|
||||
await _queueJobFailureService.QueueJobFailure(job.JobId,
|
||||
dispatcherReturnResponse.WorkItemId,
|
||||
response.ResponseStack == null || response.ResponseStack.Any() ? response.ResponseStack : new List<string>());
|
||||
return;
|
||||
}
|
||||
|
||||
var workItem = await _workItemService.GetByIdAsync(dispatcherReturnResponse.WorkItemId);
|
||||
if (workItem.CurrentWorkItemExecution == null)
|
||||
{
|
||||
Log.Warning($"CurrentWorkitemExecution was null for WorkItemId:{dispatcherReturnResponse.WorkItemId}");
|
||||
return;
|
||||
}
|
||||
await _executionMessageService.SaveExecutionMessages(workItem.CurrentWorkItemExecution.WorkItemExecutionId, response.ResponseStack);
|
||||
|
||||
if (workItem.CanceledJobCounter > 0)
|
||||
{
|
||||
workItem = await _workItemService.ResetCanceledJobCounter(workItem);
|
||||
}
|
||||
|
||||
job.JobStatus = JobStatus.DeterminingOperation;
|
||||
var updatedJob = await _jobService.UpdateAsync(job);
|
||||
|
||||
await _queueDetermineOperationService.QueueDetermineOperation(updatedJob, workItem, SetDetermineOperationType(dispatcherReturnResponse.Skipped));
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
var exceptionMessage = $"Failed to perform completion. Job may be stuck in an inoperable status. WorkItemId: {dispatcherReturnResponse.WorkItemId}";
|
||||
Log.Error(exception, exceptionMessage);
|
||||
throw new ServiceException(exceptionMessage);
|
||||
}
|
||||
}
|
||||
|
||||
public DetermineOperationType SetDetermineOperationType(bool isSkipped)
|
||||
{
|
||||
if(isSkipped)
|
||||
{
|
||||
return DetermineOperationType.Skipped;
|
||||
}
|
||||
|
||||
return DetermineOperationType.Completed;
|
||||
}
|
||||
}
|
||||
+82
@@ -0,0 +1,82 @@
|
||||
using LCU.Api.Internal.Dispatcher.Services.QueueServices;
|
||||
using LCU.Api.Internal.MaidBotLib.Models;
|
||||
using LCU.Library.Internal.DispatcherModels.Entities;
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices
|
||||
{
|
||||
public class DetermineOperationService : IDetermineOperationService
|
||||
{
|
||||
private readonly IJobService _jobService;
|
||||
private readonly IWorkItemService _workItemService;
|
||||
private readonly IWorkItemExecutionService _workItemExecutionService;
|
||||
private readonly IQueueExecutionService _queueExecutionService;
|
||||
|
||||
public DetermineOperationService(IJobService jobService,
|
||||
IWorkItemService workItemService,
|
||||
IWorkItemExecutionService workItemExecutionService,
|
||||
IQueueExecutionService queueExecutionService)
|
||||
{
|
||||
_jobService = jobService;
|
||||
_workItemService = workItemService;
|
||||
_workItemExecutionService = workItemExecutionService;
|
||||
_queueExecutionService = queueExecutionService;
|
||||
}
|
||||
|
||||
public async Task HandleDetermineOperationRequest(DetermineOperationRequest determineOperationRequest)
|
||||
{
|
||||
try
|
||||
{
|
||||
switch (determineOperationRequest.DetermineOperationType)
|
||||
{
|
||||
case DetermineOperationType.Failed:
|
||||
await HandleFailedWorkItem(determineOperationRequest);
|
||||
break;
|
||||
case DetermineOperationType.Skipped:
|
||||
await HandleWorkItemCompletion(determineOperationRequest, StatusType.Skipped);
|
||||
break;
|
||||
case DetermineOperationType.Completed:
|
||||
await HandleWorkItemCompletion(determineOperationRequest, StatusType.Completed);
|
||||
break;
|
||||
default:
|
||||
throw new NotImplementedException();
|
||||
}
|
||||
return;
|
||||
}
|
||||
catch (ServiceException serviceException)
|
||||
{
|
||||
Log.Error(serviceException, $"Dispatcher Determine Operation Failure. " +
|
||||
$"The Dispatcher Determine Operation is placed on LCU.Dispatcher.DetermineOperation_error.");
|
||||
throw new ServiceException($"Failed to handle DetermineOperationRequest.");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failed to handle DetermineOperationRequest.");
|
||||
throw new ServiceException($"Failed to handle DetermineOperationRequest.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task HandleFailedWorkItem(DetermineOperationRequest determineOperationRequest)
|
||||
{
|
||||
await _jobService.CompleteJobAndWorkItem(determineOperationRequest.CurrentJob, determineOperationRequest.CurrentWorkItem.CurrentWorkItemExecution, StatusType.Failed);
|
||||
}
|
||||
|
||||
public async Task HandleWorkItemCompletion(DetermineOperationRequest determineOperationRequest, StatusType status)
|
||||
{
|
||||
var nextWorkItem = await _workItemService.GetNextWorkItemByJobId(determineOperationRequest.CurrentJob.JobId, determineOperationRequest.CurrentWorkItem.Order);
|
||||
if (nextWorkItem == null)
|
||||
{
|
||||
await _jobService.CompleteJobAndWorkItem(determineOperationRequest.CurrentJob, determineOperationRequest.CurrentWorkItem.CurrentWorkItemExecution, status);
|
||||
return;
|
||||
}
|
||||
|
||||
await _jobService.CompleteWorkItemAndContinueProcessingJob(determineOperationRequest.CurrentJob,
|
||||
determineOperationRequest.CurrentWorkItem.CurrentWorkItemExecution,
|
||||
status);
|
||||
var newWorkItemExecution = await _workItemExecutionService.CreateWorkItemExecution(nextWorkItem.WorkItemId);
|
||||
|
||||
await _queueExecutionService.QueueExecution(newWorkItemExecution);
|
||||
}
|
||||
}
|
||||
}
|
||||
+96
@@ -0,0 +1,96 @@
|
||||
using LCU.Api.Internal.Dispatcher.Services.QueueServices;
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using LCU.Library.Internal.LcuHttpClient.HttpRequest;
|
||||
using System.Collections.Generic;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices;
|
||||
public class ExecutionService : IExecutionService
|
||||
{
|
||||
private readonly IPostRequest _postRequest;
|
||||
private readonly IWorkItemService _workItemService;
|
||||
private readonly IJobService _jobService;
|
||||
private readonly IQueueCompletionService _queueCompletionService;
|
||||
|
||||
private const string HttRequestIsNullMessage = "Http Request is null";
|
||||
|
||||
public ExecutionService(IPostRequest postRequest,
|
||||
IWorkItemService workItemService,
|
||||
|
||||
IJobService jobService,
|
||||
IQueueCompletionService queueCompletionService)
|
||||
{
|
||||
_postRequest = postRequest;
|
||||
_workItemService = workItemService;
|
||||
_jobService = jobService;
|
||||
_queueCompletionService = queueCompletionService;
|
||||
}
|
||||
|
||||
public async Task HandleExecutionRequest(WorkItemExecution request)
|
||||
{
|
||||
try
|
||||
{
|
||||
var executionResponse = await ExecuteRequest(request);
|
||||
if (!executionResponse.IsSuccess)
|
||||
{
|
||||
await _queueCompletionService.QueueCompletion(new DispatcherReturnResponse() { WorkItemId = executionResponse.WorkItemExecution.WorkItemId },
|
||||
executionResponse.Messages, executionResponse.IsSuccess);
|
||||
}
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failed to perform execution. WorkItemId: {request.WorkItemId}");
|
||||
throw new ServiceException($"Failed to perform execution. WorkItemId: {request.WorkItemId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<ExecutionResponse> ExecuteRequest(WorkItemExecution request)
|
||||
{
|
||||
try
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(request);
|
||||
|
||||
var requestedWorkItem = await _workItemService.GetByIdAsync(request.WorkItemId);
|
||||
|
||||
await UpdateJobStatusToProcessing(requestedWorkItem);
|
||||
|
||||
var result = await _postRequest.AsyncWithNewtonSerializedNoThrow(new Uri(requestedWorkItem.Url), requestedWorkItem.SerializedPayload);
|
||||
if (!result.IsSuccessStatusCode)
|
||||
{
|
||||
var content = await result.Content.ReadAsStringAsync();
|
||||
return new ExecutionResponse(false, request, content);
|
||||
}
|
||||
return new ExecutionResponse(true, request);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
var errorMessage = $"Exception during ExecuteRequest. Exception Message: {ex.Message}. Inner Exception: {ex.InnerException}";
|
||||
Log.Error(errorMessage);
|
||||
return new ExecutionResponse(false, request, errorMessage);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task UpdateJobStatusToProcessing(WorkItem requestedWorkItem)
|
||||
{
|
||||
var job = await _jobService.GetByWorkItemIdExcludingWorkItemsAsync(requestedWorkItem.WorkItemId);
|
||||
if (job.JobStatus == JobStatus.Processing)
|
||||
{
|
||||
return;
|
||||
}
|
||||
job.JobStatus = JobStatus.Processing;
|
||||
await _jobService.UpdateAsync(job);
|
||||
Log.Information("Job Id {JobId} was saved with status {Status}", job.JobId, job.JobStatus);
|
||||
}
|
||||
|
||||
public struct ExecutionResponse
|
||||
{
|
||||
public ExecutionResponse(bool isSuccess, WorkItemExecution workItemExecution, string messages = null)
|
||||
{
|
||||
IsSuccess = isSuccess;
|
||||
WorkItemExecution = workItemExecution;
|
||||
Messages = messages != null ? new List<string>() { messages } : null;
|
||||
}
|
||||
|
||||
public bool IsSuccess { get; set; }
|
||||
public List<string> Messages { get; set; }
|
||||
public WorkItemExecution WorkItemExecution { get; set; }
|
||||
}
|
||||
}
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
using System.Threading;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices;
|
||||
|
||||
public interface IBatchErroredJobsService
|
||||
{
|
||||
Task PerformBatchErroredJobs(CancellationToken cancellationToken);
|
||||
}
|
||||
+8
@@ -0,0 +1,8 @@
|
||||
using System.Threading;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices;
|
||||
|
||||
public interface ICleanUpJobsService
|
||||
{
|
||||
Task PerformCleanUpJobs(CancellationToken cancellationToken);
|
||||
}
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
using LCU.Library.Internal.ApiResponse.Models;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices;
|
||||
|
||||
public interface ICompletionService
|
||||
{
|
||||
Task HandleCompletionRequest(Response response);
|
||||
}
|
||||
+9
@@ -0,0 +1,9 @@
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices
|
||||
{
|
||||
public interface IDetermineOperationService
|
||||
{
|
||||
Task HandleDetermineOperationRequest(DetermineOperationRequest determineOperationRequest);
|
||||
}
|
||||
}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices;
|
||||
|
||||
public interface IExecutionService
|
||||
{
|
||||
Task HandleExecutionRequest(WorkItemExecution request);
|
||||
}
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices;
|
||||
|
||||
public interface IIngestionService
|
||||
{
|
||||
Task HandleIngestionRequest(Job request);
|
||||
Task<bool> PerformIngestion(Job queueRequest);
|
||||
}
|
||||
+9
@@ -0,0 +1,9 @@
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices
|
||||
{
|
||||
public interface IJobFailureService
|
||||
{
|
||||
Task HandleJobFailureRequest(JobFailureRequest jobFailureRequest);
|
||||
}
|
||||
}
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
using System.Threading;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices;
|
||||
|
||||
public interface IJobScheduleService
|
||||
{
|
||||
Task PerformSchedule(CancellationToken cancellationToken);
|
||||
}
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices
|
||||
{
|
||||
public interface ISendErroredJobAlertService
|
||||
{
|
||||
Task HandleSendingErroredJobAlerts(List<BatchedErroredJobAlertRequest> batchedErroredJobAlertRequestList);
|
||||
}
|
||||
}
|
||||
+159
@@ -0,0 +1,159 @@
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using System.Collections.Generic;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices;
|
||||
|
||||
public class IngestionService : IIngestionService
|
||||
{
|
||||
private readonly IJobService _jobService;
|
||||
private readonly IWorkItemService _workItemService;
|
||||
private readonly IJobErrorRecipientService _jobErrorRecipientService;
|
||||
private readonly IJobTagService _jobTagService;
|
||||
|
||||
public IngestionService(IJobService jobService,
|
||||
IWorkItemService workItemService,
|
||||
IJobErrorRecipientService jobErrorRecipientService,
|
||||
IJobTagService jobTagService)
|
||||
{
|
||||
_jobService = jobService;
|
||||
_workItemService = workItemService;
|
||||
_jobErrorRecipientService = jobErrorRecipientService;
|
||||
_jobTagService = jobTagService;
|
||||
}
|
||||
|
||||
public async Task HandleIngestionRequest(Job request)
|
||||
{
|
||||
try
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(request);
|
||||
var result = await PerformIngestion(request);
|
||||
|
||||
if (!result)
|
||||
{
|
||||
throw new ServiceException("Failure to PerformIngestion.");
|
||||
}
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failed to perform execution.");
|
||||
throw new ServiceException($"Failed to perform ingestion.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<bool> PerformIngestion(Job request)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (request.IsNewJob)
|
||||
{
|
||||
return await SaveNewRequest(request);
|
||||
}
|
||||
|
||||
return await UpdateRequest(request);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Log.Information(ex, $"Failed to perform ingestion.");
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<bool> SaveNewRequest(Job request)
|
||||
{
|
||||
request = await _jobService.SaveAsync(request);
|
||||
if (request == null)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
var updatedWorkItems = await _workItemService.PopulateActionFieldsAndUpdateList(request.WorkItems);
|
||||
if (updatedWorkItems == null)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
public async Task<bool> UpdateRequest(Job request)
|
||||
{
|
||||
var existingJob = await _jobService.GetByIdAsync(request.JobId);
|
||||
|
||||
await UpdateRequestWorkItems(request, existingJob);
|
||||
UpdateRequestErrorRecipients(request, existingJob);
|
||||
await UpdateRequestJobTags(request, existingJob);
|
||||
request.JobStatus = (request.NextRunTime == null && request.CronSchedule == null) ? JobStatus.Inactive : existingJob.JobStatus;
|
||||
|
||||
var updatedJob = await _jobService.UpdateAsync(request);
|
||||
if (updatedJob == null)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
public async Task UpdateRequestWorkItems(Job request, Job existingJob)
|
||||
{
|
||||
var removedWorkItems = existingJob.WorkItems.Where(wi =>
|
||||
!request.WorkItems.Any(newWI =>
|
||||
newWI.WorkItemId == wi.WorkItemId)).ToList();
|
||||
await _workItemService.DeleteList(removedWorkItems);
|
||||
|
||||
var newWorkItems = request.WorkItems.Where(wi => wi.IsNewWorkItem).ToList();
|
||||
_workItemService.SaveList(newWorkItems);
|
||||
|
||||
var updatedWorkItems = await _workItemService.PopulateActionFieldsAndUpdateList(request.WorkItems);
|
||||
if (updatedWorkItems == null)
|
||||
{
|
||||
throw new Exception($"WorkItem Error while Populating Action Fields and updating.");
|
||||
}
|
||||
}
|
||||
|
||||
public void UpdateRequestErrorRecipients(Job request, Job existingJob)
|
||||
{
|
||||
var removedJobErrorRecipients = existingJob.JobErrorRecipients.Where(recipient =>
|
||||
!request.JobErrorRecipients.Any(newRecipient =>
|
||||
newRecipient.Recipient == recipient.Recipient)).ToList();
|
||||
_jobErrorRecipientService.DeleteList(removedJobErrorRecipients);
|
||||
|
||||
var newJobErrorRecipients = new List<JobErrorRecipient>();
|
||||
foreach (var recipient in request.JobErrorRecipients)
|
||||
{
|
||||
var jobErrorRecipient = existingJob.JobErrorRecipients.FirstOrDefault(existing =>
|
||||
existing.Recipient == recipient.Recipient);
|
||||
|
||||
if (jobErrorRecipient == null)
|
||||
{
|
||||
newJobErrorRecipients.Add(recipient);
|
||||
}
|
||||
}
|
||||
|
||||
_jobErrorRecipientService.SaveList(newJobErrorRecipients);
|
||||
}
|
||||
|
||||
public async Task UpdateRequestJobTags(Job request, Job existingJob)
|
||||
{
|
||||
var removedJobTags = existingJob.JobTags.Where(tag =>
|
||||
!request.JobTags.Any(jobTags =>
|
||||
jobTags.TagId == tag.TagId)).ToList();
|
||||
|
||||
foreach (var tag in removedJobTags)
|
||||
{
|
||||
await _jobTagService.Delete(tag.JobTagId);
|
||||
}
|
||||
|
||||
var newJobTags = new List<JobTag>();
|
||||
foreach (var tag in request.JobTags)
|
||||
{
|
||||
var existingTag = existingJob.JobTags.FirstOrDefault(existing =>
|
||||
existing.TagId == tag.TagId);
|
||||
|
||||
if (existingTag == null)
|
||||
{
|
||||
newJobTags.Add(tag);
|
||||
var result = await _jobTagService.SaveAsync(tag);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
+91
@@ -0,0 +1,91 @@
|
||||
using LCU.Api.Internal.Dispatcher.Services.QueueServices;
|
||||
using LCU.Api.Internal.MaidBotLib.Models;
|
||||
using LCU.Library.Internal.DispatcherModels.Entities;
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using LCU.Library.Internal.DispatcherModels.Models;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
using LCU.Library.Internal.DocumentIngestorModels.Models.IndexFileDocuments;
|
||||
using LCU.Library.Internal.Exceptions;
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices
|
||||
{
|
||||
public class JobFailureService : IJobFailureService
|
||||
{
|
||||
private readonly IWorkItemService _workItemService;
|
||||
private readonly IWorkItemExecutionService _workItemExecutionService;
|
||||
private readonly IJobService _jobService;
|
||||
private readonly IExecutionMessageService _executionMessageService;
|
||||
private readonly IQueueDetermineOperationService _queueDetermineOperationService;
|
||||
private readonly IQueueSendErroredJobAlertService _queueSendErroredJobAlertService;
|
||||
|
||||
public JobFailureService(IWorkItemService workItemService,
|
||||
IWorkItemExecutionService workItemExecutionService,
|
||||
IJobService jobService,
|
||||
IExecutionMessageService executionMessageService,
|
||||
IQueueDetermineOperationService queueDetermineOperationService,
|
||||
IQueueSendErroredJobAlertService queueSendErroredJobAlertService)
|
||||
{
|
||||
_workItemService = workItemService;
|
||||
_workItemExecutionService = workItemExecutionService;
|
||||
_jobService = jobService;
|
||||
_executionMessageService = executionMessageService;
|
||||
_queueDetermineOperationService = queueDetermineOperationService;
|
||||
_queueSendErroredJobAlertService = queueSendErroredJobAlertService;
|
||||
}
|
||||
|
||||
public async Task HandleJobFailureRequest(JobFailureRequest jobFailureRequest)
|
||||
{
|
||||
try
|
||||
{
|
||||
var workItem = await _workItemService.GetByIdAsync(jobFailureRequest.WorkItemId);
|
||||
|
||||
var updatedWorkItemExecution = await _workItemExecutionService.UpdateFailedWorkItemExecution(workItem.CurrentWorkItemExecution);
|
||||
|
||||
if (workItem.CancelJobOnFailure)
|
||||
{
|
||||
workItem = await _workItemService.IncreaseCanceledJobCounter(workItem);
|
||||
}
|
||||
|
||||
var job = await _jobService.GetByWorkItemIdExcludingWorkItemsIncludeErrorRecipientsAsync(jobFailureRequest.WorkItemId);
|
||||
|
||||
if (workItem.RequiresJobCancellation)
|
||||
{
|
||||
var message = $"This Job will be canceled and if scheduleable, rescheduled. The " +
|
||||
$"WorkItem's canceled Job counter is currently at {workItem.CanceledJobCounter} and has " +
|
||||
$"not yet went over the cancel Job limit of {workItem.CancelJobLimit}.";
|
||||
jobFailureRequest.ExecutionMessages.Add(message);
|
||||
job.JobStatus = JobStatus.DeterminingOperation;
|
||||
}
|
||||
else
|
||||
{
|
||||
job.JobStatus = JobStatus.Failed;
|
||||
}
|
||||
|
||||
await _executionMessageService.SaveExecutionMessages(workItem.CurrentWorkItemExecution.WorkItemExecutionId, jobFailureRequest.ExecutionMessages);
|
||||
var updatedJob = await _jobService.UpdateAsync(job);
|
||||
|
||||
if (updatedJob.JobStatus == JobStatus.DeterminingOperation)
|
||||
{
|
||||
await _queueDetermineOperationService.QueueDetermineOperation(updatedJob, workItem, DetermineOperationType.Failed);
|
||||
}
|
||||
else
|
||||
{
|
||||
await _queueSendErroredJobAlertService.QueueSendErroredJobAlert(updatedJob, workItem);
|
||||
}
|
||||
}
|
||||
catch (ServiceException serviceException)
|
||||
{
|
||||
Log.Error(serviceException, $"Dispatcher Job Failure Request Failure. " +
|
||||
$"The Dispatcher Job Failure Request is placed on LCU.Dispatcher.JobFailure_error.");
|
||||
throw new ServiceException($"Failed to handle JobFailureRequest.");
|
||||
}
|
||||
catch(Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failed to handle JobFailureRequest.");
|
||||
throw new ServiceException($"Failed to handle JobFailureRequest.");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+67
@@ -0,0 +1,67 @@
|
||||
using System.Threading;
|
||||
using LCU.Api.Internal.MaidBotLib.Models;
|
||||
using LCU.Api.Internal.Dispatcher.Services.QueueServices;
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using System;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices;
|
||||
|
||||
public class JobScheduleService : IJobScheduleService
|
||||
{
|
||||
private readonly IQueueExecutionService _queueExecutionService;
|
||||
private readonly IJobService _jobService;
|
||||
private readonly IWorkItemService _workItemService;
|
||||
private readonly IWorkItemExecutionService _workItemExecutionService;
|
||||
|
||||
public JobScheduleService(IQueueExecutionService queueExecutionService,
|
||||
IJobService jobService,
|
||||
IWorkItemService workItemService,
|
||||
IWorkItemExecutionService workItemExecutionService)
|
||||
{
|
||||
_queueExecutionService = queueExecutionService;
|
||||
_jobService = jobService;
|
||||
_workItemService = workItemService;
|
||||
_workItemExecutionService = workItemExecutionService;
|
||||
}
|
||||
|
||||
public async Task PerformSchedule(CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
var readyToScheduleJobs = await _jobService.GetJobsReadyToBeScheduled();
|
||||
|
||||
foreach (var job in readyToScheduleJobs)
|
||||
{
|
||||
var workItem = await _workItemService.GetFirstOrDefaultWorkItemByJobId(job.JobId);
|
||||
|
||||
if (workItem == null)
|
||||
{
|
||||
// THIS SHOULD NEVER HAPPEN
|
||||
Log.ForContext(Enum.GetName(typeof(PropertyEnum), PropertyEnum.Group),
|
||||
Enum.GetName(typeof(GroupEnum), GroupEnum.ITDevelopment))
|
||||
.Error($"Failure in Dispatcher Job Scheduler. Attempting to schedule Job: {job.JobId} - {job.Name} and NO workItems were found to process.");
|
||||
continue;
|
||||
}
|
||||
|
||||
var workItemExecution = await _workItemExecutionService.CreateWorkItemExecution(workItem.WorkItemId);
|
||||
|
||||
job.JobStatus = JobStatus.PendingToRun;
|
||||
|
||||
if (job.SoftDeleteDate != null)
|
||||
{
|
||||
job.SoftDeleteDate = null;
|
||||
}
|
||||
|
||||
await _jobService.UpdateAsync(job);
|
||||
Log.Information("Job Id {JobId} was saved with status {Status}", job.JobId, job.JobStatus);
|
||||
|
||||
await _queueExecutionService.QueueExecution(workItemExecution);
|
||||
}
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.ForContext(Enum.GetName(typeof(PropertyEnum), PropertyEnum.Group),
|
||||
Enum.GetName(typeof(GroupEnum), GroupEnum.ITDevelopment))
|
||||
.Error(exception, $"Failed to perform schedule.");
|
||||
}
|
||||
}
|
||||
}
|
||||
+146
@@ -0,0 +1,146 @@
|
||||
using LCU.Api.Internal.Dispatcher.Configurations;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.DispatcherOperationServices
|
||||
{
|
||||
public class SendErroredJobAlertService : ISendErroredJobAlertService
|
||||
{
|
||||
private readonly ApiConfiguration _apiConfiguration;
|
||||
private readonly INotificationService _notificationService;
|
||||
|
||||
public SendErroredJobAlertService(ApiConfiguration apiConfiguration, INotificationService notificationService)
|
||||
{
|
||||
_apiConfiguration = apiConfiguration;
|
||||
_notificationService = notificationService;
|
||||
}
|
||||
|
||||
public Task HandleSendingErroredJobAlerts(List<BatchedErroredJobAlertRequest> batchedErroredJobAlertRequestList)
|
||||
{
|
||||
var erroredJobAlertRequests = CombineFromBatch(batchedErroredJobAlertRequestList);
|
||||
|
||||
return SendErroredJobsReport(erroredJobAlertRequests);
|
||||
}
|
||||
|
||||
public async Task SendErroredJobsReport(List<ErroredJobAlertRequest> erroredJobAlertRequests)
|
||||
{
|
||||
var groupedJobsByRecipients = GroupJobsByRecipients(erroredJobAlertRequests);
|
||||
var result = true;
|
||||
foreach (var recipientGroup in groupedJobsByRecipients)
|
||||
{
|
||||
var successful = await HandleBatchErroredJobs(recipientGroup.Value, recipientGroup.Key);
|
||||
if (!successful)
|
||||
{
|
||||
result = false;
|
||||
}
|
||||
}
|
||||
|
||||
if (!result)
|
||||
{
|
||||
throw new ServiceException($"Failure to Send Errored Jobs Report. Result: {result}");
|
||||
}
|
||||
}
|
||||
|
||||
public List<ErroredJobAlertRequest> CombineFromBatch(List<BatchedErroredJobAlertRequest> batchedErroredJobAlertRequestList)
|
||||
{
|
||||
var erroredJobAlertRequests = new List<ErroredJobAlertRequest>();
|
||||
foreach (var batch in batchedErroredJobAlertRequestList)
|
||||
{
|
||||
erroredJobAlertRequests.AddRange(batch.ErroredJobAlertRequests);
|
||||
}
|
||||
return erroredJobAlertRequests;
|
||||
}
|
||||
|
||||
public Dictionary<string, List<ErroredJobAlertRequest>> GroupJobsByRecipients(List<ErroredJobAlertRequest> erroredJobAlertRequests)
|
||||
{
|
||||
var groupedRecipients = new Dictionary<string, List<ErroredJobAlertRequest>>();
|
||||
foreach (var erroredJobAlertRequest in erroredJobAlertRequests)
|
||||
{
|
||||
AddRecipients(erroredJobAlertRequest, groupedRecipients);
|
||||
}
|
||||
|
||||
return groupedRecipients;
|
||||
}
|
||||
|
||||
public void AddRecipients(ErroredJobAlertRequest erroredJbAlertRequest, Dictionary<string, List<ErroredJobAlertRequest>> groupedRecipients)
|
||||
{
|
||||
if (erroredJbAlertRequest.CurrentJob.JobErrorRecipients == null ||
|
||||
erroredJbAlertRequest.CurrentJob.JobErrorRecipients.Count == 0)
|
||||
{
|
||||
AddRecipient(_apiConfiguration.DefaultErrorRecipient, erroredJbAlertRequest, groupedRecipients);
|
||||
return;
|
||||
}
|
||||
|
||||
foreach (var jobErrorRecipient in erroredJbAlertRequest.CurrentJob.JobErrorRecipients)
|
||||
{
|
||||
AddRecipient(jobErrorRecipient.Recipient, erroredJbAlertRequest, groupedRecipients);
|
||||
}
|
||||
}
|
||||
|
||||
public void AddRecipient(string defaultErrorRecipient, ErroredJobAlertRequest erroredJobAlertRequest, Dictionary<string, List<ErroredJobAlertRequest>> groupedRecipients)
|
||||
{
|
||||
var recipientLower = defaultErrorRecipient.ToLower();
|
||||
if (!groupedRecipients.ContainsKey(recipientLower))
|
||||
{
|
||||
groupedRecipients.Add(recipientLower, new List<ErroredJobAlertRequest>());
|
||||
}
|
||||
|
||||
groupedRecipients[recipientLower].Add(erroredJobAlertRequest);
|
||||
}
|
||||
|
||||
public async Task<bool> HandleBatchErroredJobs(List<ErroredJobAlertRequest> erroredJobAlertRequests, string recipient)
|
||||
{
|
||||
string rowsOfErrors = "";
|
||||
foreach (ErroredJobAlertRequest erroredJobAlertRequest in erroredJobAlertRequests)
|
||||
{
|
||||
var jobErrorDetail = SetUpJobErrorDetails(erroredJobAlertRequest);
|
||||
if (jobErrorDetail != null)
|
||||
{
|
||||
rowsOfErrors += CreateErrorRow(jobErrorDetail);
|
||||
}
|
||||
}
|
||||
|
||||
var emailTemplate = CreateBatchErroredJobsEmail(rowsOfErrors);
|
||||
var successfullySent = await _notificationService.SendEmailAsync(recipient,
|
||||
_apiConfiguration.BatchErroredJobsSubject,
|
||||
emailTemplate);
|
||||
return successfullySent;
|
||||
}
|
||||
|
||||
public JobErrorDetail SetUpJobErrorDetails(ErroredJobAlertRequest erroredJobAlertRequest)
|
||||
{
|
||||
var jobErrorDetail = new JobErrorDetail()
|
||||
{
|
||||
JobName = erroredJobAlertRequest?.CurrentJob?.Name,
|
||||
WorkItemName = erroredJobAlertRequest?.CurrentWorkItem?.Name,
|
||||
JobStatus = erroredJobAlertRequest?.CurrentJob?.JobStatus.ToString(),
|
||||
StartTime = erroredJobAlertRequest?.CurrentWorkItem?.CurrentWorkItemExecution?.StartedTimestamp.ToString()
|
||||
};
|
||||
|
||||
return jobErrorDetail;
|
||||
}
|
||||
|
||||
public string CreateErrorRow(JobErrorDetail jobErrorDetails)
|
||||
{
|
||||
using var errorRowReader = new StreamReader(_apiConfiguration.ErrorRowTemplate);
|
||||
string errorRowOutput = errorRowReader.ReadToEnd();
|
||||
|
||||
foreach (var property in jobErrorDetails.GetType().GetProperties())
|
||||
{
|
||||
string value = property.GetValue(jobErrorDetails) == null ? string.Empty :
|
||||
property.GetValue(jobErrorDetails).ToString();
|
||||
errorRowOutput = errorRowOutput.Replace($"{{{property.Name}}}", value);
|
||||
}
|
||||
|
||||
return errorRowOutput;
|
||||
}
|
||||
|
||||
public string CreateBatchErroredJobsEmail(string errorRows)
|
||||
{
|
||||
using var batchErroredJobsReader = new StreamReader(_apiConfiguration.BatchErroredJobsTemplate);
|
||||
string batchErroredJobsOutput = batchErroredJobsReader.ReadToEnd();
|
||||
return batchErroredJobsOutput.Replace("{TableRows}", errorRows);
|
||||
}
|
||||
}
|
||||
}
|
||||
+90
@@ -0,0 +1,90 @@
|
||||
using System.Collections.Generic;
|
||||
using LCU.Api.Internal.DispatcherLib.Repositories;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public class ExecutionMessageService : IExecutionMessageService
|
||||
{
|
||||
private readonly IUnitOfWork _unitOfWork;
|
||||
private readonly IExecutionMessageRepository _executionMessageRepository;
|
||||
|
||||
public ExecutionMessageService(IUnitOfWork unitOfWork,
|
||||
IExecutionMessageRepository executionMessageRepository)
|
||||
{
|
||||
_unitOfWork = unitOfWork;
|
||||
_executionMessageRepository = executionMessageRepository;
|
||||
}
|
||||
|
||||
public async Task SaveExecutionMessages(List<ExecutionMessage> executionMessageList)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (executionMessageList == null || !executionMessageList.Any())
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
foreach (var message in executionMessageList)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(message.Message))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
var savedExecutionMessage = await SaveAsync(message);
|
||||
ArgumentNullException.ThrowIfNull(savedExecutionMessage);
|
||||
}
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "During SaveExecutionMessages there was an error. ");
|
||||
throw new ServiceException($"During SaveExecutionMessages there was an error. " +
|
||||
"ExceptionMessage: {exception.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task SaveExecutionMessages(int workItemExecutionId, List<string> messageList)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (messageList == null || !messageList.Any())
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
foreach (var message in messageList)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(message))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
var executionMessage = new ExecutionMessage(workItemExecutionId, message);
|
||||
var savedExecutionMessage = await SaveAsync(executionMessage);
|
||||
ArgumentNullException.ThrowIfNull(savedExecutionMessage);
|
||||
}
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"During SaveExecutionMessages there was an error. " +
|
||||
$"WorkItemExecutionId: {workItemExecutionId}");
|
||||
throw new ServiceException($"During SaveExecutionMessages there was an error. " +
|
||||
$"WorkItemExecutionId: {workItemExecutionId}, " +
|
||||
$"ExceptionMessage: {exception.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<ExecutionMessage> SaveAsync(ExecutionMessage workItemExecution)
|
||||
{
|
||||
try
|
||||
{
|
||||
var result = await _executionMessageRepository.AddAsync(workItemExecution);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
return result;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure saving dispatcher Execution Messages", workItemExecution);
|
||||
throw new ServiceException($"Failure saving dispatcher Execution Messages. ExceptionMessage: { exception.Message }");
|
||||
}
|
||||
}
|
||||
}
|
||||
+13
@@ -0,0 +1,13 @@
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services
|
||||
{
|
||||
public interface ICleanUpJobSettingService
|
||||
{
|
||||
Task<IEnumerable<CleanUpJobSetting>> GetAsync();
|
||||
Task<CleanUpJobSetting> GetByCleanUpJobSettingIdAsync(int cleanUpJobSettingId);
|
||||
Task<CleanUpJobSetting> SaveAsync(CleanUpJobSetting request);
|
||||
Task<CleanUpJobSetting> UpdateAsync(CleanUpJobSetting request);
|
||||
Task DeleteAsync(int cleanUpJobSettingId);
|
||||
}
|
||||
}
|
||||
+13
@@ -0,0 +1,13 @@
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public interface ICleanUpJobTagService
|
||||
{
|
||||
Task<IEnumerable<CleanUpJobTag>> GetAsync();
|
||||
Task<CleanUpJobTag> GetByCleanUpJobTagIdAsync(int cleanUpJobTagId);
|
||||
Task<IEnumerable<CleanUpJobTag>> GetByCleanUpJobSettingIdAsync(int cleanUpJobSettingId);
|
||||
Task<CleanUpJobTag> SaveAsync(CleanUpJobTag request);
|
||||
Task DeleteAsync(int cleanUpJobTagId);
|
||||
Task DeleteByCleanUpJobSettingIdAsync(int cleanUpJobSettingId);
|
||||
}
|
||||
+9
@@ -0,0 +1,9 @@
|
||||
using System.Collections.Generic;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public interface IExecutionMessageService
|
||||
{
|
||||
Task SaveExecutionMessages(List<ExecutionMessage> executionMessageList);
|
||||
Task SaveExecutionMessages(int workItemExecutionId, List<string> messageList);
|
||||
Task<ExecutionMessage> SaveAsync(ExecutionMessage workItemExecution);
|
||||
}
|
||||
+9
@@ -0,0 +1,9 @@
|
||||
using System.Collections.Generic;
|
||||
using LCU.Library.Internal.DispatcherModels.Entities;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public interface IJobErrorRecipientService
|
||||
{
|
||||
void SaveList(ICollection<JobErrorRecipient> jobErrorRecipient);
|
||||
void DeleteList(ICollection<JobErrorRecipient> jobErrorRecipient);
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
using System.Collections.Generic;
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Results;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public interface IJobService
|
||||
{
|
||||
Task<PagedResult<Job>> GetFilteredJobs(Filter filter);
|
||||
Task<IEnumerable<Job>> GetAsync();
|
||||
Task<IEnumerable<Job>> GetInactiveAsync();
|
||||
Task<IEnumerable<Job>> GetInactiveExcludeSoftDeletedTrackedAsync();
|
||||
Task<IEnumerable<Job>> GetSoftDeletedTrackedAsync(DateTime cutOffDate);
|
||||
Task<Job> GetByIdAsync(int id);
|
||||
Task<Job> GetByIdExcludingWorkItemsAsync(int id);
|
||||
Task<Job> GetByWorkItemIdAsync(int workItemId);
|
||||
Task<Job> GetByWorkItemIdExcludingWorkItemsAsync(int workItemId);
|
||||
Task<Job> GetByWorkItemIdExcludingWorkItemsIncludeErrorRecipientsAsync(int workItemId);
|
||||
Task<IEnumerable<Job>> GetJobsReadyToBeScheduled();
|
||||
Task<IEnumerable<Job>> GetErrorMonitoringJobs();
|
||||
Task<Job> SaveAsync(Job job);
|
||||
Task CompleteJobAndWorkItem(Job job, WorkItemExecution currentWorkItemExecution, StatusType finalStatus);
|
||||
Task CompleteWorkItemAndContinueProcessingJob(Job job, WorkItemExecution currentWorkItemExecution, StatusType finalStatus);
|
||||
Task<Job> UpdateAsync(Job job);
|
||||
Task<Job> UpdateExcludingWorkItemsAsyncWithoutSave(Job job);
|
||||
Task<Job> RemoveSoftDeleteById(int id);
|
||||
Task<Job> RunJobNowAsync(Job job);
|
||||
Task<Job> RunJobNowByIdAsync(int id);
|
||||
Task<Job> DeactivateJobAsync(Job job);
|
||||
Task<Job> RescheduleJobAsync(Job job);
|
||||
Task DeleteAsync(int jobId);
|
||||
Task DeleteFromCleanupAsync(Job job);
|
||||
Task<Job> CopyJobAsync(CopyJobRequest request);
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
using System.Collections.Generic;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public interface IJobTagService
|
||||
{
|
||||
Task<IEnumerable<JobTag>> GetAsync();
|
||||
Task<IEnumerable<JobTag>> GetByTagIdAsync(int tagId);
|
||||
Task<JobTag> SaveAsync(JobTag jobTag);
|
||||
Task Delete(int jobTagId);
|
||||
}
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
using System.Threading.Tasks;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public interface INotificationService
|
||||
{
|
||||
Task<bool> SendEmailAsync(string recipient, string subject, string emailTemplate);
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
using System.Collections.Generic;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public interface ITagService
|
||||
{
|
||||
Task<IEnumerable<Tag>> GetAsync();
|
||||
Task<Tag> GetByIdAsync(int tagId);
|
||||
Task<Tag> SaveAsync(Tag tag);
|
||||
Task<Tag> UpdateAsync(Tag tag);
|
||||
Task DeleteAsync(int tagId);
|
||||
}
|
||||
+17
@@ -0,0 +1,17 @@
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using System.Collections.Generic;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public interface IWorkItemExecutionService
|
||||
{
|
||||
Task<WorkItemExecution> GetByIdAsync(int workItemExecutionId);
|
||||
Task<WorkItemExecution> UpdateCompleteWorkItemExecution(WorkItemExecution workItemExecution, StatusType workItemFinalStatus);
|
||||
Task<WorkItemExecution> UpdateCompleteWorkItemExecutionWithoutSave(WorkItemExecution workItemExecution, StatusType workItemFinalStatus);
|
||||
Task<WorkItemExecution> UpdateAsync(WorkItemExecution workItemExecution);
|
||||
Task<WorkItemExecution> CreateWorkItemExecution(int workItemId);
|
||||
Task<WorkItemExecution> SaveAsync(WorkItemExecution workItemExecution);
|
||||
Task<WorkItemExecution> UpdateProcessingWorkItemExecution(WorkItemExecution workItemExecution);
|
||||
Task<WorkItemExecution> UpdateFailedWorkItemExecution(WorkItemExecution workItemExecution);
|
||||
Task<IEnumerable<WorkItemExecution>> GetAllUnFinishedWorkItemExecutions(int workItemExecutionId);
|
||||
Task<IEnumerable<WorkItemExecution>> GetAllByWorkItemId(int workItemId);
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
using System.Collections.Generic;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public interface IWorkItemService
|
||||
{
|
||||
Task<WorkItem> GetByIdAsync(int workItemId);
|
||||
Task<IEnumerable<WorkItem>> GetByJobIdAsync(int jobId);
|
||||
Task<IEnumerable<WorkItem>> GetByJobIdWithExecutionsAsync(int jobId);
|
||||
Task<WorkItem> GetFirstOrDefaultWorkItemByJobId(int jobId);
|
||||
Task<WorkItem> GetNextWorkItemByJobId(int jobId, int currentWorkItemOrder);
|
||||
void SaveList(ICollection<WorkItem> workItems);
|
||||
Task<ICollection<WorkItem>> PopulateActionFieldsAndUpdateList(ICollection<WorkItem> workItems);
|
||||
Task<WorkItem> ResetCanceledJobCounter(WorkItem workItem);
|
||||
Task<WorkItem> IncreaseCanceledJobCounter(WorkItem workItem);
|
||||
Task<WorkItem> UpdateAsync(WorkItem workItem);
|
||||
Task DeleteList(ICollection<WorkItem> workItems);
|
||||
WorkItem PopulateActionFields(WorkItem workItem);
|
||||
}
|
||||
+55
@@ -0,0 +1,55 @@
|
||||
using System.Collections.Generic;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using LCU.Api.Internal.DispatcherLib.Repositories;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public class JobErrorRecipientService : IJobErrorRecipientService
|
||||
{
|
||||
private readonly IUnitOfWork _unitOfWork;
|
||||
private readonly IJobErrorRecipientRepository _jobErrorRecipientRepository;
|
||||
|
||||
public JobErrorRecipientService(IUnitOfWork unitOfWork,
|
||||
IJobErrorRecipientRepository jobErrorRecipientRepository)
|
||||
{
|
||||
_unitOfWork = unitOfWork;
|
||||
_jobErrorRecipientRepository = jobErrorRecipientRepository;
|
||||
}
|
||||
|
||||
public void SaveList(ICollection<JobErrorRecipient> jobErrorRecipient)
|
||||
{
|
||||
try
|
||||
{
|
||||
_jobErrorRecipientRepository.AddList(jobErrorRecipient);
|
||||
_unitOfWork.Complete();
|
||||
}
|
||||
catch (SqlException ex)
|
||||
{
|
||||
Log.Error(ex, $"SQL Failure saving list of job error recipients.");
|
||||
throw new ServiceException($"SQL Failure saving list of job error recipients.");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Log.Error(ex, $"Failure saving list of job error recipients.");
|
||||
throw new ServiceException($"Failure saving list of job error recipients.");
|
||||
}
|
||||
}
|
||||
|
||||
public void DeleteList(ICollection<JobErrorRecipient> jobErrorRecipient)
|
||||
{
|
||||
try
|
||||
{
|
||||
_jobErrorRecipientRepository.RemoveList(jobErrorRecipient);
|
||||
_unitOfWork.Complete();
|
||||
}
|
||||
catch (SqlException ex)
|
||||
{
|
||||
Log.Error(ex, $"SQL Failure removing list of job error recipients.");
|
||||
throw new ServiceException($"SQL Failure removing list of job error recipients.");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
Log.Error(ex, $"Failure removing list of job error recipients.");
|
||||
throw new ServiceException($"Failure removing list of job error recipients.");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,583 @@
|
||||
using System.Collections.Generic;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using Newtonsoft.Json;
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using LCU.Api.Internal.DispatcherLib.Repositories;
|
||||
using LCU.Api.Internal.DispatcherSharedLib.Utilities;
|
||||
using LCU.Api.Internal.Dispatcher.Configurations;
|
||||
using LCU.Library.Internal.TypeExtensions.ObjectExtensions;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Results;
|
||||
using LCU.Library.Internal.Logging.Performance;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public class JobService : IJobService
|
||||
{
|
||||
private readonly IJobRepository _jobRepository;
|
||||
private readonly IWorkItemService _workItemService;
|
||||
private readonly IWorkItemExecutionService _workItemExecutionService;
|
||||
private readonly IUnitOfWork _unitOfWork;
|
||||
private readonly ApiConfiguration _apiConfiguration;
|
||||
|
||||
public JobService(IUnitOfWork unitOfWork,
|
||||
IJobRepository jobRepository,
|
||||
IWorkItemService workItemService,
|
||||
IWorkItemExecutionService workItemExecutionService,
|
||||
ApiConfiguration apiConfiguration)
|
||||
{
|
||||
_unitOfWork = unitOfWork;
|
||||
_jobRepository = jobRepository;
|
||||
_workItemService = workItemService;
|
||||
_workItemExecutionService = workItemExecutionService;
|
||||
_apiConfiguration = apiConfiguration;
|
||||
}
|
||||
|
||||
public async Task<PagedResult<Job>> GetFilteredJobs(Filter filter)
|
||||
{
|
||||
try
|
||||
{
|
||||
var performanceMonitorId = Guid.NewGuid();
|
||||
var getFilteredJobsQueryMonitor = new PerformanceMonitor("GetFilteredJobs", performanceMonitorId);
|
||||
|
||||
var filteredJobs = await _jobRepository.FilteredListAsync(filter,
|
||||
_apiConfiguration.NumberOfWorkItemExecutions,
|
||||
_apiConfiguration.NumberOfExecutionMessages);
|
||||
|
||||
getFilteredJobsQueryMonitor.LogPerformance();
|
||||
|
||||
return filteredJobs;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure retrieving filtered dispatcher jobs.");
|
||||
throw new ServiceException("Failure retrieving filtered dispatcher jobs.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<Job>> GetAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
return await _jobRepository.ListAsync(_apiConfiguration.NumberOfWorkItemExecutions, _apiConfiguration.NumberOfExecutionMessages);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure retrieving dispatcher jobs.");
|
||||
throw new ServiceException("Failure retrieving dispatcher jobs.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<Job>> GetInactiveAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
return await _jobRepository.ListInactiveAsync(_apiConfiguration.NumberOfWorkItemExecutions, _apiConfiguration.NumberOfExecutionMessages);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure retrieving inactive dispatcher jobs.");
|
||||
throw new ServiceException("Failure retrieving inactive dispatcher jobs.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<Job>> GetInactiveExcludeSoftDeletedTrackedAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
return await _jobRepository.ListInactiveExcludeSoftDeletedTrackedAsync(_apiConfiguration.NumberOfWorkItemExecutions, _apiConfiguration.NumberOfExecutionMessages);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure retrieving non-SoftDeleted inactive dispatcher jobs.");
|
||||
throw new ServiceException("Failure retrieving non-SoftDeleted inactive dispatcher jobs.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<Job>> GetSoftDeletedTrackedAsync(DateTime cutOffDate)
|
||||
{
|
||||
try
|
||||
{
|
||||
return await _jobRepository.ListSoftDeletedTrackedAsync(_apiConfiguration.NumberOfWorkItemExecutions, _apiConfiguration.NumberOfExecutionMessages, cutOffDate);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure retrieving soft deleted dispatcher jobs.");
|
||||
throw new ServiceException("Failure retrieving soft deleted dispatcher jobs.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Job> GetByIdAsync(int id)
|
||||
{
|
||||
try
|
||||
{
|
||||
var job = await _jobRepository.FindByIdAsync(id,
|
||||
_apiConfiguration.NumberOfWorkItemExecutions,
|
||||
_apiConfiguration.NumberOfExecutionMessages);
|
||||
ArgumentNullException.ThrowIfNull(job);
|
||||
return job;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher jobs with id {id}.");
|
||||
throw new ServiceException($"Failure retrieving dispatcher jobs with id {id}.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Job> GetByIdExcludingWorkItemsAsync(int id)
|
||||
{
|
||||
try
|
||||
{
|
||||
var job = await _jobRepository.FindByIdExcludingWorkItemsAsync(id);
|
||||
ArgumentNullException.ThrowIfNull(job);
|
||||
return job;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher jobs with id {id}.");
|
||||
throw new ServiceException($"Failure retrieving dispatcher jobs with id {id}.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Job> GetByWorkItemIdAsync(int workItemId)
|
||||
{
|
||||
try
|
||||
{
|
||||
var job = await _jobRepository.FindByWorkItemIdAsync(workItemId,
|
||||
_apiConfiguration.NumberOfWorkItemExecutions,
|
||||
_apiConfiguration.NumberOfExecutionMessages);
|
||||
ArgumentNullException.ThrowIfNull(job);
|
||||
return job;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher jobs by WorkItemId {workItemId}.");
|
||||
throw new ServiceException($"Failure retrieving dispatcher jobs by WorkItemId {workItemId}.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Job> GetByWorkItemIdExcludingWorkItemsAsync(int workItemId)
|
||||
{
|
||||
try
|
||||
{
|
||||
var job = await _jobRepository.FindByWorkItemIdExcludingWorkItemsAsync(workItemId);
|
||||
ArgumentNullException.ThrowIfNull(job);
|
||||
return job;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher jobs excluding WorkItems by WorkItemId {workItemId}.");
|
||||
throw new ServiceException($"Failure retrieving dispatcher jobs excluding WorkItems by WorkItemId {workItemId}.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Job> GetByWorkItemIdExcludingWorkItemsIncludeErrorRecipientsAsync(int workItemId)
|
||||
{
|
||||
try
|
||||
{
|
||||
var job = await _jobRepository.FindByWorkItemIdExcludingWorkItemsIncludeErrorRecipientsAsync(workItemId);
|
||||
ArgumentNullException.ThrowIfNull(job);
|
||||
return job;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher jobs excluding WorkItems by WorkItemId {workItemId}.");
|
||||
throw new ServiceException($"Failure retrieving dispatcher jobs excluding WorkItems by WorkItemId {workItemId}.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<Job>> GetJobsReadyToBeScheduled()
|
||||
{
|
||||
try
|
||||
{
|
||||
var result = await _jobRepository.FindJobsReadyToBeScheduled(DateTime.Now);
|
||||
return result;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure getting pending Jobs");
|
||||
throw new ServiceException("Failure getting pending Jobs");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<Job>> GetErrorMonitoringJobs()
|
||||
{
|
||||
try
|
||||
{
|
||||
var result = await _jobRepository.FindJobsPastDateByStatuses(DateTime.Now,
|
||||
_apiConfiguration.NumberOfWorkItemExecutions,
|
||||
_apiConfiguration.NumberOfExecutionMessages,
|
||||
_apiConfiguration.JobStatusesToMonitorByTimeLimit);
|
||||
ArgumentNullException.ThrowIfNull(result);
|
||||
var erroredAndStuckJobs = result.Where(job => job.JobStatus == JobStatus.Failed).ToList();
|
||||
if (_apiConfiguration.JobStatusesToMonitorByTimeLimit is not null && _apiConfiguration.JobStatusesToMonitorByTimeLimit.Count() > 0)
|
||||
{
|
||||
erroredAndStuckJobs.AddRange(GetJobsInStatusLongerThanTimeLimit(result.Except(erroredAndStuckJobs).ToList(),
|
||||
_apiConfiguration.ProcessingLimitInMinutes));
|
||||
}
|
||||
|
||||
return erroredAndStuckJobs;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure getting next errored and stuck in processing Jobs");
|
||||
throw new ServiceException("Failure getting next errored and stuck in processing Jobs");
|
||||
}
|
||||
}
|
||||
|
||||
public List<Job> GetJobsInStatusLongerThanTimeLimit(List<Job> matchingStatusJobs, int processingLimitInMinutes)
|
||||
{
|
||||
var stuckInStatusJobs = new List<Job>();
|
||||
|
||||
foreach (var job in matchingStatusJobs)
|
||||
{
|
||||
var currentWorkItem = job.WorkItems.FirstOrDefault(wi => wi.CurrentWorkItemExecution != null);
|
||||
if (currentWorkItem == null)
|
||||
{
|
||||
Log.Warning($"Current Work Item came back as null when attempting to get stuck in processing Jobs. JobId: {job.JobId}");
|
||||
continue;
|
||||
}
|
||||
|
||||
var startTime = Convert.ToDateTime(currentWorkItem.CurrentWorkItemExecution.StartedTimestamp);
|
||||
var minutesInStatus = DateTime.Now.Subtract(startTime).TotalMinutes;
|
||||
if (minutesInStatus > processingLimitInMinutes)
|
||||
{
|
||||
stuckInStatusJobs.Add(job);
|
||||
}
|
||||
}
|
||||
|
||||
return stuckInStatusJobs;
|
||||
}
|
||||
|
||||
public async Task<Job> SaveAsync(Job job)
|
||||
{
|
||||
try
|
||||
{
|
||||
var result = await _jobRepository.AddAsync(job);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
ArgumentNullException.ThrowIfNull(result);
|
||||
return result;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure saving dispatcher job", job);
|
||||
throw new ServiceException("Failure saving dispatcher job");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task CompleteJobAndWorkItem(Job job, WorkItemExecution currentWorkItemExecution, StatusType workItemFinalStatus)
|
||||
{
|
||||
try
|
||||
{
|
||||
job.ResetRunTimeAndStatus();
|
||||
CronHelpers.SetNextScheduledRunTime(ref job);
|
||||
var updatedJob = await UpdateExcludingWorkItemsAsyncWithoutSave(job);
|
||||
if (updatedJob == null)
|
||||
{
|
||||
throw new Exception($"Failure to setup update for job. Job:{JsonConvert.SerializeObject(job)}");
|
||||
}
|
||||
|
||||
var updatedWorkItemExecution = await _workItemExecutionService.UpdateCompleteWorkItemExecutionWithoutSave(currentWorkItemExecution, workItemFinalStatus);
|
||||
if (updatedWorkItemExecution == null)
|
||||
{
|
||||
throw new Exception($"Failure to setup update for WorkItemExecution. WorkItemExecution:{JsonConvert.SerializeObject(currentWorkItemExecution)}");
|
||||
}
|
||||
await _unitOfWork.CompleteAsync();
|
||||
Log.Information("Job Id {JobId} was saved with status {Status}", job.JobId, job.JobStatus);
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
var message = $"SQL Failure to Complete Job and WorkItemExecution. JobId: {job.JobId}, WorkItemExecutionId: {currentWorkItemExecution.WorkItemExecutionId}.";
|
||||
Log.Error(sqlException, message);
|
||||
throw new ServiceException($"{message} ExceptionMessage: {sqlException.Message}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
var message = $"Failure to Complete Job and WorkItemExecution. JobId: {job.JobId}, WorkItemExecutionId: {currentWorkItemExecution.WorkItemExecutionId}.";
|
||||
Log.Error(exception, message);
|
||||
throw new ServiceException($"{message} ExceptionMessage: {exception.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task CompleteWorkItemAndContinueProcessingJob(Job job, WorkItemExecution currentWorkItemExecution, StatusType workItemFinalStatus)
|
||||
{
|
||||
try
|
||||
{
|
||||
job.JobStatus = JobStatus.Processing;
|
||||
var updatedJob = await UpdateExcludingWorkItemsAsyncWithoutSave(job);
|
||||
if (updatedJob == null)
|
||||
{
|
||||
throw new Exception($"Failure to setup update for job. Job:{JsonConvert.SerializeObject(job)}");
|
||||
}
|
||||
|
||||
var updatedWorkItemExecution = await _workItemExecutionService.UpdateCompleteWorkItemExecutionWithoutSave(currentWorkItemExecution, workItemFinalStatus);
|
||||
if (updatedWorkItemExecution == null)
|
||||
{
|
||||
throw new Exception($"Failure to setup update for WorkItemExecution. WorkItemExecution:{JsonConvert.SerializeObject(currentWorkItemExecution)}");
|
||||
}
|
||||
await _unitOfWork.CompleteAsync();
|
||||
Log.Information("Job Id {JobId} was saved with status {Status}", job.JobId, job.JobStatus);
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
var message = $"SQL Failure to Complete Job and WorkItemExecution. JobId: {job.JobId}, WorkItemExecutionId: {currentWorkItemExecution.WorkItemExecutionId}.";
|
||||
Log.Error(sqlException, message);
|
||||
throw new ServiceException($"{message} ExceptionMessage: {sqlException.Message}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
var message = $"Failure to Complete Job and WorkItemExecution. JobId: {job.JobId}, WorkItemExecutionId: {currentWorkItemExecution.WorkItemExecutionId}.";
|
||||
Log.Error(exception, message);
|
||||
throw new ServiceException($"{message} ExceptionMessage: {exception.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Job> UpdateAsync(Job job)
|
||||
{
|
||||
try
|
||||
{
|
||||
var updatedJob = await _jobRepository.Update(job);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
|
||||
return updatedJob;
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, "SQL Failure updating dispatcher job", job);
|
||||
throw new ServiceException("SQL Failure updating dispatcher job");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure updating dispatcher job", job);
|
||||
throw new ServiceException("Failure updating dispatcher job");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Job> UpdateExcludingWorkItemsAsyncWithoutSave(Job job)
|
||||
{
|
||||
return await _jobRepository.Update(job);
|
||||
}
|
||||
|
||||
public async Task<Job> RemoveSoftDeleteById(int id)
|
||||
{
|
||||
try
|
||||
{
|
||||
var existing = await GetByIdAsync(id);
|
||||
ArgumentNullException.ThrowIfNull(existing);
|
||||
|
||||
existing.SoftDeleteDate = null;
|
||||
_jobRepository.Update(existing);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
|
||||
return existing;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure to remove soft delete date for dispatcher job. JobId: {id}");
|
||||
throw new ServiceException($"Failure to remove soft delete date for dispatcher job. JobId: {id}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task DeleteAsync(int jobId)
|
||||
{
|
||||
try
|
||||
{
|
||||
var job = await _jobRepository.FindByIdAsync(jobId,
|
||||
_apiConfiguration.NumberOfWorkItemExecutions,
|
||||
_apiConfiguration.NumberOfExecutionMessages);
|
||||
ArgumentNullException.ThrowIfNull(job);
|
||||
|
||||
_jobRepository.Remove(job);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, $"SQL Failure removing dispatcher job. JobId: {jobId}");
|
||||
throw new ServiceException($"SQL Failure removing dispatcher job. JobId: {jobId}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure removing dispatcher job. JobId: {jobId}");
|
||||
throw new ServiceException($"Failure removing dispatcher job. JobId: {jobId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task DeleteFromCleanupAsync(Job job)
|
||||
{
|
||||
try
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(job);
|
||||
|
||||
_jobRepository.Remove(job);
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, $"SQL Failure removing dispatcher job. JobId: {job.JobId}");
|
||||
throw new ServiceException($"SQL Failure removing dispatcher job. JobId: {job.JobId}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure removing dispatcher job. JobId: {job.JobId}");
|
||||
throw new ServiceException($"Failure removing dispatcher job. JobId: {job.JobId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Job> RunJobNowAsync(Job job)
|
||||
{
|
||||
try
|
||||
{
|
||||
job = await ResetJob(job);
|
||||
job.NextRunTime = DateTime.Now;
|
||||
job.JobStatus = JobStatus.ReadyToSchedule;
|
||||
var updatedJob = await UpdateAsync(job);
|
||||
return updatedJob;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure to set job to run now.");
|
||||
throw new ServiceException("Failure to set job to run now.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Job> RunJobNowByIdAsync(int id)
|
||||
{
|
||||
try
|
||||
{
|
||||
var job = await GetByIdAsync(id);
|
||||
var updatedJob = await RunJobNowAsync(job);
|
||||
return updatedJob;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure to set job to run now by id. Id: {id}");
|
||||
throw new ServiceException($"Failure to set job to run now by id. Id: {id}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Job> DeactivateJobAsync(Job job)
|
||||
{
|
||||
try
|
||||
{
|
||||
job = await ResetJob(job);
|
||||
var updatedJob = await UpdateAsync(job);
|
||||
return updatedJob;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
|
||||
Log.Error(exception, "Failure to deactivate job.");
|
||||
throw new ServiceException("Failure to deactivate job.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Job> RescheduleJobAsync(Job job)
|
||||
{
|
||||
try
|
||||
{
|
||||
job = await ResetJob(job);
|
||||
CronHelpers.SetNextScheduledRunTime(ref job);
|
||||
var updatedJob = await UpdateAsync(job);
|
||||
Log.Information("Job Id {JobId} was saved with status {Status}", job.JobId, job.JobStatus);
|
||||
return updatedJob;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure to reschedule job.");
|
||||
throw new ServiceException("Failure to reschedule job.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Job> ResetJob(Job job)
|
||||
{
|
||||
foreach (var workItem in job.WorkItems)
|
||||
{
|
||||
var executionsToReset = await _workItemExecutionService.GetAllUnFinishedWorkItemExecutions(workItem.WorkItemId);
|
||||
foreach (var workItemExecution in executionsToReset)
|
||||
{
|
||||
workItemExecution.StartedTimestamp = workItemExecution.StartedTimestamp == null ?
|
||||
DateTime.Now :
|
||||
workItemExecution.StartedTimestamp;
|
||||
workItemExecution.CompletedTimestamp = workItemExecution.CompletedTimestamp == null ?
|
||||
DateTime.Now :
|
||||
workItemExecution.CompletedTimestamp;
|
||||
|
||||
var updatedExecution = await _workItemExecutionService.UpdateAsync(workItemExecution);
|
||||
ArgumentNullException.ThrowIfNull(updatedExecution);
|
||||
}
|
||||
}
|
||||
|
||||
job.ResetRunTimeAndStatus();
|
||||
return job;
|
||||
}
|
||||
|
||||
public async Task<Job> CopyJobAsync(CopyJobRequest request)
|
||||
{
|
||||
try
|
||||
{
|
||||
var job = (await GetByIdAsync(request.JobId)).DeepCopy();
|
||||
|
||||
if (job == null)
|
||||
{
|
||||
throw new Exception($"Job with id {request.JobId} not found.");
|
||||
}
|
||||
|
||||
job.Name = $"(COPY) {job.Name}";
|
||||
job.JobId = 0;
|
||||
job.CronSchedule = null;
|
||||
job.CreatedAt = DateTime.Now;
|
||||
job.ModifiedDate = null;
|
||||
job.ResetRunTimeAndStatus();
|
||||
|
||||
foreach (var errorRecipient in job.JobErrorRecipients)
|
||||
{
|
||||
errorRecipient.JobErrorRecipientId = 0;
|
||||
errorRecipient.JobId = 0;
|
||||
}
|
||||
|
||||
if (request.ChangeEnvironment)
|
||||
{
|
||||
job.JobTags = null;
|
||||
|
||||
foreach (var workItem in job.WorkItems)
|
||||
{
|
||||
workItem.WorkItemId = 0;
|
||||
workItem.JobId = 0;
|
||||
workItem.Url = null;
|
||||
workItem.WorkItemExecutions = null;
|
||||
}
|
||||
|
||||
return job;
|
||||
}
|
||||
|
||||
foreach (var workItem in job.WorkItems)
|
||||
{
|
||||
workItem.WorkItemId = 0;
|
||||
workItem.JobId = 0;
|
||||
workItem.Url = null;
|
||||
workItem.WorkItemExecutions = null;
|
||||
}
|
||||
|
||||
foreach (var jobTag in job.JobTags)
|
||||
{
|
||||
jobTag.JobTagId = 0;
|
||||
jobTag.JobId = 0;
|
||||
jobTag.Tag = null;
|
||||
}
|
||||
|
||||
var savedJob = await SaveAsync(job);
|
||||
|
||||
if (savedJob == null)
|
||||
{
|
||||
throw new Exception("Failure to save new job.");
|
||||
}
|
||||
|
||||
return savedJob;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
var message = $"Failure to copy dispatcher job. Request: {JsonConvert.SerializeObject(request)}";
|
||||
Log.Error(exception, message);
|
||||
throw new ServiceException($"{message}. Exception: {exception.Message}");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
using System.Collections.Generic;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using LCU.Api.Internal.DispatcherLib.Repositories;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public class JobTagService : IJobTagService
|
||||
{
|
||||
private readonly IJobTagRepository _jobTagRepository;
|
||||
private readonly IUnitOfWork _unitOfWork;
|
||||
|
||||
public JobTagService(IUnitOfWork unitOfWork,
|
||||
IJobTagRepository jobTagRepository)
|
||||
{
|
||||
_unitOfWork = unitOfWork;
|
||||
_jobTagRepository = jobTagRepository;
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<JobTag>> GetAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
return await _jobTagRepository.ListAsync();
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure retrieving JobTags.");
|
||||
throw new ServiceException("Failure retrieving JobTags.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<JobTag>> GetByTagIdAsync(int tagId)
|
||||
{
|
||||
try
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(tagId);
|
||||
return await _jobTagRepository.FindByTagIdAsync(tagId);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher Job Tags. TagId: {tagId}");
|
||||
throw new ServiceException($"Failure retrieving dispatcher Job Tags. TagId: {tagId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<JobTag> SaveAsync(JobTag jobTag)
|
||||
{
|
||||
try
|
||||
{
|
||||
var result = await _jobTagRepository.AddAsync(jobTag);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
ArgumentNullException.ThrowIfNull(result);
|
||||
return result;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure saving dispatcher JobTag. JobTagId: {jobTag.JobTagId}");
|
||||
throw new ServiceException($"Failure saving dispatcher JobTag. JobTagId: {jobTag.JobTagId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task Delete(int jobTagId)
|
||||
{
|
||||
try
|
||||
{
|
||||
var jobTag = await _jobTagRepository.FindByIdAsync(jobTagId);
|
||||
ArgumentNullException.ThrowIfNull(jobTag);
|
||||
_jobTagRepository.Remove(jobTag);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, $"SQL Failure removing dispatcher JobTag. JobTagId: {jobTagId}");
|
||||
throw new ServiceException($"SQL Failure removing dispatcher JobTag. JobTagId: {jobTagId}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure removing dispatcher JobTag. JobTagId: {jobTagId}");
|
||||
throw new ServiceException($"Failure removing dispatcher JobTag. JobTagId: {jobTagId}");
|
||||
}
|
||||
}
|
||||
}
|
||||
+40
@@ -0,0 +1,40 @@
|
||||
using System.Collections.Generic;
|
||||
using Newtonsoft.Json;
|
||||
using LCU.Library.Internal.NotificationsModels.Models;
|
||||
using LCU.Api.Internal.DispatcherSharedLib.Configurations;
|
||||
using LCU.Api.Internal.DispatcherSharedLib.Utilities;
|
||||
using LCU.Api.Internal.Dispatcher.DataAccess;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public class NotificationService : INotificationService
|
||||
{
|
||||
private readonly INotificationsApiDao _notificationsApiDao;
|
||||
|
||||
public NotificationService(INotificationsApiDao notificationsApiDao)
|
||||
{
|
||||
_notificationsApiDao = notificationsApiDao;
|
||||
}
|
||||
|
||||
public async Task<bool> SendEmailAsync(string recipient, string subject, string emailTemplate)
|
||||
{
|
||||
try
|
||||
{
|
||||
var emailRequest = new EmailRequest
|
||||
{
|
||||
Recipient = new List<string>() { recipient },
|
||||
Subject = subject,
|
||||
Body = emailTemplate
|
||||
};
|
||||
|
||||
await _notificationsApiDao.SendEmailAsync(emailRequest);
|
||||
|
||||
return true;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure to send email from Dispatcher");
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
+8
@@ -0,0 +1,8 @@
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.QueueServices;
|
||||
|
||||
public interface IQueueCompletionService
|
||||
{
|
||||
Task QueueCompletion(DispatcherReturnResponse request, List<string> executionMessages = null, bool successful = true);
|
||||
}
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.QueueServices
|
||||
{
|
||||
public interface IQueueDetermineOperationService
|
||||
{
|
||||
Task QueueDetermineOperation(Job job, WorkItem workItem, DetermineOperationType type);
|
||||
}
|
||||
}
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.QueueServices;
|
||||
|
||||
public interface IQueueExecutionService
|
||||
{
|
||||
Task QueueExecution(WorkItemExecution request);
|
||||
Task QueueRetryExecution(WorkItemExecution request);
|
||||
}
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.QueueServices
|
||||
{
|
||||
public interface IQueueJobFailureService
|
||||
{
|
||||
public Task QueueJobFailure(int jobId, int workItemId, List<string> messages);
|
||||
}
|
||||
}
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.QueueServices
|
||||
{
|
||||
public interface IQueueSendErroredJobAlertService
|
||||
{
|
||||
public Task QueueSendErroredJobAlert(BatchedErroredJobAlertRequest request);
|
||||
public Task QueueSendErroredJobAlert(Job job, WorkItem workItem);
|
||||
}
|
||||
}
|
||||
+45
@@ -0,0 +1,45 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using MassTransit;
|
||||
using Newtonsoft.Json;
|
||||
using ApiResponse = LCU.Library.Internal.ApiResponse.Models;
|
||||
using LCU.Api.Internal.MaidBotLib.Models;
|
||||
using LCU.Api.Internal.Dispatcher.Configurations;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.QueueServices;
|
||||
|
||||
public class QueueCompletionService : IQueueCompletionService
|
||||
{
|
||||
private readonly ISendEndpoint _sendEndpoint;
|
||||
private readonly ReceiveEndpointConfigurations _queueReceiveEndpointsConfiguration;
|
||||
|
||||
public QueueCompletionService(IBusControl busControl,
|
||||
ReceiveEndpointConfigurations queueReceiveEndpointsConfiguration)
|
||||
{
|
||||
_queueReceiveEndpointsConfiguration = queueReceiveEndpointsConfiguration;
|
||||
_sendEndpoint = busControl.GetSendEndpoint(_queueReceiveEndpointsConfiguration.CompletionReceiveEndpoint.ReceiveEndpointUri).Result;
|
||||
}
|
||||
|
||||
public async Task QueueCompletion(DispatcherReturnResponse request, List<string> messages = null, bool successful = true)
|
||||
{
|
||||
try
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(request);
|
||||
|
||||
var response = new ApiResponse.Response()
|
||||
{
|
||||
Successful = successful,
|
||||
ResponseStack = messages == null ? new List<string>() : messages,
|
||||
ResponsePayload = JsonConvert.SerializeObject(request)
|
||||
};
|
||||
var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
|
||||
await _sendEndpoint.Send(response, cancellationTokenSource.Token);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.ForContext(Enum.GetName(typeof(PropertyEnum), PropertyEnum.Group),
|
||||
Enum.GetName(typeof(GroupEnum), GroupEnum.ITDevelopment))
|
||||
.Fatal(exception, $"Failed to queue completion workitem request. WorkItemId: {request?.WorkItemId}");
|
||||
throw new ServiceException($"Failed to queue completion workitem request. WorkItemId: {request?.WorkItemId}");
|
||||
}
|
||||
}
|
||||
}
|
||||
+48
@@ -0,0 +1,48 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using LCU.Api.Internal.Dispatcher.Configurations;
|
||||
using LCU.Api.Internal.MaidBotLib.Models;
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
using MassTransit;
|
||||
using Newtonsoft.Json;
|
||||
using ApiResponse = LCU.Library.Internal.ApiResponse.Models;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.QueueServices
|
||||
{
|
||||
public class QueueDetermineOperationService : IQueueDetermineOperationService
|
||||
{
|
||||
private readonly ISendEndpoint _sendEndpoint;
|
||||
private readonly ReceiveEndpointConfigurations _queueReceiveEndpointsConfiguration;
|
||||
|
||||
public QueueDetermineOperationService(IBusControl busControl,
|
||||
ReceiveEndpointConfigurations queueReceiveEndpointsConfiguration)
|
||||
{
|
||||
_queueReceiveEndpointsConfiguration = queueReceiveEndpointsConfiguration;
|
||||
_sendEndpoint = busControl.GetSendEndpoint(_queueReceiveEndpointsConfiguration.DetermineOperationReceiveEndpoint.ReceiveEndpointUri).Result;
|
||||
}
|
||||
|
||||
public async Task QueueDetermineOperation(Job job, WorkItem workItem, DetermineOperationType type)
|
||||
{
|
||||
try
|
||||
{
|
||||
var request = new DetermineOperationRequest
|
||||
{
|
||||
CurrentJob = job,
|
||||
CurrentWorkItem = workItem,
|
||||
DetermineOperationType = type
|
||||
};
|
||||
|
||||
var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
|
||||
await _sendEndpoint.Send(request, cancellationTokenSource.Token);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.ForContext(Enum.GetName(typeof(PropertyEnum), PropertyEnum.Group),
|
||||
Enum.GetName(typeof(GroupEnum), GroupEnum.ITDevelopment))
|
||||
.Fatal(exception, $"Failed to queue determine operation request. JobName: {job.Name} WorkItemId: {workItem.WorkItemId}");
|
||||
throw new ServiceException($"Failed to queue determine operation request. JobName: {job.Name} WorkItemId: {workItem.WorkItemId}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+55
@@ -0,0 +1,55 @@
|
||||
using System.Threading;
|
||||
using MassTransit;
|
||||
using LCU.Api.Internal.MaidBotLib.Models;
|
||||
using LCU.Api.Internal.Dispatcher.Configurations;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.QueueServices;
|
||||
|
||||
public class QueueExecutionService : IQueueExecutionService
|
||||
{
|
||||
private readonly ISendEndpoint _sendEndpoint;
|
||||
private readonly IWorkItemExecutionService _workItemExecutionService;
|
||||
private readonly ReceiveEndpointConfigurations _queueReceiveEndpointsConfiguration;
|
||||
|
||||
public QueueExecutionService(IBusControl busControl,
|
||||
IWorkItemExecutionService workItemExecutionService,
|
||||
ReceiveEndpointConfigurations queueReceiveEndpointsConfiguration)
|
||||
{
|
||||
_queueReceiveEndpointsConfiguration = queueReceiveEndpointsConfiguration;
|
||||
_workItemExecutionService = workItemExecutionService;
|
||||
_sendEndpoint = busControl.GetSendEndpoint(_queueReceiveEndpointsConfiguration.ExecutionReceiveEndpoint.ReceiveEndpointUri).Result;
|
||||
}
|
||||
|
||||
public async Task QueueExecution(WorkItemExecution request)
|
||||
{
|
||||
try
|
||||
{
|
||||
var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
|
||||
await _sendEndpoint.Send(request, cancellationTokenSource.Token);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.ForContext(Enum.GetName(typeof(PropertyEnum), PropertyEnum.Group),
|
||||
Enum.GetName(typeof(GroupEnum), GroupEnum.ITDevelopment))
|
||||
.Fatal(exception, $"Failed to queue workitemexecution request. WorkItemExecutionId: {request?.WorkItemExecutionId}");
|
||||
throw new ServiceException($"Failed to queue workitemexecution request. WorkItemExecutionId: {request?.WorkItemExecutionId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task QueueRetryExecution(WorkItemExecution request)
|
||||
{
|
||||
try
|
||||
{
|
||||
var updatedWorkItemExection = _workItemExecutionService.UpdateProcessingWorkItemExecution(request);
|
||||
|
||||
var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
|
||||
await _sendEndpoint.Send(request, cancellationTokenSource.Token);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.ForContext(Enum.GetName(typeof(PropertyEnum), PropertyEnum.Group),
|
||||
Enum.GetName(typeof(GroupEnum), GroupEnum.ITDevelopment))
|
||||
.Fatal(exception, $"Failed to queue workitemexecution request. WorkItemExecutionId: {request?.WorkItemExecutionId}");
|
||||
throw new ServiceException($"Failed to queue workitemexecution request. WorkItemExecutionId: {request?.WorkItemExecutionId}");
|
||||
}
|
||||
}
|
||||
}
|
||||
+49
@@ -0,0 +1,49 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using LCU.Api.Internal.Dispatcher.Configurations;
|
||||
using LCU.Api.Internal.MaidBotLib.Models;
|
||||
using LCU.Library.Internal.ApiResponse.Models;
|
||||
using LCU.Library.Internal.DispatcherModels.Entities;
|
||||
using LCU.Library.Internal.DispatcherModels.Models;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
using MassTransit;
|
||||
using Newtonsoft.Json;
|
||||
using ApiResponse = LCU.Library.Internal.ApiResponse.Models;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.QueueServices
|
||||
{
|
||||
public class QueueJobFailureService : IQueueJobFailureService
|
||||
{
|
||||
private readonly ISendEndpoint _sendEndpoint;
|
||||
private readonly ReceiveEndpointConfigurations _queueReceiveEndpointsConfiguration;
|
||||
|
||||
public QueueJobFailureService(IBusControl busControl,
|
||||
ReceiveEndpointConfigurations queueReceiveEndpointsConfiguration)
|
||||
{
|
||||
_queueReceiveEndpointsConfiguration = queueReceiveEndpointsConfiguration;
|
||||
_sendEndpoint = busControl.GetSendEndpoint(_queueReceiveEndpointsConfiguration.JobFailureReceiveEndpoint.ReceiveEndpointUri).Result;
|
||||
}
|
||||
public async Task QueueJobFailure(int jobId, int workItemId, List<string> messages)
|
||||
{
|
||||
try
|
||||
{
|
||||
var request = new JobFailureRequest()
|
||||
{
|
||||
JobId = jobId,
|
||||
WorkItemId = workItemId,
|
||||
ExecutionMessages = messages
|
||||
};
|
||||
|
||||
var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
|
||||
await _sendEndpoint.Send(request, cancellationTokenSource.Token);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.ForContext(Enum.GetName(typeof(PropertyEnum), PropertyEnum.Group),
|
||||
Enum.GetName(typeof(GroupEnum), GroupEnum.ITDevelopment))
|
||||
.Fatal(exception, $"Failed to queue job failure request. JobId: {jobId}, WorkItemId: {workItemId}");
|
||||
throw new ServiceException($"Failed to queue job failure request. JobId: {jobId}, WorkItemId: {workItemId}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+63
@@ -0,0 +1,63 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using LCU.Api.Internal.Dispatcher.Configurations;
|
||||
using LCU.Api.Internal.MaidBotLib.Models;
|
||||
using LCU.Library.Internal.DispatcherModels.Models.Requests;
|
||||
using MassTransit;
|
||||
using Newtonsoft.Json;
|
||||
using ApiResponse = LCU.Library.Internal.ApiResponse.Models;
|
||||
|
||||
namespace LCU.Api.Internal.Dispatcher.Services.QueueServices
|
||||
{
|
||||
public class QueueSendErroredJobAlertService : IQueueSendErroredJobAlertService
|
||||
{
|
||||
private readonly ISendEndpoint _sendEndpoint;
|
||||
private readonly ReceiveEndpointConfigurations _queueReceiveEndpointsConfiguration;
|
||||
|
||||
public QueueSendErroredJobAlertService(IBusControl busControl,
|
||||
ReceiveEndpointConfigurations queueReceiveEndpointConfigurations)
|
||||
{
|
||||
_queueReceiveEndpointsConfiguration = queueReceiveEndpointConfigurations;
|
||||
_sendEndpoint = busControl.GetSendEndpoint(_queueReceiveEndpointsConfiguration.SendErroredJobAlertReceiveEndpoint.ReceiveEndpointUri).Result;
|
||||
}
|
||||
public async Task QueueSendErroredJobAlert(BatchedErroredJobAlertRequest request)
|
||||
{
|
||||
try
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(request);
|
||||
var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(30));
|
||||
await _sendEndpoint.Send(request, cancellationTokenSource.Token);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
var workItemIds = string.Join(",",
|
||||
request.ErroredJobAlertRequests
|
||||
.Where(x => x.CurrentWorkItem != null)
|
||||
.Select(x => x.CurrentWorkItem.WorkItemId));
|
||||
|
||||
Log.ForContext(Enum.GetName(typeof(PropertyEnum), PropertyEnum.Group),
|
||||
Enum.GetName(typeof(GroupEnum), GroupEnum.ITDevelopment))
|
||||
.Fatal(exception, $"Failed to queue send errored job alert workitem requests. WorkItemIds: {workItemIds}");
|
||||
throw new ServiceException($"Failed to queue send errored job alert workitem requests. WorkItemIds: {workItemIds}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task QueueSendErroredJobAlert(Job job, WorkItem workItem)
|
||||
{
|
||||
var request = new ErroredJobAlertRequest()
|
||||
{
|
||||
CurrentJob = job,
|
||||
CurrentWorkItem = workItem
|
||||
};
|
||||
var requestList = new List<ErroredJobAlertRequest>()
|
||||
{
|
||||
request
|
||||
};
|
||||
var batchedRequest = new BatchedErroredJobAlertRequest()
|
||||
{
|
||||
ErroredJobAlertRequests = requestList
|
||||
};
|
||||
await QueueSendErroredJobAlert(batchedRequest);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
using System.Collections.Generic;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using LCU.Api.Internal.DispatcherLib.Repositories;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public class TagService : ITagService
|
||||
{
|
||||
private readonly ITagRepository _tagRepository;
|
||||
private readonly IUnitOfWork _unitOfWork;
|
||||
private readonly IJobTagService _jobTagService;
|
||||
|
||||
public TagService(IUnitOfWork unitOfWork,
|
||||
ITagRepository tagRepository, IJobTagService jobTagService)
|
||||
{
|
||||
_unitOfWork = unitOfWork;
|
||||
_tagRepository = tagRepository;
|
||||
_jobTagService = jobTagService;
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<Tag>> GetAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
return await _tagRepository.ListAsync();
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure retrieving dispatcher Tags.");
|
||||
throw new ServiceException("Failure retrieving dispatcher Tags.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Tag> GetByIdAsync(int id)
|
||||
{
|
||||
try
|
||||
{
|
||||
var tag = await _tagRepository.FindByIdAsync(id);
|
||||
ArgumentNullException.ThrowIfNull(tag);
|
||||
return tag;
|
||||
}
|
||||
catch (ArgumentNullException nullEx)
|
||||
{
|
||||
Log.Error(nullEx, $"Failure retrieving dispatcher Tags. TagId: {id}");
|
||||
throw new ArgumentNullException($"Failure retrieving dispatcher Tags. TagId: {id}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher Tags. TagId: {id}");
|
||||
throw new ServiceException($"Failure retrieving dispatcher Tags. TagId: {id}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Tag> SaveAsync(Tag tag)
|
||||
{
|
||||
try
|
||||
{
|
||||
var result = await _tagRepository.AddAsync(tag);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
ArgumentNullException.ThrowIfNull(result);
|
||||
return result;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure saving dispatcher Tag. TagId: {tag.TagId}");
|
||||
throw new ServiceException($"Failure saving dispatcher Tag. TagId: {tag.TagId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<Tag> UpdateAsync(Tag tag)
|
||||
{
|
||||
try
|
||||
{
|
||||
_tagRepository.Update(tag);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
|
||||
return tag;
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, $"SQL Failure updating dispatcher Tag. TagId: {tag.TagId}");
|
||||
throw new ServiceException($"SQL Failure updating dispatcher Tag. TagId: {tag.TagId}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"SQL Failure updating dispatcher Tag. TagId: {tag.TagId}");
|
||||
throw new ServiceException($"SQL Failure updating dispatcher Tag. TagId: {tag.TagId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task DeleteAsync(int tagId)
|
||||
{
|
||||
try
|
||||
{
|
||||
var jobTags = await _jobTagService.GetByTagIdAsync(tagId);
|
||||
if (jobTags == null || !jobTags.Any())
|
||||
{
|
||||
await RemoveJobTags(jobTags.ToList());
|
||||
}
|
||||
|
||||
var tag = await _tagRepository.FindByIdAsync(tagId);
|
||||
ArgumentNullException.ThrowIfNull(tag);
|
||||
_tagRepository.Remove(tag);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, $"SQL Failure removing dispatcher Tag. TagId: {tagId}");
|
||||
throw new ServiceException($"SQL Failure removing dispatcher Tag. TagId: {tagId}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure removing dispatcher Tag. TagId: {tagId}");
|
||||
throw new ServiceException($"Failure removing dispatcher Tag. TagId: {tagId}");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<bool> RemoveJobTags(List<JobTag> jobTags)
|
||||
{
|
||||
bool isSuccessful = false;
|
||||
try
|
||||
{
|
||||
foreach (var jobTag in jobTags)
|
||||
{
|
||||
await _jobTagService.Delete(jobTag.JobTagId);
|
||||
isSuccessful = true;
|
||||
}
|
||||
return isSuccessful;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure to remove job tags.");
|
||||
throw new ServiceException("Failure to remove job tags.");
|
||||
}
|
||||
}
|
||||
}
|
||||
+150
@@ -0,0 +1,150 @@
|
||||
using System.Collections.Generic;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using LCU.Library.Internal.DispatcherModels.Enums;
|
||||
using LCU.Api.Internal.DispatcherLib.Repositories;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public class WorkItemExecutionService : IWorkItemExecutionService
|
||||
{
|
||||
private readonly IUnitOfWork _unitOfWork;
|
||||
private readonly IWorkItemExecutionRepository _workItemExecutionRepository;
|
||||
|
||||
public WorkItemExecutionService(IUnitOfWork unitOfWork,
|
||||
IWorkItemExecutionRepository workItemExecutionRepository)
|
||||
{
|
||||
_unitOfWork = unitOfWork;
|
||||
_workItemExecutionRepository = workItemExecutionRepository;
|
||||
}
|
||||
|
||||
public async Task<WorkItemExecution> GetByIdAsync(int id)
|
||||
{
|
||||
WorkItemExecution workItemExecution = null;
|
||||
try
|
||||
{
|
||||
workItemExecution = await _workItemExecutionRepository.FindByIdAsync(id);
|
||||
ArgumentNullException.ThrowIfNull(workItemExecution);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher WorkItemExecution with id {id}.");
|
||||
throw new ServiceException($"Failure retrieving dispatcher WorkItemExecution with id {id}.");
|
||||
}
|
||||
|
||||
return workItemExecution;
|
||||
}
|
||||
|
||||
public async Task<WorkItemExecution> UpdateCompleteWorkItemExecution(WorkItemExecution workItemExecution, StatusType finalStatus)
|
||||
{
|
||||
workItemExecution = WorkItemExecution.SetAsFinal(workItemExecution, finalStatus);
|
||||
var updatedFailedWorkItem = await UpdateAsync(workItemExecution);
|
||||
return updatedFailedWorkItem;
|
||||
}
|
||||
|
||||
public async Task<WorkItemExecution> UpdateCompleteWorkItemExecutionWithoutSave(WorkItemExecution workItemExecution, StatusType finalStatus)
|
||||
{
|
||||
workItemExecution = WorkItemExecution.SetAsFinal(workItemExecution, finalStatus);
|
||||
var updatedFailedWorkItem = await UpdateAsyncWithoutSave(workItemExecution);
|
||||
return updatedFailedWorkItem;
|
||||
}
|
||||
|
||||
public async Task<WorkItemExecution> UpdateProcessingWorkItemExecution(WorkItemExecution workItemExecution)
|
||||
{
|
||||
workItemExecution.Status = StatusType.Processing;
|
||||
return await UpdateAsync(workItemExecution);
|
||||
}
|
||||
public async Task<WorkItemExecution> UpdateFailedWorkItemExecution(WorkItemExecution workItemExecution)
|
||||
{
|
||||
workItemExecution.Status = StatusType.Failed;
|
||||
return await UpdateAsync(workItemExecution);
|
||||
}
|
||||
|
||||
public async Task<WorkItemExecution> UpdateAsync(WorkItemExecution workItemExecution)
|
||||
{
|
||||
try
|
||||
{
|
||||
var updatedWorkItemExecution = await UpdateAsyncWithoutSave(workItemExecution);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
|
||||
return updatedWorkItemExecution;
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, $"SQL Failure updating dispatcher WorkItemExecution with id {workItemExecution.WorkItemId}");
|
||||
throw new ServiceException($"SQL Failure updating dispatcher WorkItemExecution with id {workItemExecution.WorkItemId}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure updating dispatcher WorkItemExecution with id {workItemExecution.WorkItemId}");
|
||||
throw new ServiceException($"Failure updating dispatcher WorkItemExecution with id {workItemExecution.WorkItemId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<WorkItemExecution> UpdateAsyncWithoutSave(WorkItemExecution workItemExecution)
|
||||
{
|
||||
try
|
||||
{
|
||||
_workItemExecutionRepository.Update(workItemExecution);
|
||||
|
||||
return workItemExecution;
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, $"SQL Failure updating dispatcher WorkItemExecution with id {workItemExecution.WorkItemId}");
|
||||
throw new ServiceException($"SQL Failure updating dispatcher WorkItemExecution with id {workItemExecution.WorkItemId}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure updating dispatcher WorkItemExecution with id {workItemExecution.WorkItemId}");
|
||||
throw new ServiceException($"Failure updating dispatcher WorkItemExecution with id {workItemExecution.WorkItemId}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<WorkItemExecution> CreateWorkItemExecution(int workItemId)
|
||||
{
|
||||
var newWorkItemExecution = new WorkItemExecution(workItemId);
|
||||
return await SaveAsync(newWorkItemExecution);
|
||||
}
|
||||
|
||||
public async Task<WorkItemExecution> SaveAsync(WorkItemExecution workItemExecution)
|
||||
{
|
||||
try
|
||||
{
|
||||
var result = await _workItemExecutionRepository.AddAsync(workItemExecution);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
return result;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, "Failure saving dispatcher WorkItemExecution", workItemExecution);
|
||||
throw new ServiceException("Failure saving dispatcher WorkItemExecution");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<WorkItemExecution>> GetAllUnFinishedWorkItemExecutions(int workItemId)
|
||||
{
|
||||
try
|
||||
{
|
||||
var result = await _workItemExecutionRepository.FindUnFinishedWorkItemExecutions(workItemId);
|
||||
return result;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher WorkItemExecutions with id {workItemId}.");
|
||||
throw new ServiceException($"Failure retrieving dispatcher WorkItemExecutions with id {workItemId}.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<WorkItemExecution>> GetAllByWorkItemId(int workItemId)
|
||||
{
|
||||
try
|
||||
{
|
||||
var result = await _workItemExecutionRepository.FindAllByWorkItemId(workItemId);
|
||||
return result;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher WorkItemExecutions with WorkItemId {workItemId}.");
|
||||
throw new ServiceException($"Failure retrieving dispatcher WorkItemExecutions with WorkItemId {workItemId}.");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,258 @@
|
||||
using LCU.Api.Internal.Dispatcher.Configurations;
|
||||
using LCU.Api.Internal.DispatcherLib.Repositories;
|
||||
using LCU.Api.Internal.DispatcherSharedLib.BuilderServices;
|
||||
using LCU.Library.Internal.ApiResponse.Models;
|
||||
using LCU.Library.Internal.LcuMassTransitModels;
|
||||
using Microsoft.Data.SqlClient;
|
||||
using Newtonsoft.Json;
|
||||
using System.Collections.Generic;
|
||||
namespace LCU.Api.Internal.Dispatcher.Services;
|
||||
|
||||
public class WorkItemService : IWorkItemService
|
||||
{
|
||||
private readonly IUnitOfWork _unitOfWork;
|
||||
private readonly IWorkItemRepository _workItemRepository;
|
||||
private readonly IWorkItemExecutionService _workItemExecutionService;
|
||||
private readonly IActionBuilderService _actionBuilderService;
|
||||
private readonly ConnectionConfiguration _connectionConfiguration;
|
||||
private readonly ReceiveEndpointConfiguration _completionReceiveEndpoint;
|
||||
private readonly ApiConfiguration _apiConfiguration;
|
||||
|
||||
public WorkItemService(IUnitOfWork unitOfWork,
|
||||
IWorkItemRepository workItemRepository,
|
||||
IWorkItemExecutionService workItemExecutionService,
|
||||
IActionBuilderService actionBuilderService,
|
||||
ConnectionConfiguration connectionConfiguration,
|
||||
ReceiveEndpointConfigurations queueReceiveEndpointsConfigurations,
|
||||
ApiConfiguration apiConfiguration)
|
||||
{
|
||||
_unitOfWork = unitOfWork;
|
||||
_workItemRepository = workItemRepository;
|
||||
_workItemExecutionService = workItemExecutionService;
|
||||
_actionBuilderService = actionBuilderService;
|
||||
_connectionConfiguration = connectionConfiguration;
|
||||
_completionReceiveEndpoint = queueReceiveEndpointsConfigurations.CompletionReceiveEndpoint;
|
||||
_apiConfiguration = apiConfiguration;
|
||||
}
|
||||
|
||||
public async Task<WorkItem> GetByIdAsync(int id)
|
||||
{
|
||||
WorkItem workItem = null;
|
||||
try
|
||||
{
|
||||
workItem = await _workItemRepository.FindByIdAsync(id);
|
||||
ArgumentNullException.ThrowIfNull(workItem);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher workitem with id {id}.");
|
||||
throw new ServiceException($"Failure retrieving dispatcher workitem with id {id}.");
|
||||
}
|
||||
|
||||
return workItem;
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<WorkItem>> GetByJobIdAsync(int jobId)
|
||||
{
|
||||
IEnumerable<WorkItem> workItems = null;
|
||||
try
|
||||
{
|
||||
workItems = await _workItemRepository.FindByJobIdAsync(jobId);
|
||||
ArgumentNullException.ThrowIfNull(workItems);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher workitems with JobId {jobId}.");
|
||||
throw new ServiceException($"Failure retrieving dispatcher workitems with JobId {jobId}.");
|
||||
}
|
||||
|
||||
return workItems;
|
||||
}
|
||||
|
||||
|
||||
public async Task<IEnumerable<WorkItem>> GetByJobIdWithExecutionsAsync(int jobId)
|
||||
{
|
||||
IEnumerable<WorkItem> workItems = null;
|
||||
try
|
||||
{
|
||||
workItems = await _workItemRepository.FindByJobIdWithExecutionsAsync(jobId,
|
||||
_apiConfiguration.NumberOfWorkItemExecutions,
|
||||
_apiConfiguration.NumberOfExecutionMessages);
|
||||
ArgumentNullException.ThrowIfNull(workItems);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving dispatcher workitems and executions with JobId {jobId}.");
|
||||
throw new ServiceException($"Failure retrieving dispatcher workitems and executions with JobId {jobId}.");
|
||||
}
|
||||
|
||||
return workItems;
|
||||
}
|
||||
|
||||
public async Task<WorkItem> GetFirstOrDefaultWorkItemByJobId(int jobId)
|
||||
{
|
||||
WorkItem workItem = null;
|
||||
try
|
||||
{
|
||||
workItem = await _workItemRepository.GetFirstOrDefaultWorkItemByJobId(jobId);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving first or default workItem with JobId: {jobId}.");
|
||||
throw new ServiceException($"Failure retrieving first or default workItem with JobId: {jobId}.");
|
||||
}
|
||||
|
||||
return workItem;
|
||||
}
|
||||
|
||||
public async Task<WorkItem> GetNextWorkItemByJobId(int jobId, int currentWorkItemOrder)
|
||||
{
|
||||
WorkItem workItem = null;
|
||||
try
|
||||
{
|
||||
workItem = await _workItemRepository.GetNextWorkItemByJobId(jobId, currentWorkItemOrder);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure retrieving next workItem with JobId: {jobId}.");
|
||||
throw new ServiceException($"Failure retrieving next workItem with JobId: {jobId}.");
|
||||
}
|
||||
|
||||
return workItem;
|
||||
}
|
||||
|
||||
public async Task<ICollection<WorkItem>> PopulateActionFieldsAndUpdateList(ICollection<WorkItem> workItems)
|
||||
{
|
||||
try
|
||||
{
|
||||
var updatedWorkItems = new List<WorkItem>();
|
||||
foreach (var workItem in workItems)
|
||||
{
|
||||
var newWorkItem = workItem;
|
||||
newWorkItem = PopulateActionFields(newWorkItem);
|
||||
newWorkItem = await UpdateAsync(newWorkItem);
|
||||
updatedWorkItems.Add(newWorkItem);
|
||||
}
|
||||
|
||||
return updatedWorkItems;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure populating action fields and updating dispatcher workitems.");
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
public async Task<WorkItem> ResetCanceledJobCounter(WorkItem workItem)
|
||||
{
|
||||
var trackedWorkItem = await _workItemRepository.GetTrackedWorkItem(workItem.WorkItemId);
|
||||
trackedWorkItem.CanceledJobCounter = 0;
|
||||
await _unitOfWork.CompleteAsync();
|
||||
return await _workItemRepository.FindByIdAsync(trackedWorkItem.WorkItemId);
|
||||
}
|
||||
|
||||
public async Task<WorkItem> IncreaseCanceledJobCounter(WorkItem workItem)
|
||||
{
|
||||
var trackedWorkItem = await _workItemRepository.GetTrackedWorkItem(workItem.WorkItemId);
|
||||
trackedWorkItem.CanceledJobCounter++;
|
||||
await _unitOfWork.CompleteAsync();
|
||||
return await _workItemRepository.FindByIdAsync(trackedWorkItem.WorkItemId);
|
||||
}
|
||||
|
||||
public async Task<WorkItem> UpdateAsync(WorkItem workItem)
|
||||
{
|
||||
try
|
||||
{
|
||||
_workItemRepository.Update(workItem);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
|
||||
return workItem;
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, $"SQL Failure updating dispatcher workitem with id {workItem.WorkItemId}");
|
||||
throw new ServiceException($"SQL Failure updating dispatcher workitem with id {workItem.WorkItemId}");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure updating dispatcher workitem with id {workItem.WorkItemId}");
|
||||
throw new ServiceException($"Failure updating dispatcher workitem with id {workItem.WorkItemId}");
|
||||
}
|
||||
}
|
||||
|
||||
public void SaveList(ICollection<WorkItem> workItems)
|
||||
{
|
||||
try
|
||||
{
|
||||
_workItemRepository.AddList(workItems);
|
||||
_unitOfWork.Complete();
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure saving list of workitems.");
|
||||
throw new ServiceException($"Failure saving list of workitems.");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task DeleteList(ICollection<WorkItem> workItems)
|
||||
{
|
||||
try
|
||||
{
|
||||
foreach (var workItem in workItems)
|
||||
{
|
||||
var workItemExecutions = await _workItemExecutionService.GetAllByWorkItemId(workItem.WorkItemId);
|
||||
workItem.WorkItemExecutions = workItemExecutions.ToList();
|
||||
}
|
||||
|
||||
_workItemRepository.RemoveList(workItems);
|
||||
await _unitOfWork.CompleteAsync();
|
||||
}
|
||||
catch (SqlException sqlException)
|
||||
{
|
||||
Log.Error(sqlException, $"SQL Failure removing dispatcher workitems");
|
||||
throw new ServiceException($"SQL Failure removing dispatcher workitems");
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
Log.Error(exception, $"Failure removing dispatcher workitems");
|
||||
throw new ServiceException($"Failure removing dispatcher workitems");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public WorkItem PopulateActionFields(WorkItem workItem)
|
||||
{
|
||||
var actionBuilder = _actionBuilderService.BuildAction(workItem.ActionType);
|
||||
var responseItem = CreateResponseItem(workItem.WorkItemId);
|
||||
workItem = actionBuilder.AddUrlAndHttpMethod(workItem);
|
||||
workItem = actionBuilder.AddWorkItemReturnResponse(workItem, responseItem);
|
||||
return workItem;
|
||||
}
|
||||
|
||||
private ResponseItem CreateResponseItem(int workItemId)
|
||||
{
|
||||
var dispatcherReturnResponse = new DispatcherReturnResponse
|
||||
{
|
||||
WorkItemId = workItemId
|
||||
};
|
||||
var serializeDispatcherReturnResponse = JsonConvert.SerializeObject(dispatcherReturnResponse);
|
||||
|
||||
var newReturnItem = new ResponseItem
|
||||
{
|
||||
ResponsePayload = serializeDispatcherReturnResponse,
|
||||
ReturnQueueConnectionConfig = new QueueConnectionConfig()
|
||||
{
|
||||
Url = _connectionConfiguration.HostUrl,
|
||||
Username = _connectionConfiguration.UserName,
|
||||
Password = _connectionConfiguration.Password
|
||||
},
|
||||
ReturnQueueConsumerConfig = new QueueConsumerConfig()
|
||||
{
|
||||
ExchangeName = _completionReceiveEndpoint.ReceiveEndpoint
|
||||
}
|
||||
};
|
||||
|
||||
return newReturnItem;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user