From 032cd1d24012505499cb21855737acb70bf2cd2f Mon Sep 17 00:00:00 2001 From: voltsrage Date: Wed, 24 Jun 2026 01:19:03 +0800 Subject: [PATCH] test phase 23 verification script --- VigilCare.WardGateway/appsettings.json | 2 +- .../DataLakeEventParserTests.cs | 49 +++++++ .../TrendDetectorTests.cs | 19 +++ .../TrendAnalyzerService.cs | 6 +- .../DataLake/DataLakeEventParser.cs | 137 ++++++++++++++++++ .../DataLake/DataLakeWriterService.cs | 86 ++--------- .../Domains/Enums/TrendOutcome.cs | 3 +- VigilCareClinicalAPI/Trend/TrendDetector.cs | 11 ++ docker-compose.yml | 2 +- scripts/demo-network-partition.sh | 4 +- scripts/run-phase20-verification.sh | 2 +- scripts/run-phase21-verification.sh | 2 +- scripts/run-phase22-verification.sh | 2 +- scripts/run-phase23-verification.sh | 1 + 14 files changed, 240 insertions(+), 86 deletions(-) create mode 100644 VigilCareClinicalAPI.Tests/DataLakeEventParserTests.cs create mode 100644 VigilCareClinicalAPI/DataLake/DataLakeEventParser.cs diff --git a/VigilCare.WardGateway/appsettings.json b/VigilCare.WardGateway/appsettings.json index 1540ca5..307dafa 100644 --- a/VigilCare.WardGateway/appsettings.json +++ b/VigilCare.WardGateway/appsettings.json @@ -13,7 +13,7 @@ "PagingAckTimeoutMs": 300000 }, "CentralApi": { - "BaseUrl": "http://localhost:5080" + "BaseUrl": "http://localhost:5270" }, "Gateway": { "GatewayId": "22222222-2222-2222-2222-222222222222", diff --git a/VigilCareClinicalAPI.Tests/DataLakeEventParserTests.cs b/VigilCareClinicalAPI.Tests/DataLakeEventParserTests.cs new file mode 100644 index 0000000..788921e --- /dev/null +++ b/VigilCareClinicalAPI.Tests/DataLakeEventParserTests.cs @@ -0,0 +1,49 @@ +using FluentAssertions; + +public class DataLakeEventParserTests +{ + [Fact] + public void ParseObservationRow_LegacyMinimalPayload_UsesFallbackFields() + { + var encounterId = Guid.NewGuid(); + var payload = $$"""{"code":"HEART_RATE","value":80,"encounterId":"{{encounterId}}"}"""; + + var row = DataLakeEventParser.ParseObservationRow(payload, offset: 42, partition: 2); + + row.EncounterId.Should().Be(encounterId.ToString()); + row.ObservationCode.Should().Be("HEART_RATE"); + row.Value.Should().Be(80); + row.ObservationId.Should().Be($"legacy:{encounterId}:42"); + row.PatientId.Should().BeEmpty(); + row.Unit.Should().BeEmpty(); + row.Source.Should().BeEmpty(); + row.RecordedAt.Should().BeEmpty(); + row.KafkaPartition.Should().Be(2); + row.KafkaOffset.Should().Be(42); + } + + [Fact] + public void ParseObservationRow_FullPayload_ParsesAllFields() + { + var payload = """ + { + "observationId": "20facb3e-3dcb-46dd-8b81-0fa98cf097e6", + "encounterId": "d4a0bfa6-3f3c-4dc3-a0d5-f08b574a7cb4", + "patientId": "eff74ea8-8b02-4a45-8c85-e28604247cc0", + "mrn": "MRN-000001", + "observationCode": "HEART_RATE", + "value": 165, + "unit": "bpm", + "source": "DEVICE", + "recordedAt": "2026-06-23T17:08:20+00:00" + } + """; + + var row = DataLakeEventParser.ParseObservationRow(payload, offset: 99, partition: 0); + + row.ObservationId.Should().Be("20facb3e-3dcb-46dd-8b81-0fa98cf097e6"); + row.Mrn.Should().Be("MRN-000001"); + row.Unit.Should().Be("bpm"); + row.RecordedAt.Should().Be("2026-06-23T17:08:20+00:00"); + } +} diff --git a/VigilCareClinicalAPI.Tests/TrendDetectorTests.cs b/VigilCareClinicalAPI.Tests/TrendDetectorTests.cs index e92ca9e..e8d164e 100644 --- a/VigilCareClinicalAPI.Tests/TrendDetectorTests.cs +++ b/VigilCareClinicalAPI.Tests/TrendDetectorTests.cs @@ -116,6 +116,25 @@ public class TrendDetectorTests : IAsyncLifetime (await redis.GetDatabase().KeyExistsAsync(key)).Should().BeFalse(); } + [Fact] + public async Task UnknownEncounter_SkipsAlert() + { + using var scope = _fixture.Services.CreateScope(); + var detector = scope.ServiceProvider.GetRequiredService(); + var unknownEncounterId = Guid.NewGuid(); + + await detector.ProcessObservationAsync( + unknownEncounterId, _patientId, "HEART_RATE", 72m, BaseTime); + var result = await detector.ProcessObservationAsync( + unknownEncounterId, _patientId, "HEART_RATE", 95m, BaseTime.AddMinutes(10)); + + result.Outcome.Should().Be(TrendOutcome.EncounterNotFound); + result.AlertCreated.Should().BeFalse(); + + var db = scope.ServiceProvider.GetRequiredService(); + (await db.ClinicalAlerts.CountAsync()).Should().Be(0); + } + [Fact] public async Task DuplicateTrendAlert_Idempotent() { diff --git a/VigilCareClinicalAPI/BackgroundServices/TrendAnalyzerService.cs b/VigilCareClinicalAPI/BackgroundServices/TrendAnalyzerService.cs index 4045ff6..0e5f491 100644 --- a/VigilCareClinicalAPI/BackgroundServices/TrendAnalyzerService.cs +++ b/VigilCareClinicalAPI/BackgroundServices/TrendAnalyzerService.cs @@ -59,7 +59,11 @@ public class TrendAnalyzerService : BackgroundService evt.RecordedAt, stoppingToken); - if (outcome.Outcome == TrendOutcome.RapidDeterioration) + if (outcome.Outcome == TrendOutcome.EncounterNotFound) + _logger.LogWarning( + "Skipping stale observation.recorded event — encounter={Id} code={Code} offset={Offset}", + evt.EncounterId, outcome.ObservationCode, result.Offset.Value); + else if (outcome.Outcome == TrendOutcome.RapidDeterioration) _logger.LogInformation( "RAPID_DETERIORATION alert via consumer — encounter={Id} code={Code} rate={Rate}/min", evt.EncounterId, outcome.ObservationCode, outcome.RatePerMinute); diff --git a/VigilCareClinicalAPI/DataLake/DataLakeEventParser.cs b/VigilCareClinicalAPI/DataLake/DataLakeEventParser.cs new file mode 100644 index 0000000..e72bd6c --- /dev/null +++ b/VigilCareClinicalAPI/DataLake/DataLakeEventParser.cs @@ -0,0 +1,137 @@ +using System.Text.Json; + +/// +/// Best-effort deserializers for Kafka payloads written to Parquet. +/// Tolerates legacy/minimal observation shapes (e.g. test outbox rows using +/// code instead of observationCode). +/// +public static class DataLakeEventParser +{ + public static ObservationRow ParseObservationRow(string payload, long offset, int partition) + { + var d = JsonDocument.Parse(payload).RootElement; + var encounterId = GetString(d, "encounterId"); + var observationId = GetString(d, "observationId"); + if (string.IsNullOrEmpty(observationId)) + observationId = $"legacy:{encounterId}:{offset}"; + + return new ObservationRow( + ObservationId : observationId, + EncounterId : encounterId, + PatientId : GetString(d, "patientId"), + Mrn : GetString(d, "mrn"), + ObservationCode : GetString(d, "observationCode", "code"), + Value : GetDouble(d, "value"), + Unit : GetString(d, "unit"), + Source : GetString(d, "source"), + RecordedAt : GetTimestampString(d, "recordedAt"), + KafkaPartition : partition, + KafkaOffset : offset); + } + + public static AlertRow ParseAlertRow(string payload, long offset, int partition) + { + var d = JsonDocument.Parse(payload).RootElement; + return new AlertRow( + AlertId : GetString(d, "alertId"), + EncounterId : GetString(d, "encounterId"), + PatientId : GetString(d, "patientId"), + AlertType : GetString(d, "alertType"), + Severity : GetString(d, "severity"), + Details : GetString(d, "details"), + TriggeredAt : GetTimestampString(d, "triggeredAt"), + KafkaPartition : partition, + KafkaOffset : offset); + } + + public static EncounterStatusRow ParseEncounterRow(string payload, long offset, int partition) + { + var d = JsonDocument.Parse(payload).RootElement; + return new EncounterStatusRow( + EncounterId : GetString(d, "encounterId"), + PatientId : GetString(d, "patientId"), + PreviousStatus : GetString(d, "previousStatus"), + NewStatus : GetString(d, "newStatus"), + ChangedAt : GetTimestampString(d, "changedAt"), + KafkaPartition : partition, + KafkaOffset : offset); + } + + public static string ExtractDatePath(string topic, string payload, KafkaTopicOptions topics) + { + try + { + var doc = JsonDocument.Parse(payload); + var ts = topic switch + { + var t when t == topics.ObservationRecorded => GetTimestamp(doc.RootElement, "recordedAt"), + var t when t == topics.AlertGenerated => GetTimestamp(doc.RootElement, "triggeredAt"), + var t when t == topics.EncounterStatusChanged => GetTimestamp(doc.RootElement, "changedAt"), + _ => DateTimeOffset.UtcNow, + }; + return $"{ts.Year:D4}/{ts.Month:D2}/{ts.Day:D2}"; + } + catch + { + var now = DateTimeOffset.UtcNow; + return $"{now.Year:D4}/{now.Month:D2}/{now.Day:D2}"; + } + } + + private static string GetString(JsonElement d, string primary, string? alternate = null) + { + if (TryGetProperty(d, primary, out var prop)) + return ElementToString(prop); + if (alternate is not null && TryGetProperty(d, alternate, out prop)) + return ElementToString(prop); + return ""; + } + + private static double GetDouble(JsonElement d, string name) + { + if (!TryGetProperty(d, name, out var prop)) + return 0; + + return prop.ValueKind switch + { + JsonValueKind.Number => prop.GetDouble(), + JsonValueKind.String => double.TryParse(prop.GetString(), out var v) ? v : 0, + _ => 0 + }; + } + + private static string GetTimestampString(JsonElement d, string name) + { + if (!TryGetProperty(d, name, out var prop)) + return ""; + + if (prop.ValueKind == JsonValueKind.String) + return prop.GetString() ?? ""; + + return prop.TryGetDateTimeOffset(out var dt) ? dt.ToString("O") : ""; + } + + private static DateTimeOffset GetTimestamp(JsonElement d, string name) + { + if (!TryGetProperty(d, name, out var prop)) + return DateTimeOffset.UtcNow; + + if (prop.ValueKind == JsonValueKind.String) + return prop.GetDateTimeOffset(); + + return prop.TryGetDateTimeOffset(out var dt) ? dt : DateTimeOffset.UtcNow; + } + + private static bool TryGetProperty(JsonElement d, string name, out JsonElement prop) => + d.TryGetProperty(name, out prop) && prop.ValueKind != JsonValueKind.Null; + + private static string ElementToString(JsonElement prop) => + prop.ValueKind switch + { + JsonValueKind.String => prop.GetString() ?? "", + JsonValueKind.Number => prop.GetRawText(), + _ when prop.ValueKind == JsonValueKind.True || prop.ValueKind == JsonValueKind.False + => prop.GetBoolean().ToString(), + _ => prop.GetRawText() + }; +} diff --git a/VigilCareClinicalAPI/DataLake/DataLakeWriterService.cs b/VigilCareClinicalAPI/DataLake/DataLakeWriterService.cs index 6c769c5..3bd24f8 100644 --- a/VigilCareClinicalAPI/DataLake/DataLakeWriterService.cs +++ b/VigilCareClinicalAPI/DataLake/DataLakeWriterService.cs @@ -191,22 +191,28 @@ public sealed class DataLakeWriterService : BackgroundService } } + private string ExtractDatePath(string topic, string payload) => + DataLakeEventParser.ExtractDatePath(topic, payload, _kafkaOptions.Topics); + private async Task BuildParquetAsync( string topic, List events, int partition) { if (topic == _kafkaOptions.Topics.ObservationRecorded) { - var rows = events.Select(e => ParseObservationRow(e, partition)).ToList(); + var rows = events.Select(e => DataLakeEventParser.ParseObservationRow( + e.Payload, e.Offset, partition)).ToList(); return await ParquetFileBuilder.BuildObservationsAsync(rows); } if (topic == _kafkaOptions.Topics.AlertGenerated) { - var rows = events.Select(e => ParseAlertRow(e, partition)).ToList(); + var rows = events.Select(e => DataLakeEventParser.ParseAlertRow( + e.Payload, e.Offset, partition)).ToList(); return await ParquetFileBuilder.BuildAlertsAsync(rows); } if (topic == _kafkaOptions.Topics.EncounterStatusChanged) { - var rows = events.Select(e => ParseEncounterRow(e, partition)).ToList(); + var rows = events.Select(e => DataLakeEventParser.ParseEncounterRow( + e.Payload, e.Offset, partition)).ToList(); return await ParquetFileBuilder.BuildEncountersAsync(rows); } throw new InvalidOperationException($"Unknown topic: {topic}"); @@ -233,80 +239,6 @@ public sealed class DataLakeWriterService : BackgroundService return $"{folder}/{key.DatePath}/partition-{key.Partition}-offset-{firstOffset:D10}.parquet"; } - private string ExtractDatePath(string topic, string payload) - { - try - { - var doc = JsonDocument.Parse(payload); - var ts = topic switch - { - var t when t == _kafkaOptions.Topics.ObservationRecorded => doc.RootElement.GetProperty("recordedAt").GetDateTimeOffset(), - var t when t == _kafkaOptions.Topics.AlertGenerated => doc.RootElement.GetProperty("triggeredAt").GetDateTimeOffset(), - var t when t == _kafkaOptions.Topics.EncounterStatusChanged => doc.RootElement.GetProperty("changedAt").GetDateTimeOffset(), - _ => DateTimeOffset.UtcNow, - }; - return $"{ts.Year:D4}/{ts.Month:D2}/{ts.Day:D2}"; - } - catch - { - // Malformed payload: use today so the event is not lost. - var now = DateTimeOffset.UtcNow; - return $"{now.Year:D4}/{now.Month:D2}/{now.Day:D2}"; - } - } - - // --------------------------------------------------------------------------- - // Payload deserializers — each reads only the fields needed for the Parquet row. - // --------------------------------------------------------------------------- - - private static ObservationRow ParseObservationRow(BufferedEvent e, int partition) - { - var d = JsonDocument.Parse(e.Payload).RootElement; - return new ObservationRow( - ObservationId : d.GetProperty("observationId").GetString() ?? "", - EncounterId : d.GetProperty("encounterId").GetString() ?? "", - PatientId : d.GetProperty("patientId").GetString() ?? "", - Mrn : d.TryGetProperty("mrn", out var mrn) ? mrn.GetString() ?? "" : "", - ObservationCode : d.GetProperty("observationCode").GetString() ?? "", - Value : d.GetProperty("value").GetDouble(), - Unit : d.GetProperty("unit").GetString() ?? "", - Source : d.GetProperty("source").GetString() ?? "", - RecordedAt : d.GetProperty("recordedAt").GetString() ?? "", - KafkaPartition : partition, - KafkaOffset : e.Offset - ); - } - - private static AlertRow ParseAlertRow(BufferedEvent e, int partition) - { - var d = JsonDocument.Parse(e.Payload).RootElement; - return new AlertRow( - AlertId : d.GetProperty("alertId").GetString() ?? "", - EncounterId : d.GetProperty("encounterId").GetString() ?? "", - PatientId : d.GetProperty("patientId").GetString() ?? "", - AlertType : d.GetProperty("alertType").GetString() ?? "", - Severity : d.GetProperty("severity").GetString() ?? "", - Details : d.TryGetProperty("details", out var det) ? det.GetString() ?? "" : "", - TriggeredAt : d.GetProperty("triggeredAt").GetString() ?? "", - KafkaPartition : partition, - KafkaOffset : e.Offset - ); - } - - private static EncounterStatusRow ParseEncounterRow(BufferedEvent e, int partition) - { - var d = JsonDocument.Parse(e.Payload).RootElement; - return new EncounterStatusRow( - EncounterId : d.GetProperty("encounterId").GetString() ?? "", - PatientId : d.GetProperty("patientId").GetString() ?? "", - PreviousStatus : d.GetProperty("previousStatus").GetString() ?? "", - NewStatus : d.GetProperty("newStatus").GetString() ?? "", - ChangedAt : d.GetProperty("changedAt").GetString() ?? "", - KafkaPartition : partition, - KafkaOffset : e.Offset - ); - } - // --------------------------------------------------------------------------- // MinIO upload // --------------------------------------------------------------------------- diff --git a/VigilCareClinicalAPI/Domains/Enums/TrendOutcome.cs b/VigilCareClinicalAPI/Domains/Enums/TrendOutcome.cs index 11ffa05..7fa3b9d 100644 --- a/VigilCareClinicalAPI/Domains/Enums/TrendOutcome.cs +++ b/VigilCareClinicalAPI/Domains/Enums/TrendOutcome.cs @@ -4,5 +4,6 @@ public enum TrendOutcome InsufficientHistory, Stable, RapidDeterioration, - AlertAlreadyOpen + AlertAlreadyOpen, + EncounterNotFound } \ No newline at end of file diff --git a/VigilCareClinicalAPI/Trend/TrendDetector.cs b/VigilCareClinicalAPI/Trend/TrendDetector.cs index f6c187a..4891b03 100644 --- a/VigilCareClinicalAPI/Trend/TrendDetector.cs +++ b/VigilCareClinicalAPI/Trend/TrendDetector.cs @@ -76,6 +76,17 @@ public class TrendDetector if (!TrendCalculator.ExceedsThreshold(observationCode, rate.Value, threshold)) return new TrendResult(TrendOutcome.Stable, observationCode, rate); + using (var scope = _services.CreateScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + if (!await db.Encounters.AnyAsync(e => e.Id == encounterId, ct)) + { + _logger.LogWarning( + "Skipping trend alert for unknown encounter {EncounterId}", encounterId); + return new TrendResult(TrendOutcome.EncounterNotFound, observationCode, rate); + } + } + var details = TrendCalculator.DescribeTrend(observationCode, rate.Value, value); var created = await TryCreateAlertAsync( encounterId, patientId, observationCode, details, rate.Value, ct); diff --git a/docker-compose.yml b/docker-compose.yml index 78c9395..f6fda86 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -203,7 +203,7 @@ services: RabbitMq__Host: ward-gateway-rabbitmq RabbitMq__Port: 5672 RabbitMq__PagingAckTimeoutMs: 300000 - CentralApi__BaseUrl: "http://host.docker.internal:5080" + CentralApi__BaseUrl: "http://host.docker.internal:5270" Gateway__GatewayId: "22222222-2222-2222-2222-222222222222" Gateway__SiteId: "11111111-1111-1111-1111-111111111111" Gateway__Department: "ICU" diff --git a/scripts/demo-network-partition.sh b/scripts/demo-network-partition.sh index ac1b80d..50a734d 100755 --- a/scripts/demo-network-partition.sh +++ b/scripts/demo-network-partition.sh @@ -1,7 +1,7 @@ #!/usr/bin/env bash set -euo pipefail -CENTRAL="${CENTRAL_URL:-http://localhost:5080}" +CENTRAL="${CENTRAL_URL:-http://localhost:5270}" GATEWAY="${GATEWAY_URL:-http://localhost:5081}" JWT="${ADMIN_JWT:?Set ADMIN_JWT to a valid admin bearer token}" GATEWAY_JWT="${GATEWAY_JWT:?Set GATEWAY_JWT to a valid gateway dashboard JWT}" @@ -24,7 +24,7 @@ docker network disconnect "$NETWORK" "$GATEWAY_CONTAINER" 2>/dev/null || true echo "Partition active — posting observation to gateway..." ENCOUNTER_ID=$(curl -sf "$GATEWAY/api/v1/encounters?status=ACTIVE&department=ICU" \ - -H "Authorization: Bearer $GATEWAY_JWT" | jq -r '.data.items[0].id') + -H "Authorization: Bearer $GATEWAY_JWT" | jq -r '.data.items[0].encounterId') curl -sf -X POST "$GATEWAY/api/v1/encounters/$ENCOUNTER_ID/observations" \ -H "Authorization: Bearer $GATEWAY_JWT" \ diff --git a/scripts/run-phase20-verification.sh b/scripts/run-phase20-verification.sh index 50739b2..bb7ea23 100755 --- a/scripts/run-phase20-verification.sh +++ b/scripts/run-phase20-verification.sh @@ -11,7 +11,7 @@ echo "==> Run gateway registry tests" dotnet test VigilCareClinicalAPI.Tests --filter "FullyQualifiedName~GatewayRegistry" --no-build 2>/dev/null \ || dotnet test VigilCareClinicalAPI.Tests --filter "FullyQualifiedName~GatewayRegistry" -API="${API_BASE_URL:-http://localhost:5080}" +API="${API_BASE_URL:-http://localhost:5270}" KEY="${GATEWAY_API_KEY:-dev-gateway-key-change-in-production}" # Obtain admin JWT (adjust to your local login endpoint) diff --git a/scripts/run-phase21-verification.sh b/scripts/run-phase21-verification.sh index 8970da7..54e2563 100755 --- a/scripts/run-phase21-verification.sh +++ b/scripts/run-phase21-verification.sh @@ -5,7 +5,7 @@ ROOT="$(cd "$(dirname "$0")/.." && pwd)" cd "$ROOT" GW="${GATEWAY_URL:-http://localhost:5081}" -CENTRAL="${CENTRAL_URL:-http://localhost:5080}" +CENTRAL="${CENTRAL_URL:-http://localhost:5270}" SITE_ID="${SITE_ID:-11111111-1111-1111-1111-111111111111}" GW_ID="${GATEWAY_ID:-22222222-2222-2222-2222-222222222222}" ADMIN_JWT="${ADMIN_JWT:?Set ADMIN_JWT (central admin JWT)}" diff --git a/scripts/run-phase22-verification.sh b/scripts/run-phase22-verification.sh index bbba034..c4d2a4d 100755 --- a/scripts/run-phase22-verification.sh +++ b/scripts/run-phase22-verification.sh @@ -4,7 +4,7 @@ set -euo pipefail ROOT="$(cd "$(dirname "$0")/.." && pwd)" cd "$ROOT" -CENTRAL="${CENTRAL_URL:-http://localhost:5080}" +CENTRAL="${CENTRAL_URL:-http://localhost:5270}" GATEWAY="${GATEWAY_URL:-http://localhost:5081}" KEY="${GATEWAY_API_KEY:-dev-gateway-key-change-in-production}" GW_ID="${GATEWAY_ID:-22222222-2222-2222-2222-222222222222}" diff --git a/scripts/run-phase23-verification.sh b/scripts/run-phase23-verification.sh index 5e4e739..7861c79 100755 --- a/scripts/run-phase23-verification.sh +++ b/scripts/run-phase23-verification.sh @@ -12,6 +12,7 @@ cd vigilcare-dashboard && npm run test -- GatewayOperations.spec.js && cd .. echo "==> Optional partition demo (skip with SKIP_PARTITION=1)" if [[ "${SKIP_PARTITION:-0}" != "1" ]]; then + export CENTRAL_URL="${CENTRAL_URL:-http://localhost:5270}" ./scripts/demo-network-partition.sh fi