diff --git a/README.md b/README.md index 476748b..4acd9be 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ A production-quality clinical backend built with ASP.NET Core 8, PostgreSQL, Apa ## Domain Model — How It Maps to a Real Clinical System -In a hospital, a patient presents for care and an encounter is opened. Bedside monitors and lab systems post observations continuously against that encounter. A rules engine evaluates each observation against configured thresholds and flags abnormal values as clinical alerts. Clinicians acknowledge and resolve alerts. If a critical alert goes unacknowledged for five minutes, the system escalates to the on-call backup. All events flow through Kafka so the Elasticsearch dashboard, sepsis engine, and data lake each consume the same stream independently. +In a hospital, a patient presents for care and an encounter is opened. Bedside monitors and lab systems post observations continuously against that encounter. A rules engine evaluates each observation against configured thresholds and flags abnormal values as clinical alerts. Clinicians acknowledge and resolve alerts. If a critical alert goes unacknowledged for five minutes, the system escalates to the on-call backup. All events flow through Kafka so the Elasticsearch dashboard, sepsis engine, and data lake writer consume the same stream independently. ``` Patient ─────────────────────────── one patient = one MRN, many lifetime encounters @@ -55,7 +55,7 @@ An `OutboxEvent` is written in the same transaction as any observation or alert, - **Elasticsearch CQRS Projection** — `EsIndexerService` consumer group upserts `patient_encounters` documents, appends to the `observations` index, and updates `openAlertCount` on alert events; patient/encounter search; per-encounter observation trend (hourly avg/min/max); alert volume summary by department and severity; population query (numeric range aggregation across all patients) - **Sepsis Early Warning Engine** — `SepsisEngineService` Kafka consumer evaluates SIRS criteria (temperature, heart rate, respiratory rate, WBC) per encounter using Redis keys with a 30-minute TTL sliding window; on ≥2 active criteria, inserts a `SEPSIS_WARNING / CRITICAL` alert idempotently (`INSERT WHERE NOT EXISTS`) - **RabbitMQ Notification Workers** — `NotificationPublisherService` reads `alert.generated` from Kafka and publishes paging jobs to `alerts.paging.queue`; `PagingWorkerService` sends the page and waits for acknowledgment; if no ack arrives before timeout it NACKs to `alerts.paging.dlq` with `x-message-ttl = 300000ms`; if the host is stopping, in-flight paging messages are NACKed with `requeue=true` so they are retried after restart and do not false-escalate; `EscalationWorkerService` pages the on-call backup and sets alert status to `escalated`; `DischargeSummaryWorkerService` reads `encounter.status.changed`, generates a discharge summary, and stores it in MinIO under `/discharge-summaries/{encounterId}/summary.pdf` -- **Data Lake Writer** — Kafka consumer writing partitioned Parquet files to MinIO (`/observations/`, `/alerts/`, `/encounters/` by date); flush policy: 1,000 events or 5 minutes, whichever comes first; columnar format for 10-year regulatory retention +- **Data Lake Writer (Phase 9 - in progress)** — `DataLakeWriterService` Kafka consumer buffers `observation.recorded`, `alert.generated`, and `encounter.status.changed` events, flushes partitioned Parquet files to MinIO (`/observations/`, `/alerts/`, `/encounters/` by date), and commits offsets after successful uploads - **Reconciliation Jobs** — three scheduled checks: (1) unacknowledged CRITICAL alerts older than 30 minutes, (2) pending orders without results after 4 hours, (3) active inpatients with no observation in 2 hours; each finding creates a `reconciliation_alerts` row and publishes to RabbitMQ - **Standard Envelope** — all responses use a consistent `{ success, statusCode, data, error }` wrapper; validation errors use the same shape; `ApiBehaviorOptions` overridden so model validation also produces the standard envelope - **Observability** — Serilog structured logging enriched with `correlationId`, `encounterId`, `patientId` on alert paths; Seq sink (`http://localhost:5345`); Prometheus (`http://localhost:9101`) scrapes `GET /metrics`; Grafana dashboards (`http://localhost:3101`, admin/admin) for clinical metrics including `alerts_unacknowledged_gauge`; per-request correlation IDs in request logs and response headers @@ -87,6 +87,7 @@ IHostedServices (background): PagingWorkerService → RabbitMQ paging.queue → log page → NACK on timeout (or requeue on shutdown) EscalationWorkerService → RabbitMQ escalation.queue → update alert status DischargeSummaryWorkerService → RabbitMQ discharge.queue → MinIO PDF + DataLakeWriterService → Kafka (data-lake-writer) → Parquet files in MinIO ReconciliationScheduler → three scheduled safety checks → reconciliation_alerts + RabbitMQ ``` @@ -182,6 +183,14 @@ VigilCareClinicalAPI/ │ └── RabbitMqTopologyProvisioner.cs # Declares exchange, queues, DLQ bindings on startup ├── Storage/ │ └── MinioClientFactory.cs +├── DataLake/ +│ ├── DataLakeOptions.cs # Flush thresholds and bucket settings +│ ├── DataLakeWriterService.cs # consumer group: data-lake-writer; Kafka → Parquet → MinIO +│ └── ParquetFileBuilder.cs # Topic row models → Parquet byte arrays +├── Models/Records/ +│ ├── Observation/ObservationRow.cs # Parquet row contract for observation events +│ ├── Alert/AlertRow.cs # Parquet row contract for alert events +│ └── Encounter/EncounterStatusRow.cs # Parquet row contract for encounter status events ├── Data/ │ ├── AppDbContext.cs # EF Core context — entity configs, indexes, constraints │ ├── Configurations/ # IEntityTypeConfiguration per entity @@ -213,7 +222,9 @@ tests/ ├── SirsDetectorTests.cs # Redis SIRS state SET/DEL/MGET logic ├── SirsEvaluatorTests.cs # Per-code criterion evaluation ├── NotificationPipelineTests.cs # RabbitMQ topology, DLQ routing - └── ReconciliationTests.cs # Three reconciliation checks, deduplication, RabbitMQ publish + ├── ReconciliationTests.cs # Three reconciliation checks, deduplication, RabbitMQ publish + ├── ObservabilityPhase8Tests.cs # /metrics families and correlation header behavior + └── DataLakePhase9Tests.cs # Kafka → MinIO Parquet flow and schema checks ``` --- @@ -817,4 +828,4 @@ Observation history uses cursor pagination on `(recorded_at DESC, id DESC)`. Off | 6 | RabbitMQ exchange and queue topology; `NotificationPublisherService`; `PagingWorkerService`; DLQ escalation (`EscalationWorkerService`); discharge summary (`DischargeSummaryWorkerService` → MinIO); integration tests | Done | | 7 | Reconciliation scheduler — unacknowledged critical alerts, stale pending orders, disconnected monitors; `reconciliation_alerts` table; RabbitMQ publish; integration tests | Done | | 8 | Prometheus metrics (`GET /metrics`); Grafana dashboards; eight application metric families | In progress | -| 9 | Data lake writer — Kafka consumer group `data-lake-writer`; Parquet flush to MinIO | Planned | +| 9 | Data lake writer — Kafka consumer group `data-lake-writer`; Parquet flush to MinIO; integration tests (`DataLakePhase9Tests`) | In progress | diff --git a/VigilCareClinicalAPI.Tests/DataLakePhase9Tests.cs b/VigilCareClinicalAPI.Tests/DataLakePhase9Tests.cs new file mode 100644 index 0000000..899abee --- /dev/null +++ b/VigilCareClinicalAPI.Tests/DataLakePhase9Tests.cs @@ -0,0 +1,249 @@ +using System.Net.Http.Json; +using System.Text.Json; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Minio.DataModel.Args; +using Parquet; +using StackExchange.Redis; + +public class DataLakePhase9Tests : IClassFixture +{ + private readonly ApiFixture _fixture; + private readonly HttpClient _http; + private readonly MinioOptions _minioOpts; + + public DataLakePhase9Tests(ApiFixture fixture) + { + _fixture = fixture; + _http = fixture.CreateClient(); + _minioOpts = fixture.Services + .GetRequiredService>().Value; + } + + [Fact] + public async Task ObservationEvents_FlushesToParquetInMinIO() + { + await EnsureThresholdsAsync(); + var patientId = await CreatePatientAsync(); + var encounterId = await CreateActiveEncounterAsync(patientId); + + for (var i = 0; i < 3; i++) + { + var resp = await _http.PostAsJsonAsync( + $"/api/v1/encounters/{encounterId}/observations", + new BatchIngestRequest(new List + { + new("HEART_RATE", 72 + i, "bpm", ObservationSource.Device, DateTimeOffset.UtcNow, null) + })); + resp.EnsureSuccessStatusCode(); + } + + await Task.Delay(TimeSpan.FromSeconds(20)); + + var keys = await ListMinioObjectsAsync("observations/"); + Assert.True(keys.Count > 0, + $"No Parquet files found under observations/ in MinIO bucket '{_minioOpts.BucketName}'."); + Assert.All(keys, k => Assert.EndsWith(".parquet", k)); + } + + [Fact] + public async Task AlertEvents_FlushesToParquetInMinIO() + { + await EnsureThresholdsAsync(); + for (var i = 0; i < 3; i++) + { + var pid = await CreatePatientAsync(); + var eid = await CreateActiveEncounterAsync(pid); + + var resp = await _http.PostAsJsonAsync( + $"/api/v1/encounters/{eid}/observations", + new BatchIngestRequest(new List + { + new("POTASSIUM_MEQ_L", 2.1m, "mEq/L", ObservationSource.Device, DateTimeOffset.UtcNow, null) + })); + resp.EnsureSuccessStatusCode(); + } + + await Task.Delay(TimeSpan.FromSeconds(20)); + + var keys = await ListMinioObjectsAsync("alerts/"); + Assert.True(keys.Count > 0, + $"No Parquet files found under alerts/ in MinIO bucket '{_minioOpts.BucketName}'."); + } + + [Fact] + public async Task EncounterStatusEvents_FlushesToParquetInMinIO() + { + await EnsureThresholdsAsync(); + for (var i = 0; i < 3; i++) + { + var pid = await CreatePatientAsync(); + _ = await CreateActiveEncounterAsync(pid); + } + + await Task.Delay(TimeSpan.FromSeconds(20)); + + var keys = await ListMinioObjectsAsync("encounters/"); + Assert.True(keys.Count > 0, + $"No Parquet files found under encounters/ in MinIO bucket '{_minioOpts.BucketName}'."); + } + + [Fact] + public async Task ObservationParquetFile_ContainsCorrectColumns() + { + await EnsureThresholdsAsync(); + var patientId = await CreatePatientAsync(); + var encounterId = await CreateActiveEncounterAsync(patientId); + + for (var i = 0; i < 3; i++) + { + var resp = await _http.PostAsJsonAsync( + $"/api/v1/encounters/{encounterId}/observations", + new BatchIngestRequest(new List + { + new("TEMP_C", 37.5m, "°C", ObservationSource.Device, DateTimeOffset.UtcNow, null) + })); + resp.EnsureSuccessStatusCode(); + } + + await Task.Delay(TimeSpan.FromSeconds(20)); + + var keys = await ListMinioObjectsAsync("observations/"); + Assert.True(keys.Count > 0); + + var bytes = await DownloadMinioObjectAsync(keys[0]); + Assert.True(bytes.Length > 0); + + using var ms = new MemoryStream(bytes); + using var reader = await ParquetReader.CreateAsync(ms); + + var columnNames = reader.Schema.DataFields.Select(f => f.Name).ToHashSet(); + Assert.Contains("observation_id", columnNames); + Assert.Contains("encounter_id", columnNames); + Assert.Contains("observation_code", columnNames); + Assert.Contains("value", columnNames); + Assert.Contains("kafka_partition", columnNames); + Assert.Contains("kafka_offset", columnNames); + } + + private async Task> ListMinioObjectsAsync(string prefix) + { + var client = MinioClientFactory.Build(_minioOpts); + var keys = new List(); + + var listArgs = new ListObjectsArgs() + .WithBucket(_minioOpts.BucketName) + .WithPrefix(prefix) + .WithRecursive(true); + + await foreach (var item in client.ListObjectsEnumAsync(listArgs)) + { + keys.Add(item.Key); + } + + return keys; + } + + private async Task DownloadMinioObjectAsync(string objectKey) + { + var client = MinioClientFactory.Build(_minioOpts); + using var ms = new MemoryStream(); + + await client.GetObjectAsync(new GetObjectArgs() + .WithBucket(_minioOpts.BucketName) + .WithObject(objectKey) + .WithCallbackStream(stream => stream.CopyTo(ms))); + + return ms.ToArray(); + } + + private async Task CreatePatientAsync() + { + var resp = await _http.PostAsJsonAsync("/api/v1/patients", new + { + firstName = "DataLake", + lastName = "Test", + dateOfBirth = "1980-11-12", + gender = "Female", + }); + resp.EnsureSuccessStatusCode(); + + var body = await resp.Content.ReadFromJsonAsync(); + return body!.RootElement.GetProperty("data").GetProperty("id").GetGuid(); + } + + private async Task CreateActiveEncounterAsync(Guid patientId) + { + var resp = await _http.PostAsJsonAsync( + $"/api/v1/patients/{patientId}/encounters", + new + { + encounterType = "INPATIENT", + department = "ICU", + attendingPhysician = "Dr. Osei", + admittedAt = DateTimeOffset.UtcNow, + }); + resp.EnsureSuccessStatusCode(); + + var body = await resp.Content.ReadFromJsonAsync(); + var encounterId = body!.RootElement.GetProperty("data").GetProperty("id").GetGuid(); + + // Encounter may be created as Scheduled depending on fixture seed path; move it + // to Active if required, but tolerate Conflict when already Active. + var activateResp = await _http.PatchAsJsonAsync( + $"/api/v1/encounters/{encounterId}/status", + new { status = "Active" }); + if (activateResp.StatusCode != System.Net.HttpStatusCode.Conflict) + activateResp.EnsureSuccessStatusCode(); + + return encounterId; + } + + private async Task EnsureThresholdsAsync() + { + using var scope = _fixture.Services.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + + await EnsureThresholdAsync(db, "HEART_RATE", "Heart Rate", "bpm", 30, 50, 100, 150); + await EnsureThresholdAsync(db, "POTASSIUM_MEQ_L", "Serum Potassium", "mEq/L", 2.5m, 3.5m, 5.0m, 6.5m); + await EnsureThresholdAsync(db, "TEMP_C", "Temperature", "°C", 34m, 36m, 37.8m, 40m); + await db.SaveChangesAsync(); + + var redis = scope.ServiceProvider.GetRequiredService(); + var cache = redis.GetDatabase(1); + await cache.StringSetAsync("threshold:HEART_RATE", + """{"ObservationCode":"HEART_RATE","CriticalLow":30,"WarningLow":50,"WarningHigh":100,"CriticalHigh":150}"""); + await cache.StringSetAsync("threshold:POTASSIUM_MEQ_L", + """{"ObservationCode":"POTASSIUM_MEQ_L","CriticalLow":2.5,"WarningLow":3.5,"WarningHigh":5.0,"CriticalHigh":6.5}"""); + await cache.StringSetAsync("threshold:TEMP_C", + """{"ObservationCode":"TEMP_C","CriticalLow":34,"WarningLow":36,"WarningHigh":37.8,"CriticalHigh":40}"""); + } + + private static async Task EnsureThresholdAsync( + AppDbContext db, + string code, + string displayName, + string unit, + decimal criticalLow, + decimal warningLow, + decimal warningHigh, + decimal criticalHigh) + { + var existing = await db.AlertThresholds.FirstOrDefaultAsync(t => t.ObservationCode == code); + if (existing is not null) return; + + db.AlertThresholds.Add(new AlertThreshold + { + Id = Guid.NewGuid(), + ObservationCode = code, + DisplayName = displayName, + Unit = unit, + CriticalLow = criticalLow, + WarningLow = warningLow, + WarningHigh = warningHigh, + CriticalHigh = criticalHigh, + CreatedAt = DateTimeOffset.UtcNow + }); + } +} diff --git a/VigilCareClinicalAPI/DataLake/DataLakeOptions.cs b/VigilCareClinicalAPI/DataLake/DataLakeOptions.cs new file mode 100644 index 0000000..4b9a8e8 --- /dev/null +++ b/VigilCareClinicalAPI/DataLake/DataLakeOptions.cs @@ -0,0 +1,17 @@ +public sealed class DataLakeOptions +{ + public const string Section = "DataLake"; + + // Maximum events buffered before a forced flush. At ~17 obs/sec steady state + // this fires approximately once per minute. Set low in tests (3–5) to avoid + // waiting for real traffic. + public int FlushCount { get; init; } = 1_000; + + // Maximum age of the oldest buffered event before a time-based flush is forced. + // Production: 300 seconds (5 minutes). Tests: 10 seconds. + public int FlushIntervalSeconds { get; init; } = 300; + + // MinIO bucket. The discharge summary worker from Phase 6 uses the same bucket + // under a different prefix — all VigilCare data stays in one bucket. + public string BucketName { get; init; } = "vigilcare"; +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/DataLake/DataLakeWriterService.cs b/VigilCareClinicalAPI/DataLake/DataLakeWriterService.cs new file mode 100644 index 0000000..1e1b46c --- /dev/null +++ b/VigilCareClinicalAPI/DataLake/DataLakeWriterService.cs @@ -0,0 +1,302 @@ +using System.Text.Json; +using Confluent.Kafka; +using Microsoft.Extensions.Options; +using Minio.DataModel.Args; + +public sealed class DataLakeWriterService : BackgroundService +{ + private readonly KafkaOptions _kafkaOptions; + private readonly DataLakeOptions _opts; + private readonly MinioOptions _minioOpts; + private readonly ILogger _logger; + + // Buffer key: identifies one Parquet file-to-be. + // Events sharing a topic, date, and Kafka partition land in the same file. + private record BufferKey(string Topic, string DatePath, int Partition); + + private record BufferedEvent(string Payload, long Offset); + + private readonly Dictionary> _buffer = new(); + // Track the highest offset per topic-partition for post-flush commit. + private readonly Dictionary _highWatermarks = new(); + + public DataLakeWriterService( + IOptions kafkaOptions, + IOptions opts, + IOptions minioOpts, + ILogger logger) + { + _kafkaOptions = kafkaOptions.Value; + _opts = opts.Value; + _minioOpts = minioOpts.Value; + _logger = logger; + } + + protected override async Task ExecuteAsync(CancellationToken ct) + { + var consumerConfig = new ConsumerConfig + { + BootstrapServers = _kafkaOptions.BootstrapServers, + GroupId = "data-lake-writer", + AutoOffsetReset = AutoOffsetReset.Earliest, + EnableAutoCommit = false, + }; + + using var consumer = new ConsumerBuilder(consumerConfig).Build(); + consumer.Subscribe(new[] + { + _kafkaOptions.Topics.ObservationRecorded, + _kafkaOptions.Topics.AlertGenerated, + _kafkaOptions.Topics.EncounterStatusChanged, + }); + + _logger.LogInformation( + "DataLakeWriterService started. FlushCount={FlushCount} FlushIntervalSeconds={FlushInterval}", + _opts.FlushCount, _opts.FlushIntervalSeconds); + + var lastFlush = DateTimeOffset.UtcNow; + + try + { + while (!ct.IsCancellationRequested) + { + ConsumeResult? result; + try + { + result = consumer.Consume(TimeSpan.FromMilliseconds(500)); + } + catch (ConsumeException ex) + { + _logger.LogError(ex, "DataLakeWriter consume error"); + continue; + } + + if (result is not null) + AddToBuffer(result); + + var totalBuffered = _buffer.Values.Sum(v => v.Count); + var shouldFlushCount = totalBuffered >= _opts.FlushCount; + var shouldFlushTime = DateTimeOffset.UtcNow - lastFlush + >= TimeSpan.FromSeconds(_opts.FlushIntervalSeconds); + + if ((shouldFlushCount || shouldFlushTime) && totalBuffered > 0) + { + await FlushAsync(consumer, ct); + lastFlush = DateTimeOffset.UtcNow; + } + } + } + finally + { + // Final flush on shutdown so buffered events are not lost. + if (_buffer.Values.Sum(v => v.Count) > 0) + { + try { await FlushAsync(consumer, ct); } + catch (Exception ex) + { + _logger.LogError(ex, "DataLakeWriter shutdown flush failed — some events may be re-read on next start"); + } + } + consumer.Close(); + } + } + + private void AddToBuffer(ConsumeResult result) + { + var datePath = ExtractDatePath(result.Topic, result.Message.Value); + var key = new BufferKey(result.Topic, datePath, result.Partition.Value); + + if (!_buffer.TryGetValue(key, out var list)) + { + list = new List(); + _buffer[key] = list; + } + list.Add(new BufferedEvent(result.Message.Value, result.Offset.Value)); + + // Track highest offset per topic-partition for post-flush commit. + var tp = new TopicPartition(result.Topic, result.Partition); + _highWatermarks[tp] = new TopicPartitionOffset(tp, result.Offset + 1); + } + + private async Task FlushAsync(IConsumer consumer, CancellationToken ct) + { + var filesWritten = 0; + + foreach (var (key, events) in _buffer) + { + if (events.Count == 0) continue; + + try + { + var firstOffset = events.Min(e => e.Offset); + var objectKey = BuildObjectKey(key, firstOffset); + var bytes = await BuildParquetAsync(key.Topic, events, key.Partition); + + await UploadToMinioAsync(objectKey, bytes, ct); + filesWritten++; + + _logger.LogInformation( + "[DATA-LAKE] Wrote {Count} events → {ObjectKey} ({Bytes} bytes)", + events.Count, objectKey, bytes.Length); + } + catch (Exception ex) + { + // Log and continue — a failed file for one key must not prevent other + // keys from flushing. The uncommitted offsets will cause reprocessing. + _logger.LogError(ex, "[DATA-LAKE] Failed to write file for key {Key}", key); + } + } + + // Commit only after all files are uploaded. + // Events for any key that failed above will be re-read on next startup. + if (_highWatermarks.Any()) + { + consumer.Commit(_highWatermarks.Values); + _logger.LogInformation( + "[DATA-LAKE] Committed offsets for {PartitionCount} partitions after flushing {FileCount} files", + _highWatermarks.Count, filesWritten); + } + + _buffer.Clear(); + _highWatermarks.Clear(); + } + + private async Task BuildParquetAsync( + string topic, List events, int partition) + { + if (topic == _kafkaOptions.Topics.ObservationRecorded) + { + var rows = events.Select(e => ParseObservationRow(e, partition)).ToList(); + return await ParquetFileBuilder.BuildObservationsAsync(rows); + } + if (topic == _kafkaOptions.Topics.AlertGenerated) + { + var rows = events.Select(e => ParseAlertRow(e, partition)).ToList(); + return await ParquetFileBuilder.BuildAlertsAsync(rows); + } + if (topic == _kafkaOptions.Topics.EncounterStatusChanged) + { + var rows = events.Select(e => ParseEncounterRow(e, partition)).ToList(); + return await ParquetFileBuilder.BuildEncountersAsync(rows); + } + throw new InvalidOperationException($"Unknown topic: {topic}"); + } + + // --------------------------------------------------------------------------- + // Object key and date partition helpers + // --------------------------------------------------------------------------- + + // File path: observations/2025/01/15/partition-0-offset-0000001000.parquet + // The date comes from the event timestamp, not the wall clock. + // Events from the same encounter that cross midnight are written into the date + // bucket matching their recorded_at timestamp — consistent with how Athena and + // Spark partition-prune by event time, not ingest time. + private string BuildObjectKey(BufferKey key, long firstOffset) + { + var folder = key.Topic switch + { + var t when t == _kafkaOptions.Topics.ObservationRecorded => "observations", + var t when t == _kafkaOptions.Topics.AlertGenerated => "alerts", + var t when t == _kafkaOptions.Topics.EncounterStatusChanged => "encounters", + _ => "unknown", + }; + 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.GetProperty("details").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 + // --------------------------------------------------------------------------- + + private async Task UploadToMinioAsync(string objectKey, byte[] bytes, CancellationToken ct) + { + var client = MinioClientFactory.Build(_minioOpts); + var bucket = _opts.BucketName; + + var exists = await client.BucketExistsAsync( + new BucketExistsArgs().WithBucket(bucket), ct); + if (!exists) + await client.MakeBucketAsync(new MakeBucketArgs().WithBucket(bucket), ct); + + await client.PutObjectAsync(new PutObjectArgs() + .WithBucket(bucket) + .WithObject(objectKey) + .WithStreamData(new MemoryStream(bytes)) + .WithObjectSize(bytes.Length) + .WithContentType("application/octet-stream"), + ct); + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/DataLake/ParquetFileBuilder.cs b/VigilCareClinicalAPI/DataLake/ParquetFileBuilder.cs new file mode 100644 index 0000000..54aeb33 --- /dev/null +++ b/VigilCareClinicalAPI/DataLake/ParquetFileBuilder.cs @@ -0,0 +1,102 @@ +using Parquet; +using Parquet.Data; +using Parquet.Schema; + +public static class ParquetFileBuilder +{ + public static async Task BuildObservationsAsync(IReadOnlyList rows) + { + var schema = new ParquetSchema( + new DataField("observation_id"), + new DataField("encounter_id"), + new DataField("patient_id"), + new DataField("mrn"), + new DataField("observation_code"), + new DataField("value"), + new DataField("unit"), + new DataField("source"), + new DataField("recorded_at"), + new DataField("kafka_partition"), + new DataField("kafka_offset") + ); + + using var ms = new MemoryStream(); + using (var writer = await ParquetWriter.CreateAsync(schema, ms)) + using (var rg = writer.CreateRowGroup()) + { + var f = schema.DataFields; + await rg.WriteColumnAsync(new DataColumn(f[0], rows.Select(r => r.ObservationId).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[1], rows.Select(r => r.EncounterId).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[2], rows.Select(r => r.PatientId).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[3], rows.Select(r => r.Mrn).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[4], rows.Select(r => r.ObservationCode).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[5], rows.Select(r => r.Value).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[6], rows.Select(r => r.Unit).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[7], rows.Select(r => r.Source).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[8], rows.Select(r => r.RecordedAt).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[9], rows.Select(r => r.KafkaPartition).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[10], rows.Select(r => r.KafkaOffset).ToArray())); + } + return ms.ToArray(); + } + + public static async Task BuildAlertsAsync(IReadOnlyList rows) + { + var schema = new ParquetSchema( + new DataField("alert_id"), + new DataField("encounter_id"), + new DataField("patient_id"), + new DataField("alert_type"), + new DataField("severity"), + new DataField("details"), + new DataField("triggered_at"), + new DataField("kafka_partition"), + new DataField("kafka_offset") + ); + + using var ms = new MemoryStream(); + using (var writer = await ParquetWriter.CreateAsync(schema, ms)) + using (var rg = writer.CreateRowGroup()) + { + var f = schema.DataFields; + await rg.WriteColumnAsync(new DataColumn(f[0], rows.Select(r => r.AlertId).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[1], rows.Select(r => r.EncounterId).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[2], rows.Select(r => r.PatientId).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[3], rows.Select(r => r.AlertType).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[4], rows.Select(r => r.Severity).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[5], rows.Select(r => r.Details).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[6], rows.Select(r => r.TriggeredAt).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[7], rows.Select(r => r.KafkaPartition).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[8], rows.Select(r => r.KafkaOffset).ToArray())); + } + return ms.ToArray(); + } + + public static async Task BuildEncountersAsync(IReadOnlyList rows) + { + var schema = new ParquetSchema( + new DataField("encounter_id"), + new DataField("patient_id"), + new DataField("previous_status"), + new DataField("new_status"), + new DataField("changed_at"), + new DataField("kafka_partition"), + new DataField("kafka_offset") + ); + + using var ms = new MemoryStream(); + using (var writer = await ParquetWriter.CreateAsync(schema, ms)) + using (var rg = writer.CreateRowGroup()) + { + var f = schema.DataFields; + await rg.WriteColumnAsync(new DataColumn(f[0], rows.Select(r => r.EncounterId).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[1], rows.Select(r => r.PatientId).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[2], rows.Select(r => r.PreviousStatus).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[3], rows.Select(r => r.NewStatus).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[4], rows.Select(r => r.ChangedAt).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[5], rows.Select(r => r.KafkaPartition).ToArray())); + await rg.WriteColumnAsync(new DataColumn(f[6], rows.Select(r => r.KafkaOffset).ToArray())); + } + return ms.ToArray(); + } +} \ No newline at end of file diff --git a/VigilCareClinicalAPI/Models/Records/Alert/AlertRow.cs b/VigilCareClinicalAPI/Models/Records/Alert/AlertRow.cs new file mode 100644 index 0000000..7e04659 --- /dev/null +++ b/VigilCareClinicalAPI/Models/Records/Alert/AlertRow.cs @@ -0,0 +1,11 @@ +public sealed record AlertRow( + string AlertId, + string EncounterId, + string PatientId, + string AlertType, + string Severity, + string Details, + string TriggeredAt, + int KafkaPartition, + long KafkaOffset +); \ No newline at end of file diff --git a/VigilCareClinicalAPI/Models/Records/Encounter/EncounterStatusRow.cs b/VigilCareClinicalAPI/Models/Records/Encounter/EncounterStatusRow.cs new file mode 100644 index 0000000..77051d4 --- /dev/null +++ b/VigilCareClinicalAPI/Models/Records/Encounter/EncounterStatusRow.cs @@ -0,0 +1,9 @@ +public sealed record EncounterStatusRow( + string EncounterId, + string PatientId, + string PreviousStatus, + string NewStatus, + string ChangedAt, + int KafkaPartition, + long KafkaOffset +); \ No newline at end of file diff --git a/VigilCareClinicalAPI/Models/Records/Observation/ObservationRow.cs b/VigilCareClinicalAPI/Models/Records/Observation/ObservationRow.cs new file mode 100644 index 0000000..a417764 --- /dev/null +++ b/VigilCareClinicalAPI/Models/Records/Observation/ObservationRow.cs @@ -0,0 +1,13 @@ +public sealed record ObservationRow( + string ObservationId, + string EncounterId, + string PatientId, + string Mrn, + string ObservationCode, + double Value, + string Unit, + string Source, + string RecordedAt, + int KafkaPartition, + long KafkaOffset +); \ No newline at end of file diff --git a/VigilCareClinicalAPI/Program.cs b/VigilCareClinicalAPI/Program.cs index 6c9d2e1..617aea4 100644 --- a/VigilCareClinicalAPI/Program.cs +++ b/VigilCareClinicalAPI/Program.cs @@ -55,6 +55,9 @@ try builder.Services.Configure( builder.Configuration.GetSection(ReconciliationJobOptions.Section)); + builder.Services.Configure( + builder.Configuration.GetSection(DataLakeOptions.Section)); + builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); @@ -84,6 +87,7 @@ try builder.Services.AddHostedService(); builder.Services.AddHostedService(); builder.Services.AddHostedService(); + builder.Services.AddHostedService(); builder.Services.AddControllers() .AddJsonOptions(opts => diff --git a/VigilCareClinicalAPI/VigilCareClinicalAPI.csproj b/VigilCareClinicalAPI/VigilCareClinicalAPI.csproj index 0350c92..d6ac878 100644 --- a/VigilCareClinicalAPI/VigilCareClinicalAPI.csproj +++ b/VigilCareClinicalAPI/VigilCareClinicalAPI.csproj @@ -18,6 +18,7 @@ + diff --git a/VigilCareClinicalAPI/appsettings.Testing.json b/VigilCareClinicalAPI/appsettings.Testing.json index a6617e6..4d89225 100644 --- a/VigilCareClinicalAPI/appsettings.Testing.json +++ b/VigilCareClinicalAPI/appsettings.Testing.json @@ -13,5 +13,9 @@ }, "RabbitMq": { "PagingAckTimeoutMs": 5000 + }, + "DataLake": { + "FlushCount": 3, + "FlushIntervalSeconds": 10 } } diff --git a/VigilCareClinicalAPI/appsettings.json b/VigilCareClinicalAPI/appsettings.json index e3d3f05..6b1276c 100644 --- a/VigilCareClinicalAPI/appsettings.json +++ b/VigilCareClinicalAPI/appsettings.json @@ -74,5 +74,9 @@ "UnacknowledgedAlertThresholdMinutes": 30, "PendingOrderThresholdHours": 4, "NoObservationThresholdHours": 2 + }, + "DataLake": { + "FlushCount": 1000, + "FlushIntervalSeconds": 300 } } diff --git a/docs/decisions/data-lake-design.md b/docs/decisions/data-lake-design.md new file mode 100644 index 0000000..ab8521a --- /dev/null +++ b/docs/decisions/data-lake-design.md @@ -0,0 +1,87 @@ +```markdown +# Decision: Data Lake Architecture — Parquet on MinIO + +## Context + +Healthcare regulations in most jurisdictions require medical records to be retained +for 7–25 years. A medium hospital with 200 concurrent inpatients generates roughly +17 observations per second at steady state. Over 10 years that is approximately +5 billion observation rows. Keeping this in PostgreSQL would require continuous +partitioning and archival work; running population health queries against it would +compete with live ingest writes. + +## Decision + +A separate data lake tier handles long-term retention. The Kafka `data-lake-writer` +consumer group reads all three topics and writes Parquet files to MinIO under +date-partitioned prefixes. + +## Why Parquet over JSON or CSV + +Parquet is columnar. A population health query — "give me all heart rate values for +ICU patients in 2024" — reads only the `observation_code` and `value` columns without +deserializing `encounter_id`, `patient_id`, `unit`, `source`, or `recorded_at`. + +The repetition of `observation_code` values across millions of rows (thousands of rows +all with `HEART_RATE`) compresses at 5–10× vs JSON via dictionary encoding. At 10-year +scale this difference is measured in terabytes of storage cost. + +CSV is row-oriented like JSON and does not support schema evolution — adding a new +column requires rewriting every historical file. + +## Why MinIO (S3-compatible) over PostgreSQL or Elasticsearch for archive + +PostgreSQL is the operational write layer. Running 10-year-scale queries on the same +instance that serves live ingest introduces contention. Adding partitioning and tiered +storage to PostgreSQL adds operational complexity without solving the fundamental +problem: it is still a row store. + +Elasticsearch is optimized for search and aggregations, not for full-table scans or +columnar projections. Storing 5 billion observation rows in Elasticsearch would require +enormous index memory and produce no benefit for the analytics pattern that justifies +the archive (population health over multi-year windows). + +MinIO is S3-compatible. The files it stores can be queried directly by DuckDB +(single analyst), Apache Spark (cluster analytics), and AWS Athena (serverless queries +over S3) without any data movement. Switching from MinIO to S3 in production requires +changing one endpoint URL. + +## Why not Apache Flink or Spark Streaming for the lake writer + +At a single-hospital scale — 50 obs/sec peak, 6 Kafka partitions — the consumer lag +from a simple .NET BackgroundService with in-memory buffering and a 5-minute flush is +negligible. The overhead of deploying and operating a Flink cluster adds operational +cost that is not justified by the throughput. + +The scale inflection point: approximately 10,000+ events/sec sustained, or the need +for exactly-once semantics at the storage layer. Below that, the simple consumer is +correct. Above it, Flink's checkpoint-based exactly-once delivery to Parquet +(via its FileSystem sink) becomes worth the operational cost. + +## Flush policy rationale + +**Count-based flush (1,000 events):** Bounds memory. At 17 obs/sec steady state this +fires approximately once per minute. At 50 obs/sec peak it fires every 20 seconds. + +**Time-based flush (5 minutes):** Bounds latency. At low load (night shift, few +monitors active), the count threshold might not fire for hours. A 5-minute ceiling +means the most recent data is always queryable within 5 minutes of arriving in Kafka. + +**At-least-once delivery:** Kafka offsets are committed only after the Parquet file +is successfully uploaded. A process crash between buffering and uploading produces +duplicate rows on the next startup — the same observation appears in two files with +different `kafka_offset` values. For a regulatory archive this is acceptable. +Downstream queries can deduplicate on `observation_id`. Data loss is not acceptable; +duplicates are. + +## Date partition structure + +Files are partitioned by the **event timestamp** from the payload, not by wall clock +time at consumption. An observation `recorded_at 2025-01-14T23:59:59Z` consumed at +`2025-01-15T00:01:00Z` is written to `observations/2025/01/14/...`. + +This matches how Athena and Spark apply partition pruning: a query with +`WHERE recorded_at BETWEEN '2025-01-14' AND '2025-01-14'` reads only the +`observations/2025/01/14/` prefix. Partitioning by ingest time instead would cause +the query to miss that row. +``` \ No newline at end of file