接入实时通知能力：Server-Sent Events (SSE) 高性能流式推送

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

namespace BitzOrcas.Platform.Notifications.Infrastructure.Channels;

/// <summary>
/// 租户级实时通知流式分发通道中心
/// </summary>
public sealed class NotificationChannelHub
{
    // 双层字典保证租户数据隔离：TenantId -> (UserId -> Channel)
    private readonly ConcurrentDictionary<string, ConcurrentDictionary<string, Channel<NotificationPayload>>> _tenantChannels = new();

    /// <summary>
    /// 为特定用户注册并获取流式读取通道
    /// </summary>
    public ChannelReader<NotificationPayload> Subscribe(string tenantId, string userId)
    {
        var userMap = _tenantChannels.GetOrAdd(tenantId, _ => new ConcurrentDictionary<string, Channel<NotificationPayload>>());

        // 使用单读单写的高吞吐 bounded 内存通道，设置背压缓冲容量为 100
        var channel = userMap.GetOrAdd(userId, _ => Channel.CreateBounded<NotificationPayload>(new BoundedChannelOptions(100)
        {
            FullMode = BoundedChannelFullMode.DropOldest
        }));

        return channel.Reader;
    }

    /// <summary>
    /// 注销用户长连接会话
    /// </summary>
    public void Unsubscribe(string tenantId, string userId)
    {
        if (_tenantChannels.TryGetValue(tenantId, out var userMap))
        {
            userMap.TryRemove(userId, out _);
        }
    }

    /// <summary>
    /// 向指定租户下的目标用户推送增量通知
    /// </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))
        {
            // 写入通道，若用户在线立即触发 HTTP 流式刷新
            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. 设置 Server-Sent Events 标准协议响应标头
            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
            {
                // 发送连接就绪事件
                await context.Response.WriteAsync("event: connected\ndata: {\"status\":\"ok\"}\n\n", cancellationToken);
                await context.Response.Body.FlushAsync(cancellationToken);

                // 2. 循环消费通道中的实时消息流
                while (!cancellationToken.IsCancellationRequested)
                {
                    // 设置 15 秒心跳保活检测，防止云厂商 NAT 网关或反向代理单向超时断开
                    using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
                    cts.CancelAfter(TimeSpan.FromSeconds(15));

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

                        // 格式化为 SSE 标准包体：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)
                    {
                        // 15 秒内无业务消息，推送 SSE 空白注释包保活心跳
                        await context.Response.WriteAsync(":ping\n\n", cancellationToken);
                        await context.Response.Body.FlushAsync(cancellationToken);
                    }
                }
            }
            finally
            {
                // 3. 连接断开时确保资源回收注销
                channelHub.Unsubscribe(tenantId, userId);
            }
        })
        .RequireAuthorization()
        .WithTags("Notifications")
        .WithSummary("建立租户级实时通知 SSE 流式长连接");
    }
}

import { useEffect } from 'react';

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

    // 建立基于标准 HTTP 的 SSE 长连接 (通过 URL 参数或 Cookie 传递鉴权凭证)
    const eventSource = new EventSource(`/api/notifications/stream?access_token=${encodeURIComponent(accessToken)}`);

    // 监听立案审批提醒事件
    eventSource.addEventListener('matter_alert', (event: MessageEvent) => {
      const data = JSON.parse(event.data);
      console.info('[SSE] 收到待办与利益冲突预警:', data);

      // 触发 UI 侧全局 Toast 提醒并增量刷新未读数角标
      window.dispatchEvent(new CustomEvent('bitz:notification:received', { detail: data }));
    });

    // 原生处理网络故障与自动重连
    eventSource.onerror = (err) => {
      console.warn('[SSE] 连接中断，浏览器将自动根据指数退避策略重试...', err);
    };

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