nexus/backend/Controllers/DomainEventsController.cs

using System.Text.Json;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc;
using Microsoft.EntityFrameworkCore;
using Nexus.Api.Data;
using Nexus.Api.Models;
using Nexus.Api.Observability;
using Nexus.Api.Services;

namespace Nexus.Api.Controllers;

[Authorize]
[ApiController]
[Route("api/v1/events")]
public sealed class DomainEventsController(
    NexusDbContext db,
    IDomainEventStreamService eventStream,
    ILogger<DomainEventsController> logger) : ControllerBase
{
    private const int MaximumReplay = 512;

    [HttpGet]
    public async Task Get(
        [FromQuery] string? channels,
        [FromQuery] long? afterSequence,
        CancellationToken cancellationToken)
    {
        Response.Headers.ContentType = "text/event-stream";
        Response.Headers.CacheControl = "no-cache, no-store, must-revalidate";
        Response.Headers.Connection = "keep-alive";
        Response.Headers["X-Accel-Buffering"] = "no";

        var requestedChannels = ParseChannels(channels);
        var cursor = ResolveCursor(afterSequence);
        await using var subscription = eventStream.Subscribe(requestedChannels);
        var lastSent = cursor ?? subscription.StartingSequence;

        if (cursor.HasValue)
        {
            var earliest = await db.OutboxEvents
                .AsNoTracking()
                .Where(item => item.PublishedAt != null)
                .OrderBy(item => item.Sequence)
                .Select(item => (long?)item.Sequence)
                .FirstOrDefaultAsync(cancellationToken);
            if (earliest.HasValue && cursor.Value < earliest.Value - 1)
            {
                await WriteResyncRequiredAsync(
                    cursor.Value,
                    "retention_gap",
                    cancellationToken);
                return;
            }

            var replay = await db.OutboxEvents
                .AsNoTracking()
                .Where(item =>
                    item.PublishedAt != null &&
                    item.Sequence > cursor.Value &&
                    item.Sequence <= subscription.StartingSequence)
                .OrderBy(item => item.Sequence)
                .Take(MaximumReplay + 1)
                .ToListAsync(cancellationToken);

            if (replay.Count > MaximumReplay)
            {
                await WriteResyncRequiredAsync(
                    cursor.Value,
                    "replay_limit",
                    cancellationToken);
                return;
            }

            foreach (var item in replay)
            {
                var domainEvent = DomainEventStreamService.Map(item);
                if (!MatchesChannel(domainEvent, requestedChannels))
                    continue;
                await WriteEventAsync("domain", domainEvent.Sequence, domainEvent, cancellationToken);
                lastSent = domainEvent.Sequence;
            }
        }

        try
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                using var iteration = CancellationTokenSource.CreateLinkedTokenSource(
                    cancellationToken);
                var readTask = subscription.Reader
                    .WaitToReadAsync(iteration.Token)
                    .AsTask();
                var heartbeatTask = Task.Delay(
                    TimeSpan.FromSeconds(20),
                    iteration.Token);
                var completed = await Task.WhenAny(readTask, heartbeatTask);
                if (completed == heartbeatTask)
                {
                    iteration.Cancel();
                    await WriteEventAsync(
                        "heartbeat",
                        lastSent,
                        new
                        {
                            sequence = eventStream.CurrentSequence,
                            timestamp = DateTimeOffset.UtcNow
                        },
                        cancellationToken);
                    continue;
                }

                iteration.Cancel();
                if (!await readTask)
                    break;
                while (subscription.Reader.TryRead(out var domainEvent))
                {
                    if (domainEvent.Sequence <= lastSent)
                        continue;
                    await WriteEventAsync(
                        "domain",
                        domainEvent.Sequence,
                        domainEvent,
                        cancellationToken);
                    lastSent = domainEvent.Sequence;
                }
            }
        }
        catch (DomainEventSubscriberOverflowException)
        {
            await WriteResyncRequiredAsync(lastSent, "subscriber_overflow", cancellationToken);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            // Normal browser disconnect.
        }
        catch (Exception exception)
        {
            logger.LogWarning(exception, "Domain event stream ended unexpectedly");
        }
    }

    private long? ResolveCursor(long? queryCursor)
    {
        if (queryCursor.HasValue)
            return queryCursor.Value >= 0 ? queryCursor : null;
        if (!Request.Headers.TryGetValue("Last-Event-ID", out var header))
            return null;
        return long.TryParse(header.FirstOrDefault(), out var parsed) && parsed >= 0
            ? parsed
            : null;
    }

    private async Task WriteResyncRequiredAsync(
        long staleSequence,
        string reason,
        CancellationToken cancellationToken)
    {
        NexusTelemetry.SseResyncs.Add(1);
        // Resume after the latest sequence that was visible when the resync
        // decision was made. Re-emitting the stale cursor would make a client
        // reconnect into the same retention/replay gap forever.
        var resumeSequence = eventStream.CurrentSequence;
        var payload = JsonSerializer.SerializeToElement(new
        {
            reason,
            staleSequence,
            resumeSequence
        });
        var domainEvent = new DomainEventDto(
            resumeSequence,
            DomainEventTypes.ResyncRequired,
            new EntityRefDto("event-stream", "*"),
            0,
            DateTimeOffset.UtcNow,
            payload);
        await WriteEventAsync(
            DomainEventTypes.ResyncRequired,
            resumeSequence,
            domainEvent,
            cancellationToken);
    }

    private async Task WriteEventAsync(
        string eventName,
        long sequence,
        object payload,
        CancellationToken cancellationToken)
    {
        await Response.WriteAsync($"id: {sequence}\n", cancellationToken);
        await Response.WriteAsync($"event: {eventName}\n", cancellationToken);
        await Response.WriteAsync(
            $"data: {JsonSerializer.Serialize(payload, JsonOptions)}\n\n",
            cancellationToken);
        await Response.Body.FlushAsync(cancellationToken);
    }

    private static readonly JsonSerializerOptions JsonOptions =
        new(JsonSerializerDefaults.Web);

    private static IReadOnlySet<string> ParseChannels(string? channels)
    {
        if (string.IsNullOrWhiteSpace(channels))
            return new HashSet<string>(["*"], StringComparer.OrdinalIgnoreCase);

        var parsed = channels
            .Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries)
            .Select(channel => channel.ToLowerInvariant())
            .Where(channel => channel.All(character =>
                char.IsLetterOrDigit(character) || character is '-' or '_'))
            .Take(16)
            .ToHashSet(StringComparer.OrdinalIgnoreCase);
        return parsed.Count == 0
            ? new HashSet<string>(["*"], StringComparer.OrdinalIgnoreCase)
            : parsed;
    }

    private static bool MatchesChannel(
        DomainEventDto domainEvent,
        IReadOnlySet<string> channels)
    {
        if (channels.Contains("*"))
            return true;
        var channel = domainEvent.Entity.Type switch
        {
            "agent-proposal" => "agents",
            "task" => "tasks",
            "run" => "runs",
            "cron" => "cron",
            _ => $"{domainEvent.Entity.Type}s"
        };
        return channels.Contains(channel);
    }
}