Fix SSE integration test race conditions and timeouts

Resolved test failures in CI environment by:
- Added ClientCount property to ISseStream/SseStream for subscription tracking
- Wait for client connection before publishing events to eliminate race conditions
- Increased timeouts from 2/5 seconds to 15/30 seconds for slower CI environments
- Added connection stabilization delay after client count verification
- Improved event reading logic to handle stream buffering correctly

All 76 tests now pass reliably in both local and CI environments.

Author: ANcpLua

SWEN3.Paperless.RabbitMq.Tests/Integration/GenAIEventStreamIntegrationTests.cs

@@ -13,23 +13,26 @@ public async Task MapGenAIEventStream_ShouldEmitCorrectEventType(string status,
 
         var readTask = Task.Run(async () =>
         {
-            var response = await client.GetAsync("/api/v1/events/genai", HttpCompletionOption.ResponseHeadersRead);
+            using var response = await client.GetAsync("/api/v1/events/genai", HttpCompletionOption.ResponseHeadersRead);
             await using var stream = await response.Content.ReadAsStreamAsync();
             using var reader = new StreamReader(stream);
 
-            var line = await reader.ReadLineAsync();
-            response.Dispose();
-            return line;
+            while (true)
+            {
+                var line = await reader.ReadLineAsync();
+                if (line == null) return null;
+                if (line.StartsWith("event:")) return line;
+            }
         });
 
-        await Task.Delay(100, TestContext.Current.CancellationToken);
+        while (sseStream.ClientCount == 0) await Task.Delay(50, TestContext.Current.CancellationToken);
 
         var genAiEvent = status == "Completed"
             ? new GenAIEvent(Guid.NewGuid(), "Test summary", DateTimeOffset.UtcNow)
             : new GenAIEvent(Guid.NewGuid(), string.Empty, DateTimeOffset.UtcNow, "Service error");
         sseStream.Publish(genAiEvent);
 
-        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(2));
+        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
         var eventLine = await readTask.WaitAsync(cts.Token);
 
         eventLine.Should().Be($"event: {expectedEventType}");
@@ -42,16 +45,31 @@ public async Task MapGenAIEventStream_WithMultipleEvents_ShouldStreamInOrder()
         using var server = SseTestHelpers.CreateSseTestServer(sseStream, endpoints => endpoints.MapGenAIEventStream());
 
         var client = server.CreateClient();
-        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
         var token = cts.Token;
         var readTask = Task.Run(() => ReadEventsAsync(client, 3, token), token);
 
-        await Task.Delay(300, TestContext.Current.CancellationToken);
+        // Wait for client to connect with timeout
+        var waitStart = DateTime.UtcNow;
+        while (sseStream.ClientCount == 0)
+        {
+            if (DateTime.UtcNow - waitStart > TimeSpan.FromSeconds(10))
+                throw new TimeoutException("Client did not connect within timeout");
+            await Task.Delay(50, TestContext.Current.CancellationToken);
+        }
+
+        // Give the HTTP connection a moment to stabilize
+        await Task.Delay(100, TestContext.Current.CancellationToken);
+
         sseStream.Publish(new GenAIEvent(Guid.NewGuid(), "Summary 1", DateTimeOffset.UtcNow));
+        await Task.Delay(50, TestContext.Current.CancellationToken);
+
         sseStream.Publish(new GenAIEvent(Guid.NewGuid(), "Summary 2", DateTimeOffset.UtcNow));
+        await Task.Delay(50, TestContext.Current.CancellationToken);
+
         sseStream.Publish(new GenAIEvent(Guid.NewGuid(), string.Empty, DateTimeOffset.UtcNow, "Error occurred"));
 
-        var events = await readTask.WaitAsync(TimeSpan.FromSeconds(2), TestContext.Current.CancellationToken);
+        var events = await readTask.WaitAsync(TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken);
 
         events[0].Event.Should().Be("event: genai-completed");
         events[0].Data.Should().Contain("Summary 1");
@@ -73,16 +91,29 @@ public async Task MapGenAIEventStream_WithMultipleEvents_ShouldStreamInOrder()
             using var reader = new StreamReader(stream);
 
             var events = new List<(string Event, string Data)>();
-            for (var i = 0; i < count; i++)
-            {
-                var eventLine = await reader.ReadLineAsync(cancellationToken);
-                var dataLine = await reader.ReadLineAsync(cancellationToken);
-
-                if (eventLine == null || dataLine == null)
-                    throw new InvalidOperationException($"Unexpected end of stream while reading event {i + 1} of {count}");
+            string? currentEvent = null;
+            string? currentData = null;
 
-                events.Add((eventLine, dataLine));
-                await reader.ReadLineAsync(cancellationToken);
+            while (events.Count < count)
+            {
+                var line = await reader.ReadLineAsync(cancellationToken);
+                if (line == null) break;
+
+                if (string.IsNullOrWhiteSpace(line))
+                {
+                    if (currentEvent != null && currentData != null)
+                    {
+                        events.Add((currentEvent, currentData));
+                        currentEvent = null;
+                        currentData = null;
+                    }
+                    continue;
+                }
+
+                if (line.StartsWith("event:"))
+                    currentEvent = line;
+                else if (line.StartsWith("data:"))
+                    currentData = line;
             }
 
             return events;

SWEN3.Paperless.RabbitMq.Tests/Integration/OcrEventStreamIntegrationTests.cs

@@ -11,25 +11,38 @@ public async Task MapOcrEventStream_ShouldEmitCorrectEventType(string status, st
         var sseStream = new SseStream<OcrEvent>();
         var server = SseTestHelpers.CreateSseTestServer(sseStream, endpoints => endpoints.MapOcrEventStream());
 
+        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
         var readTask = Task.Run(async () =>
         {
             using var client = server.CreateClient();
-            var response = await client.GetAsync("/api/v1/ocr-results", HttpCompletionOption.ResponseHeadersRead);
-            await using var stream = await response.Content.ReadAsStreamAsync();
+            using var response = await client.GetAsync("/api/v1/ocr-results", HttpCompletionOption.ResponseHeadersRead, cts.Token);
+            await using var stream = await response.Content.ReadAsStreamAsync(cts.Token);
             using var reader = new StreamReader(stream);
 
-            var line = await reader.ReadLineAsync();
-            response.Dispose();
-            return line;
-        });
+            while (true)
+            {
+                var line = await reader.ReadLineAsync(cts.Token);
+                if (line == null) return null;
+                if (line.StartsWith("event:")) return line;
+            }
+        }, cts.Token);
 
-        await Task.Delay(300, TestContext.Current.CancellationToken);
+        // Wait for client to connect with timeout
+        var waitStart = DateTime.UtcNow;
+        while (sseStream.ClientCount == 0)
+        {
+            if (DateTime.UtcNow - waitStart > TimeSpan.FromSeconds(10))
+                throw new TimeoutException("Client did not connect within timeout");
+            await Task.Delay(50, TestContext.Current.CancellationToken);
+        }
+
+        // Give the HTTP connection a moment to stabilize
+        await Task.Delay(100, TestContext.Current.CancellationToken);
 
         var ocrEvent = new OcrEvent(Guid.NewGuid(), status, status is "Completed" ? "Text" : null,
             DateTimeOffset.UtcNow);
         sseStream.Publish(ocrEvent);
 
-        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(2));
         var eventLine = await readTask.WaitAsync(cts.Token);
 
         eventLine.Should().Be($"event: {expectedEventType}");

SWEN3.Paperless.RabbitMq/Sse/ISseStream.cs

@@ -29,4 +29,9 @@ public interface ISseStream<T>
     /// </summary>
     /// <param name="item">The event to publish.</param>
     void Publish(T item);
+
+    /// <summary>
+    ///     Gets the number of currently subscribed clients.
+    /// </summary>
+    int ClientCount { get; }
 }

SWEN3.Paperless.RabbitMq/Sse/SseStream.cs

@@ -23,4 +23,6 @@ public void Publish(T item)
     {
         foreach (var channel in _channels.Values) channel.Writer.TryWrite(item);
     }
+
+    public int ClientCount => _channels.Count;
 }