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"), ExplanationJson: GetJsonObjectString(d, "explanation"), 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 GetJsonObjectString(JsonElement d, string name) { if (!TryGetProperty(d, name, out var prop) || prop.ValueKind != JsonValueKind.Object) return ""; return prop.GetRawText(); } 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() }; }