feat: Add ProfileWork with Hangfire job synchronization
- ProfileWork.cs: Business logic for DB-to-Hangfire sync - Fetches active profiles from database via MediatR - Registers/updates Hangfire recurring jobs with local timezone - Removes stale jobs (deleted from DB) - O(n) performance optimization using HashSet for lookups - Uses ProfileWorker's stopping token for graceful shutdown Job lifecycle: - ProfileWorker stops → All running jobs cancelled via token closure - Jobs use RecurringJobOptions with TimeZoneInfo.Local - Automatic retry support via Hangfire [AutomaticRetry] attribute
This commit is contained in:
60
ECMJobRunner.WebCron/ProfileWorker/ProfileWork.cs
Normal file
60
ECMJobRunner.WebCron/ProfileWorker/ProfileWork.cs
Normal file
@@ -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<ProfileWork> 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<IMediator>(
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user