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.
This commit is contained in:
2026-11-02 10:10:51 +01:00
parent 76fc005b02
commit be881ea7c9
5 changed files with 123 additions and 0 deletions

View File

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

View File

@@ -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" />

View File

@@ -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."));
}
}

View File

@@ -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."));
}
}

View File

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