Deal/src/core/Deal.Api/Endpoints/EventsEndpoint.cs

using Deal.Api.Events;
using Deal.Api.Extensions;
using Deal.Api.Models;
using Deal.Api.Services;

namespace Deal.Api.Endpoints;

/// <summary>
/// SSE-поток событий канбана
/// </summary>
public static class EventsEndpoint
{
    private const string EventsPath = "/api/events";

    private const string OpenApiTag = "events";

    private const string EventStreamContentType = "text/event-stream";

    private const string NoCacheHeaderValue = "no-cache";

    private const string NoBufferingHeaderValue = "no";

    // Ping-комментарий: строки протокола SSE, начинающиеся с ':', клиент игнорирует.
    private const string PingComment = ": ping\n\n";

    private static readonly TimeSpan PingInterval = TimeSpan.FromSeconds(15);

    /// <summary>
    /// Регистрирует GET /api/events.
    /// </summary>
    /// <param name="app">Построитель маршрутов приложения.</param>
    /// <returns>Построитель маршрутов для цепочки вызовов.</returns>
    public static IEndpointRouteBuilder MapEventsEndpoint(this IEndpointRouteBuilder app)
    {
        app.MapGet(EventsPath, StreamEventsAsync).WithTags(OpenApiTag);
        return app;
    }

    // GET /api/events: поток text/event-stream канала тенанта сессии.
    // context: Контекст запроса (сессия — HttpContext.Items).
    // ct: Отмена запроса: клиент отвалился — завершаем поток и отписываемся.
    private static async Task StreamEventsAsync(HttpContext context, CancellationToken ct)
    {
        CurrentUser? user = context.GetCurrentUser();
        if (user is null)
        {
            await EndpointResults.Unauthorized(AuthHelpers.UnauthorizedDetail).ExecuteAsync(context);
            return;
        }

        SseBroker broker = context.RequestServices.GetRequiredService<SseBroker>();
        SseSubscription subscription = broker.Subscribe(user.TenantId);
        try
        {
            HttpResponse response = context.Response;
            response.ContentType = EventStreamContentType;
            response.Headers["Cache-Control"] = NoCacheHeaderValue;
            response.Headers["X-Accel-Buffering"] = NoBufferingHeaderValue;

            while (true)
            {
                using var pingTimeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
                pingTimeout.CancelAfter(PingInterval);
                try
                {
                    await subscription.Events.WaitToReadAsync(pingTimeout.Token);
                }
                catch (OperationCanceledException) when (!ct.IsCancellationRequested)
                {
                    // Тишина 15 с — ping держит соединение; отмену клиента ловит внешний catch.
                    await WriteFrameAsync(response, PingComment, ct);
                    continue;
                }

                while (subscription.Events.TryRead(out SseEvent? sseEvent))
                {
                    await WriteFrameAsync(response, sseEvent.RenderFrame(), ct);
                }
            }
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested)
        {
            // Клиент закрыл соединение — штатное завершение потока.
        }
        catch (IOException)
        {
            // Сброс соединения клиентом (закрытая вкладка/обрыв сети): ответ уже не доставить.
        }
        finally
        {
            broker.Unsubscribe(user.TenantId, subscription.Id);
        }
    }

    // Пишет frame в поток ответа и сбрасывает буфер — события уходят сразу (не пачкой).
    // response: Ответ (stream уже начат).
    // frame: Frame протокола SSE.
    // ct: Токен отмены запроса.
    private static async Task WriteFrameAsync(
        HttpResponse response,
        string frame,
        CancellationToken ct)
    {
        await response.WriteAsync(frame, ct);
        await response.Body.FlushAsync(ct);
    }
}