nexus/backend/Services/DomainEventStreamService.cs

using System.Collections.Concurrent;
using System.Data;
using System.Text.Json;
using System.Threading.Channels;
using Microsoft.EntityFrameworkCore;
using Nexus.Api.Data;
using Nexus.Api.Models;
using Nexus.Api.Observability;

namespace Nexus.Api.Services;

/// <summary>
/// Publishes the PostgreSQL transactional outbox to bounded in-process SSE
/// subscribers. PostgreSQL remains authoritative: the in-process channel only
/// reduces latency, while reconnect replay is always read from the database.
/// </summary>
public sealed class DomainEventStreamService(
    IServiceScopeFactory scopeFactory,
    ILogger<DomainEventStreamService> logger) :
    BackgroundService,
    IDomainEventStreamService
{
    private const int BatchSize = 128;
    private const int SubscriberCapacity = 64;
    private const int MinimumRetainedSequences = 10_000;
    private static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(1);
    private static readonly TimeSpan Retention = TimeSpan.FromHours(24);

    private readonly ConcurrentDictionary<Guid, Subscriber> subscribers = new();
    private long currentSequence;
    private long reportedBacklog;
    private DateTimeOffset nextRetentionSweep = DateTimeOffset.UtcNow.AddMinutes(10);

    public long CurrentSequence => Interlocked.Read(ref currentSequence);

    public DomainEventSubscription Subscribe(IReadOnlySet<string> channels)
    {
        var id = Guid.NewGuid();
        var queue = Channel.CreateBounded<DomainEventDto>(
            new BoundedChannelOptions(SubscriberCapacity)
            {
                FullMode = BoundedChannelFullMode.Wait,
                SingleReader = true,
                SingleWriter = true,
                AllowSynchronousContinuations = false
            });
        subscribers[id] = new Subscriber(queue, channels);
        NexusTelemetry.SseSubscribers.Add(1);

        return new DomainEventSubscription(
            queue.Reader,
            CurrentSequence,
            () =>
            {
                if (subscribers.TryRemove(id, out var removed))
                {
                    removed.Queue.Writer.TryComplete();
                    NexusTelemetry.SseSubscribers.Add(-1);
                }

                return ValueTask.CompletedTask;
            });
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await InitializeSequenceAsync(stoppingToken);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                var published = await PublishNextBatchAsync(stoppingToken);
                if (DateTimeOffset.UtcNow >= nextRetentionSweep)
                {
                    await PruneRetainedEventsAsync(stoppingToken);
                    nextRetentionSweep = DateTimeOffset.UtcNow.AddHours(1);
                }

                if (!published)
                    await Task.Delay(PollInterval, stoppingToken);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception exception)
            {
                logger.LogError(exception, "Domain outbox iteration failed");
                await Task.Delay(PollInterval, stoppingToken);
            }
        }

        foreach (var (id, subscriber) in subscribers)
        {
            if (subscribers.TryRemove(id, out _))
            {
                subscriber.Queue.Writer.TryComplete();
                NexusTelemetry.SseSubscribers.Add(-1);
            }
        }

        var backlog = Interlocked.Exchange(ref reportedBacklog, 0);
        if (backlog != 0)
            NexusTelemetry.OutboxBacklog.Add(-backlog);
    }

    internal static DomainEventDto Map(OutboxEvent item)
    {
        using var document = JsonDocument.Parse(item.PayloadJson);
        var entityType = NormalizeEntityType(item.AggregateType);
        return new DomainEventDto(
            item.Sequence,
            item.Type,
            new EntityRefDto(entityType, item.AggregateId),
            item.AggregateRevision,
            item.OccurredAt,
            document.RootElement.Clone());
    }

    private async Task InitializeSequenceAsync(CancellationToken cancellationToken)
    {
        await using var scope = scopeFactory.CreateAsyncScope();
        var db = scope.ServiceProvider.GetRequiredService<NexusDbContext>();
        var latest = await db.OutboxEvents
            .AsNoTracking()
            .Where(item => item.PublishedAt != null)
            .Select(item => (long?)item.Sequence)
            .MaxAsync(cancellationToken) ?? 0;
        Interlocked.Exchange(ref currentSequence, latest);
    }

    private async Task<bool> PublishNextBatchAsync(CancellationToken cancellationToken)
    {
        await using var scope = scopeFactory.CreateAsyncScope();
        var db = scope.ServiceProvider.GetRequiredService<NexusDbContext>();
        List<OutboxEvent> pending;

        if (db.Database.IsRelational())
        {
            await using var transaction = await db.Database.BeginTransactionAsync(
                IsolationLevel.ReadCommitted,
                cancellationToken);
            pending = await db.OutboxEvents
                .FromSqlInterpolated($"""
                    SELECT * FROM "OutboxEvents"
                    WHERE "PublishedAt" IS NULL
                    ORDER BY "Sequence"
                    LIMIT {BatchSize}
                    FOR UPDATE SKIP LOCKED
                    """)
                .ToListAsync(cancellationToken);

            var publishedAt = DateTimeOffset.UtcNow;
            foreach (var item in pending)
            {
                item.PublishedAt = publishedAt;
                item.PublishAttempts++;
                item.LastErrorCode = null;
            }

            await db.SaveChangesAsync(cancellationToken);
            await transaction.CommitAsync(cancellationToken);
        }
        else
        {
            pending = await db.OutboxEvents
                .Where(item => item.PublishedAt == null)
                .OrderBy(item => item.Sequence)
                .Take(BatchSize)
                .ToListAsync(cancellationToken);
            var publishedAt = DateTimeOffset.UtcNow;
            foreach (var item in pending)
            {
                item.PublishedAt = publishedAt;
                item.PublishAttempts++;
                item.LastErrorCode = null;
            }

            await db.SaveChangesAsync(cancellationToken);
        }

        foreach (var item in pending)
        {
            var domainEvent = Map(item);
            Interlocked.Exchange(ref currentSequence, domainEvent.Sequence);
            Publish(domainEvent);
            NexusTelemetry.OutboxPublished.Add(1);
        }

        var nextBacklogEstimate = pending.Count == BatchSize ? BatchSize : 0;
        var previousBacklog = Interlocked.Exchange(
            ref reportedBacklog,
            nextBacklogEstimate);
        NexusTelemetry.OutboxBacklog.Add(nextBacklogEstimate - previousBacklog);
        return pending.Count > 0;
    }

    private void Publish(DomainEventDto domainEvent)
    {
        var channel = ChannelFor(domainEvent);
        foreach (var (id, subscriber) in subscribers)
        {
            if (!subscriber.Channels.Contains("*") &&
                !subscriber.Channels.Contains(channel))
            {
                continue;
            }

            if (subscriber.Queue.Writer.TryWrite(domainEvent))
                continue;

            if (subscribers.TryRemove(id, out var removed))
            {
                removed.Queue.Writer.TryComplete(
                    new DomainEventSubscriberOverflowException());
                NexusTelemetry.SseSubscribers.Add(-1);
                NexusTelemetry.SseResyncs.Add(1);
            }
        }
    }

    private async Task PruneRetainedEventsAsync(CancellationToken cancellationToken)
    {
        await using var scope = scopeFactory.CreateAsyncScope();
        var db = scope.ServiceProvider.GetRequiredService<NexusDbContext>();
        var maximum = await db.OutboxEvents
            .Where(item => item.PublishedAt != null)
            .Select(item => (long?)item.Sequence)
            .MaxAsync(cancellationToken) ?? 0;
        var sequenceCutoff = Math.Max(0, maximum - MinimumRetainedSequences);
        var timeCutoff = DateTimeOffset.UtcNow - Retention;
        if (sequenceCutoff == 0)
            return;

        if (db.Database.IsRelational())
        {
            await db.OutboxEvents
                .Where(item =>
                    item.PublishedAt != null &&
                    item.Sequence < sequenceCutoff &&
                    item.OccurredAt < timeCutoff)
                .ExecuteDeleteAsync(cancellationToken);
            return;
        }

        var expired = await db.OutboxEvents
            .Where(item =>
                item.PublishedAt != null &&
                item.Sequence < sequenceCutoff &&
                item.OccurredAt < timeCutoff)
            .ToListAsync(cancellationToken);
        db.OutboxEvents.RemoveRange(expired);
        await db.SaveChangesAsync(cancellationToken);
    }

    private static string ChannelFor(DomainEventDto domainEvent) =>
        domainEvent.Entity.Type switch
        {
            "agent-proposal" => "agents",
            "task" => "tasks",
            "run" => "runs",
            "cron" => "cron",
            "notification" => "notifications",
            "incident" => "incidents",
            _ => $"{domainEvent.Entity.Type}s"
        };

    private static string NormalizeEntityType(string aggregateType)
    {
        var normalized = aggregateType
            .Trim()
            .Replace("_", "-", StringComparison.Ordinal)
            .ToLowerInvariant();
        return normalized switch
        {
            "agentproposal" or "agent-proposal" => "agent-proposal",
            "worktask" or "task" => "task",
            "openclawrun" or "run" => "run",
            "cronjob" or "cron" => "cron",
            _ => normalized
        };
    }

    private sealed record Subscriber(
        Channel<DomainEventDto> Queue,
        IReadOnlySet<string> Channels);
}