SSE, WebSockets, SignalR or Polling

using System.Net.WebSockets;
using System.Text.Json;

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddSingleton<StockLevels>();
var app = builder.Build();

// WebSockets bypass CORS: with cookie auth, an unchecked Origin lets any site open this socket.
var webSockets = new WebSocketOptions { KeepAliveInterval = TimeSpan.FromSeconds(20) };
webSockets.AllowedOrigins.Add("https://orders.example.com");
app.UseWebSockets(webSockets);

app.Map("/ws/stock/{sku}", async (string sku, HttpContext context, StockLevels stock) =>
{
    if (!context.WebSockets.IsWebSocketRequest)
    {
        context.Response.StatusCode = StatusCodes.Status400BadRequest;
        return;
    }

    using var socket = await context.WebSockets.AcceptWebSocketAsync();
    var ct = context.RequestAborted;
    var closed = WaitForCloseAsync(socket, ct);

    using var timer = new PeriodicTimer(TimeSpan.FromSeconds(5));
    try
    {
        // Check after the wait: the client may have closed during it, and sending then would throw.
        while (await timer.WaitForNextTickAsync(ct) && !closed.IsCompleted)
        {
            var payload = JsonSerializer.SerializeToUtf8Bytes(new { sku, available = stock.Available(sku), at = DateTimeOffset.UtcNow });
            await socket.SendAsync(payload, WebSocketMessageType.Text, endOfMessage: true, ct);
        }
        await closed;
    }
    catch (Exception ex) when (ex is WebSocketException or OperationCanceledException)
    {
        // The client vanished without a close frame or closed mid-send: an ordinary end, not an error.
    }
});

app.Run();

static async Task WaitForCloseAsync(WebSocket socket, CancellationToken ct)
{
    var buffer = new byte[1024];
    while (true)
    {
        var result = await socket.ReceiveAsync(buffer, ct);
        if (result.MessageType == WebSocketMessageType.Close)
        {
            await socket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "closing", ct);
            return;
        }
    }
}

public sealed class StockLevels
{
    private readonly Dictionary<string, int> levels = new() { ["KETTLE-BLUE-1L"] = 140, ["MUG-WHITE-350"] = 820 };

    public int Available(string sku) => levels.GetValueOrDefault(sku);
}

# In the http block: send "Connection: upgrade" only when the client asked for an upgrade.
map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      close;
}

upstream orders_api {
    server orders-api:8080;
    keepalive 32;
}

server {
    listen 443 ssl;
    http2 on;
    server_name orders.example.com;

    # Server-Sent Events: pass each event through as soon as it is written.
    location /orders/ {
        proxy_pass http://orders_api;
        proxy_http_version 1.1;
        proxy_set_header Connection "";
        proxy_buffering off;
        proxy_read_timeout 120s;
    }

    # SignalR and raw WebSockets: forward the upgrade handshake.
    location /hubs/ {
        proxy_pass http://orders_api;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;
        proxy_read_timeout 120s;
    }
}

using System.Collections.Concurrent;
using System.Net.ServerSentEvents;
using System.Runtime.CompilerServices;
using System.Security.Claims;
using System.Text.Json;
using System.Threading.Channels;
using Microsoft.AspNetCore.Authentication.Cookies;

var builder = WebApplication.CreateBuilder(args);
builder.WebHost.ConfigureKestrel(kestrel =>
{
    kestrel.ListenAnyIP(8080); // public, behind the ingress
    kestrel.ListenAnyIP(8081); // cluster-internal only, never exposed by the ingress
});
builder.Services.AddAuthentication(CookieAuthenticationDefaults.AuthenticationScheme).AddCookie();
builder.Services.AddAuthorization(options =>
    // The local port, not the spoofable Host header that RequireHost matches.
    options.AddPolicy("internal", policy => policy.RequireAssertion(context =>
        context.Resource is HttpContext http && http.Connection.LocalPort == 8081)));
builder.Services.AddSingleton<OrderNotifications>();
var app = builder.Build();
app.UseAuthentication();
app.UseAuthorization();

// The customer comes from the auth cookie, never from the URL, so nobody can watch another customer's orders.
app.MapGet("/me/notifications",
    (ClaimsPrincipal user, HttpContext http, OrderNotifications notifications, IHostApplicationLifetime lifetime) =>
{
    if (!int.TryParse(user.FindFirstValue("customer_id"), out var customerId)) return Results.Forbid();
    http.Response.Headers["X-Accel-Buffering"] = "no";
    // End streams when the host stops, or shutdown waits for them until ShutdownTimeout.
    var stop = CancellationTokenSource.CreateLinkedTokenSource(http.RequestAborted, lifetime.ApplicationStopping);
    http.Response.RegisterForDispose(stop);
    var reader = notifications.Subscribe(customerId, stop.Token);
    return TypedResults.ServerSentEvents(StreamAsync(reader, TimeSpan.FromSeconds(20), stop.Token));
}).RequireAuthorization();

app.MapPost("/internal/notifications", (OrderNotification notification, OrderNotifications notifications) =>
{
    notifications.Publish(notification);
    return TypedResults.Accepted((string?)null);
}).RequireAuthorization("internal");

app.Run();

static async IAsyncEnumerable<SseItem<string>> StreamAsync(
    ChannelReader<OrderNotification> reader, TimeSpan heartbeat, [EnumeratorCancellation] CancellationToken ct)
{
    while (true)
    {
        var ready = await WaitAsync(reader, heartbeat, ct);
        if (ready is null)
        {
            yield return new SseItem<string>("{}", "heartbeat"); // keeps idle-timeout proxies from closing the stream
            continue;
        }
        if (ready == false) yield break;
        while (reader.TryRead(out var notification))
        {
            yield return new SseItem<string>(JsonSerializer.Serialize(notification), "order");
        }
    }
}

static async Task<bool?> WaitAsync(ChannelReader<OrderNotification> reader, TimeSpan heartbeat, CancellationToken ct)
{
    using var limit = CancellationTokenSource.CreateLinkedTokenSource(ct);
    limit.CancelAfter(heartbeat);
    try
    {
        return await reader.WaitToReadAsync(limit.Token);
    }
    catch (OperationCanceledException) when (!ct.IsCancellationRequested)
    {
        return null;
    }
}

public sealed record OrderNotification(int CustomerId, int OrderId, string Status);

public sealed class OrderNotifications
{
    private readonly ConcurrentDictionary<Guid, (int CustomerId, Channel<OrderNotification> Channel)> subscribers = new();

    public ChannelReader<OrderNotification> Subscribe(int customerId, CancellationToken ct)
    {
        var id = Guid.NewGuid();
        // A slow client loses old notifications instead of growing memory without limit.
        var channel = Channel.CreateBounded<OrderNotification>(new BoundedChannelOptions(100)
        {
            FullMode = BoundedChannelFullMode.DropOldest
        });
        subscribers[id] = (customerId, channel);
        ct.Register(() =>
        {
            subscribers.TryRemove(id, out _);
            channel.Writer.TryComplete();
        });
        return channel.Reader;
    }

    public void Publish(OrderNotification notification)
    {
        foreach (var (customerId, channel) in subscribers.Values)
        {
            if (customerId == notification.CustomerId) channel.Writer.TryWrite(notification);
        }
    }
}