vigilcare-clinical/docs/guides/12-event-driven-background-services.md

Guide 12: Event-Driven Background Services in .NET

What is a Background Service?

In a web API, most code runs in response to HTTP requests — a request arrives, your code processes it, a response goes back. But many tasks need to run continuously without a request triggering them:

• Polling the outbox table every second to relay events to Kafka
• Consuming Kafka messages to compute clinical scores
• Checking every 30 seconds whether any critical alerts have gone unacknowledged
• Scanning every 5 minutes for overdue sepsis bundles

In .NET, these long-running tasks are called background services. They start when the application starts, run continuously in the background, and stop when the application shuts down.

.NET provides two base classes for this:

• IHostedService: Has StartAsync (called once when the app starts) and StopAsync (called once when the app stops). Good for one-time initialization tasks.
• BackgroundService: Extends IHostedService with an ExecuteAsync method that runs for the lifetime of the application. Good for continuous processing loops.

---

Why Background Services in This Project?

VigilCareClinical has 25+ background services running inside the API process. They handle everything from event relay to clinical scoring to metrics collection. Without them, the API would only be able to do work when an HTTP request arrives — and most of the important work (computing NEWS2 scores, detecting sepsis, escalating alerts) happens asynchronously.

---

Registration in Program.cs

Every background service is registered with dependency injection in Program.cs:

// One-time initialization services
builder.Services.AddHostedService<ThresholdCacheLoader>();
builder.Services.AddHostedService<KafkaTopicProvisioner>();
builder.Services.AddHostedService<ElasticIndexProvisioner>();
builder.Services.AddHostedService<PatientPhiMigrationService>();

// Event relay
builder.Services.AddHostedService<OutboxRelayService>();

// Kafka consumers (clinical scoring engines)
builder.Services.AddHostedService<SepsisEngineService>();
builder.Services.AddHostedService<News2ScoringService>();
builder.Services.AddHostedService<GcsScoringService>();
builder.Services.AddHostedService<SofaScoringService>();
builder.Services.AddHostedService<TrendAnalyzerService>();
builder.Services.AddHostedService<WarningAlertService>();

// Kafka consumers (infrastructure)
builder.Services.AddHostedService<EsIndexerService>();
builder.Services.AddHostedService<DataLakeWriterService>();
builder.Services.AddHostedService<NotificationPublisherService>();

// RabbitMQ consumers
builder.Services.AddHostedService<PagingWorkerService>();
builder.Services.AddHostedService<EscalationWorkerService>();
builder.Services.AddHostedService<DischargeSummaryWorkerService>();

// Periodic scanners
builder.Services.AddHostedService<ReconciliationScheduler>();
builder.Services.AddHostedService<SepsisBundleMonitorService>();
builder.Services.AddHostedService<GatewayStaleDetectorService>();
builder.Services.AddHostedService<AlertQualityAggregatorService>();

// Metrics collectors
builder.Services.AddHostedService<AlertsUnacknowledgedCollector>();
builder.Services.AddHostedService<OutboxPendingCollector>();
builder.Services.AddHostedService<KafkaConsumerLagCollector>();
builder.Services.AddHostedService<WardGatewayMetricsCollector>();

AddHostedService<T>() tells .NET: "create an instance of this class and call its StartAsync/ExecuteAsync when the application starts." All hosted services run concurrently within the same process.

---

The Four Patterns

Every background service in this project follows one of four patterns. Understanding these patterns makes it easy to read any service's code.

Pattern 1: One-Time Initializer (IHostedService)

These services run once at startup and then stop. They prepare the system for operation.

public class ThresholdCacheLoader : IHostedService
{
    public async Task StartAsync(CancellationToken cancellationToken)
    {
        // Load all alert thresholds from PostgreSQL into Redis
        var thresholds = await db.AlertThresholds.ToListAsync(cancellationToken);

        for (var attempt = 0; attempt < 3; attempt++)
        {
            try
            {
                var batch = cache.CreateBatch();
                foreach (var t in thresholds)
                    _ = batch.StringSetAsync($"threshold:{t.ObservationCode}", json);
                batch.Execute();
                return;  // success — done
            }
            catch (RedisException ex)
            {
                _logger.LogWarning(ex, "Redis unavailable — attempt {Attempt}/3",
                    attempt + 1);
                await Task.Delay(backoffMs[attempt], cancellationToken);
            }
        }

        _logger.LogError("Failed to load thresholds — app will start without cache");
    }

    public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
}

Other initializers:

• KafkaTopicProvisioner — creates Kafka topics with 6 partitions
• ElasticIndexProvisioner — creates Elasticsearch indexes with mappings
• RabbitMqTopologyProvisioner — declares exchanges, queues, and bindings
• PatientPhiMigrationService — one-time migration to add search tokens to existing patients

Key characteristic: StartAsync runs to completion, then the service does nothing until shutdown. The work is done during startup, before the application starts accepting HTTP requests.

Pattern 2: Kafka Consumer (BackgroundService)

These services consume messages from Kafka topics and process them. They're the core of the event-driven architecture.

public class SepsisEngineService : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // 1. Configure the consumer
        var config = new ConsumerConfig
        {
            BootstrapServers = _kafkaOptions.BootstrapServers,
            GroupId = "sepsis-engine",
            AutoOffsetReset = AutoOffsetReset.Earliest,
            EnableAutoCommit = false
        };

        using var consumer = new ConsumerBuilder<string, string>(config).Build();
        consumer.Subscribe(_kafkaOptions.Topics.ObservationRecorded);

        // 2. Create a poison pill guard
        var guard = new PoisonPillGuard("sepsis-engine",
            _kafkaOptions.MaxPoisonRetries, _logger);

        try
        {
            // 3. Consume loop — runs for the lifetime of the application
            while (!stoppingToken.IsCancellationRequested)
            {
                ConsumeResult<string, string>? result = null;
                try
                {
                    result = consumer.Consume(stoppingToken);

                    // 4. Process the message
                    var evt = JsonSerializer.Deserialize<SepsisObservationEvent>(
                        result.Message.Value)!;
                    using var scope = _services.CreateScope();
                    var detector = scope.ServiceProvider
                        .GetRequiredService<QsofaDetector>();
                    await detector.ProcessObservationAsync(
                        evt.EncounterId, evt.PatientId,
                        evt.ObservationCode, evt.Value, stoppingToken);

                    // 5. Commit offset (tell Kafka "I'm done with this message")
                    consumer.Commit(result);
                    guard.OnSuccess();
                }
                catch (OperationCanceledException) { break; }
                catch (Exception ex)
                {
                    // 6. Poison pill handling
                    if (result is not null && guard.ShouldSkip(result, ex))
                    {
                        consumer.Commit(result);
                        continue;
                    }
                    _logger.LogError(ex, "SepsisEngine failed — will retry");
                    await Task.Delay(2000, stoppingToken);
                }
            }
        }
        finally
        {
            consumer.Close();  // 7. Clean group leave on shutdown
        }
    }
}

Other Kafka consumers follow the same structure: News2ScoringService, GcsScoringService, SofaScoringService, TrendAnalyzerService, WarningAlertService, EsIndexerService, DataLakeWriterService, NotificationPublisherService.

Key characteristics:

• Infinite while loop broken only by CancellationToken
• Manual Kafka offset commit after successful processing
• PoisonPillGuard prevents stuck consumers
• DI scope created per message (database contexts are scoped, not singleton)
• consumer.Close() in finally for clean group leave

Pattern 3: Periodic Timer (BackgroundService)

These services wake up on a fixed schedule, do some work, then go back to sleep. They monitor system state and collect metrics.

public sealed class GatewayStaleDetectorService : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        // PeriodicTimer fires at a fixed interval
        using var timer = new PeriodicTimer(
            TimeSpan.FromMinutes(_opts.PollIntervalMinutes));  // 5 minutes

        while (await timer.WaitForNextTickAsync(ct))
            await DetectStaleAsync(ct);
    }

    private async Task DetectStaleAsync(CancellationToken ct)
    {
        // Create a DI scope for each tick
        await using var scope = _scopes.CreateAsyncScope();
        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();

        var cutoff = DateTimeOffset.UtcNow.AddMinutes(-_opts.StaleThresholdMinutes);

        var stale = await db.WardGateways
            .Where(g => g.Status != GatewayStatus.Offline
                     && g.LastHeartbeatAt < cutoff)
            .ToListAsync(ct);

        foreach (var gateway in stale)
        {
            gateway.MarkOffline();
            _logger.LogWarning("Gateway {Code} marked OFFLINE", gateway.GatewayCode);
        }

        if (stale.Count > 0)
            await db.SaveChangesAsync(ct);
    }
}

Other periodic services:

• AlertsUnacknowledgedCollector — every 30 seconds, counts open critical alerts
• OutboxPendingCollector — every 30 seconds, counts pending outbox events
• KafkaConsumerLagCollector — every 30 seconds, checks Kafka consumer lag
• WardGatewayMetricsCollector — every 60 seconds, reports gateway status
• ReconciliationScheduler — every 30 minutes, runs safety checks
• SepsisBundleMonitorService — every 5 minutes, marks overdue bundles
• AlertQualityAggregatorService — every 60 minutes, computes alert quality metrics

Key characteristics:

• PeriodicTimer or Task.Delay for the interval
• DI scope created per tick (fresh database context each time)
• Errors are caught and logged, not propagated (the service continues to the next tick)
• No message consumption — these services query the database directly

PeriodicTimer vs Task.Delay: Both work for periodic execution. PeriodicTimer (introduced in .NET 6) is slightly more precise because it accounts for the time spent doing work — if your work takes 2 seconds and the interval is 30 seconds, the next tick fires 28 seconds after the work finishes, maintaining a true 30-second cadence. Task.Delay would wait 30 seconds after the work finishes, making the actual interval 32 seconds.

Pattern 4: RabbitMQ Consumer (BackgroundService)

These services consume messages from RabbitMQ queues using the event-driven consumer model (push-based, not pull-based like Kafka):

public sealed class PagingWorkerService : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // 1. Wait for topology to be ready
        await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);

        // 2. Create connection and channel
        var factory = RabbitMqConnectionFactory.Create(o, dispatchConsumersAsync: true);
        using var connection = factory.CreateConnection("paging-worker");
        using var channel = connection.CreateModel();

        // 3. Set prefetch (how many unacked messages at once)
        channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

        // 4. Register an async event handler
        var consumer = new AsyncEventingBasicConsumer(channel);
        consumer.Received += async (sender, ea) =>
        {
            await HandlePageAsync(channel, ea, stoppingToken);
        };

        // 5. Start consuming
        channel.BasicConsume("alerts.paging.queue", autoAck: false, consumer);

        // 6. Block until shutdown
        await Task.Delay(Timeout.Infinite, stoppingToken);
    }
}

Why Task.Delay(Timeout.Infinite)? Unlike Kafka consumers (which pull messages in a loop), RabbitMQ consumers are push-based — RabbitMQ delivers messages to the Received event handler. The ExecuteAsync method just needs to stay alive (not return) so the connection and event handler remain active. Task.Delay(Timeout.Infinite, stoppingToken) blocks forever until the cancellation token is triggered by application shutdown.

Other RabbitMQ consumers: EscalationWorkerService, DischargeSummaryWorkerService, ClinicalSyncBatchConsumer.

---

The DI Scope Problem

Why do background services need to create scopes?

In .NET dependency injection, services have different lifetimes:

• Singleton: One instance for the entire application
• Scoped: One instance per "scope" (in a web app, one per HTTP request)
• Transient: A new instance every time it's requested

AppDbContext (the database context) is registered as scoped — each HTTP request gets its own instance to avoid thread-safety issues and stale data. But background services are singletons — they're created once and live forever.

If a background service tries to inject a scoped service directly, .NET throws an error. The solution: create a new scope for each unit of work:

// Wrong — would fail because BackgroundService is a singleton
public class MyService : BackgroundService
{
    private readonly AppDbContext _db;  // scoped — can't inject into singleton
}

// Correct — create a scope per tick/message
public class MyService : BackgroundService
{
    private readonly IServiceScopeFactory _scopes;

    private async Task DoWorkAsync(CancellationToken ct)
    {
        await using var scope = _scopes.CreateAsyncScope();
        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
        // use db within this scope...
    }  // scope is disposed, db is disposed
}

The scope acts like a mini HTTP request — it creates and disposes the database context cleanly.

---

Startup Ordering

Background services start in the order they're registered. Services that provision infrastructure run first:

// Phase 1: Infrastructure provisioning (these must complete first)
builder.Services.AddHostedService<ThresholdCacheLoader>();      // Redis cache
builder.Services.AddHostedService<KafkaTopicProvisioner>();     // Kafka topics
builder.Services.AddHostedService<ElasticIndexProvisioner>();   // ES indexes
// RabbitMqTopologyProvisioner runs as a singleton, registered separately

// Phase 2: Event processing (depends on Phase 1)
builder.Services.AddHostedService<OutboxRelayService>();        // needs Kafka topics
builder.Services.AddHostedService<EsIndexerService>();          // needs ES indexes
builder.Services.AddHostedService<SepsisEngineService>();       // needs Kafka topics

// Phase 3: Notification workers (depends on RabbitMQ topology)
builder.Services.AddHostedService<PagingWorkerService>();
builder.Services.AddHostedService<EscalationWorkerService>();

Some services add explicit delays to wait for infrastructure:

// PagingWorkerService — wait 5 seconds for RabbitMQ topology provisioning
await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);

// KafkaConsumerLagCollector — wait 20 seconds for Kafka to be reachable
await Task.Delay(TimeSpan.FromSeconds(20), ct);

This is a pragmatic approach for a single-process application. In a microservices architecture, you'd use health checks and readiness probes instead.

---

Graceful Shutdown

When the application shuts down (e.g., Ctrl+C or a deployment), .NET cancels the CancellationToken passed to each service. Well-behaved services respond to this:

Kafka consumers close cleanly:

finally
{
    consumer.Close();  // tells Kafka broker to rebalance partitions immediately
}

RabbitMQ consumers requeue in-flight messages:

catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
    channel.BasicNack(ea.DeliveryTag, multiple: false, requeue: true);
}

The data lake writer flushes buffered events:

finally
{
    if (_buffer.Values.Sum(v => v.Count) > 0)
        await FlushAsync(consumer, CancellationToken.None);
    consumer.Close();
}

---

Background Service Summary

Service	Pattern	Interval/Trigger	Purpose
ThresholdCacheLoader	Initializer	Once at startup	Pre-load Redis cache
KafkaTopicProvisioner	Initializer	Once at startup	Create Kafka topics
ElasticIndexProvisioner	Initializer	Once at startup	Create ES indexes
PatientPhiMigrationService	Initializer	Once at startup	Migrate PHI search tokens
OutboxRelayService	Periodic	Every 1 second	Outbox → Kafka relay
SepsisEngineService	Kafka consumer	Per message	qSOFA screening
News2ScoringService	Kafka consumer	Per message	NEWS2 score computation
GcsScoringService	Kafka consumer	Per message	GCS aggregation
SofaScoringService	Kafka consumer	Per message	SOFA organ scoring
TrendAnalyzerService	Kafka consumer	Per message	Rate-of-change detection
WarningAlertService	Kafka consumer	Per message	Warning threshold alerts
EsIndexerService	Kafka consumer	Per message	CQRS projection to ES
DataLakeWriterService	Kafka consumer	Buffered flush	Parquet files to MinIO
NotificationPublisherService	Kafka consumer	Per message	Kafka → RabbitMQ bridge
PagingWorkerService	RabbitMQ consumer	Per message	Page physician, wait for ACK
EscalationWorkerService	RabbitMQ consumer	Per message	Escalate unacked alerts
DischargeSummaryWorkerService	RabbitMQ consumer	Per message	Generate discharge PDFs
ClinicalSyncBatchConsumer	RabbitMQ consumer	Per message	Process gateway sync batches
ReconciliationScheduler	Periodic timer	Every 30 minutes	Safety checks
SepsisBundleMonitorService	Periodic timer	Every 5 minutes	Mark overdue bundles
GatewayStaleDetectorService	Periodic timer	Every 5 minutes	Detect offline gateways
AlertQualityAggregatorService	Periodic timer	Every 60 minutes	Compute quality metrics
AlertsUnacknowledgedCollector	Periodic timer	Every 30 seconds	Prometheus gauge
OutboxPendingCollector	Periodic timer	Every 30 seconds	Prometheus gauge
KafkaConsumerLagCollector	Periodic timer	Every 30 seconds	Prometheus gauge
WardGatewayMetricsCollector	Periodic timer	Every 60 seconds	Prometheus gauge

---

Key Takeaways

• Background services are how .NET applications do work outside HTTP requests — consuming messages, polling databases, collecting metrics.
• Four patterns cover all use cases: one-time initializers, Kafka consumers, periodic timers, and RabbitMQ consumers. Learn these four and you can read any service in the codebase.
• Always create a DI scope per unit of work — background services are singletons, but database contexts are scoped. The IServiceScopeFactory pattern bridges this gap.
• Handle errors gracefully — catch, log, and continue to the next cycle/message. A transient database timeout should not kill a background service permanently.
• Respond to cancellation — when the application shuts down, clean up resources: close Kafka consumers, requeue RabbitMQ messages, flush buffers.
• Startup ordering matters — provisioners (topics, indexes, topology) must run before consumers that depend on them. Use registration order and startup delays.