diff --git a/ECMJobRunner.WebCron/ProfileWorker/ProfileWork.cs b/ECMJobRunner.WebCron/ProfileWorker/ProfileWork.cs new file mode 100644 index 0000000..40a220b --- /dev/null +++ b/ECMJobRunner.WebCron/ProfileWorker/ProfileWork.cs @@ -0,0 +1,60 @@ +using ECMJobRunner.Application.DEXJob.Queries; +using ECMJobRunner.WebCron.Extensions; +using Hangfire; +using MediatR; + +namespace ECMJobRunner.WebCron.ProfileWorker; + + +public class ProfileWork(ILogger Logger, IRecurringJobManager JobManager, IMediator Mediator, ProfileCache ProfileCache) +{ + public async Task ExecuteAsync(CancellationToken stoppingToken) + { + // Fetch active profiles from database + var profiles = await Mediator.Send(new GetProfileQuery() + { + Active = true, + IncludeSqlJobs = true + }, stoppingToken); + + // Create HashSet for O(1) lookup performance + var profileJobIds = profiles.Select(p => p.JobId()).ToHashSet(); + + // Remove jobs from Hangfire that no longer exist in the database + foreach (var cachedJobId in ProfileCache.GetAllJobIds()) + { + if (!profileJobIds.Contains(cachedJobId)) + { + // Remove job from Hangfire if it no longer exists in the database + JobManager.RemoveIfExists(cachedJobId); + ProfileCache.TryRemove(cachedJobId, out _); + Logger.LogInformation("Job {JobId} removed", cachedJobId); + } + } + + // Add or update jobs in Hangfire based on the database profiles + foreach (var profile in profiles) + { + if (ProfileCache.TryGetValue(profile.JobId(), out var currentProfile) + && currentProfile!.Schedule == profile.Schedule) + continue; + + // Add or update recurring job using MediatR command + JobManager.AddOrUpdate( + profile.JobId(), + mediator => mediator.Send(profile.ToJob(), stoppingToken), + profile.Schedule, + new RecurringJobOptions + { + TimeZone = TimeZoneInfo.Local + } + ); + + // Store/update in local cache + ProfileCache.AddOrUpdate(profile.JobId(), profile); + + Logger.LogInformation("Job {JobId} registered with schedule: {Schedule}", + profile.JobId(), profile.Schedule); + } + } +}