diff --git a/src/Agent/AutopilotMonitor.Agent/buildcounter.txt b/src/Agent/AutopilotMonitor.Agent/buildcounter.txt
index c0556fb2..4d0e90cb 100644
--- a/src/Agent/AutopilotMonitor.Agent/buildcounter.txt
+++ b/src/Agent/AutopilotMonitor.Agent/buildcounter.txt
@@ -1 +1 @@
-511
+512
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/IngestCriticalPathTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/IngestCriticalPathTests.cs
new file mode 100644
index 00000000..95ef7eb8
--- /dev/null
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/IngestCriticalPathTests.cs
@@ -0,0 +1,221 @@
+using AutopilotMonitor.Functions.Functions.Ingest;
+using AutopilotMonitor.Functions.Security;
+using AutopilotMonitor.Shared.Models;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+///
+/// Tests for the critical ingest processing path.
+///
+/// REGRESSION GUARD: On 2026-03-19 the ingest endpoint returned 500 for all devices because
+/// individual events did not have TenantId/SessionId set before being passed to
+/// StoreEventsBatchAsync, which validates them as GUIDs.
+/// StampServerFields (extracted from ProcessIngestAsync) is the fix — these tests ensure
+/// it can never be accidentally removed or bypassed.
+///
+public class IngestCriticalPathTests
+{
+ private static readonly string ValidTenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
+ private static readonly string ValidSessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
+
+ // -------------------------------------------------------------------------
+ // StampServerFields — THE regression test
+ // -------------------------------------------------------------------------
+
+ [Fact]
+ public void StampServerFields_SetsReceivedAtOnAllEvents()
+ {
+ var receivedAt = DateTime.UtcNow;
+ var events = MakeEvents(3);
+
+ IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, receivedAt);
+
+ Assert.All(events, evt => Assert.Equal(receivedAt, evt.ReceivedAt));
+ }
+
+ [Fact]
+ public void StampServerFields_SetsTenantIdOnAllEvents()
+ {
+ var events = MakeEvents(3);
+
+ IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, DateTime.UtcNow);
+
+ Assert.All(events, evt => Assert.Equal(ValidTenantId, evt.TenantId));
+ }
+
+ [Fact]
+ public void StampServerFields_SetsSessionIdOnAllEvents()
+ {
+ var events = MakeEvents(3);
+
+ IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, DateTime.UtcNow);
+
+ Assert.All(events, evt => Assert.Equal(ValidSessionId, evt.SessionId));
+ }
+
+ [Fact]
+ public void StampServerFields_OverridesNullTenantId()
+ {
+ // Regression: agent may not set TenantId per-event (e.g. new event types, early startup)
+ var events = new List
+ {
+ new() { TenantId = null, SessionId = null },
+ };
+
+ IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, DateTime.UtcNow);
+
+ Assert.Equal(ValidTenantId, events[0].TenantId);
+ Assert.Equal(ValidSessionId, events[0].SessionId);
+ }
+
+ [Fact]
+ public void StampServerFields_OverridesWrongTenantId()
+ {
+ // Regression: agent running in unusual context may send wrong or empty TenantId per-event
+ var events = new List
+ {
+ new() { TenantId = "wrong-tenant-value", SessionId = "wrong-session" },
+ };
+
+ IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, DateTime.UtcNow);
+
+ Assert.Equal(ValidTenantId, events[0].TenantId);
+ Assert.Equal(ValidSessionId, events[0].SessionId);
+ }
+
+ [Fact]
+ public void StampServerFields_AllEventsReceiveSameTenantId()
+ {
+ // All events in one batch belong to the same tenant — never mixed
+ var events = new List
+ {
+ new() { TenantId = null },
+ new() { TenantId = "different-1" },
+ new() { TenantId = "different-2" },
+ new() { TenantId = ValidTenantId }, // already correct
+ };
+
+ IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, DateTime.UtcNow);
+
+ Assert.All(events, evt => Assert.Equal(ValidTenantId, evt.TenantId));
+ }
+
+ [Fact]
+ public void StampServerFields_EmptyList_DoesNotThrow()
+ {
+ var ex = Record.Exception(() =>
+ IngestEventsFunction.StampServerFields([], ValidTenantId, ValidSessionId, DateTime.UtcNow));
+ Assert.Null(ex);
+ }
+
+ // -------------------------------------------------------------------------
+ // Guard: after StampServerFields, EnsureValidGuid must pass for all events
+ // (simulates the contract between ProcessIngestAsync and StoreEventsBatchAsync)
+ // -------------------------------------------------------------------------
+
+ [Fact]
+ public void StampServerFields_ThenEnsureValidGuid_DoesNotThrow()
+ {
+ // This is the exact sequence in ProcessIngestAsync → StoreEventsBatchAsync.
+ // If StampServerFields is removed or broken, this test fails with ArgumentException.
+ var events = MakeEvents(5, tenantId: null, sessionId: null);
+
+ IngestEventsFunction.StampServerFields(events, ValidTenantId, ValidSessionId, DateTime.UtcNow);
+
+ // Simulate what StoreEventsBatchAsync does:
+ foreach (var evt in events)
+ {
+ SecurityValidator.EnsureValidGuid(evt.TenantId, "TenantId");
+ SecurityValidator.EnsureValidGuid(evt.SessionId, "SessionId");
+ }
+ // No exception = pass
+ }
+
+ [Fact]
+ public void WithoutStampServerFields_EnsureValidGuid_Throws()
+ {
+ // Documents WHY StampServerFields is required: without it, storage validation explodes.
+ var events = MakeEvents(1, tenantId: null, sessionId: null);
+
+ Assert.Throws(() =>
+ SecurityValidator.EnsureValidGuid(events[0].TenantId, "TenantId"));
+ }
+
+ // -------------------------------------------------------------------------
+ // NDJSON → Stamp → Validate: full pipeline test (no I/O, pure logic)
+ // -------------------------------------------------------------------------
+
+ [Fact]
+ public void FullPipeline_ParseNdjsonThenStamp_AllEventsHaveValidGuids()
+ {
+ // Simulates what ProcessIngestAsync does from parse to storage:
+ // 1. Parse NDJSON (events may have null/wrong TenantId from agent)
+ // 2. StampServerFields (backend overrides with validated tenantId)
+ // 3. EnsureValidGuid passes → no 500
+
+ var ndjsonLines = new[]
+ {
+ Newtonsoft.Json.JsonConvert.SerializeObject(
+ new { SessionId = ValidSessionId, TenantId = ValidTenantId }),
+ Newtonsoft.Json.JsonConvert.SerializeObject(
+ new EnrollmentEvent { EventType = "phase_changed" }), // no TenantId
+ Newtonsoft.Json.JsonConvert.SerializeObject(
+ new EnrollmentEvent { EventType = "hardware_spec", TenantId = "" }), // empty TenantId
+ };
+ var ndjson = string.Join('\n', ndjsonLines);
+
+ var (sessionId, tenantId, events) = NdjsonParser.ParseNdjson(ndjson);
+
+ // Stamp (as ProcessIngestAsync does)
+ IngestEventsFunction.StampServerFields(events, tenantId, sessionId, DateTime.UtcNow);
+
+ // Validate (as StoreEventsBatchAsync does)
+ foreach (var evt in events)
+ {
+ SecurityValidator.EnsureValidGuid(evt.TenantId, "TenantId");
+ SecurityValidator.EnsureValidGuid(evt.SessionId, "SessionId");
+ }
+ // No exception = the regression is fixed
+ }
+
+ // -------------------------------------------------------------------------
+ // TenantId mismatch guard (body vs header)
+ // -------------------------------------------------------------------------
+
+ [Fact]
+ public void TenantIdMismatch_DifferentCasing_IsDetected()
+ {
+ // OrdinalIgnoreCase is used — same GUID in different case should NOT be a mismatch
+ var header = ValidTenantId.ToLowerInvariant();
+ var body = ValidTenantId.ToUpperInvariant();
+
+ Assert.True(string.Equals(header, body, StringComparison.OrdinalIgnoreCase),
+ "Same GUID in different casing must be treated as equal (prevents false 403)");
+ }
+
+ [Fact]
+ public void TenantIdMismatch_DifferentGuids_IsDetected()
+ {
+ var headerTenant = ValidTenantId;
+ var bodyTenant = "c3d4e5f6-a7b8-9012-cdef-012345678901";
+
+ Assert.False(string.Equals(headerTenant, bodyTenant, StringComparison.OrdinalIgnoreCase),
+ "Different GUIDs must be detected as a mismatch (prevents TenantId spoofing)");
+ }
+
+ // -------------------------------------------------------------------------
+ // Helpers
+ // -------------------------------------------------------------------------
+
+ private static List MakeEvents(int count, string? tenantId = "original-tenant", string? sessionId = "original-session")
+ {
+ return Enumerable.Range(0, count)
+ .Select(i => new EnrollmentEvent
+ {
+ EventType = $"test_event_{i}",
+ TenantId = tenantId,
+ SessionId = sessionId
+ })
+ .ToList();
+ }
+}
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/NdjsonParserTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/NdjsonParserTests.cs
new file mode 100644
index 00000000..09c815c3
--- /dev/null
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/NdjsonParserTests.cs
@@ -0,0 +1,190 @@
+using AutopilotMonitor.Functions.Functions.Ingest;
+using AutopilotMonitor.Shared.Models;
+using Newtonsoft.Json;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+///
+/// Tests for NdjsonParser — the NDJSON+gzip parsing contract.
+/// If the wire format between agent and backend ever drifts, these tests will catch it.
+///
+public class NdjsonParserTests
+{
+ private static readonly string ValidTenantId = "a1b2c3d4-e5f6-7890-abcd-ef1234567890";
+ private static readonly string ValidSessionId = "b2c3d4e5-f6a7-8901-bcde-f12345678901";
+
+ // --- ParseNdjson (plain string, no gzip) ---
+
+ [Fact]
+ public void ParseNdjson_MetadataOnly_ReturnsCorrectTenantAndSession()
+ {
+ var ndjson = JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId });
+
+ var (sessionId, tenantId, events) = NdjsonParser.ParseNdjson(ndjson);
+
+ Assert.Equal(ValidSessionId, sessionId);
+ Assert.Equal(ValidTenantId, tenantId);
+ Assert.Empty(events);
+ }
+
+ [Fact]
+ public void ParseNdjson_WithEvents_ReturnsAllEvents()
+ {
+ var lines = new[]
+ {
+ JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId }),
+ JsonConvert.SerializeObject(new EnrollmentEvent { EventType = "phase_changed", Message = "Device Setup" }),
+ JsonConvert.SerializeObject(new EnrollmentEvent { EventType = "app_install_started", Message = "MyApp" }),
+ };
+ var ndjson = string.Join('\n', lines);
+
+ var (_, _, events) = NdjsonParser.ParseNdjson(ndjson);
+
+ Assert.Equal(2, events.Count);
+ Assert.Equal("phase_changed", events[0].EventType);
+ Assert.Equal("app_install_started", events[1].EventType);
+ }
+
+ [Fact]
+ public void ParseNdjson_EventsDoNotInheritTenantIdFromMetadata()
+ {
+ // Critical contract: the parser does NOT stamp TenantId onto events.
+ // That is done by StampServerFields — these are two separate responsibilities.
+ var lines = new[]
+ {
+ JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId }),
+ JsonConvert.SerializeObject(new EnrollmentEvent { EventType = "test_event" }),
+ };
+ var ndjson = string.Join('\n', lines);
+
+ var (_, _, events) = NdjsonParser.ParseNdjson(ndjson);
+
+ // Event has no TenantId from agent JSON → will be null until StampServerFields is called
+ Assert.Null(events[0].TenantId);
+ Assert.Null(events[0].SessionId);
+ }
+
+ [Fact]
+ public void ParseNdjson_EventWithAgentSuppliedTenantId_PreservesItUntilStamped()
+ {
+ var agentTenantId = "c3d4e5f6-a7b8-9012-cdef-012345678901";
+ var lines = new[]
+ {
+ JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId }),
+ JsonConvert.SerializeObject(new EnrollmentEvent { EventType = "test_event", TenantId = agentTenantId }),
+ };
+ var ndjson = string.Join('\n', lines);
+
+ var (_, _, events) = NdjsonParser.ParseNdjson(ndjson);
+
+ // Parser preserves what the agent sent — StampServerFields will override it
+ Assert.Equal(agentTenantId, events[0].TenantId);
+ }
+
+ [Fact]
+ public void ParseNdjson_TrailingNewlines_AreIgnored()
+ {
+ var ndjson = JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId })
+ + "\n\n\n";
+
+ var (sessionId, tenantId, events) = NdjsonParser.ParseNdjson(ndjson);
+
+ Assert.Equal(ValidSessionId, sessionId);
+ Assert.Empty(events);
+ }
+
+ [Fact]
+ public void ParseNdjson_NullOrEmptyLine_IsSkipped()
+ {
+ var lines = new[]
+ {
+ JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId }),
+ "",
+ JsonConvert.SerializeObject(new EnrollmentEvent { EventType = "real_event" }),
+ };
+ // Note: Split('\n', RemoveEmptyEntries) already removes empty lines —
+ // this test verifies the parser handles extra whitespace lines gracefully.
+ var ndjson = string.Join('\n', lines);
+
+ var (_, _, events) = NdjsonParser.ParseNdjson(ndjson);
+
+ Assert.Single(events);
+ Assert.Equal("real_event", events[0].EventType);
+ }
+
+ [Fact]
+ public void ParseNdjson_EventDataWithNestedObjects_IsNormalized()
+ {
+ var lines = new[]
+ {
+ JsonConvert.SerializeObject(new { SessionId = ValidSessionId, TenantId = ValidTenantId }),
+ """{"EventType":"hardware_spec","Data":{"cpuName":"Intel i7","cores":8}}""",
+ };
+ var ndjson = string.Join('\n', lines);
+
+ var (_, _, events) = NdjsonParser.ParseNdjson(ndjson);
+
+ // After normalization, values should be native .NET types, not JValue/JObject
+ Assert.Single(events);
+ var cpuName = events[0].Data["cpuName"];
+ Assert.IsNotType(cpuName);
+ Assert.Equal("Intel i7", cpuName?.ToString());
+ }
+
+ [Fact]
+ public void ParseNdjson_EmptyString_ThrowsInvalidOperationException()
+ {
+ Assert.Throws(() => NdjsonParser.ParseNdjson(""));
+ }
+
+ // --- ParseGzipAsync ---
+
+ [Fact]
+ public async Task ParseGzipAsync_ValidPayload_ReturnsCorrectData()
+ {
+ var payload = NdjsonParser.BuildGzipPayload(ValidSessionId, ValidTenantId,
+ [
+ new EnrollmentEvent { EventType = "phase_changed" },
+ new EnrollmentEvent { EventType = "app_install_started" },
+ ]);
+
+ using var stream = new MemoryStream(payload);
+ var (sessionId, tenantId, events) = await NdjsonParser.ParseGzipAsync(stream, 5 * 1024 * 1024);
+
+ Assert.Equal(ValidSessionId, sessionId);
+ Assert.Equal(ValidTenantId, tenantId);
+ Assert.Equal(2, events.Count);
+ }
+
+ [Fact]
+ public async Task ParseGzipAsync_PayloadExceedsMaxSize_Throws()
+ {
+ var payload = NdjsonParser.BuildGzipPayload(ValidSessionId, ValidTenantId,
+ Enumerable.Range(0, 100).Select(_ => new EnrollmentEvent
+ {
+ EventType = "test",
+ Message = new string('x', 1000)
+ }));
+
+ using var stream = new MemoryStream(payload);
+
+ // Restrict to 1 byte — will throw immediately
+ await Assert.ThrowsAsync(
+ () => NdjsonParser.ParseGzipAsync(stream, maxSizeBytes: 1));
+ }
+
+ // --- BuildGzipPayload (test helper verification) ---
+
+ [Fact]
+ public async Task BuildGzipPayload_RoundTrip_PreservesMetadata()
+ {
+ var payload = NdjsonParser.BuildGzipPayload(ValidSessionId, ValidTenantId, []);
+
+ using var stream = new MemoryStream(payload);
+ var (sessionId, tenantId, events) = await NdjsonParser.ParseGzipAsync(stream, 5 * 1024 * 1024);
+
+ Assert.Equal(ValidSessionId, sessionId);
+ Assert.Equal(ValidTenantId, tenantId);
+ Assert.Empty(events);
+ }
+}
diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/SecurityValidatorTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/SecurityValidatorTests.cs
new file mode 100644
index 00000000..d359a4b3
--- /dev/null
+++ b/src/Backend/AutopilotMonitor.Functions.Tests/SecurityValidatorTests.cs
@@ -0,0 +1,124 @@
+using AutopilotMonitor.Functions.Security;
+
+namespace AutopilotMonitor.Functions.Tests;
+
+///
+/// Tests for SecurityValidator GUID validation.
+/// StoreEventsBatchAsync relies on EnsureValidGuid to reject invalid TenantId/SessionId values —
+/// these tests guard that contract.
+///
+public class SecurityValidatorTests
+{
+ // --- IsValidGuid ---
+
+ [Fact]
+ public void IsValidGuid_WithValidLowercaseGuid_ReturnsTrue()
+ {
+ Assert.True(SecurityValidator.IsValidGuid("a1b2c3d4-e5f6-7890-abcd-ef1234567890"));
+ }
+
+ [Fact]
+ public void IsValidGuid_WithValidUppercaseGuid_ReturnsTrue()
+ {
+ Assert.True(SecurityValidator.IsValidGuid("A1B2C3D4-E5F6-7890-ABCD-EF1234567890"));
+ }
+
+ [Fact]
+ public void IsValidGuid_WithValidMixedCaseGuid_ReturnsTrue()
+ {
+ Assert.True(SecurityValidator.IsValidGuid("a1B2c3D4-e5F6-7890-AbCd-Ef1234567890"));
+ }
+
+ [Fact]
+ public void IsValidGuid_WithNewGuid_ReturnsTrue()
+ {
+ Assert.True(SecurityValidator.IsValidGuid(Guid.NewGuid().ToString()));
+ }
+
+ [Fact]
+ public void IsValidGuid_WithNull_ReturnsFalse()
+ {
+ Assert.False(SecurityValidator.IsValidGuid(null));
+ }
+
+ [Fact]
+ public void IsValidGuid_WithEmptyString_ReturnsFalse()
+ {
+ Assert.False(SecurityValidator.IsValidGuid(""));
+ }
+
+ [Fact]
+ public void IsValidGuid_WithWhitespace_ReturnsFalse()
+ {
+ Assert.False(SecurityValidator.IsValidGuid(" "));
+ }
+
+ [Fact]
+ public void IsValidGuid_WithPlainString_ReturnsFalse()
+ {
+ Assert.False(SecurityValidator.IsValidGuid("not-a-guid"));
+ }
+
+ [Fact]
+ public void IsValidGuid_WithDeviceName_ReturnsFalse()
+ {
+ // Regression: agent running as defaultuser0 could send "DESKTOP-DIU8038\defaultuser0" as TenantId
+ Assert.False(SecurityValidator.IsValidGuid(@"DESKTOP-DIU8038\defaultuser0"));
+ }
+
+ [Fact]
+ public void IsValidGuid_WithGuidWithoutDashes_ReturnsFalse()
+ {
+ // No dashes → not in standard format
+ Assert.False(SecurityValidator.IsValidGuid("a1b2c3d4e5f67890abcdef1234567890"));
+ }
+
+ [Fact]
+ public void IsValidGuid_WithGuidInBraces_ReturnsFalse()
+ {
+ // Braced format {…} is not accepted — we require plain hyphenated form
+ Assert.False(SecurityValidator.IsValidGuid("{a1b2c3d4-e5f6-7890-abcd-ef1234567890}"));
+ }
+
+ // --- EnsureValidGuid ---
+
+ [Fact]
+ public void EnsureValidGuid_WithValidGuid_DoesNotThrow()
+ {
+ var ex = Record.Exception(() =>
+ SecurityValidator.EnsureValidGuid(Guid.NewGuid().ToString(), "TenantId"));
+ Assert.Null(ex);
+ }
+
+ [Fact]
+ public void EnsureValidGuid_WithNull_ThrowsArgumentException()
+ {
+ var ex = Assert.Throws(() =>
+ SecurityValidator.EnsureValidGuid(null, "TenantId"));
+ Assert.Contains("TenantId", ex.Message);
+ }
+
+ [Fact]
+ public void EnsureValidGuid_WithEmptyString_ThrowsArgumentException()
+ {
+ var ex = Assert.Throws(() =>
+ SecurityValidator.EnsureValidGuid("", "SessionId"));
+ Assert.Contains("SessionId", ex.Message);
+ }
+
+ [Fact]
+ public void EnsureValidGuid_WithInvalidString_ThrowsArgumentException()
+ {
+ var ex = Assert.Throws(() =>
+ SecurityValidator.EnsureValidGuid("invalid-tenant", "TenantId"));
+ Assert.Equal("TenantId", ex.ParamName);
+ }
+
+ [Fact]
+ public void EnsureValidGuid_ErrorMessage_MentionsParameterName()
+ {
+ var ex = Assert.Throws(() =>
+ SecurityValidator.EnsureValidGuid("bad", "MyParam"));
+ Assert.Contains("MyParam", ex.Message);
+ }
+}
diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestEventsFunction.Helpers.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestEventsFunction.Helpers.cs
index 02064089..b2fe68fc 100644
--- a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestEventsFunction.Helpers.cs
+++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/IngestEventsFunction.Helpers.cs
@@ -1,16 +1,12 @@
-using System.IO.Compression;
using System.Linq;
using System;
using System.Net;
-using System.Text;
using AutopilotMonitor.Functions.Services;
using AutopilotMonitor.Shared.Models;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Http;
using Microsoft.Azure.Functions.Worker.Extensions.SignalRService;
using Microsoft.Extensions.Logging;
-using Newtonsoft.Json;
-using Newtonsoft.Json.Linq;
namespace AutopilotMonitor.Functions.Functions.Ingest
{
@@ -21,109 +17,24 @@ namespace AutopilotMonitor.Functions.Functions.Ingest
{
private async Task ParseNdjsonGzipRequest(Stream body, string? tenantId = null)
{
- // Get configuration for payload size limit (use default if tenantId not available yet)
var config = !string.IsNullOrEmpty(tenantId)
? await _configService.GetConfigurationAsync(tenantId)
: null;
var maxPayloadSizeBytes = (config?.MaxNdjsonPayloadSizeMB ?? 5) * 1024 * 1024;
- // Decompress gzip with size limit protection
- using var decompressed = new MemoryStream();
- using (var gzip = new GZipStream(body, CompressionMode.Decompress, leaveOpen: true))
- {
- // Copy with size limit to prevent memory exhaustion attacks
- var buffer = new byte[8192];
- int bytesRead;
- long totalBytesRead = 0;
-
- while ((bytesRead = await gzip.ReadAsync(buffer, 0, buffer.Length)) > 0)
- {
- totalBytesRead += bytesRead;
-
- // Check if we've exceeded the maximum payload size
- if (totalBytesRead > maxPayloadSizeBytes)
- {
- throw new InvalidOperationException(
- $"NDJSON payload size exceeds maximum allowed size of {config?.MaxNdjsonPayloadSizeMB ?? 5} MB (decompressed). " +
- $"Current size: {totalBytesRead / 1024.0 / 1024.0:F2} MB"
- );
- }
-
- await decompressed.WriteAsync(buffer, 0, bytesRead);
- }
-
- _logger.LogDebug($"NDJSON payload decompressed: {totalBytesRead / 1024.0:F2} KB (limit: {maxPayloadSizeBytes / 1024.0 / 1024.0} MB)");
- }
-
- decompressed.Position = 0;
- var ndjson = await new StreamReader(decompressed, Encoding.UTF8).ReadToEndAsync();
-
- // Parse NDJSON (newline-delimited JSON)
- var lines = ndjson.Split('\n', StringSplitOptions.RemoveEmptyEntries);
-
- if (lines.Length < 1)
- {
- throw new InvalidOperationException("NDJSON must contain at least metadata line");
- }
-
- // First line: metadata
- var metadata = JsonConvert.DeserializeObject(lines[0]);
- if (metadata == null)
- {
- throw new InvalidOperationException("Failed to parse NDJSON metadata");
- }
-
- // Subsequent lines: events
- var events = new List();
- for (int i = 1; i < lines.Length; i++)
- {
- var evt = JsonConvert.DeserializeObject(lines[i]);
- if (evt != null)
- {
- NormalizeEventData(evt);
- events.Add(evt);
- }
- }
+ var (sessionId, metaTenantId, events) = await NdjsonParser.ParseGzipAsync(body, maxPayloadSizeBytes);
+ _logger.LogDebug("NDJSON payload parsed: {EventCount} events (limit: {LimitMB} MB)",
+ events.Count, config?.MaxNdjsonPayloadSizeMB ?? 5);
return new IngestEventsRequest
{
- SessionId = metadata.SessionId,
- TenantId = metadata.TenantId,
+ SessionId = sessionId,
+ TenantId = metaTenantId,
Events = events
};
}
- ///
- /// Normalizes event Data dictionary by converting Newtonsoft JToken objects to native .NET types.
- /// Required because Newtonsoft.Json deserializes nested objects as JObject/JArray, which
- /// System.Text.Json (used by SignalR) cannot serialize correctly - producing [[[]]] instead of real values.
- ///
- private static void NormalizeEventData(EnrollmentEvent evt)
- {
- if (evt.Data == null || evt.Data.Count == 0) return;
- var normalized = new Dictionary();
- foreach (var kvp in evt.Data)
- normalized[kvp.Key] = ConvertJTokenToNative(kvp.Value);
- evt.Data = normalized;
- }
-
- private static object ConvertJTokenToNative(object value)
- {
- if (value is JArray jArray)
- return jArray.Select(item => ConvertJTokenToNative(item)).ToList