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> 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> 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> 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> 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> 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 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 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 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 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 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> 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> 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 GetJobsInStatusLongerThanTimeLimit(List matchingStatusJobs, int processingLimitInMinutes) { var stuckInStatusJobs = new List(); 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 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 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 UpdateExcludingWorkItemsAsyncWithoutSave(Job job) { return await _jobRepository.Update(job); } public async Task 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 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 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 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 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 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 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}"); } } }