From be881ea7c9d92bd90056cde86d39c185c8f3b98a Mon Sep 17 00:00:00 2001
From: TekH <h.tek@digitaldata.works>
Date: Mon, 2 Nov 2026 10:10:51 +0100
Subject: [PATCH] feat(infrastructure): add health checks for RabbitMQ and
 email sync worker

Introduce singleton health checks that background workers update at runtime:

- ConsumerInitHealthCheck reports whether the RabbitMQ consumer pool started
- RabbitMqConnectionHealthCheck opens a channel to verify broker reachability
- EmailSyncWorkerHealthCheck detects stalled or erroring sync cycles

Register all three in the infrastructure DI container so the API layer can wire them into the health check pipeline.
---
 .../DependencyInjection.cs                    |  7 +++
 ...ata.MessagingService.Infrastructure.csproj |  1 +
 .../Background/ConsumerInitHealthCheck.cs     | 29 ++++++++++
 .../Background/EmailSyncWorkerHealthCheck.cs  | 53 +++++++++++++++++++
 .../RabbitMqConnectionHealthCheck.cs          | 33 ++++++++++++
 5 files changed, 123 insertions(+)
 create mode 100644 src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/ConsumerInitHealthCheck.cs
 create mode 100644 src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/EmailSyncWorkerHealthCheck.cs
 create mode 100644 src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/RabbitMqConnectionHealthCheck.cs

diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs b/src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs
index 9e1f166..f6b5789 100644
--- a/src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs
+++ b/src/infrastructure/DigitalData.MessagingService.Infrastructure/DependencyInjection.cs
@@ -58,6 +58,13 @@ public static class DependencyInjection
         // --- RabbitMQ Configuration ---
         services.AddRabbitMqConnectionFactory(configuration);
 
+        // ConsumerInitHealthCheck, RabbitMqConnectionHealthCheck and EmailSyncWorkerHealthCheck are
+        // registered as singletons here so background workers can inject and update them.
+        // Health check pipeline is wired up in the API layer via AddHealthChecks().
+        services.AddSingleton<ConsumerInitHealthCheck>();
+        services.AddSingleton<RabbitMqConnectionHealthCheck>();
+        services.AddSingleton<EmailSyncWorkerHealthCheck>();
+
         // Register Background Workers
         services.AddHostedService<ConsumerInitWorker>();
         services.AddHostedService<EmailSyncWorker>();
diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/DigitalData.MessagingService.Infrastructure.csproj b/src/infrastructure/DigitalData.MessagingService.Infrastructure/DigitalData.MessagingService.Infrastructure.csproj
index 91eaa7c..d7d6d9d 100644
--- a/src/infrastructure/DigitalData.MessagingService.Infrastructure/DigitalData.MessagingService.Infrastructure.csproj
+++ b/src/infrastructure/DigitalData.MessagingService.Infrastructure/DigitalData.MessagingService.Infrastructure.csproj
@@ -22,6 +22,7 @@
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
       <PrivateAssets>all</PrivateAssets>
     </PackageReference>
+    <PackageReference Include="Microsoft.Extensions.Diagnostics.HealthChecks" Version="8.0.11" />
     <PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.10" />
     <PackageReference Include="Microsoft.Extensions.Http" Version="8.0.1" />
     <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="10.0.10" />
diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/ConsumerInitHealthCheck.cs b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/ConsumerInitHealthCheck.cs
new file mode 100644
index 0000000..e37b88c
--- /dev/null
+++ b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/ConsumerInitHealthCheck.cs
@@ -0,0 +1,29 @@
+using Microsoft.Extensions.Diagnostics.HealthChecks;
+
+namespace DigitalData.MessagingService.Infrastructure.Services.Background;
+
+/// <summary>
+/// Reports the initialization status of <see cref="ConsumerInitWorker"/>.
+/// Marked as <see cref="HealthStatus.Unhealthy"/> if the RabbitMQ consumer pool
+/// failed to start, causing the health endpoint to return HTTP 503.
+/// </summary>
+public class ConsumerInitHealthCheck : IHealthCheck
+{
+    private volatile bool _isHealthy = true;
+    private string? _failureMessage;
+
+    internal void MarkUnhealthy(string message)
+    {
+        _isHealthy = false;
+        _failureMessage = message;
+    }
+
+    public Task<HealthCheckResult> CheckHealthAsync(
+        HealthCheckContext context,
+        CancellationToken cancellationToken = default)
+    {
+        return Task.FromResult(_isHealthy
+            ? HealthCheckResult.Healthy("RabbitMQ consumer pool is running.")
+            : HealthCheckResult.Unhealthy(_failureMessage ?? "RabbitMQ consumer pool failed to initialize."));
+    }
+}
diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/EmailSyncWorkerHealthCheck.cs b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/EmailSyncWorkerHealthCheck.cs
new file mode 100644
index 0000000..131caca
--- /dev/null
+++ b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/EmailSyncWorkerHealthCheck.cs
@@ -0,0 +1,53 @@
+using Microsoft.Extensions.Diagnostics.HealthChecks;
+
+namespace DigitalData.MessagingService.Infrastructure.Services.Background;
+
+/// <summary>
+/// Reports the operational status of <see cref="EmailSyncWorker"/>.
+/// Becomes <see cref="HealthStatus.Unhealthy"/> if the sync loop has not completed
+/// a cycle within twice the configured sync interval, indicating a stalled worker.
+/// Becomes <see cref="HealthStatus.Degraded"/> if the last sync cycle had errors.
+/// </summary>
+public class EmailSyncWorkerHealthCheck : IHealthCheck
+{
+    private DateTimeOffset? _lastCycleCompletedAt;
+    private int _consecutiveErrorCount;
+    private TimeSpan _syncInterval = TimeSpan.FromSeconds(300);
+    private bool _started;
+
+    internal void OnWorkerStarted(TimeSpan syncInterval)
+    {
+        _syncInterval = syncInterval;
+        _started = true;
+        _lastCycleCompletedAt = DateTimeOffset.UtcNow;
+    }
+
+    internal void OnCycleCompleted(int errorCount)
+    {
+        _lastCycleCompletedAt = DateTimeOffset.UtcNow;
+        _consecutiveErrorCount = errorCount;
+    }
+
+    public Task<HealthCheckResult> CheckHealthAsync(
+        HealthCheckContext context,
+        CancellationToken cancellationToken = default)
+    {
+        if (!_started)
+            return Task.FromResult(HealthCheckResult.Degraded("EmailSyncWorker has not started yet."));
+
+        var deadline = _lastCycleCompletedAt!.Value + (_syncInterval * 2);
+
+        if (DateTimeOffset.UtcNow > deadline)
+            return Task.FromResult(HealthCheckResult.Unhealthy(
+                $"EmailSyncWorker has not completed a cycle since {_lastCycleCompletedAt:HH:mm:ss} UTC. " +
+                $"Expected within {_syncInterval.TotalSeconds * 2}s."));
+
+        if (_consecutiveErrorCount > 0)
+            return Task.FromResult(HealthCheckResult.Degraded(
+                $"EmailSyncWorker last cycle had {_consecutiveErrorCount} account error(s). " +
+                $"Last cycle: {_lastCycleCompletedAt:HH:mm:ss} UTC."));
+
+        return Task.FromResult(HealthCheckResult.Healthy(
+            $"EmailSyncWorker is running. Last cycle: {_lastCycleCompletedAt:HH:mm:ss} UTC."));
+    }
+}
diff --git a/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/RabbitMqConnectionHealthCheck.cs b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/RabbitMqConnectionHealthCheck.cs
new file mode 100644
index 0000000..76b420b
--- /dev/null
+++ b/src/infrastructure/DigitalData.MessagingService.Infrastructure/Services/Background/RabbitMqConnectionHealthCheck.cs
@@ -0,0 +1,33 @@
+using DigitalData.MessagingService.RabbitMQ;
+using Microsoft.Extensions.Diagnostics.HealthChecks;
+
+namespace DigitalData.MessagingService.Infrastructure.Services.Background;
+
+/// <summary>
+/// Verifies the RabbitMQ connection is open and reachable by attempting to open a channel.
+/// Returns <see cref="HealthStatus.Unhealthy"/> if the broker is unreachable or the connection is closed.
+/// </summary>
+public class RabbitMqConnectionHealthCheck(RabbitMqConnectionFactory ConnectionFactory) : IHealthCheck
+{
+    public async Task<HealthCheckResult> CheckHealthAsync(
+        HealthCheckContext context,
+        CancellationToken cancellationToken = default)
+    {
+        try
+        {
+            var connection = await ConnectionFactory.GetDefaultConnectionAsync();
+
+            if (!connection.IsOpen)
+                return HealthCheckResult.Unhealthy("RabbitMQ connection is closed.");
+
+            // Open and immediately close a lightweight channel to confirm broker is responsive
+            await using var channel = await connection.CreateChannelAsync(cancellationToken: cancellationToken);
+
+            return HealthCheckResult.Healthy("RabbitMQ connection is open.");
+        }
+        catch (Exception ex)
+        {
+            return HealthCheckResult.Unhealthy("RabbitMQ connection check failed.", ex);
+        }
+    }
+}