Files
2026-03-24 15:26:03 -05:00

583 lines
22 KiB
C#

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}");
}
}
}