263 lines
11 KiB
C#
263 lines
11 KiB
C#
using System.Text.Json;
|
|
using Confluent.Kafka;
|
|
using Elastic.Clients.Elasticsearch;
|
|
using Microsoft.Extensions.Options;
|
|
|
|
public class EsIndexerService : BackgroundService
|
|
{
|
|
private static readonly JsonSerializerOptions EventJsonOptions = new()
|
|
{
|
|
PropertyNameCaseInsensitive = true
|
|
};
|
|
|
|
private readonly ElasticsearchClient _elastic;
|
|
private readonly KafkaOptions _kafkaOptions;
|
|
private readonly ElasticsearchOptions _esOptions;
|
|
private readonly ILogger<EsIndexerService> _logger;
|
|
|
|
public EsIndexerService(
|
|
ElasticsearchClient elastic,
|
|
IOptions<KafkaOptions> kafkaOptions,
|
|
IOptions<ElasticsearchOptions> esOptions,
|
|
ILogger<EsIndexerService> logger)
|
|
{
|
|
_elastic = elastic;
|
|
_kafkaOptions = kafkaOptions.Value;
|
|
_esOptions = esOptions.Value;
|
|
_logger = logger;
|
|
}
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
var config = new ConsumerConfig
|
|
{
|
|
BootstrapServers = _kafkaOptions.BootstrapServers,
|
|
GroupId = "es-indexer",
|
|
AutoOffsetReset = AutoOffsetReset.Earliest,
|
|
// Manual commit: offset is only committed after a successful Elasticsearch write.
|
|
// If the process crashes between ES write and commit, the message is reprocessed.
|
|
// Consumers must be idempotent. See idempotency contract above.
|
|
EnableAutoCommit = false,
|
|
EnablePartitionEof = false
|
|
};
|
|
|
|
using var consumer = new ConsumerBuilder<string, string>(config).Build();
|
|
|
|
consumer.Subscribe(new[]
|
|
{
|
|
_kafkaOptions.Topics.ObservationRecorded,
|
|
_kafkaOptions.Topics.AlertGenerated,
|
|
_kafkaOptions.Topics.EncounterStatusChanged
|
|
});
|
|
|
|
_logger.LogInformation("EsIndexerService started. Subscribed to 3 topics.");
|
|
|
|
try
|
|
{
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
ConsumeResult<string, string>? result = null;
|
|
try
|
|
{
|
|
result = consumer.Consume(stoppingToken);
|
|
await DispatchAsync(result.Topic, result.Message.Value, stoppingToken);
|
|
consumer.Commit(result);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex,
|
|
"EsIndexer failed processing topic={Topic} offset={Offset} — not committing",
|
|
result?.Topic, result?.Offset.Value);
|
|
// Do not commit: message will be redelivered on restart
|
|
await Task.Delay(1000, stoppingToken);
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
consumer.Close();
|
|
}
|
|
}
|
|
|
|
private Task DispatchAsync(string topic, string payload, CancellationToken ct) => topic switch
|
|
{
|
|
var t when t == _kafkaOptions.Topics.EncounterStatusChanged =>
|
|
HandleEncounterStatusChangedAsync(payload, ct),
|
|
var t when t == _kafkaOptions.Topics.ObservationRecorded =>
|
|
HandleObservationRecordedAsync(payload, ct),
|
|
var t when t == _kafkaOptions.Topics.AlertGenerated =>
|
|
HandleAlertGeneratedAsync(payload, ct),
|
|
_ => Task.CompletedTask
|
|
};
|
|
|
|
// --- encounter.status.changed ---
|
|
// Upserts the patient_encounters document. DocAsUpsert=true means:
|
|
// if the document does not exist, it is created; if it exists, it is replaced.
|
|
// Idempotent: applying the same event twice produces the same document.
|
|
private async Task HandleEncounterStatusChangedAsync(string payload, CancellationToken ct)
|
|
{
|
|
var evt = JsonSerializer.Deserialize<EncounterStatusChangedEvent>(payload, EventJsonOptions)!;
|
|
|
|
var doc = new PatientEncounterDocument
|
|
{
|
|
EncounterId = evt.EncounterId.ToString(),
|
|
PatientId = evt.PatientId.ToString(),
|
|
Mrn = evt.Mrn,
|
|
PatientName = evt.PatientName,
|
|
Department = evt.Department,
|
|
Status = evt.NewStatus,
|
|
AttendingPhysician = evt.AttendingPhysician,
|
|
RoomBed = evt.RoomBed,
|
|
AdmissionReason = evt.AdmissionReason,
|
|
AdmittedAt = evt.AdmittedAt,
|
|
OpenAlertCount = 0,
|
|
LastObservationAt = null
|
|
};
|
|
|
|
var resp = await _elastic.UpdateAsync<PatientEncounterDocument, PatientEncounterDocument>(
|
|
_esOptions.Indices.PatientEncounters,
|
|
evt.EncounterId.ToString(),
|
|
u => u.Doc(doc).DocAsUpsert(true),
|
|
ct);
|
|
|
|
if (!resp.IsValidResponse)
|
|
throw new InvalidOperationException(
|
|
$"ES upsert failed for encounter {evt.EncounterId}: {resp.DebugInformation}");
|
|
|
|
_logger.LogDebug("Upserted patient_encounters for encounter {Id} → status={Status}",
|
|
evt.EncounterId, evt.NewStatus);
|
|
}
|
|
|
|
// --- observation.recorded ---
|
|
// Indexes the observation by observationId — idempotent PUT.
|
|
// Also updates lastObservationAt on the parent encounter document using a script.
|
|
private async Task HandleObservationRecordedAsync(string payload, CancellationToken ct)
|
|
{
|
|
var evt = JsonSerializer.Deserialize<ObservationRecordedEvent>(payload, EventJsonOptions)!;
|
|
|
|
// Index into observations — document ID is observationId
|
|
var doc = new ObservationDocument
|
|
{
|
|
ObservationId = evt.ObservationId.ToString(),
|
|
EncounterId = evt.EncounterId.ToString(),
|
|
PatientId = evt.PatientId.ToString(),
|
|
Mrn = evt.Mrn ?? string.Empty,
|
|
ObservationCode = evt.ObservationCode,
|
|
Value = (double)evt.Value,
|
|
Unit = evt.Unit,
|
|
Source = evt.Source,
|
|
RecordedAt = evt.RecordedAt
|
|
};
|
|
|
|
var indexResp = await _elastic.IndexAsync(
|
|
doc,
|
|
i => i.Index(_esOptions.Indices.Observations).Id(doc.ObservationId),
|
|
ct);
|
|
|
|
if (!indexResp.IsValidResponse)
|
|
throw new InvalidOperationException(
|
|
$"ES index failed for observation {evt.ObservationId}: {indexResp.DebugInformation}");
|
|
|
|
// Update lastObservationAt on patient_encounters using a conditional script:
|
|
// only update if the new recordedAt is later than the stored value.
|
|
// This handles out-of-order delivery: an older observation re-processed after a
|
|
// newer one must not overwrite lastObservationAt with a stale timestamp.
|
|
var updateResp = await _elastic.UpdateAsync<PatientEncounterDocument, object>(
|
|
_esOptions.Indices.PatientEncounters,
|
|
evt.EncounterId.ToString(),
|
|
u => u
|
|
.Script(new Script(new InlineScript
|
|
{
|
|
Source = """
|
|
if (ctx._source.lastObservationAt == null ||
|
|
params.recordedAt > ctx._source.lastObservationAt) {
|
|
ctx._source.lastObservationAt = params.recordedAt;
|
|
}
|
|
""",
|
|
Language = ScriptLanguage.Painless,
|
|
Params = new Dictionary<string, object>
|
|
{
|
|
["recordedAt"] = evt.RecordedAt.ToString("O")
|
|
}
|
|
}))
|
|
.RetryOnConflict(3),
|
|
ct);
|
|
|
|
if (!updateResp.IsValidResponse && updateResp.Result != Result.NotFound)
|
|
_logger.LogWarning(
|
|
"Could not update lastObservationAt for encounter {Id} — encounter document may not exist yet",
|
|
evt.EncounterId);
|
|
}
|
|
|
|
// --- alert.generated ---
|
|
// Indexes the alert by alertId. Also increments openAlertCount on patient_encounters.
|
|
// TRADE-OFF: openAlertCount increment is not idempotent for partial reprocessing.
|
|
// It is correct for full replay from offset 0 (the stated recovery procedure).
|
|
// For production, use a set of counted alert IDs in the script to enforce idempotency.
|
|
private async Task HandleAlertGeneratedAsync(string payload, CancellationToken ct)
|
|
{
|
|
var evt = JsonSerializer.Deserialize<AlertGeneratedEvent>(payload, EventJsonOptions)!;
|
|
using var payloadDoc = JsonDocument.Parse(payload);
|
|
var root = payloadDoc.RootElement;
|
|
|
|
var alertDoc = new ClinicalAlertDocument
|
|
{
|
|
AlertId = evt.AlertId.ToString(),
|
|
EncounterId = evt.EncounterId.ToString(),
|
|
PatientId = evt.PatientId.ToString(),
|
|
Department = evt.Department ?? string.Empty,
|
|
AlertType = evt.AlertType,
|
|
Severity = evt.Severity,
|
|
Status = "Open",
|
|
TriggeredAt = evt.TriggeredAt
|
|
};
|
|
|
|
var indexResp = await _elastic.IndexAsync(
|
|
alertDoc,
|
|
i => i.Index(_esOptions.Indices.ClinicalAlerts).Id(alertDoc.AlertId),
|
|
ct);
|
|
|
|
if (!indexResp.IsValidResponse)
|
|
throw new InvalidOperationException(
|
|
$"ES index failed for alert {evt.AlertId}: {indexResp.DebugInformation}");
|
|
|
|
// Increment openAlertCount on the parent encounter document.
|
|
// NEWS2 alerts also carry news2Score/news2RiskLevel in the Kafka payload;
|
|
// stamp those on patient_encounters so ward dashboards can filter by acuity.
|
|
var scriptLines = new List<string> { "ctx._source.openAlertCount += 1" };
|
|
Dictionary<string, object>? scriptParams = null;
|
|
|
|
if (root.TryGetProperty("news2Score", out var scoreElem) &&
|
|
root.TryGetProperty("news2RiskLevel", out var riskElem))
|
|
{
|
|
scriptLines.Add("ctx._source.news2Score = params.score");
|
|
scriptLines.Add("ctx._source.news2RiskLevel = params.riskLevel");
|
|
scriptParams = new Dictionary<string, object>
|
|
{
|
|
["score"] = scoreElem.GetInt32(),
|
|
["riskLevel"] = riskElem.GetString()!
|
|
};
|
|
}
|
|
|
|
var updateResp = await _elastic.UpdateAsync<PatientEncounterDocument, object>(
|
|
_esOptions.Indices.PatientEncounters,
|
|
evt.EncounterId.ToString(),
|
|
u => u
|
|
.Script(new Script(new InlineScript
|
|
{
|
|
Source = string.Join(";\n", scriptLines),
|
|
Language = ScriptLanguage.Painless,
|
|
Params = scriptParams
|
|
}))
|
|
.RetryOnConflict(3),
|
|
ct);
|
|
|
|
if (!updateResp.IsValidResponse && updateResp.Result != Result.NotFound)
|
|
_logger.LogWarning(
|
|
"Could not update patient_encounters for alert on encounter {Id}", evt.EncounterId);
|
|
}
|
|
} |