Added Mediator and Quartz components

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;

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<long> RequestCounter = Meter.CreateCounter<long>("mediator_requests_total");
+    public static readonly Histogram<double> RequestDuration = Meter.CreateHistogram<double>("mediator_request_duration_ms");
+}

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<QuartzOptions>(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<RetryJobListener>();
+        services.AddTransient<IJobOrchestrator, JobOrchestrator>();
+        
+        return services;
+    }
+}

PostFundManagement.Domain/Mediator/LoggingPipelineBehavior.cs

@@ -0,0 +1,29 @@
+namespace PostFundManagement.Domain.Mediator;
+
+public sealed class LoggingPipelineBehavior<TRequest, TResponse>(ILogger<LoggingPipelineBehavior<TRequest, TResponse>> logger) :
+    IPipelineBehavior<TRequest, TResponse?>
+    where TRequest : IRequest<TResponse>
+    where TResponse : ResultBase, new()
+{
+    public async ValueTask<TResponse?> Handle(TRequest message, MessageHandlerDelegate<TRequest, TResponse?> 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;
+    }
+}

PostFundManagement.Domain/Mediator/TelemetryPipelineBehavior.cs

@@ -0,0 +1,69 @@
+using System.Diagnostics;
+using PostFundManagement.Domain.Extensions;
+
+namespace PostFundManagement.Domain.Mediator;
+
+public sealed class TelemetryPipelineBehavior<TRequest, TResponse> :
+    IPipelineBehavior<TRequest, TResponse?>
+    where TRequest : IRequest<TResponse>
+    where TResponse : ResultBase, new()
+{
+    public async ValueTask<TResponse?> Handle(TRequest message, MessageHandlerDelegate<TRequest, TResponse?> 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);
+        }
+    }
+}

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>(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<MediatorJob<TNotification>>()
+            .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<ITrigger> { trigger }.AsReadOnly(), replace: true, cancellationToken);
+    }
+
+    public async ValueTask ScheduleAsync<TNotification>(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<MediatorJob<TNotification>>()
+            .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<ITrigger> { trigger }.AsReadOnly(), replace: true, cancellationToken);
+    }
+
+    public async ValueTask<bool> 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);
+    }
+}

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<TNotification>(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<TNotification>(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;
+        }
+    }
+}

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

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;
 

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;
 

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