From f4c680d19d5fcaf333964d26473731c4b74a3247 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Thu, 20 Aug 2026 16:32:13 +0200 Subject: [PATCH] Added Mediator and Quartz components --- .../Extensions/Constants.cs | 2 + .../Extensions/MediatorTelemetry.cs | 15 +++ .../Extensions/Quartz.cs | 104 ++++++++++++++++++ .../Mediator/LoggingPipelineBehavior.cs | 29 +++++ .../Mediator/TelemetryPipelineBehavior.cs | 69 ++++++++++++ .../Quartz/JobOrchestrator.cs | 87 +++++++++++++++ .../Quartz/MediatorJob.cs | 52 +++++++++ .../Quartz/RetryJobListener.cs | 21 ++++ .../Factories/ApplicationDbContextFactory.cs | 2 +- .../DataProtectionDbContextFactory.cs | 2 +- .../Extensions/Postgres.cs | 3 +- 11 files changed, 382 insertions(+), 4 deletions(-) create mode 100644 PostFundManagement.Domain/Extensions/MediatorTelemetry.cs create mode 100644 PostFundManagement.Domain/Extensions/Quartz.cs create mode 100644 PostFundManagement.Domain/Mediator/LoggingPipelineBehavior.cs create mode 100644 PostFundManagement.Domain/Mediator/TelemetryPipelineBehavior.cs create mode 100644 PostFundManagement.Domain/Quartz/JobOrchestrator.cs create mode 100644 PostFundManagement.Domain/Quartz/MediatorJob.cs create mode 100644 PostFundManagement.Domain/Quartz/RetryJobListener.cs diff --git a/PostFundManagement.Domain/Extensions/Constants.cs b/PostFundManagement.Domain/Extensions/Constants.cs index 6d7291a..392a3fd 100644 --- a/PostFundManagement.Domain/Extensions/Constants.cs +++ b/PostFundManagement.Domain/Extensions/Constants.cs @@ -2,6 +2,8 @@ namespace PostFundManagement.Domain.Extensions; public static class Constants { + public const string DatabaseConfigName = "PfmDatabase"; + public const int LabelLength = 256; public const int ShortLabelLength = 100; diff --git a/PostFundManagement.Domain/Extensions/MediatorTelemetry.cs b/PostFundManagement.Domain/Extensions/MediatorTelemetry.cs new file mode 100644 index 0000000..812ec2b --- /dev/null +++ b/PostFundManagement.Domain/Extensions/MediatorTelemetry.cs @@ -0,0 +1,15 @@ +using System.Diagnostics; +using System.Diagnostics.Metrics; + +namespace PostFundManagement.Domain.Extensions; + +public static class MediatorTelemetry +{ + public const string ServiceName = "LiteCharms.Mediator"; + + public static readonly ActivitySource Source = new(ServiceName); + public static readonly Meter Meter = new(ServiceName); + + public static readonly Counter RequestCounter = Meter.CreateCounter("mediator_requests_total"); + public static readonly Histogram RequestDuration = Meter.CreateHistogram("mediator_request_duration_ms"); +} \ No newline at end of file diff --git a/PostFundManagement.Domain/Extensions/Quartz.cs b/PostFundManagement.Domain/Extensions/Quartz.cs new file mode 100644 index 0000000..f8aeaec --- /dev/null +++ b/PostFundManagement.Domain/Extensions/Quartz.cs @@ -0,0 +1,104 @@ +using PostFundManagement.Domain.Abstractions; +using PostFundManagement.Domain.Quartz; +using static PostFundManagement.Domain.Extensions.Constants; + +namespace PostFundManagement.Domain.Extensions; + +public static class Quartz +{ + public const string DefaultSchedulerName = "tech-shop"; + + public static IServiceCollection AddQuartzSchedulerClient(this IServiceCollection services, string schedulerName, IConfiguration configuration) + { + var connectionString = configuration.GetConnectionString(DatabaseConfigName); + + services.ConfigureCommon(); + + services.AddQuartz(config => + { + config.SchedulerName = schedulerName; + config.SchedulerId = "AUTO"; + + config.UseSimpleTypeLoader(); + config.UseDefaultThreadPool(options => options.MaxConcurrency = 0); + config.UseTimeZoneConverter(); + + config.UsePersistentStore(storage => + { + storage.PerformSchemaValidation = false; + + storage.UseSystemTextJsonSerializer(); + storage.SetProperty("quartz.jobStore.clustered", "true"); + storage.SetProperty("quartz.jobStore.tablePrefix", "qrtz_"); + + storage.UsePostgres(connectionString!); + storage.UseClustering(cluster => + { + cluster.CheckinInterval = TimeSpan.FromSeconds(30); + cluster.CheckinMisfireThreshold = TimeSpan.FromSeconds(90); + }); + }); + }); + + return services; + } + + public static IServiceCollection AddQuartzScheduler(this IServiceCollection services, string schedulerName, IConfiguration configuration) + { + var connectionString = configuration.GetConnectionString(DatabaseConfigName); + + services.ConfigureCommon(); + + services.AddQuartzHostedService(options => options.WaitForJobsToComplete = true); + + services.AddQuartz(config => + { + config.SchedulerName = schedulerName; + config.SchedulerId = "AUTO"; + config.InterruptJobsOnShutdown = true; + config.InterruptJobsOnShutdownWithWait = true; + config.MaxBatchSize = 5; + + config.UseSimpleTypeLoader(); + config.UseDefaultThreadPool(options => options.MaxConcurrency = 1); + config.UseTimeZoneConverter(); + + config.SetProperty("quartz.jobStore.misfireThreshold", TimeSpan.FromMinutes(2).TotalMilliseconds.ToString(CultureInfo.InvariantCulture)); + + config.UsePersistentStore(storage => + { + storage.PerformSchemaValidation = false; + + storage.UseSystemTextJsonSerializer(); + storage.SetProperty("quartz.jobStore.clustered", "true"); + storage.SetProperty("quartz.jobStore.tablePrefix", "qrtz_"); + + storage.UsePostgres(connectionString!); + storage.UseClustering(cluster => + { + cluster.CheckinInterval = TimeSpan.FromSeconds(30); + cluster.CheckinMisfireThreshold = TimeSpan.FromSeconds(90); + }); + }); + }); + + return services; + } + + private static IServiceCollection ConfigureCommon(this IServiceCollection services) + { + services.Configure(options => + { + options.Scheduling.IgnoreDuplicates = true; + options.Scheduling.OverWriteExistingData = true; + + options["quartz.plugin.jobHistory.type"] = "Quartz.Plugin.History.LoggingJobHistoryPlugin, Quartz.Plugins"; + options["quartz.plugin.triggerHistory.type"] = "Quartz.Plugin.History.LoggingTriggerHistoryPlugin, Quartz.Plugins"; + }); + + services.AddTransient(); + services.AddTransient(); + + return services; + } +} \ No newline at end of file diff --git a/PostFundManagement.Domain/Mediator/LoggingPipelineBehavior.cs b/PostFundManagement.Domain/Mediator/LoggingPipelineBehavior.cs new file mode 100644 index 0000000..750e6a7 --- /dev/null +++ b/PostFundManagement.Domain/Mediator/LoggingPipelineBehavior.cs @@ -0,0 +1,29 @@ +namespace PostFundManagement.Domain.Mediator; + +public sealed class LoggingPipelineBehavior(ILogger> logger) : + IPipelineBehavior + where TRequest : IRequest + where TResponse : ResultBase, new() +{ + public async ValueTask Handle(TRequest message, MessageHandlerDelegate next, CancellationToken cancellationToken) + { + TResponse? response = await next(message, cancellationToken); + + if (response is null) + logger.LogCritical("{Request} {TypeName} was returned as null", typeof(TRequest).Name, typeof(TRequest).Name); + + if(response?.IsFailed == true || response?.Errors?.Any() == true) + { + foreach (var error in response.Errors) + { + if (!string.IsNullOrWhiteSpace(error.Message)) + logger.LogWarning("{Request} {Error}", typeof(TRequest).Name, error.Message); + + if (error?.Reasons?.Count > 0) + error.Reasons.ForEach(r => logger.LogError("{Request} {Reason}", typeof(TRequest).Name, r.ToString())); + } + } + + return response; + } +} \ No newline at end of file diff --git a/PostFundManagement.Domain/Mediator/TelemetryPipelineBehavior.cs b/PostFundManagement.Domain/Mediator/TelemetryPipelineBehavior.cs new file mode 100644 index 0000000..07774d6 --- /dev/null +++ b/PostFundManagement.Domain/Mediator/TelemetryPipelineBehavior.cs @@ -0,0 +1,69 @@ +using System.Diagnostics; +using PostFundManagement.Domain.Extensions; + +namespace PostFundManagement.Domain.Mediator; + +public sealed class TelemetryPipelineBehavior : + IPipelineBehavior + where TRequest : IRequest + where TResponse : ResultBase, new() +{ + public async ValueTask Handle(TRequest message, MessageHandlerDelegate next, CancellationToken cancellationToken) + { + var requestName = typeof(TRequest).Name; + + using var activity = MediatorTelemetry.Source.StartActivity(requestName); + + activity?.SetTag("mediator.request_type", typeof(TRequest).FullName); + + var stopWatch = Stopwatch.StartNew(); + var status = "Success"; + + try + { + TResponse? response = await next(message, cancellationToken); + + if (response is null) + { + status = "NullResponse"; + activity?.SetStatus(ActivityStatusCode.Error, "Response was null"); + + return response; + } + + if (response.IsFailed) + { + status = "Failed"; + activity?.SetStatus(ActivityStatusCode.Error, "Request failed"); + + var firstError = response.Errors.FirstOrDefault()?.Message ?? "Unknown Error"; + activity?.SetTag("error.message", firstError); + + foreach (var error in response.Errors) + activity?.AddEvent(new ActivityEvent("Result Error", tags: new() { { "message", error.Message } })); + } + else + activity?.SetStatus(ActivityStatusCode.Ok); + + return response; + } + catch (Exception ex) + { + status = "Exception"; + activity?.SetStatus(ActivityStatusCode.Error, ex.Message); + + activity?.AddException(ex); + + throw; + } + finally + { + stopWatch.Stop(); + + var tags = new TagList { { "request", requestName }, { "status", status } }; + + MediatorTelemetry.RequestCounter.Add(1, tags); + MediatorTelemetry.RequestDuration.Record(stopWatch.Elapsed.TotalMilliseconds, tags); + } + } +} \ No newline at end of file diff --git a/PostFundManagement.Domain/Quartz/JobOrchestrator.cs b/PostFundManagement.Domain/Quartz/JobOrchestrator.cs new file mode 100644 index 0000000..48cb645 --- /dev/null +++ b/PostFundManagement.Domain/Quartz/JobOrchestrator.cs @@ -0,0 +1,87 @@ +using PostFundManagement.Domain.Abstractions; + +namespace PostFundManagement.Domain.Quartz; + +public sealed class JobOrchestrator(ISchedulerFactory schedulerFactory) : IJobOrchestrator +{ + public async ValueTask SendAsync(TNotification notification, CancellationToken cancellationToken = default) + where TNotification : IEvent + { + var chainedJobGroup = "onetime-jobs"; + + var scheduler = await schedulerFactory.GetScheduler(cancellationToken); + var jobKey = new JobKey($"{notification.Name.ToLower(CultureInfo.InvariantCulture)}-{notification.CorrelationId.ToLower(CultureInfo.InvariantCulture)}", chainedJobGroup); + var triggerKey = new TriggerKey($"{jobKey.Name}-trigger", chainedJobGroup); + + var job = JobBuilder.Create>() + .WithIdentity(jobKey) + .WithDescription($"Correlation ID: {notification.CorrelationId}") + .UsingJobData(new JobDataMap { ["Payload"] = JsonSerializer.Serialize(notification) }) + .DisallowConcurrentExecution() + .RequestRecovery() + .Build(); + + var trigger = global::Quartz.TriggerBuilder.Create() + .WithIdentity(triggerKey) + .StartNow() + .Build(); + + await scheduler.ScheduleJob(job, new List { trigger }.AsReadOnly(), replace: true, cancellationToken); + } + + public async ValueTask ScheduleAsync(TNotification notification, string cronExpression, CancellationToken cancellationToken = default) + where TNotification : IEvent + { + var chainedJobGroup = "scheduled-jobs"; + + var scheduler = await schedulerFactory.GetScheduler(cancellationToken); + var jobKey = new JobKey($"{notification.Name.ToLower(CultureInfo.InvariantCulture)}", chainedJobGroup); + var triggerKey = new TriggerKey($"{jobKey.Name}-trigger", chainedJobGroup); + + var job = JobBuilder.Create>() + .WithIdentity(jobKey) + .WithDescription($"Correlation ID: {notification.CorrelationId}") + .UsingJobData(new JobDataMap { ["Payload"] = JsonSerializer.Serialize(notification) }) + .DisallowConcurrentExecution() + .StoreDurably() + .Build(); + + var now = DateTime.UtcNow; + + var trigger = global::Quartz.TriggerBuilder.Create() + .WithIdentity(triggerKey) + .WithDescription($"Scheduled via Main Job at {now:g}") + .WithCronSchedule(cronExpression, cron => cron + .WithMisfireHandlingInstructionIgnoreMisfires()) + .StartAt((DateTimeOffset)now) + .Build(); + + await scheduler.AddJob(job, replace: true, cancellationToken); + + if (await scheduler.CheckExists(triggerKey, cancellationToken)) + await scheduler.RescheduleJob(triggerKey, trigger, cancellationToken); + else + await scheduler.ScheduleJob(job, new List { trigger }.AsReadOnly(), replace: true, cancellationToken); + } + + public async ValueTask InterruptAsync(string eventName, string? correlationId = null, CancellationToken cancellationToken = default) + { + var scheduler = await schedulerFactory.GetScheduler(cancellationToken); + + var jobKeyName = string.Empty; + var jobGroup = string.Empty; + + if (!string.IsNullOrWhiteSpace(correlationId)) + { + jobKeyName = $"{eventName.ToLower(CultureInfo.InvariantCulture)}-{correlationId.ToLower(CultureInfo.InvariantCulture)}"; + jobGroup = "onetime-jobs"; + } + else + { + jobKeyName = eventName.ToLower(CultureInfo.InvariantCulture); + jobGroup = "scheduled-jobs"; + } + + return await scheduler.Interrupt(JobKey.Create(jobKeyName, jobGroup), cancellationToken); + } +} \ No newline at end of file diff --git a/PostFundManagement.Domain/Quartz/MediatorJob.cs b/PostFundManagement.Domain/Quartz/MediatorJob.cs new file mode 100644 index 0000000..f42c2d5 --- /dev/null +++ b/PostFundManagement.Domain/Quartz/MediatorJob.cs @@ -0,0 +1,52 @@ +using System.Diagnostics; +using PostFundManagement.Domain.Abstractions; +using PostFundManagement.Domain.Extensions; + +namespace PostFundManagement.Domain.Quartz; + +[DisallowConcurrentExecution] +public sealed class MediatorJob(IMediator mediator) : IJob where TNotification : IEvent +{ + public async Task Execute(IJobExecutionContext context) + { + if (context.Recovering) + Trace.WriteLine($"CRITICAL RECOVERY: Resurrecting job '{typeof(TNotification).Name}' after a previous cluster node crashed mid-execution."); + + var data = context.MergedJobDataMap["Payload"] as string; + + if (string.IsNullOrWhiteSpace(data)) + { + Trace.WriteLine("Job Payload missing, job ended"); + + return; + } + + var notification = JsonSerializer.Deserialize(data); + + if (notification is null) + { + Trace.WriteLine("Notification could not be Json converted from data string, job ended"); + + return; + } + + using var activity = MediatorTelemetry.Source.StartActivity(typeof(TNotification).Name); + + activity?.SetTag("event.correlation_id", notification.CorrelationId); + + try + { + await mediator.Publish(notification, context.CancellationToken); + + Trace.WriteLine("Job published successfully"); + } + catch (OperationCanceledException) when (context.CancellationToken.IsCancellationRequested) + { + Trace.WriteLine($"Job '{typeof(TNotification).Name}' was gracefully interrupted by the cluster control plane."); + + activity?.SetStatus(ActivityStatusCode.Ok); + + return; + } + } +} \ No newline at end of file diff --git a/PostFundManagement.Domain/Quartz/RetryJobListener.cs b/PostFundManagement.Domain/Quartz/RetryJobListener.cs new file mode 100644 index 0000000..9565773 --- /dev/null +++ b/PostFundManagement.Domain/Quartz/RetryJobListener.cs @@ -0,0 +1,21 @@ +namespace PostFundManagement.Domain.Quartz; + +public sealed class RetryJobListener : IJobListener +{ + public string Name => "RetryJobListener"; + + public int RetryCount { get; set; } = 3; + + public Task JobExecutionVetoed(IJobExecutionContext context, CancellationToken cancellationToken = default) => Task.CompletedTask; + + public Task JobToBeExecuted(IJobExecutionContext context, CancellationToken cancellationToken = default) => Task.CompletedTask; + + public async Task JobWasExecuted(IJobExecutionContext context, JobExecutionException? jobException, CancellationToken cancellationToken = default) + { + if (context.CancellationToken.IsCancellationRequested) + return; + + if (jobException is not null && context.RefireCount < RetryCount) + jobException.RefireImmediately = true; + } +} \ No newline at end of file diff --git a/PostFundManagement.Infrastructure/Database/Factories/ApplicationDbContextFactory.cs b/PostFundManagement.Infrastructure/Database/Factories/ApplicationDbContextFactory.cs index a0f4aea..7d06651 100644 --- a/PostFundManagement.Infrastructure/Database/Factories/ApplicationDbContextFactory.cs +++ b/PostFundManagement.Infrastructure/Database/Factories/ApplicationDbContextFactory.cs @@ -1,4 +1,4 @@ -using static PostFundManagement.Infrastructure.Extensions.Postgres; +using static PostFundManagement.Domain.Extensions.Constants; namespace PostFundManagement.Infrastructure.Database.Factories; diff --git a/PostFundManagement.Infrastructure/Database/Factories/DataProtectionDbContextFactory.cs b/PostFundManagement.Infrastructure/Database/Factories/DataProtectionDbContextFactory.cs index d71cad2..b54ff53 100644 --- a/PostFundManagement.Infrastructure/Database/Factories/DataProtectionDbContextFactory.cs +++ b/PostFundManagement.Infrastructure/Database/Factories/DataProtectionDbContextFactory.cs @@ -1,4 +1,4 @@ -using static PostFundManagement.Infrastructure.Extensions.Postgres; +using static PostFundManagement.Domain.Extensions.Constants; namespace PostFundManagement.Infrastructure.Database.Factories; diff --git a/PostFundManagement.Infrastructure/Extensions/Postgres.cs b/PostFundManagement.Infrastructure/Extensions/Postgres.cs index 00dd5cf..c12d4b5 100644 --- a/PostFundManagement.Infrastructure/Extensions/Postgres.cs +++ b/PostFundManagement.Infrastructure/Extensions/Postgres.cs @@ -1,11 +1,10 @@ using PostFundManagement.Infrastructure.Database; +using static PostFundManagement.Domain.Extensions.Constants; namespace PostFundManagement.Infrastructure.Extensions; public static class Postgres { - public const string DatabaseConfigName = "PfmDatabase"; - public static IServiceCollection AddDataProtectionDatabase(this IServiceCollection services, IConfiguration configuration) { var connectionString = configuration.GetConnectionString(DatabaseConfigName);