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