From fde3d564849ae65f080d464006344d3ec637e688 Mon Sep 17 00:00:00 2001 From: voltsrage Date: Wed, 17 Jun 2026 00:05:37 +0800 Subject: [PATCH] feature: Elasticsearch CQRS Projection and Analytics Endpoints --- .../ElasticIndexProvisioner.cs | 94 ++++ .../BackgroundServices/EsIndexerService.cs | 241 +++++++++ .../Controllers/AnalyticsController.cs | 107 ++++ .../Configurations/ElasticIndexOptions.cs | 6 + .../Configurations/ElasticsearchOptions.cs | 6 + .../Documents/ClinicalAlertDocument.cs | 11 + .../Documents/ObservationDocument.cs | 12 + .../Documents/PatientEncounterDocument.cs | 13 + .../Records/Alert/AlertGeneratedEvent.cs | 3 + .../Encounter/EncounterStatusChangedEvent.cs | 4 + .../Observation/ObservationRecordedEvent.cs | 4 + VigilCareClinicalAPI/Program.cs | 14 + .../Services/AnalyticsService.cs | 212 ++++++++ .../Services/EncounterService.cs | 10 +- .../Services/Interfaces/IAnalyticsService.cs | 11 + .../Services/ObservationService.cs | 3 + .../Services/PatientService.cs | 23 + .../VigilCareClinicalAPI.csproj | 1 + VigilCareClinicalAPI/appsettings.json | 8 + docker-compose.yml | 22 +- scripts/run-elasticsearch-analytics-tests.sh | 478 ++++++++++++++++++ 21 files changed, 1281 insertions(+), 2 deletions(-) create mode 100644 VigilCareClinicalAPI/BackgroundServices/ElasticIndexProvisioner.cs create mode 100644 VigilCareClinicalAPI/BackgroundServices/EsIndexerService.cs create mode 100644 VigilCareClinicalAPI/Controllers/AnalyticsController.cs create mode 100644 VigilCareClinicalAPI/Data/Configurations/ElasticIndexOptions.cs create mode 100644 VigilCareClinicalAPI/Data/Configurations/ElasticsearchOptions.cs create mode 100644 VigilCareClinicalAPI/Elasticsearch/Documents/ClinicalAlertDocument.cs create mode 100644 VigilCareClinicalAPI/Elasticsearch/Documents/ObservationDocument.cs create mode 100644 VigilCareClinicalAPI/Elasticsearch/Documents/PatientEncounterDocument.cs create mode 100644 VigilCareClinicalAPI/Models/Records/Alert/AlertGeneratedEvent.cs create mode 100644 VigilCareClinicalAPI/Models/Records/Encounter/EncounterStatusChangedEvent.cs create mode 100644 VigilCareClinicalAPI/Models/Records/Observation/ObservationRecordedEvent.cs create mode 100644 VigilCareClinicalAPI/Services/AnalyticsService.cs create mode 100644 VigilCareClinicalAPI/Services/Interfaces/IAnalyticsService.cs create mode 100755 scripts/run-elasticsearch-analytics-tests.sh diff --git a/VigilCareClinicalAPI/BackgroundServices/ElasticIndexProvisioner.cs b/VigilCareClinicalAPI/BackgroundServices/ElasticIndexProvisioner.cs new file mode 100644 index 0000000..e15b0b6 --- /dev/null +++ b/VigilCareClinicalAPI/BackgroundServices/ElasticIndexProvisioner.cs @@ -0,0 +1,94 @@ +using Elastic.Clients.Elasticsearch; +using Elastic.Clients.Elasticsearch.IndexManagement; +using Microsoft.Extensions.Options; + +public class ElasticIndexProvisioner : IHostedService +{ + private readonly ElasticsearchClient _elastic; + private readonly ElasticsearchOptions _options; + private readonly ILogger _logger; + + public ElasticIndexProvisioner( + ElasticsearchClient elastic, + IOptions options, + ILogger logger) + { + _elastic = elastic; + _options = options.Value; + _logger = logger; + } + + public async Task StartAsync(CancellationToken cancellationToken) + { + await EnsureIndexAsync( + _options.Indices.PatientEncounters, BuildPatientEncountersMapping()); + await EnsureIndexAsync( + _options.Indices.Observations, BuildObservationsMapping()); + await EnsureIndexAsync( + _options.Indices.ClinicalAlerts, BuildClinicalAlertsMapping()); + } + + private async Task EnsureIndexAsync( + string indexName, + Action> configure) where T : class + { + var exists = await _elastic.Indices.ExistsAsync(indexName); + if (exists.Exists) + { + _logger.LogInformation("Elasticsearch index '{Index}' already exists — skipping", indexName); + return; + } + + var resp = await _elastic.Indices.CreateAsync(indexName, configure); + if (!resp.IsValidResponse) + throw new InvalidOperationException( + $"Failed to create Elasticsearch index '{indexName}': {resp.DebugInformation}"); + + _logger.LogInformation("Created Elasticsearch index '{Index}'", indexName); + } + + private Action> BuildPatientEncountersMapping() => + d => d.Mappings(m => m.Properties(p => p + .Keyword(k => k.EncounterId) + .Keyword(k => k.PatientId) + .Keyword(k => k.Mrn) + // text for full-text search + keyword sub-field for exact sort/filter + .Text(t => t.PatientName, tf => tf + .Fields(f => f.Keyword(k => k.PatientName))) + .Keyword(k => k.Department) + .Keyword(k => k.Status) + .Keyword(k => k.AttendingPhysician) + .Date(d => d.AdmittedAt) + .IntegerNumber(i => i.OpenAlertCount) + .Date(d => d.LastObservationAt!) + )); + + private Action> BuildObservationsMapping() => + d => d.Mappings(m => m.Properties(p => p + .Keyword(k => k.ObservationId) + .Keyword(k => k.EncounterId) + .Keyword(k => k.PatientId) + .Keyword(k => k.Mrn) + .Keyword(k => k.ObservationCode) + // float: observations are decimal values like 97.3, 2.5, 118.0 + // auto-mapped 'long' would truncate fractional parts silently + .FloatNumber(f => f.Value) + .Keyword(k => k.Unit) + .Keyword(k => k.Source) + .Date(d => d.RecordedAt) + )); + + private Action> BuildClinicalAlertsMapping() => + d => d.Mappings(m => m.Properties(p => p + .Keyword(k => k.AlertId) + .Keyword(k => k.EncounterId) + .Keyword(k => k.PatientId) + .Keyword(k => k.Department) + .Keyword(k => k.AlertType) + .Keyword(k => k.Severity) + .Keyword(k => k.Status) + .Date(d => d.TriggeredAt) + )); + + public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; +} diff --git a/VigilCareClinicalAPI/BackgroundServices/EsIndexerService.cs b/VigilCareClinicalAPI/BackgroundServices/EsIndexerService.cs new file mode 100644 index 0000000..4a51526 --- /dev/null +++ b/VigilCareClinicalAPI/BackgroundServices/EsIndexerService.cs @@ -0,0 +1,241 @@ +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 _logger; + + public EsIndexerService( + ElasticsearchClient elastic, + IOptions kafkaOptions, + IOptions esOptions, + ILogger 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(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? 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(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, + AdmittedAt = evt.AdmittedAt, + OpenAlertCount = 0, + LastObservationAt = null + }; + + var resp = await _elastic.UpdateAsync( + _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(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( + _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 + { + ["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(payload, EventJsonOptions)!; + + var doc = 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( + doc, + i => i.Index(_esOptions.Indices.ClinicalAlerts).Id(doc.AlertId), + ct); + + if (!indexResp.IsValidResponse) + throw new InvalidOperationException( + $"ES index failed for alert {evt.AlertId}: {indexResp.DebugInformation}"); + + // Increment openAlertCount on the parent encounter document + var updateResp = await _elastic.UpdateAsync( + _esOptions.Indices.PatientEncounters, + evt.EncounterId.ToString(), + u => u + .Script(new Script(new InlineScript + { + Source = "ctx._source.openAlertCount += 1", + Language = ScriptLanguage.Painless + })) + .RetryOnConflict(3), + ct); + + if (!updateResp.IsValidResponse && updateResp.Result != Result.NotFound) + _logger.LogWarning( + "Could not increment openAlertCount for encounter {Id}", evt.EncounterId); + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Controllers/AnalyticsController.cs b/VigilCareClinicalAPI/Controllers/AnalyticsController.cs new file mode 100644 index 0000000..eb4c5a8 --- /dev/null +++ b/VigilCareClinicalAPI/Controllers/AnalyticsController.cs @@ -0,0 +1,107 @@ +using Microsoft.AspNetCore.Mvc; + + +/// +/// Elasticsearch-backed analytics queries for population, trends, alerts, and patient search. +/// +[ApiController] +[Route("api/v1/analytics")] +[Produces("application/json")] +public class AnalyticsController : ControllerBase +{ + private readonly IAnalyticsService _analytics; + + public AnalyticsController(IAnalyticsService analytics) => _analytics = analytics; + + /// + /// Counts distinct patients whose observations for a code exceed or fall below a threshold in a time window. + /// + /// Observation code (e.g. HEART_RATE). + /// Numeric threshold value. + /// above or below. + /// Optional start of recorded-at range. + /// Optional end of recorded-at range. + /// Cardinality of unique patients matching the criteria. + [HttpGet("population")] + [ProducesResponseType(typeof(ApiResponse), StatusCodes.Status200OK)] + [ProducesResponseType(typeof(ApiResponse), StatusCodes.Status400BadRequest)] + [ProducesResponseType(typeof(ApiResponse), StatusCodes.Status500InternalServerError)] + public async Task Population( + [FromQuery] string? code, + [FromQuery] decimal threshold, + [FromQuery] string direction, // "above" | "below" + [FromQuery] DateTimeOffset? from, + [FromQuery] DateTimeOffset? to) + { + if (string.IsNullOrWhiteSpace(code)) + return BadRequest(ApiResponse.Fail(400, "code is required.", "MISSING_CODE")); + + var result = await _analytics.GetPopulationAsync(code, threshold, direction, from, to); + return Ok(ApiResponse.Ok(result)); + } + + /// + /// Returns hourly observation volume for an encounter and observation code. + /// + /// Encounter id. + /// Observation code. + /// Optional start of recorded-at range. + /// Optional end of recorded-at range. + /// Hourly trend buckets with document counts. + [HttpGet("observations/trend")] + [ProducesResponseType(typeof(ApiResponse), StatusCodes.Status200OK)] + [ProducesResponseType(typeof(ApiResponse), StatusCodes.Status500InternalServerError)] + public async Task ObservationTrend( + [FromQuery] Guid encounterId, + [FromQuery] string code, + [FromQuery] DateTimeOffset? from, + [FromQuery] DateTimeOffset? to) + { + var result = await _analytics.GetObservationTrendAsync(encounterId, code, from, to); + return Ok(ApiResponse.Ok(result)); + } + + /// + /// Summarizes alert volume by department over a time window. + /// + /// Optional severity filter. + /// Optional department filter. + /// Optional start of triggered-at range. + /// Optional end of triggered-at range. + /// Department-level counts (severity breakdown omitted in current implementation). + [HttpGet("alerts/summary")] + [ProducesResponseType(typeof(ApiResponse), StatusCodes.Status200OK)] + [ProducesResponseType(typeof(ApiResponse), StatusCodes.Status500InternalServerError)] + public async Task AlertSummary( + [FromQuery] string? severity, + [FromQuery] string? department, + [FromQuery] DateTimeOffset? from, + [FromQuery] DateTimeOffset? to) + { + var result = await _analytics.GetAlertSummaryAsync(severity, department, from, to); + return Ok(ApiResponse.Ok(result)); + } + + /// + /// Searches patient/encounter documents by MRN, name, department, or status. + /// + /// Free-text query (MRN, name, or department). + /// Optional department filter. + /// Optional encounter status filter. + /// Page number (0-based). + /// Results per page. + /// Matching patient/encounter documents. + [HttpGet("patients")] + [ProducesResponseType(typeof(ApiResponse), StatusCodes.Status200OK)] + [ProducesResponseType(typeof(ApiResponse), StatusCodes.Status500InternalServerError)] + public async Task PatientSearch( + [FromQuery] string? q, + [FromQuery] string? department, + [FromQuery] string? status, + [FromQuery] int page = 0, + [FromQuery] int pageSize = 20) + { + var result = await _analytics.SearchPatientsAsync(q, department, status, page, pageSize); + return Ok(ApiResponse.Ok(result)); + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Data/Configurations/ElasticIndexOptions.cs b/VigilCareClinicalAPI/Data/Configurations/ElasticIndexOptions.cs new file mode 100644 index 0000000..be44190 --- /dev/null +++ b/VigilCareClinicalAPI/Data/Configurations/ElasticIndexOptions.cs @@ -0,0 +1,6 @@ +public class ElasticIndexOptions +{ + public string PatientEncounters { get; set; } = "patient_encounters"; + public string Observations { get; set; } = "observations"; + public string ClinicalAlerts { get; set; } = "clinical_alerts"; +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Data/Configurations/ElasticsearchOptions.cs b/VigilCareClinicalAPI/Data/Configurations/ElasticsearchOptions.cs new file mode 100644 index 0000000..75c6800 --- /dev/null +++ b/VigilCareClinicalAPI/Data/Configurations/ElasticsearchOptions.cs @@ -0,0 +1,6 @@ +public class ElasticsearchOptions +{ + public const string Section = "Elasticsearch"; + public string Uri { get; set; } = null!; + public ElasticIndexOptions Indices { get; set; } = null!; +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Elasticsearch/Documents/ClinicalAlertDocument.cs b/VigilCareClinicalAPI/Elasticsearch/Documents/ClinicalAlertDocument.cs new file mode 100644 index 0000000..428fc6d --- /dev/null +++ b/VigilCareClinicalAPI/Elasticsearch/Documents/ClinicalAlertDocument.cs @@ -0,0 +1,11 @@ +public class ClinicalAlertDocument +{ + public string AlertId { get; set; } = null!; + public string EncounterId { get; set; } = null!; + public string PatientId { get; set; } = null!; + public string Department { get; set; } = null!; + public string AlertType { get; set; } = null!; + public string Severity { get; set; } = null!; + public string Status { get; set; } = null!; + public DateTimeOffset TriggeredAt { get; set; } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Elasticsearch/Documents/ObservationDocument.cs b/VigilCareClinicalAPI/Elasticsearch/Documents/ObservationDocument.cs new file mode 100644 index 0000000..27dfc27 --- /dev/null +++ b/VigilCareClinicalAPI/Elasticsearch/Documents/ObservationDocument.cs @@ -0,0 +1,12 @@ +public class ObservationDocument +{ + public string ObservationId { get; set; } = null!; + public string EncounterId { get; set; } = null!; + public string PatientId { get; set; } = null!; + public string Mrn { get; set; } = null!; + public string ObservationCode { get; set; } = null!; + public double Value { get; set; } + public string Unit { get; set; } = null!; + public string Source { get; set; } = null!; + public DateTimeOffset RecordedAt { get; set; } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Elasticsearch/Documents/PatientEncounterDocument.cs b/VigilCareClinicalAPI/Elasticsearch/Documents/PatientEncounterDocument.cs new file mode 100644 index 0000000..33d4dd7 --- /dev/null +++ b/VigilCareClinicalAPI/Elasticsearch/Documents/PatientEncounterDocument.cs @@ -0,0 +1,13 @@ +public class PatientEncounterDocument +{ + public string EncounterId { get; set; } = null!; + public string PatientId { get; set; } = null!; + public string Mrn { get; set; } = null!; + public string PatientName { get; set; } = null!; + public string Department { get; set; } = null!; + public string Status { get; set; } = null!; + public string AttendingPhysician { get; set; } = null!; + public DateTimeOffset AdmittedAt { get; set; } + public int OpenAlertCount { get; set; } + public DateTimeOffset? LastObservationAt { get; set; } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Models/Records/Alert/AlertGeneratedEvent.cs b/VigilCareClinicalAPI/Models/Records/Alert/AlertGeneratedEvent.cs new file mode 100644 index 0000000..0f2c415 --- /dev/null +++ b/VigilCareClinicalAPI/Models/Records/Alert/AlertGeneratedEvent.cs @@ -0,0 +1,3 @@ +public record AlertGeneratedEvent( + Guid AlertId, Guid EncounterId, Guid PatientId, string? Department, + string AlertType, string Severity, DateTimeOffset TriggeredAt); \ No newline at end of file diff --git a/VigilCareClinicalAPI/Models/Records/Encounter/EncounterStatusChangedEvent.cs b/VigilCareClinicalAPI/Models/Records/Encounter/EncounterStatusChangedEvent.cs new file mode 100644 index 0000000..0352de1 --- /dev/null +++ b/VigilCareClinicalAPI/Models/Records/Encounter/EncounterStatusChangedEvent.cs @@ -0,0 +1,4 @@ +public record EncounterStatusChangedEvent( + Guid EncounterId, Guid PatientId, string Mrn, string PatientName, + string? PreviousStatus, string NewStatus, string Department, + string AttendingPhysician, DateTimeOffset AdmittedAt, DateTimeOffset ChangedAt); \ No newline at end of file diff --git a/VigilCareClinicalAPI/Models/Records/Observation/ObservationRecordedEvent.cs b/VigilCareClinicalAPI/Models/Records/Observation/ObservationRecordedEvent.cs new file mode 100644 index 0000000..0063aef --- /dev/null +++ b/VigilCareClinicalAPI/Models/Records/Observation/ObservationRecordedEvent.cs @@ -0,0 +1,4 @@ +public record ObservationRecordedEvent( + Guid ObservationId, Guid EncounterId, Guid PatientId, string? Mrn, + string ObservationCode, decimal Value, string Unit, string Source, + DateTimeOffset RecordedAt); \ No newline at end of file diff --git a/VigilCareClinicalAPI/Program.cs b/VigilCareClinicalAPI/Program.cs index fb0bc1a..2ca9f70 100644 --- a/VigilCareClinicalAPI/Program.cs +++ b/VigilCareClinicalAPI/Program.cs @@ -1,3 +1,4 @@ +using Elastic.Clients.Elasticsearch; using Microsoft.EntityFrameworkCore; using Serilog; using StackExchange.Redis; @@ -25,16 +26,29 @@ try builder.Services.Configure( builder.Configuration.GetSection(KafkaOptions.Section)); + builder.Services.Configure( + builder.Configuration.GetSection(ElasticsearchOptions.Section)); + + var esOptions = builder.Configuration + .GetSection(ElasticsearchOptions.Section) + .Get()!; + + builder.Services.AddSingleton( + new ElasticsearchClient(new Uri(esOptions.Uri))); + builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); + builder.Services.AddScoped(); builder.Services.AddHostedService(); builder.Services.AddHostedService(); builder.Services.AddHostedService(); + builder.Services.AddHostedService(); + builder.Services.AddHostedService(); builder.Services.AddControllers() .AddJsonOptions(opts => diff --git a/VigilCareClinicalAPI/Services/AnalyticsService.cs b/VigilCareClinicalAPI/Services/AnalyticsService.cs new file mode 100644 index 0000000..bd2d1dd --- /dev/null +++ b/VigilCareClinicalAPI/Services/AnalyticsService.cs @@ -0,0 +1,212 @@ +using Elastic.Clients.Elasticsearch; +using Elastic.Clients.Elasticsearch.QueryDsl; +using Microsoft.Extensions.Options; + +public class AnalyticsService : IAnalyticsService +{ + private readonly ElasticsearchClient _elastic; + private readonly ElasticsearchOptions _options; + + public AnalyticsService(ElasticsearchClient elastic, IOptions options) + { + _elastic = elastic; + _options = options.Value; + } + + public async Task GetPopulationAsync(string code, decimal threshold, string direction, + DateTimeOffset? from, DateTimeOffset? to) + { + var filters = new List>> + { + fq => fq.Term(t => t.Field(o => o.ObservationCode).Value(code)) + }; + + if (direction == "above") + filters.Add(fq => fq.Range(r => r.NumberRange(nr => + nr.Field(o => o.Value).Gt((double)threshold)))); + else + filters.Add(fq => fq.Range(r => r.NumberRange(nr => + nr.Field(o => o.Value).Lt((double)threshold)))); + + if (from.HasValue) + filters.Add(fq => fq.Range(r => r.DateRange(dr => + dr.Field(o => o.RecordedAt).Gte(from.Value.ToString("O"))))); + + if (to.HasValue) + filters.Add(fq => fq.Range(r => r.DateRange(dr => + dr.Field(o => o.RecordedAt).Lte(to.Value.ToString("O"))))); + + var resp = await _elastic.SearchAsync(s => s + .Indices(_options.Indices.Observations) + .Query(q => q.Bool(b => b.Filter(filters.ToArray()))) + .Aggregations(a => a + // cardinality: count of distinct patientId values — not a document count. + // Two observations from the same patient count as one patient. + .Add("unique_patients", agg => agg.Cardinality(c => c.Field(o => o.PatientId))) + ) + .Size(0)); // no raw hits — aggregation result only + + if (!resp.IsValidResponse) + throw new InvalidOperationException("Elasticsearch query failed."); + + var uniquePatients = resp.Aggregations?.GetCardinality("unique_patients")?.Value ?? 0; + + return new + { + observationCode = code, + threshold, + direction, + from, + to, + uniquePatientCount = uniquePatients + }; + } + + public async Task GetObservationTrendAsync(Guid encounterId, string code, + DateTimeOffset? from, DateTimeOffset? to) + { + var filters = new List>> + { + fq => fq.Term(t => t.Field(o => o.EncounterId).Value(encounterId.ToString())), + fq => fq.Term(t => t.Field(o => o.ObservationCode).Value(code)) + }; + + if (from.HasValue) + filters.Add(fq => fq.Range(r => r.DateRange(dr => + dr.Field(o => o.RecordedAt).Gte(from.Value.ToString("O"))))); + + if (to.HasValue) + filters.Add(fq => fq.Range(r => r.DateRange(dr => + dr.Field(o => o.RecordedAt).Lte(to.Value.ToString("O"))))); + + var resp = await _elastic.SearchAsync(s => s + .Indices(_options.Indices.Observations) + .Query(q => q.Bool(b => b.Filter(filters.ToArray()))) + .Aggregations(a => a + .Add("trend", agg => agg.DateHistogram(h => h + .Field(o => o.RecordedAt) + .CalendarInterval(Elastic.Clients.Elasticsearch.Aggregations.CalendarInterval.Hour) + ) + ) + ) + .Size(0)); + + if (!resp.IsValidResponse) + throw new InvalidOperationException("Elasticsearch query failed."); + + var buckets = resp.Aggregations?.GetDateHistogram("trend")?.Buckets + .Select(b => new + { + hour = b.KeyAsString, + avg = (double?)null, + min = (double?)null, + max = (double?)null, + count = b.DocCount + }) ?? Enumerable.Empty(); + + return new { encounterId, observationCode = code, trend = buckets }; + } + + public async Task GetAlertSummaryAsync(string? severity, string? department, + DateTimeOffset? from, DateTimeOffset? to) + { + var filters = new List>>(); + + if (from.HasValue) + filters.Add(fq => fq.Range(r => r.DateRange(dr => + dr.Field(a => a.TriggeredAt).Gte(from.Value.ToString("O"))))); + + if (to.HasValue) + filters.Add(fq => fq.Range(r => r.DateRange(dr => + dr.Field(a => a.TriggeredAt).Lte(to.Value.ToString("O"))))); + + if (!string.IsNullOrEmpty(severity)) + filters.Add(fq => fq.Term(t => t.Field(a => a.Severity).Value(severity))); + + if (!string.IsNullOrEmpty(department)) + filters.Add(fq => fq.Term(t => t.Field(a => a.Department).Value(department))); + + var resp = await _elastic.SearchAsync(s => s + .Indices(_options.Indices.ClinicalAlerts) + .Query(q => q.Bool(b => + { + if (filters.Count > 0) + b.Filter(filters.ToArray()); + else + b.Must(m => m.MatchAll(_ => { })); + })) + .Aggregations(a => a + .Add("by_department", agg => agg.Terms(t => t + .Field(a => a.Department).Size(50) + ) + ) + ) + .Size(0)); + + if (!resp.IsValidResponse) + throw new InvalidOperationException("Elasticsearch query failed."); + + var summary = resp.Aggregations?.GetStringTerms("by_department")?.Buckets + .Select(deptBucket => new + { + department = deptBucket.Key, + total = deptBucket.DocCount, + bySeverity = Enumerable.Empty() + }) ?? Enumerable.Empty(); + + return new { from, to, summary }; + } + + public async Task SearchPatientsAsync(string? q, string? department, string? status, + int page, int pageSize) + { + var filters = new List>>(); + + if (!string.IsNullOrEmpty(department)) + filters.Add(fq => fq.Term(t => t.Field(p => p.Department).Value(department))); + + if (!string.IsNullOrEmpty(status)) + filters.Add(fq => fq.Term(t => t.Field(p => p.Status).Value(status))); + + var resp = await _elastic.SearchAsync(s => s + .Indices(_options.Indices.PatientEncounters) + .Query(q2 => q2 + .Bool(b => + { + if (!string.IsNullOrWhiteSpace(q)) + { + b.Should( + // MRN: exact keyword match — boosted because MRN lookup is + // the most common search pattern for clinicians who know the number + s2 => s2.Term(t => t.Field(p => p.Mrn).Value(q).Boost(3)), + // Patient name: full-text across the analyzed field + s2 => s2.Match(m => m.Field(p => p.PatientName).Query(q)), + // Department: exact filter match included in should for relevance scoring + s2 => s2.Term(t => t.Field(p => p.Department).Value(q)) + ); + b.MinimumShouldMatch(1); + } + else + { + b.Must(m => m.MatchAll(_ => { })); + } + + if (filters.Count > 0) + b.Filter(filters.ToArray()); + }) + ) + .From(page * pageSize) + .Size(pageSize)); + + if (!resp.IsValidResponse) + throw new InvalidOperationException("Elasticsearch query failed."); + + return new + { + total = resp.Total, + page, + pageSize, + data = resp.Documents + }; + } +} diff --git a/VigilCareClinicalAPI/Services/EncounterService.cs b/VigilCareClinicalAPI/Services/EncounterService.cs index 83aa37f..9560d23 100644 --- a/VigilCareClinicalAPI/Services/EncounterService.cs +++ b/VigilCareClinicalAPI/Services/EncounterService.cs @@ -34,7 +34,10 @@ public class EncounterService : IEncounterService public async Task TransitionStatusAsync( Guid encounterId, EncounterStatus targetStatus) { - var encounter = await _db.Encounters.FindAsync(encounterId); + var encounter = await _db.Encounters + .Include(e => e.Patient) + .FirstOrDefaultAsync(e => e.Id == encounterId); + if (encounter is null) throw new NotFoundException("Encounter not found.", "ENCOUNTER_NOT_FOUND"); @@ -57,8 +60,13 @@ public class EncounterService : IEncounterService { encounterId, patientId = encounter.PatientId, + mrn = encounter.Patient.Mrn, + patientName = $"{encounter.Patient.FirstName} {encounter.Patient.LastName}", previousStatus = previousStatus.ToDbString(), newStatus = targetStatus.ToDbString(), + department = encounter.Department, + attendingPhysician = encounter.AttendingPhysician, + admittedAt = encounter.AdmittedAt, changedAt = DateTimeOffset.UtcNow }), PartitionKey = encounterId.ToString(), diff --git a/VigilCareClinicalAPI/Services/Interfaces/IAnalyticsService.cs b/VigilCareClinicalAPI/Services/Interfaces/IAnalyticsService.cs new file mode 100644 index 0000000..1a7009c --- /dev/null +++ b/VigilCareClinicalAPI/Services/Interfaces/IAnalyticsService.cs @@ -0,0 +1,11 @@ +public interface IAnalyticsService +{ + Task GetPopulationAsync(string code, decimal threshold, string direction, + DateTimeOffset? from, DateTimeOffset? to); + Task GetObservationTrendAsync(Guid encounterId, string code, + DateTimeOffset? from, DateTimeOffset? to); + Task GetAlertSummaryAsync(string? severity, string? department, + DateTimeOffset? from, DateTimeOffset? to); + Task SearchPatientsAsync(string? q, string? department, string? status, + int page, int pageSize); +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Services/ObservationService.cs b/VigilCareClinicalAPI/Services/ObservationService.cs index fc52aa0..93ad13b 100644 --- a/VigilCareClinicalAPI/Services/ObservationService.cs +++ b/VigilCareClinicalAPI/Services/ObservationService.cs @@ -22,6 +22,7 @@ public class ObservationService : IObservationService { // Step 1 — encounter must be active var encounter = await _db.Encounters + .Include(e => e.Patient) .AsNoTracking() .FirstOrDefaultAsync(e => e.Id == encounterId); @@ -112,6 +113,7 @@ public class ObservationService : IObservationService alertId = alert.Id, encounterId, patientId = encounter.PatientId, + department = encounter.Department, alertType = alert.AlertType.ToDbString(), severity = alert.Severity.ToDbString(), triggeredAt = alert.TriggeredAt, @@ -125,6 +127,7 @@ public class ObservationService : IObservationService observationId = observation.Id, encounterId, patientId = encounter.PatientId, + mrn = encounter.Patient?.Mrn, observationCode = req.ObservationCode, value = req.Value, unit = req.Unit, diff --git a/VigilCareClinicalAPI/Services/PatientService.cs b/VigilCareClinicalAPI/Services/PatientService.cs index 94c60c6..ecea453 100644 --- a/VigilCareClinicalAPI/Services/PatientService.cs +++ b/VigilCareClinicalAPI/Services/PatientService.cs @@ -1,3 +1,4 @@ +using System.Text.Json; using Microsoft.EntityFrameworkCore; public class PatientService : IPatientService @@ -87,6 +88,28 @@ public class PatientService : IPatientService CreatedAt = DateTimeOffset.UtcNow }; _db.Encounters.Add(encounter); + + _db.OutboxEvents.Add(new OutboxEvent + { + Id = Guid.NewGuid(), + Topic = "encounter.status.changed", + Payload = JsonSerializer.Serialize(new + { + encounterId = encounter.Id, + patientId = patient.Id, + mrn = patient.Mrn, + patientName = $"{patient.FirstName} {patient.LastName}", + previousStatus = (string?)null, + newStatus = encounter.Status.ToDbString(), + department = encounter.Department, + attendingPhysician = encounter.AttendingPhysician, + admittedAt = encounter.AdmittedAt, + changedAt = DateTimeOffset.UtcNow + }), + PartitionKey = encounter.Id.ToString(), + CreatedAt = DateTimeOffset.UtcNow + }); + await _db.SaveChangesAsync(); return encounter; } diff --git a/VigilCareClinicalAPI/VigilCareClinicalAPI.csproj b/VigilCareClinicalAPI/VigilCareClinicalAPI.csproj index dd9278d..a643762 100644 --- a/VigilCareClinicalAPI/VigilCareClinicalAPI.csproj +++ b/VigilCareClinicalAPI/VigilCareClinicalAPI.csproj @@ -10,6 +10,7 @@ + runtime; build; native; contentfiles; analyzers; buildtransitive diff --git a/VigilCareClinicalAPI/appsettings.json b/VigilCareClinicalAPI/appsettings.json index 15a3aee..ad29fdf 100644 --- a/VigilCareClinicalAPI/appsettings.json +++ b/VigilCareClinicalAPI/appsettings.json @@ -46,5 +46,13 @@ "NumPartitions": 6, "OutboxBatchSize": 100, "OutboxPollIntervalMs": 500 + }, + "Elasticsearch": { + "Uri": "http://localhost:9200", + "Indices": { + "PatientEncounters": "patient_encounters", + "Observations": "observations", + "ClinicalAlerts": "clinical_alerts" + } } } diff --git a/docker-compose.yml b/docker-compose.yml index dbdef46..c09c153 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -50,7 +50,27 @@ services: volumes: - kafka_data:/var/lib/kafka/data + elasticsearch: + image: docker.elastic.co/elasticsearch/elasticsearch:8.13.0 + environment: + - discovery.type=single-node + # Security disabled for development — no TLS negotiation overhead, no keystore setup. + # A production deployment would enable xpack.security and use HTTPS. + - xpack.security.enabled=false + - ES_JAVA_OPTS=-Xms512m -Xmx512m + - cluster.name=vigilcare-dev + ports: + - "9200:9200" + volumes: + - es_data:/usr/share/elasticsearch/data + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:9200/_cluster/health"] + interval: 10s + timeout: 5s + retries: 10 + volumes: pg_data: seq_data: - kafka_data: \ No newline at end of file + kafka_data: + es_data: \ No newline at end of file diff --git a/scripts/run-elasticsearch-analytics-tests.sh b/scripts/run-elasticsearch-analytics-tests.sh new file mode 100755 index 0000000..5e20bf5 --- /dev/null +++ b/scripts/run-elasticsearch-analytics-tests.sh @@ -0,0 +1,478 @@ +#!/usr/bin/env bash + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +COMPOSE_FILE="${COMPOSE_FILE:-${SCRIPT_DIR}/../docker-compose.yml}" + +BASE_URL="${BASE_URL:-http://localhost:5270}" +ES_URL="${ES_URL:-http://localhost:9200}" +PGHOST="${PGHOST:-localhost}" +PGPORT="${PGPORT:-5436}" +PGDATABASE="${PGDATABASE:-vigilcare}" +PGUSER="${PGUSER:-postgres}" +PGPASSWORD="${PGPASSWORD:-password}" + +ES_CONSUMER_GROUP="${ES_CONSUMER_GROUP:-es-indexer}" +INDEX_WAIT_SECS="${INDEX_WAIT_SECS:-60}" +RELAY_WAIT_SECS="${RELAY_WAIT_SECS:-45}" +KAFKA_READY_WAIT_SECS="${KAFKA_READY_WAIT_SECS:-30}" + +SCRIPT_RUN_ID="$(date -u +"%Y%m%d%H%M%S")" + +# Per-run time windows so analytics assertions are isolated from prior script runs. +_pop_month="$(echo "$SCRIPT_RUN_ID" | cut -c5-6)" +_pop_day="$(echo "$SCRIPT_RUN_ID" | cut -c7-8)" +_pop_hour="$(echo "$SCRIPT_RUN_ID" | cut -c9-10)" +_pop_min="$(echo "$SCRIPT_RUN_ID" | cut -c11-12)" +_pop_sec="$(echo "$SCRIPT_RUN_ID" | cut -c13-14)" +POPULATION_AT="2099-${_pop_month}-${_pop_day}T${_pop_hour}:${_pop_min}:${_pop_sec}Z" +POPULATION_FROM="${POPULATION_AT}" +POPULATION_TO="${POPULATION_AT}" +TREND_DAY="2099-${_pop_month}-${_pop_day}" +TREND_FROM="${TREND_DAY}T00:00:00Z" +TREND_TO="${TREND_DAY}T12:00:00Z" +TREND_AT_HOUR_1="${TREND_DAY}T01:30:00Z" +TREND_AT_HOUR_3="${TREND_DAY}T03:45:00Z" +ALERT_AT="2099-${_pop_month}-${_pop_day}T${_pop_hour}:${_pop_min}:${_pop_sec}Z" + +TMP_FILES=() + +cleanup() { + local f + for f in "${TMP_FILES[@]}"; do + rm -f "${f}" "${f}.status" 2>/dev/null || true + done +} +trap cleanup EXIT + +if ! command -v curl >/dev/null 2>&1; then + echo "Missing dependency: curl" + exit 1 +fi + +if ! command -v jq >/dev/null 2>&1; then + echo "Missing dependency: jq" + exit 1 +fi + +if ! command -v docker >/dev/null 2>&1 || [[ ! -f "${COMPOSE_FILE}" ]]; then + echo "Missing dependency: docker compose (${COMPOSE_FILE})" + exit 1 +fi + +compose() { + docker compose -f "${COMPOSE_FILE}" "$@" +} + +kafka_exec() { + compose exec -T kafka "$@" +} + +psql_cmd() { + local sql="$1" + if command -v psql >/dev/null 2>&1; then + PGPASSWORD="${PGPASSWORD}" psql -h "${PGHOST}" -p "${PGPORT}" -U "${PGUSER}" -d "${PGDATABASE}" -tAc "${sql}" + else + compose exec -T postgres psql -U "${PGUSER}" -d "${PGDATABASE}" -tAc "${sql}" + fi +} + +request() { + local method="$1" + local url="$2" + local body="${3:-}" + local tmp_body + tmp_body="$(mktemp)" + TMP_FILES+=("${tmp_body}") + local status + + if [[ -n "${body}" ]]; then + status="$(curl -sS -o "${tmp_body}" -w "%{http_code}" -X "${method}" "${url}" \ + -H "Content-Type: application/json" -d "${body}")" + else + status="$(curl -sS -o "${tmp_body}" -w "%{http_code}" -X "${method}" "${url}")" + fi + + echo "${status}" > "${tmp_body}.status" + echo "${tmp_body}" +} + +assert_status() { + local expected="$1" + local body_file="$2" + local status + status="$(cat "${body_file}.status")" + if [[ "${status}" != "${expected}" ]]; then + echo "Expected HTTP ${expected}, got ${status}" + echo "Response body:" + cat "${body_file}" + echo + return 1 + fi +} + +es_count() { + local index="$1" + curl -sS "${ES_URL}/${index}/_count" | jq -r '.count' +} + +es_encounter_source() { + local encounter_id="$1" + curl -sS "${ES_URL}/patient_encounters/_source/${encounter_id}" +} + +wait_for_kafka() { + local elapsed=0 + while (( elapsed < KAFKA_READY_WAIT_SECS )); do + if kafka_exec /opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 >/dev/null 2>&1; then + return 0 + fi + sleep 2 + elapsed=$((elapsed + 2)) + done + echo "Kafka did not become ready within ${KAFKA_READY_WAIT_SECS}s" + return 1 +} + +indexer_lag() { + kafka_exec /opt/kafka/bin/kafka-consumer-groups.sh \ + --bootstrap-server localhost:9092 \ + --describe \ + --group "${ES_CONSUMER_GROUP}" 2>/dev/null | \ + awk 'NR > 1 && $1 != "" { sum += $6 } END { print sum + 0 }' +} + +wait_for_indexer_lag_zero() { + local elapsed=0 + while (( elapsed < INDEX_WAIT_SECS )); do + local lag + lag="$(indexer_lag)" + if [[ "${lag}" == "0" ]]; then + return 0 + fi + sleep 2 + elapsed=$((elapsed + 2)) + done + echo "es-indexer lag did not reach zero within ${INDEX_WAIT_SECS}s (lag=${lag:-unknown})" + return 1 +} + +wait_for_outbox_processed() { + local outbox_id="$1" + local elapsed=0 + while (( elapsed < RELAY_WAIT_SECS )); do + local processed + processed="$(psql_cmd "SELECT processed_at IS NOT NULL FROM outbox_events WHERE id = '${outbox_id}'")" + if [[ "${processed}" == "t" ]]; then + return 0 + fi + sleep 1 + elapsed=$((elapsed + 1)) + done + echo "Outbox row ${outbox_id} was not processed within ${RELAY_WAIT_SECS}s" + return 1 +} + +assert_projection_counts_match() { + local relayed_obs relayed_enc relayed_alerts es_obs es_enc es_alerts + relayed_obs="$(psql_cmd "SELECT COUNT(*) FROM outbox_events WHERE topic = 'observation.recorded' AND processed_at IS NOT NULL")" + relayed_enc="$(psql_cmd "SELECT COUNT(*) FROM outbox_events WHERE topic = 'encounter.status.changed' AND processed_at IS NOT NULL")" + relayed_alerts="$(psql_cmd "SELECT COUNT(*) FROM outbox_events WHERE topic = 'alert.generated' AND processed_at IS NOT NULL")" + es_obs="$(es_count observations)" + es_enc="$(es_count patient_encounters)" + es_alerts="$(es_count clinical_alerts)" + + if [[ "${relayed_obs}" != "${es_obs}" ]]; then + echo "Observation projection mismatch: relayed=${relayed_obs}, Elasticsearch=${es_obs}" + return 1 + fi + if [[ "${relayed_enc}" != "${es_enc}" ]]; then + echo "Encounter projection mismatch: relayed=${relayed_enc}, Elasticsearch=${es_enc}" + return 1 + fi + if [[ "${relayed_alerts}" != "${es_alerts}" ]]; then + echo "Clinical alert projection mismatch: relayed=${relayed_alerts}, Elasticsearch=${es_alerts}" + return 1 + fi + echo "OK: Elasticsearch counts match relayed outbox events (obs=${es_obs}, enc=${es_enc}, alerts=${es_alerts})" +} + +ingest_observation() { + local encounter_id="$1" + local code="$2" + local value="$3" + local unit="$4" + local source="$5" + local recorded_at="$6" + local idempotency_key="$7" + + local payload + payload="$(jq -nc \ + --arg code "${code}" \ + --argjson value "${value}" \ + --arg unit "${unit}" \ + --arg source "${source}" \ + --arg recordedAt "${recorded_at}" \ + --arg key "${idempotency_key}" \ + '{observations:[{observationCode:$code,value:$value,unit:$unit,source:$source,recordedAt:$recordedAt,idempotencyKey:$key}]}')" + + local resp + resp="$(request POST "${BASE_URL}/api/v1/encounters/${encounter_id}/observations" "${payload}")" + assert_status "201" "${resp}" +} + +TOTAL_STEPS=17 + +echo "Running Elasticsearch + analytics verification against ${BASE_URL}" +echo "Script run id: ${SCRIPT_RUN_ID}" + +echo "" +echo "[0/${TOTAL_STEPS}] Preflight — API, Postgres, Kafka, and Elasticsearch reachable" +preflight_status="$(curl -sS -o /dev/null -w "%{http_code}" "${BASE_URL}/api/v1/alert-thresholds" || true)" +if [[ "${preflight_status}" != "200" ]]; then + echo "API not reachable at ${BASE_URL} (HTTP ${preflight_status})." + echo "Start the API with: dotnet run --project VigilCareClinicalAPI" + exit 1 +fi + +if ! psql_cmd "SELECT 1" >/dev/null 2>&1; then + echo "Postgres not reachable on ${PGHOST}:${PGPORT}." + echo "Start the stack with: docker compose up -d" + exit 1 +fi + +if ! wait_for_kafka; then + exit 1 +fi + +es_health_status="$(curl -sS "${ES_URL}/_cluster/health" | jq -r '.status' || true)" +if [[ "${es_health_status}" != "green" && "${es_health_status}" != "yellow" ]]; then + echo "Elasticsearch cluster health is '${es_health_status}' (expected green or yellow)." + echo "Start the stack with: docker compose up -d" + exit 1 +fi +echo "OK: API, Postgres, Kafka, and Elasticsearch are up (cluster=${es_health_status})" + +echo "" +echo "[1/${TOTAL_STEPS}] Verifying Elasticsearch indices exist" +for index in patient_encounters observations clinical_alerts; do + exists="$(curl -sS -o /dev/null -w "%{http_code}" "${ES_URL}/${index}" || true)" + if [[ "${exists}" != "200" ]]; then + echo "Missing Elasticsearch index: ${index} (HTTP ${exists})" + exit 1 + fi +done +echo "OK: patient_encounters, observations, clinical_alerts indices exist" + +echo "" +echo "[2/${TOTAL_STEPS}] Waiting for es-indexer consumer lag to reach zero" +if ! wait_for_indexer_lag_zero; then + exit 1 +fi +echo "OK: es-indexer lag is zero" + +echo "" +echo "[3/${TOTAL_STEPS}] Verifying Elasticsearch document counts match relayed outbox events" +assert_projection_counts_match + +echo "" +echo "[4/${TOTAL_STEPS}] Creating patient and two ICU encounters for analytics fixtures" +patient_payload='{"firstName":"Verify","lastName":"Smith","dateOfBirth":"1985-03-20","gender":"F"}' +resp="$(request POST "${BASE_URL}/api/v1/patients" "${patient_payload}")" +assert_status "201" "${resp}" +patient_id="$(jq -r '.data.id' "${resp}")" +patient_mrn="$(jq -r '.data.mrn' "${resp}")" +if [[ -z "${patient_mrn}" || "${patient_mrn}" == "null" ]]; then + echo "Could not parse patient MRN from create response." + exit 1 +fi + +enc_payload_a='{"encounterType":"Inpatient","department":"ICU","attendingPhysician":"Dr. Elastic"}' +resp="$(request POST "${BASE_URL}/api/v1/patients/${patient_id}/encounters" "${enc_payload_a}")" +assert_status "201" "${resp}" +encounter_a="$(jq -r '.data.id' "${resp}")" + +enc_payload_b='{"encounterType":"Outpatient","department":"ICU","attendingPhysician":"Dr. Elastic"}' +resp="$(request POST "${BASE_URL}/api/v1/patients/${patient_id}/encounters" "${enc_payload_b}")" +assert_status "201" "${resp}" +encounter_b="$(jq -r '.data.id' "${resp}")" +echo "OK: patient ${patient_id} (mrn=${patient_mrn}), encounters ${encounter_a}, ${encounter_b}" + +echo "" +echo "[5/${TOTAL_STEPS}] Waiting for encounter documents to appear in Elasticsearch" +elapsed=0 +while (( elapsed < INDEX_WAIT_SECS )); do + source="$(es_encounter_source "${encounter_a}" 2>/dev/null || true)" + if jq -e '.encounterId' >/dev/null 2>&1 <<< "${source}"; then + break + fi + sleep 2 + elapsed=$((elapsed + 2)) +done +if ! jq -e '.encounterId' >/dev/null 2>&1 <<< "$(es_encounter_source "${encounter_a}")"; then + echo "patient_encounters document not found for encounter ${encounter_a}" + exit 1 +fi +echo "OK: patient_encounters documents indexed" + +echo "" +echo "[6/${TOTAL_STEPS}] Verifying openAlertCount is zero before any alert" +open_before="$(es_encounter_source "${encounter_a}" | jq -r '.openAlertCount')" +if [[ "${open_before}" != "0" ]]; then + echo "Expected openAlertCount=0 before alert, got ${open_before}" + exit 1 +fi +echo "OK: openAlertCount=0 before critical ingest" + +echo "" +echo "[7/${TOTAL_STEPS}] Ingesting population, trend, and critical observations" +ingest_observation "${encounter_a}" "HEART_RATE" 104 "bpm" "DEVICE" "${POPULATION_AT}" \ + "es-pop-a1-${SCRIPT_RUN_ID}" +ingest_observation "${encounter_a}" "HEART_RATE" 112 "bpm" "DEVICE" "${POPULATION_AT}" \ + "es-pop-a2-${SCRIPT_RUN_ID}" +ingest_observation "${encounter_b}" "HEART_RATE" 82 "bpm" "DEVICE" "${POPULATION_AT}" \ + "es-pop-b1-${SCRIPT_RUN_ID}" +ingest_observation "${encounter_a}" "HEART_RATE" 90 "bpm" "DEVICE" "${TREND_AT_HOUR_1}" \ + "es-trend-h1-${SCRIPT_RUN_ID}" +ingest_observation "${encounter_a}" "HEART_RATE" 95 "bpm" "DEVICE" "${TREND_AT_HOUR_3}" \ + "es-trend-h3-${SCRIPT_RUN_ID}" + +critical_payload="$(jq -nc \ + --arg recordedAt "${ALERT_AT}" \ + --arg key "es-critical-${SCRIPT_RUN_ID}" \ + '{observations:[{observationCode:"POTASSIUM_MEQ_L",value:2.1,unit:"mEq/L",source:"LAB",recordedAt:$recordedAt,idempotencyKey:$key}]}')" +resp="$(request POST "${BASE_URL}/api/v1/encounters/${encounter_a}/observations" "${critical_payload}")" +assert_status "201" "${resp}" +if [[ "$(jq -r '.data.alertGenerated' "${resp}")" != "true" ]]; then + echo "Expected critical potassium ingest to generate an alert" + exit 1 +fi +echo "OK: five observations and one critical alert ingested" + +echo "" +echo "[8/${TOTAL_STEPS}] Waiting for outbox relay and es-indexer to catch up" +latest_outbox="$(psql_cmd "SELECT id FROM outbox_events ORDER BY created_at DESC LIMIT 1")" +if [[ -n "${latest_outbox}" ]]; then + wait_for_outbox_processed "${latest_outbox}" +fi +if ! wait_for_indexer_lag_zero; then + exit 1 +fi +echo "OK: relay and es-indexer caught up" + +echo "" +echo "[9/${TOTAL_STEPS}] Re-verifying projection counts after ingest" +assert_projection_counts_match + +echo "" +echo "[10/${TOTAL_STEPS}] Verifying population query cardinality (uniquePatientCount = 1)" +population_url="${BASE_URL}/api/v1/analytics/population?code=HEART_RATE&threshold=100&direction=above&from=${POPULATION_FROM}&to=${POPULATION_TO}" +resp="$(request GET "${population_url}")" +assert_status "200" "${resp}" +if ! jq -e '.data.uniquePatientCount == 1' "${resp}" >/dev/null; then + echo "Expected uniquePatientCount=1 for two above-threshold observations from one patient" + echo "Response body:" + cat "${resp}" + echo + exit 1 +fi +echo "OK: population query returned uniquePatientCount=1" + +echo "" +echo "[11/${TOTAL_STEPS}] Verifying observation trend hourly buckets" +trend_url="${BASE_URL}/api/v1/analytics/observations/trend?encounterId=${encounter_a}&code=HEART_RATE&from=${TREND_FROM}&to=${TREND_TO}" +resp="$(request GET "${trend_url}")" +assert_status "200" "${resp}" +trend_count="$(jq -r '.data.trend | length' "${resp}")" +if [[ "${trend_count}" -lt 2 ]]; then + echo "Expected at least two hourly trend buckets, got ${trend_count}" + echo "Response body:" + cat "${resp}" + echo + exit 1 +fi +if ! jq -e '.data.trend[] | select(.count >= 1)' "${resp}" >/dev/null; then + echo "Expected trend buckets with count >= 1" + exit 1 +fi +echo "OK: trend query returned ${trend_count} hourly bucket(s)" + +echo "" +echo "[12/${TOTAL_STEPS}] Verifying patient search by exact MRN" +search_url="${BASE_URL}/api/v1/analytics/patients?q=${patient_mrn}" +resp="$(request GET "${search_url}")" +assert_status "200" "${resp}" +if ! jq -e --arg mrn "${patient_mrn}" '.data.data[] | select(.mrn == $mrn)' "${resp}" >/dev/null; then + echo "MRN search did not return patient with mrn=${patient_mrn}" + echo "Response body:" + cat "${resp}" + echo + exit 1 +fi +echo "OK: MRN search returned exact match (${patient_mrn})" + +echo "" +echo "[13/${TOTAL_STEPS}] Verifying patient search by partial name" +resp="$(request GET "${BASE_URL}/api/v1/analytics/patients?q=smith")" +assert_status "200" "${resp}" +if ! jq -e --arg id "${patient_id}" '.data.data[] | select(.patientId == $id)' "${resp}" >/dev/null; then + echo "Name search for 'smith' did not return patient ${patient_id}" + echo "Response body:" + cat "${resp}" + echo + exit 1 +fi +echo "OK: partial name search matched Smith" + +echo "" +echo "[14/${TOTAL_STEPS}] Verifying department and status filter (no text query)" +resp="$(request GET "${BASE_URL}/api/v1/analytics/patients?department=ICU&status=ACTIVE&pageSize=100")" +assert_status "200" "${resp}" +if ! jq -e --arg enc "${encounter_a}" '.data.data[] | select(.encounterId == $enc)' "${resp}" >/dev/null; then + echo "Department/status filter did not return encounter ${encounter_a}" + echo "Response body:" + cat "${resp}" + echo + exit 1 +fi +echo "OK: ICU + ACTIVE filter returned expected encounter" + +echo "" +echo "[15/${TOTAL_STEPS}] Verifying openAlertCount increments after alert.generated" +open_after="$(es_encounter_source "${encounter_a}" | jq -r '.openAlertCount')" +if [[ "${open_after}" != "1" ]]; then + echo "Expected openAlertCount=1 after critical alert, got ${open_after}" + exit 1 +fi +echo "OK: openAlertCount incremented to 1" + +echo "" +echo "[16/${TOTAL_STEPS}] Verifying alert summary by department" +summary_url="${BASE_URL}/api/v1/analytics/alerts/summary?department=ICU" +resp="$(request GET "${summary_url}")" +assert_status "200" "${resp}" +if ! jq -e '.data.summary[] | select(.department == "ICU" and (.total | tonumber) >= 1)' "${resp}" >/dev/null; then + echo "Alert summary did not include ICU with total >= 1" + echo "Response body:" + cat "${resp}" + echo + exit 1 +fi +echo "OK: alert summary includes ICU department" + +echo "" +echo "[17/${TOTAL_STEPS}] Verifying population endpoint validates required code" +resp="$(request GET "${BASE_URL}/api/v1/analytics/population?threshold=100&direction=above")" +assert_status "400" "${resp}" +if [[ "$(jq -r '.error.code' "${resp}")" != "MISSING_CODE" ]]; then + echo "Expected MISSING_CODE for population without code parameter" + exit 1 +fi +echo "OK: missing code returns 400 MISSING_CODE" + +echo "" +echo "All ${TOTAL_STEPS} Elasticsearch + analytics checks passed." +echo "" +echo "Note: Full index replay (delete indices, reset es-indexer offsets, restart API) is documented" +echo "in docs/plans/phase-4-plan.md Step 7 and must be run manually to complete the replay checklist item."