vigilcare-clinical/VigilCareClinicalAPI/BackgroundServices/SofaScoringService.cs

using System.Text.Json;
using Confluent.Kafka;
using Microsoft.Extensions.Options;

public class SofaScoringService : BackgroundService
{
    private readonly IServiceProvider _services;
    private readonly KafkaOptions _kafkaOptions;
    private readonly ILogger<SofaScoringService> _logger;

    public SofaScoringService(
        IServiceProvider services,
        IOptions<KafkaOptions> kafkaOptions,
        ILogger<SofaScoringService> logger)
    {
        _services = services;
        _kafkaOptions = kafkaOptions.Value;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var config = new ConsumerConfig
        {
            BootstrapServers = _kafkaOptions.BootstrapServers,
            GroupId = "sofa-scoring",
            AutoOffsetReset = AutoOffsetReset.Earliest,
            EnableAutoCommit = false
        };

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

        _logger.LogInformation("SofaScoringService started — consumer group: sofa-scoring");

        try
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                ConsumeResult<string, string>? result = null;
                try
                {
                    result = consumer.Consume(stoppingToken);

                    using var scope = _services.CreateScope();
                    var detector = scope.ServiceProvider.GetRequiredService<SofaDetector>();

                    if (result.Topic == _kafkaOptions.Topics.GcsScored)
                    {
                        var gcsEvt = JsonSerializer.Deserialize<GcsScoredEvent>(
                            result.Message.Value,
                            new JsonSerializerOptions { PropertyNameCaseInsensitive = true })!;

                        var outcome = await detector.ProcessGcsScoredAsync(
                            gcsEvt.EncounterId, gcsEvt.PatientId, stoppingToken);

                        if (outcome.Outcome == SofaOutcome.EncounterNotFound)
                            _logger.LogWarning(
                                "Skipping stale gcs.scored event — encounter={Id} offset={Offset}",
                                gcsEvt.EncounterId, result.Offset.Value);
                        else if (outcome.Outcome == SofaOutcome.ScoreComputed)
                            _logger.LogInformation(
                                "SOFA re-scored via gcs.scored — encounter={Id} total={Total}",
                                gcsEvt.EncounterId, outcome.Score!.Total);
                    }
                    else
                    {
                        var evt = JsonSerializer.Deserialize<News2ObservationEvent>(
                            result.Message.Value,
                            new JsonSerializerOptions { PropertyNameCaseInsensitive = true })!;

                        var outcome = await detector.ProcessObservationAsync(
                            evt.EncounterId,
                            evt.PatientId,
                            evt.ObservationCode,
                            evt.Value,
                            DateTimeOffset.UtcNow,
                            stoppingToken);

                        if (outcome.Outcome == SofaOutcome.EncounterNotFound)
                            _logger.LogWarning(
                                "Skipping stale observation.recorded event — encounter={Id} code={Code} offset={Offset}",
                                evt.EncounterId, evt.ObservationCode, result.Offset.Value);
                        else if (outcome.Outcome == SofaOutcome.ScoreComputed)
                            _logger.LogInformation(
                                "SOFA scored via consumer — encounter={Id} total={Total}",
                                evt.EncounterId, outcome.Score!.Total);
                    }

                    consumer.Commit(result);
                }
                catch (OperationCanceledException)
                {
                    break;
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex,
                        "SofaScoringService failed on topic={Topic} offset={Offset} — not committing",
                        result?.Topic, result?.Offset.Value);
                    await Task.Delay(2000, stoppingToken);
                }
            }
        }
        finally
        {
            consumer.Close();
        }
    }
}

public record GcsScoredEvent(Guid EncounterId, Guid PatientId, int TotalScore);