From c3fbc20ddc6ca1ddb02b75b7ecd433739146ea24 Mon Sep 17 00:00:00 2001
From: voltsrage <electric5388@hotmail.com>
Date: Thu, 18 Jun 2026 16:28:58 +0800
Subject: [PATCH] fix: Shutdown causes issues in datalake and minio

---
 .../DataLake/DataLakeWriterService.cs         | 26 +++++++++++++++----
 VigilCareClinicalAPI/Sepsis/SirsDetector.cs   |  1 +
 .../Services/WarningEvaluator.cs              |  1 +
 3 files changed, 23 insertions(+), 5 deletions(-)

diff --git a/VigilCareClinicalAPI/DataLake/DataLakeWriterService.cs b/VigilCareClinicalAPI/DataLake/DataLakeWriterService.cs
index 1e1b46c..1d8ab94 100644
--- a/VigilCareClinicalAPI/DataLake/DataLakeWriterService.cs
+++ b/VigilCareClinicalAPI/DataLake/DataLakeWriterService.cs
@@ -89,9 +89,11 @@ public sealed class DataLakeWriterService : BackgroundService
         finally
         {
             // Final flush on shutdown so buffered events are not lost.
+            // Use CancellationToken.None — host shutdown cancels the execute token before
+            // MinIO uploads finish, which surfaces as TaskCanceledException noise.
             if (_buffer.Values.Sum(v => v.Count) > 0)
             {
-                try { await FlushAsync(consumer, ct); }
+                try { await FlushAsync(consumer, CancellationToken.None); }
                 catch (Exception ex)
                 {
                     _logger.LogError(ex, "DataLakeWriter shutdown flush failed — some events may be re-read on next start");
@@ -143,19 +145,33 @@ public sealed class DataLakeWriterService : BackgroundService
             {
                 // 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);
+                if (ex is OperationCanceledException && ct.IsCancellationRequested)
+                {
+                    _logger.LogInformation(
+                        "[DATA-LAKE] Flush canceled for key {Key} during shutdown", key);
+                }
+                else
+                {
+                    _logger.LogError(ex, "[DATA-LAKE] Failed to write file for key {Key}", key);
+                }
             }
         }
 
-        // Commit only after all files are uploaded.
+        // Commit only after at least one file uploaded successfully.
         // Events for any key that failed above will be re-read on next startup.
-        if (_highWatermarks.Any())
+        if (filesWritten > 0 && _highWatermarks.Any())
         {
             consumer.Commit(_highWatermarks.Values);
             _logger.LogInformation(
                 "[DATA-LAKE] Committed offsets for {PartitionCount} partitions after flushing {FileCount} files",
                 _highWatermarks.Count, filesWritten);
         }
+        else if (filesWritten == 0 && _highWatermarks.Any())
+        {
+            _logger.LogWarning(
+                "[DATA-LAKE] Skipping offset commit — no files were written ({BufferedPartitions} partitions buffered)",
+                _highWatermarks.Count);
+        }
 
         _buffer.Clear();
         _highWatermarks.Clear();
@@ -256,7 +272,7 @@ public sealed class DataLakeWriterService : BackgroundService
             PatientId      : d.GetProperty("patientId").GetString()   ?? "",
             AlertType      : d.GetProperty("alertType").GetString()   ?? "",
             Severity       : d.GetProperty("severity").GetString()    ?? "",
-            Details        : d.GetProperty("details").GetString()     ?? "",
+            Details        : d.TryGetProperty("details", out var det) ? det.GetString() ?? "" : "",
             TriggeredAt    : d.GetProperty("triggeredAt").GetString() ?? "",
             KafkaPartition : partition,
             KafkaOffset    : e.Offset
diff --git a/VigilCareClinicalAPI/Sepsis/SirsDetector.cs b/VigilCareClinicalAPI/Sepsis/SirsDetector.cs
index 4b8c57f..8305cbe 100644
--- a/VigilCareClinicalAPI/Sepsis/SirsDetector.cs
+++ b/VigilCareClinicalAPI/Sepsis/SirsDetector.cs
@@ -145,6 +145,7 @@ public class SirsDetector
                 patientId,
                 alertType = AlertType.SepsisWarning.ToDbString(),
                 severity  = "Critical",
+                details,
                 triggeredAt,
                 partitionKey = encounterId.ToString()
             }),
diff --git a/VigilCareClinicalAPI/Services/WarningEvaluator.cs b/VigilCareClinicalAPI/Services/WarningEvaluator.cs
index cbb03d7..716dbe4 100644
--- a/VigilCareClinicalAPI/Services/WarningEvaluator.cs
+++ b/VigilCareClinicalAPI/Services/WarningEvaluator.cs
@@ -111,6 +111,7 @@ public class WarningEvaluator
                 patientId,
                 alertType = alertType.ToDbString(),
                 severity = "Warning",
+                details,
                 triggeredAt,
                 partitionKey = encounterId.ToString()
             }),