Integrating Real-Time Notifications: Server-Sent Events (SSE) High-Performance Streaming

using System.Collections.Concurrent;
using System.Threading.Channels;

namespace BitzOrcas.Platform.Notifications.Infrastructure.Channels;

/// <summary>
/// Tenant-isolated real-time notification streaming hub.
/// </summary>
public sealed class NotificationChannelHub
{
    // Nested dictionary enforcing tenancy boundary: TenantId -> (UserId -> Channel)
    private readonly ConcurrentDictionary<string, ConcurrentDictionary<string, Channel<NotificationPayload>>> _tenantChannels = new();

    /// <summary>
    /// Subscribes an authenticated user to their dedicated event channel.
    /// </summary>
    public ChannelReader<NotificationPayload> Subscribe(string tenantId, string userId)
    {
        var userMap = _tenantChannels.GetOrAdd(tenantId, _ => new ConcurrentDictionary<string, Channel<NotificationPayload>>());

        // Bounded channel with backpressure dropping oldest items if reader lags
        var channel = userMap.GetOrAdd(userId, _ => Channel.CreateBounded<NotificationPayload>(new BoundedChannelOptions(100)
        {
            FullMode = BoundedChannelFullMode.DropOldest
        }));

        return channel.Reader;
    }

    /// <summary>
    /// Unsubscribes and purges an active session channel upon disconnect.
    /// </summary>
    public void Unsubscribe(string tenantId, string userId)
    {
        if (_tenantChannels.TryGetValue(tenantId, out var userMap))
        {
            userMap.TryRemove(userId, out _);
        }
    }

    /// <summary>
    /// Publishes a payload to a target user within a specific tenant partition.
    /// </summary>
    public async ValueTask PublishToUserAsync(string tenantId, string userId, NotificationPayload payload, CancellationToken cancellationToken = default)
    {
        if (_tenantChannels.TryGetValue(tenantId, out var userMap) &&
            userMap.TryGetValue(userId, out var channel))
        {
            // Writes to channel, waking the active HTTP response writer
            await channel.Writer.WriteAsync(payload, cancellationToken);
        }
    }
}

public sealed record NotificationPayload(
    string NotificationId,
    string EventType,
    string Title,
    string Content,
    DateTimeOffset CreatedAtUtc);

using System.Text.Json;
using BitzOrcas.Application.Abstractions.Tenancy;
using BitzOrcas.Application.Security;
using BitzOrcas.Platform.Notifications.Infrastructure.Channels;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Routing;

namespace BitzOrcas.Platform.Notifications.Application.Endpoints;

public static class NotificationStreamEndpoint
{
    public static void MapNotificationStream(this IEndpointRouteBuilder app)
    {
        app.MapGet("/api/notifications/stream", async (
            HttpContext context,
            ICurrentUser currentUser,
            ICurrentTenant currentTenant,
            NotificationChannelHub channelHub,
            CancellationToken cancellationToken) =>
        {
            // 1. Set standard Server-Sent Events protocol headers
            context.Response.Headers.ContentType = "text/event-stream";
            context.Response.Headers.CacheControl = "no-cache";
            context.Response.Headers.Connection = "keep-alive";

            var tenantId = currentTenant.Tenant.EffectiveTenantId;
            var userId = currentUser.UserId;
            var reader = channelHub.Subscribe(tenantId, userId);

            try
            {
                // Dispatch initial connection handshake event
                await context.Response.WriteAsync("event: connected\ndata: {\"status\":\"ok\"}\n\n", cancellationToken);
                await context.Response.Body.FlushAsync(cancellationToken);

                // 2. Consume events from bounded channel
                while (!cancellationToken.IsCancellationRequested)
                {
                    // 15-second timeout to emit ping heartbeats preventing intermediary proxy disconnects
                    using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
                    cts.CancelAfter(TimeSpan.FromSeconds(15));

                    try
                    {
                        var payload = await reader.ReadAsync(cts.Token);
                        var json = JsonSerializer.Serialize(payload);

                        // Format according to standard SSE specification: event: <name>\ndata: <json>\n\n
                        await context.Response.WriteAsync($"event: {payload.EventType}\n", cancellationToken);
                        await context.Response.WriteAsync($"data: {json}\n\n", cancellationToken);
                        await context.Response.Body.FlushAsync(cancellationToken);
                    }
                    catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
                    {
                        // Emit SSE comment line as heartbeat
                        await context.Response.WriteAsync(":ping\n\n", cancellationToken);
                        await context.Response.Body.FlushAsync(cancellationToken);
                    }
                }
            }
            finally
            {
                // 3. Clean up subscription upon client disconnection
                channelHub.Unsubscribe(tenantId, userId);
            }
        })
        .RequireAuthorization()
        .WithTags("Notifications")
        .WithSummary("Establish tenant-isolated SSE real-time notification stream");
    }
}

import { useEffect } from 'react';

export function useNotificationStream(accessToken: string) {
  useEffect(() => {
    if (!accessToken) return;

    // Connect via standard HTTP SSE endpoint
    const eventSource = new EventSource(`/api/notifications/stream?access_token=${encodeURIComponent(accessToken)}`);

    // Listen for incoming legal conflict and approval alerts
    eventSource.addEventListener('matter_alert', (event: MessageEvent) => {
      const data = JSON.parse(event.data);
      console.info('[SSE] Received real-time alert:', data);

      // Dispatch browser custom event to update badges and trigger toast UI
      window.dispatchEvent(new CustomEvent('bitz:notification:received', { detail: data }));
    });

    // Native browser automatic exponential backoff error handling
    eventSource.onerror = (err) => {
      console.warn('[SSE] Connection interrupted, browser retrying automatically...', err);
    };

    return () => {
      eventSource.close();
    };
  }, [accessToken]);
}