Merge pull request #514 from serverlessworkflow/fix-sse-streaming

Fixed both the `ClusterResourceController` and `NamespacedResourceController` by ensuring that SSE-based actions do not set the response's status code after streaming

Author: cdavernas

src/api/Synapse.Api.Http/ClusterResourceController.cs

@@ -20,15 +20,10 @@ namespace Synapse.Api.Http;
 /// <param name="mediator">The service used to mediate calls</param>
 /// <param name="jsonSerializer">The service used to serialize/deserialize data to/from JSON</param>
 public abstract class ClusterResourceController<TResource>(IMediator mediator, IJsonSerializer jsonSerializer)
-    : ResourceController<TResource>(mediator)
+    : ResourceController<TResource>(mediator, jsonSerializer)
     where TResource : class, IResource, new()
 {
 
-    /// <summary>
-    /// Gets the service used to serialize/deserialize data to/from JSON
-    /// </summary>
-    protected IJsonSerializer JsonSerializer { get; } = jsonSerializer;
-
     /// <summary>
     /// Gets the resource with the specified name
     /// </summary>
@@ -97,13 +92,17 @@ public virtual async Task<IAsyncEnumerable<IResourceWatchEvent<TResource>>> Watc
     /// </summary>
     /// <param name="labelSelector">A comma-separated list of label selectors, if any</param>
     /// <param name="cancellationToken">A <see cref="CancellationToken"/></param>
-    /// <returns>A new <see cref="IActionResult"/></returns>
+    /// <returns>A new awaitable <see cref="Task"/></returns>
     [HttpGet("watch/sse")]
     [ProducesResponseType(typeof(IAsyncEnumerable<ResourceWatchEvent>), (int)HttpStatusCode.OK)]
     [ProducesErrorResponseType(typeof(Neuroglia.ProblemDetails))]
-    public virtual async Task<IActionResult> WatchResourcesUsingSSE(string? labelSelector = null, CancellationToken cancellationToken = default)
+    public virtual async Task WatchResourcesUsingSSE(string? labelSelector = null, CancellationToken cancellationToken = default)
     {
-        if (!this.TryParseLabelSelectors(labelSelector, out var labelSelectors)) return this.InvalidLabelSelector(labelSelector!);
+        if (!this.TryParseLabelSelectors(labelSelector, out var labelSelectors))
+        {
+            await WriteInvalidLabelSelectorResponseAsync(labelSelector!, cancellationToken).ConfigureAwait(false);
+            return;
+        }
         var response = await this.Mediator.ExecuteAsync(new WatchResourcesQuery<TResource>(null, labelSelectors), cancellationToken).ConfigureAwait(false);
         this.Response.Headers.ContentType = "text/event-stream";
         this.Response.Headers.CacheControl = "no-cache";
@@ -119,7 +118,6 @@ public virtual async Task<IActionResult> WatchResourcesUsingSSE(string? labelSel
             }
         }
         catch (Exception ex) when (ex is TaskCanceledException || ex is OperationCanceledException) { }
-        return this.Ok();
     }
 
     /// <summary>
@@ -142,11 +140,11 @@ public virtual async Task<IAsyncEnumerable<IResourceWatchEvent<TResource>>> Moni
     /// </summary>
     /// <param name="name">The name of the cluster resource to monitor</param>
     /// <param name="cancellationToken">A <see cref="CancellationToken"/></param>
-    /// <returns>A new <see cref="IActionResult"/></returns>
+    /// <returns>A new awaitable <see cref="Task"/></returns>
     [HttpGet("{name}/monitor/sse")]
     [ProducesResponseType(typeof(IAsyncEnumerable<ResourceWatchEvent>), (int)HttpStatusCode.OK)]
     [ProducesErrorResponseType(typeof(Neuroglia.ProblemDetails))]
-    public virtual async Task<IActionResult> MonitorResourceUsingSSE(string name, CancellationToken cancellationToken = default)
+    public virtual async Task MonitorResourceUsingSSE(string name, CancellationToken cancellationToken = default)
     {
         var response = await this.Mediator.ExecuteAsync(new MonitorResourceQuery<TResource>(name, null), cancellationToken).ConfigureAwait(false);
         this.Response.Headers.ContentType = "text/event-stream";
@@ -163,7 +161,6 @@ public virtual async Task<IActionResult> MonitorResourceUsingSSE(string name, Ca
             }
         }
         catch (Exception ex) when (ex is TaskCanceledException || ex is OperationCanceledException) { }
-        return this.Ok();
     }
 
     /// <summary>

src/api/Synapse.Api.Http/NamespacedResourceController.cs

@@ -20,15 +20,10 @@ namespace Synapse.Api.Http;
 /// <param name="mediator">The service used to mediate calls</param>
 /// <param name="jsonSerializer">The service used to serialize/deserialize data to/from JSON</param>
 public abstract class NamespacedResourceController<TResource>(IMediator mediator, IJsonSerializer jsonSerializer)
-    : ResourceController<TResource>(mediator)
+    : ResourceController<TResource>(mediator, jsonSerializer)
     where TResource : class, IResource, new()
 {
 
-    /// <summary>
-    /// Gets the service used to serialize/deserialize data to/from JSON
-    /// </summary>
-    protected IJsonSerializer JsonSerializer { get; } = jsonSerializer;
-
     /// <summary>
     /// Gets the resource with the specified name and namespace
     /// </summary>
@@ -150,13 +145,17 @@ public virtual async Task<IAsyncEnumerable<IResourceWatchEvent<TResource>>> Watc
     /// <param name="namespace">The namespace the resources to watch belong to</param>
     /// <param name="labelSelector">A comma-separated list of label selectors, if any</param>
     /// <param name="cancellationToken">A <see cref="CancellationToken"/></param>
-    /// <returns>A new <see cref="IActionResult"/></returns>
+    /// <returns>A new awaitable <see cref="Task"/></returns>
     [HttpGet("{namespace}/watch/sse")]
     [ProducesResponseType(typeof(IAsyncEnumerable<ResourceWatchEvent>), (int)HttpStatusCode.OK)]
     [ProducesErrorResponseType(typeof(Neuroglia.ProblemDetails))]
-    public virtual async Task<IActionResult> WatchResourcesUsingSSE(string @namespace, string? labelSelector = null, CancellationToken cancellationToken = default) 
+    public virtual async Task WatchResourcesUsingSSE(string @namespace, string? labelSelector = null, CancellationToken cancellationToken = default) 
     {
-        if (!this.TryParseLabelSelectors(labelSelector, out var labelSelectors)) return this.InvalidLabelSelector(labelSelector!);
+        if (!this.TryParseLabelSelectors(labelSelector, out var labelSelectors))
+        {
+            await WriteInvalidLabelSelectorResponseAsync(labelSelector!, cancellationToken).ConfigureAwait(false);
+            return;
+        }
         var response = await this.Mediator.ExecuteAsync(new WatchResourcesQuery<TResource>(@namespace, labelSelectors), cancellationToken).ConfigureAwait(false);
         this.Response.Headers.ContentType = "text/event-stream";
         this.Response.Headers.CacheControl = "no-cache";
@@ -172,7 +171,6 @@ public virtual async Task<IActionResult> WatchResourcesUsingSSE(string @namespac
             }
         }
         catch (Exception ex) when(ex is TaskCanceledException || ex is OperationCanceledException) { }
-        return this.Ok();
     }
 
     /// <summary>
@@ -197,11 +195,11 @@ public virtual async Task<IAsyncEnumerable<IResourceWatchEvent<TResource>>> Moni
     /// <param name="namespace">The namespace the resource to monitor belongs to</param>
     /// <param name="name">The name of the resource to monitor</param>
     /// <param name="cancellationToken">A <see cref="CancellationToken"/></param>
-    /// <returns>A new <see cref="IActionResult"/></returns>
+    /// <returns>A new awaitable <see cref="Task"/></returns>
     [HttpGet("{namespace}/{name}/monitor/sse")]
     [ProducesResponseType(typeof(IAsyncEnumerable<ResourceWatchEvent>), (int)HttpStatusCode.OK)]
     [ProducesErrorResponseType(typeof(Neuroglia.ProblemDetails))]
-    public virtual async Task<IActionResult> MonitorResourceUsingSSE(string name, string @namespace, CancellationToken cancellationToken = default)
+    public virtual async Task MonitorResourceUsingSSE(string name, string @namespace, CancellationToken cancellationToken = default)
     {
         var response = await this.Mediator.ExecuteAsync(new MonitorResourceQuery<TResource>(name, @namespace), cancellationToken).ConfigureAwait(false);
         this.Response.Headers.ContentType = "text/event-stream";
@@ -218,7 +216,6 @@ public virtual async Task<IActionResult> MonitorResourceUsingSSE(string name, st
             }
         }
         catch (Exception ex) when (ex is TaskCanceledException || ex is OperationCanceledException) { }
-        return this.Ok();
     }
 
     /// <summary>

src/api/Synapse.Api.Http/ResourceController.cs

@@ -18,7 +18,8 @@ namespace Synapse.Api.Http;
 /// </summary>
 /// <typeparam name="TResource">The type of <see cref="IResource"/> to manage</typeparam>
 /// <param name="mediator">The service used to mediate calls</param>
-public abstract class ResourceController<TResource>(IMediator mediator)
+/// <param name="jsonSerializer">The service used to serialize/deserialize data to/from JSON.</param>
+public abstract class ResourceController<TResource>(IMediator mediator, IJsonSerializer jsonSerializer)
     : Controller
     where TResource : class, IResource, new()
 {
@@ -28,6 +29,11 @@ public abstract class ResourceController<TResource>(IMediator mediator)
     /// </summary>
     protected IMediator Mediator { get; } = mediator;
 
+    /// <summary>
+    /// Gets the service used to serialize/deserialize data to/from JSON.
+    /// </summary>
+    protected IJsonSerializer JsonSerializer { get; } = jsonSerializer;
+
     /// <summary>
     /// Creates a new resource of the specified type
     /// </summary>
@@ -117,4 +123,30 @@ protected IActionResult InvalidLabelSelector(string labelSelector)
         return this.ValidationProblem("Bad Request", statusCode: (int)HttpStatusCode.BadRequest, title: "Bad Request", modelStateDictionary: this.ModelState);
     }
 
+    /// <summary>
+    /// Writes to the response the description of a validation problem that occurred while processing the request.
+    /// </summary>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/>.</param>
+    /// <returns>A new awaitable <see cref="Task"/>.</returns>
+    protected virtual async Task WriteValidationProblemResponseAsync(CancellationToken cancellationToken)
+    {
+        var problem = new ValidationProblemDetails(ModelState);
+        var json = JsonSerializer.SerializeToText(problem);
+        Response.StatusCode = (int)HttpStatusCode.BadRequest;
+        Response.ContentType = MediaTypeNames.Application.Json;
+        await Response.WriteAsync(json, cancellationToken).ConfigureAwait(false);
+    }
+
+    /// <summary>
+    /// Writes to the response the description of an error that occurred while parsing the request's label selector.
+    /// </summary>
+    /// <param name="labelSelector">The invalid label selector.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/>.</param>
+    /// <returns>A new awaitable <see cref="Task"/>.</returns>
+    protected virtual async Task WriteInvalidLabelSelectorResponseAsync(string labelSelector, CancellationToken cancellationToken)
+    {
+        ModelState.AddModelError(nameof(labelSelector), $"The specified value '{labelSelector}' is not a valid comma-separated label selector list");
+        await WriteValidationProblemResponseAsync(cancellationToken).ConfigureAwait(false);
+    }
+
 }

src/api/Synapse.Api.Http/Usings.cs

@@ -27,4 +27,5 @@
 global using Synapse.Resources;
 global using System.Collections.Concurrent;
 global using System.Net;
+global using System.Net.Mime;
 global using System.Text;