Update HTTP routes and headers for dweet APIs, adjust SSE response content type, and add TODO for metrics

HoolIt.sln.DotSettings.user

@@ -1,3 +1,4 @@
 <wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
 	<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003ADebugger_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002Econfig_003FJetBrains_003FRider2025_002E3_003Fresharper_002Dhost_003FSourcesCache_003Ff9d2f95d72fa884d8b6ddefc717c56da3657fbb2d5fb683656c3589eb6587_003FDebugger_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
-	<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AHttpResponseStream_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002E_002E_003F_002E_002E_003F_002E_002E_003Fhome_003Fmahdium_003F_002Econfig_003FJetBrains_003FRider2024_002E3_003Fresharper_002Dhost_003FSourcesCache_003F912bd5c687f4cf55e0daddfb3f8eecd859debac3856d3a98f1a2ad5208413bd_003FHttpResponseStream_002Ecs/@EntryIndexedValue">ForceIncluded</s:String></wpf:ResourceDictionary>
+	<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AHttpResponseStream_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002E_002E_003F_002E_002E_003F_002E_002E_003Fhome_003Fmahdium_003F_002Econfig_003FJetBrains_003FRider2024_002E3_003Fresharper_002Dhost_003FSourcesCache_003F912bd5c687f4cf55e0daddfb3f8eecd859debac3856d3a98f1a2ad5208413bd_003FHttpResponseStream_002Ecs/@EntryIndexedValue">ForceIncluded</s:String>
+	<s:String x:Key="/Default/CodeInspection/ExcludedFiles/FilesAndFoldersToSkip2/=7020124F_002D9FFC_002D4AC3_002D8F3D_002DAAB8E0240759_002Ff_003AIHttpResponseBodyFeature_002Ecs_002Fl_003A_002E_002E_003F_002E_002E_003F_002Econfig_003FJetBrains_003FRider2025_002E3_003Fresharper_002Dhost_003FDecompilerCache_003Fdecompiler_003F0be67315c9494f8ba5a3f8abd54817d1ed10_003F40_003Fc9c44723_003FIHttpResponseBodyFeature_002Ecs/@EntryIndexedValue">ForceIncluded</s:String></wpf:ResourceDictionary>

HoolIt/HoolIt.http

@@ -1,11 +1,22 @@
-@HoolIt_HostAddress = http://localhost:5246
+@HoolIt_HostAddress = http://localhost:5030
 
-GET {{HoolIt_HostAddress}}/todos/
+### Root redirect
+
+GET {{HoolIt_HostAddress}}/
+Accept: text/html
+
+###
+
+### Create a dweet for a feed (query params become content)
+
+GET {{HoolIt_HostAddress}}/dweet/for/my-feed-id?temperature=21.5&status=ok
 Accept: application/json
 
 ###
 
-GET {{HoolIt_HostAddress}}/todos/1
-Accept: application/json
+### Listen for live dweets from a feed (server-sent style, newline-delimited JSON)
+
+GET {{HoolIt_HostAddress}}/listen/for/dweets/from/my-feed-id
+Accept: application/x-ndjson
 
 ###

HoolIt/Program.cs

@@ -1,6 +1,5 @@
+using System.Buffers;
 using System.Collections.Concurrent;
-using System.Runtime.CompilerServices;
-using System.Text;
 using System.Text.Json;
 using System.Text.Json.Serialization;
 using System.Threading.Channels;
@@ -17,97 +16,96 @@ var app = builder.Build();
 app.Urls.Clear();
 app.Urls.Add("http://0.0.0.0:5030");
 
-var subscribers = new ConcurrentDictionary<string, ConcurrentDictionary<ChannelWriter<string>, byte>>();
-var cancellationSources =
-    new ConcurrentDictionary<string, CancellationTokenSource>(); // To manage cancellation per feedId
+var topicSubscribers = new ConcurrentDictionary<string, ConcurrentDictionary<Guid, ChannelWriter<byte[]>>>();
 
 app.MapGet("/", () => Results.Redirect("https://github.com/mmahdium/HoolIt"));
 
-// HAPI!
-// https://github.com/jheising/HAPI
 var createApi = app.MapGroup("/dweet/for");
-createApi.MapGet("/{feedId}", async (HttpContext context, string feedId) =>
+createApi.MapGet("/{*feedId}", (HttpContext context, string feedId) =>
 {
-    var queryDataDic = context.Request.Query.ToDictionary(k => k.Key, v => v.Value[0]);
     var dweet = new Dweet
     {
-        Content = queryDataDic,
+        Content = context.Request.Query.ToDictionary(k => k.Key, v => v.Value[0])!,
         Created = DateTime.UtcNow,
         Thing = feedId
     };
 
+    var utf8Bytes = JsonSerializer.SerializeToUtf8Bytes(dweet, AppJsonSerializerContext.Default.Dweet);
+
+    if (topicSubscribers.TryGetValue(feedId, out var subscribers))
+    {
+        var writers = subscribers.Values.ToArray();
+
+        foreach (var w in writers) w.TryWrite(utf8Bytes);
+    }
+
+    return Results.Ok(new AddDweetSucceededResponse
+    {
+        This = "succeeded",
+        By = "dweeting",
+        The = "dweet",
+        With = dweet
+    });
+});
+// TODO: Add metrics
+// Subscribe endpoint
+var getLiveDataApi = app.MapGroup("/listen/for/dweets/from");
+getLiveDataApi.MapGet("/{*feedId}", async (HttpContext context, string feedId, CancellationToken reqCancellationToken) =>
+{
+    context.Response.StatusCode = 200;
+    context.Response.Headers.ContentType = "application/x-ndjson";
+    context.Response.Headers.CacheControl = "no-cache";
+    context.Response.Headers["X-Content-Type-Options"] = "nosniff";
+
+    context.Features.Get<Microsoft.AspNetCore.Http.Features.IHttpResponseBodyFeature>()?
+        .DisableBuffering();
+
+    var channel = Channel.CreateBounded<byte[]>(new BoundedChannelOptions(512) // tune buffer count
+    {
+        SingleReader = true,
+        SingleWriter = true,
+        FullMode = BoundedChannelFullMode.DropOldest
+    });
+
+    var writer = channel.Writer;
+    
+    var subscribers = topicSubscribers.GetOrAdd(feedId, _ => new ConcurrentDictionary<Guid, ChannelWriter<byte[]>>());
+    var subscriberId = Guid.CreateVersion7(DateTimeOffset.UtcNow);
+    subscribers.TryAdd(subscriberId, writer);
+
     try
     {
-        var jsonQueryData = JsonSerializer.Serialize(dweet, AppJsonSerializerContext.Default.Dweet);
+        var reader = channel.Reader;
+        
+        var bodyWriter = context.Response.BodyWriter;
 
-        if (subscribers.TryGetValue(feedId, out var feedSubs))
-            foreach (var writer in feedSubs.Keys)
-                await writer.WriteAsync(jsonQueryData);
-
-
-        return Results.Ok(new AddDweetSucceededResponse
-            { This = "succeeded", By = "dweeting", The = "dweet", With = dweet });
+        while (await reader.WaitToReadAsync(reqCancellationToken))
+        while (reader.TryRead(out var msgBytes))
+        {
+            var memory = bodyWriter.GetMemory(msgBytes.Length + 1); // +1 for newline
+            msgBytes.CopyTo(memory);
+            memory.Span[msgBytes.Length] = (byte)'\n';
+            bodyWriter.Advance(msgBytes.Length + 1);
+            
+            await bodyWriter.FlushAsync(reqCancellationToken);
+        }
     }
-    catch (Exception e)
+    catch (OperationCanceledException)
     {
-        return Results.Json(
-            new AddDweetFailedResponse
-            {
-                This = "failed", With = "WeMessedUp",
-                Because = "IDK, we couldnt dweet it. Report it at: https://github.com/mmahdium/HoolIt/issues"
-            }, statusCode: 500);
+        // client disconnected
+    }
+    finally
+    {
+        if (topicSubscribers.TryGetValue(feedId, out var existing))
+        {
+            existing.TryRemove(subscriberId, out var removedWriter);
+            removedWriter?.TryComplete();
+
+            if (existing.IsEmpty) topicSubscribers.TryRemove(feedId, out _);
+        }
     }
 });
 
-var getLiveDataApi = app.MapGroup("/listen/for/dweets/from");
-getLiveDataApi.MapGet("/{feedId}",
-    async (HttpContext context, string feedId, IHostApplicationLifetime appLifetime,
-        CancellationToken reqCancellationToken) =>
-    {
-        context.Response.StatusCode = 200;
-        context.Response.Headers.ContentType = "text/plain";
-        context.Response.Headers.CacheControl = "no-cache";
-        context.Response.Headers["X-Content-Type-Options"] = "nosniff";
-
-        // Disable response buffering
-        context.Features.Get<Microsoft.AspNetCore.Http.Features.IHttpResponseBodyFeature>()?
-            .DisableBuffering();
-
-        var channel = Channel.CreateUnbounded<string>();
-        var writer = channel.Writer;
-
-        var feedSubs = subscribers.GetOrAdd(feedId, _ => new ConcurrentDictionary<ChannelWriter<string>, byte>());
-        feedSubs.TryAdd(writer, 0);
-
-        //Console.WriteLine($"Added subscriber to feed {feedId}");
-
-        try
-        {
-            var reader = channel.Reader;
-
-            while (!reqCancellationToken.IsCancellationRequested &&
-                   await reader.WaitToReadAsync(reqCancellationToken))
-            while (reader.TryRead(out var msg))
-            {
-                await context.Response.WriteAsync(msg + "\n", reqCancellationToken);
-                await context.Response.Body.FlushAsync(reqCancellationToken);
-            }
-        }
-        catch (OperationCanceledException)
-        {
-            //Console.WriteLine($"Cancellation requested for feed {feedId}");
-        }
-        finally
-        {
-            if (subscribers.TryGetValue(feedId, out var list))
-            {
-                feedSubs.TryRemove(writer, out _);
-                //Console.WriteLine($"Removed subscriber from feed {feedId}");
-                if (list.Count == 0) subscribers.TryRemove(feedId, out _);
-            }
-        }
-    });
-
 app.Run();
 
 [JsonSerializable(typeof(Dweet))]