125 lines
5.0 KiB
C#
125 lines
5.0 KiB
C#
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
|
|
}.ApplySecurity(_kafkaOptions);
|
|
|
|
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");
|
|
|
|
var guard = new PoisonPillGuard("sofa-scoring", _kafkaOptions.MaxPoisonRetries, _logger);
|
|
|
|
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,
|
|
evt.RecordedAt == default ? DateTimeOffset.UtcNow : evt.RecordedAt,
|
|
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);
|
|
guard.OnSuccess();
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
if (result is not null && guard.ShouldSkip(result, ex))
|
|
{
|
|
consumer.Commit(result);
|
|
continue;
|
|
}
|
|
|
|
_logger.LogError(ex,
|
|
"SofaScoringService failed on topic={Topic} offset={Offset} — will retry",
|
|
result?.Topic, result?.Offset.Value);
|
|
await Task.Delay(2000, stoppingToken);
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
consumer.Close();
|
|
}
|
|
}
|
|
}
|
|
|
|
public record GcsScoredEvent(Guid EncounterId, Guid PatientId, int TotalScore); |