feat: Add ProfileManager for automatic Hangfire job scheduling
- Create ProfileManager background service that polls database every 1 second - Query active profiles with cron schedules via MediatR GetProfileQuery - Auto-create/update Hangfire recurring jobs using IRecurringJobManager - Jobs execute TriggeringDEXJobBatchCommand via IMediator dependency injection - Add profile caching with schedule change detection to avoid redundant updates - Create DtoExtensions with JobId() and ToJob() helper methods for profile-to-command conversion
This commit is contained in:
21
ECMJobRunner.WebCron/Extensions/DtoExtensions.cs
Normal file
21
ECMJobRunner.WebCron/Extensions/DtoExtensions.cs
Normal file
@@ -0,0 +1,21 @@
|
||||
using ECMJobRunner.Application.Common.Dtos;
|
||||
using ECMJobRunner.Application.DEXJob.Commands;
|
||||
using MediatR;
|
||||
|
||||
namespace ECMJobRunner.WebCron.Extensions;
|
||||
|
||||
public static class DtoExtensions
|
||||
{
|
||||
public static string JobId(this CfgProfileDto profile)
|
||||
{
|
||||
return $"profile-{profile.Id}-{profile.ProfileName.Replace(' ', '_').ToLowerInvariant()}";
|
||||
}
|
||||
|
||||
public static TriggeringDEXJobBatchCommand ToJob(this CfgProfileDto profile)
|
||||
{
|
||||
return new TriggeringDEXJobBatchCommand
|
||||
{
|
||||
ProfileId = profile.Id,
|
||||
};
|
||||
}
|
||||
}
|
||||
59
ECMJobRunner.WebCron/ProfileManager.cs
Normal file
59
ECMJobRunner.WebCron/ProfileManager.cs
Normal file
@@ -0,0 +1,59 @@
|
||||
using ECMJobRunner.Application.Common.Dtos;
|
||||
using ECMJobRunner.Application.DEXJob.Commands;
|
||||
using ECMJobRunner.Application.DEXJob.Queries;
|
||||
using ECMJobRunner.WebCron.Extensions;
|
||||
using Hangfire;
|
||||
using MediatR;
|
||||
using System.Collections.Concurrent;
|
||||
|
||||
namespace ECMJobRunner.WebCron
|
||||
{
|
||||
public class ProfileManager(ILogger<ProfileManager> Logger, IMediator Mediator, IRecurringJobManager JobManager) : BackgroundService
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, CfgProfileDto> Profiles = new();
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
if (Logger.IsEnabled(LogLevel.Information))
|
||||
{
|
||||
Logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
|
||||
}
|
||||
|
||||
var profiles = await Mediator.Send(new GetProfileQuery()
|
||||
{
|
||||
Active = true,
|
||||
IncludeSqlJobs = true
|
||||
}, stoppingToken);
|
||||
|
||||
foreach (var profile in profiles)
|
||||
{
|
||||
if (Profiles.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(), CancellationToken.None),
|
||||
profile.Schedule,
|
||||
new RecurringJobOptions
|
||||
{
|
||||
TimeZone = TimeZoneInfo.Local
|
||||
}
|
||||
);
|
||||
|
||||
// Store/update in local cache
|
||||
Profiles[profile.JobId()] = profile;
|
||||
|
||||
Logger.LogInformation("Job {JobId} registered with schedule: {Schedule}",
|
||||
profile.JobId(), profile.Schedule);
|
||||
}
|
||||
|
||||
await Task.Delay(1000, stoppingToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user