dotnet-cqrs/Svrnty.CQRS.Events/Subscriptions/EventDeliveryService.cs

using System;
using Svrnty.CQRS.Events.Delivery;
using Svrnty.CQRS.Events.Abstractions.EventStore;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Svrnty.CQRS.Events.Abstractions;
using Svrnty.CQRS.Events.Abstractions.Subscriptions;

namespace Svrnty.CQRS.Events.Subscriptions;

/// <summary>
/// Service responsible for filtering and delivering events to subscriptions.
/// </summary>
public sealed class EventDeliveryService : IPersistentSubscriptionDeliveryService
{
    private readonly IPersistentSubscriptionStore _subscriptionStore;
    private readonly IEventStreamStore _eventStore;
    private readonly ILogger<EventDeliveryService> _logger;

    public EventDeliveryService(
        IPersistentSubscriptionStore subscriptionStore,
        IEventStreamStore eventStore,
        ILogger<EventDeliveryService> logger)
    {
        _subscriptionStore = subscriptionStore ?? throw new ArgumentNullException(nameof(subscriptionStore));
        _eventStore = eventStore ?? throw new ArgumentNullException(nameof(eventStore));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
    }

    public async Task<int> DeliverEventAsync(
        string correlationId,
        ICorrelatedEvent @event,
        long sequence,
        CancellationToken cancellationToken = default)
    {
        // Get all active subscriptions for this correlation ID
        var subscriptions = await _subscriptionStore.GetByCorrelationIdAsync(correlationId, cancellationToken);
        var activeSubscriptions = subscriptions.Where(s => s.CanReceiveEvents).ToList();

        if (activeSubscriptions.Count == 0)
        {
            _logger.LogDebug(
                "No active subscriptions found for correlation {CorrelationId}",
                correlationId);
            return 0;
        }

        var deliveredCount = 0;

        foreach (var subscription in activeSubscriptions)
        {
            // Check if this event type should be delivered
            var eventTypeName = @event.GetType().Name;
            if (!subscription.ShouldDeliverEventType(eventTypeName))
            {
                _logger.LogDebug(
                    "Event type {EventType} not in subscription {SubscriptionId} filter",
                    eventTypeName,
                    subscription.Id);
                continue;
            }

            // Check delivery mode
            if (subscription.DeliveryMode == DeliveryMode.OnReconnect)
            {
                _logger.LogDebug(
                    "Subscription {SubscriptionId} is OnReconnect mode, skipping immediate delivery",
                    subscription.Id);
                // Still update sequence for catch-up tracking
                subscription.MarkDelivered(sequence);
                await _subscriptionStore.UpdateAsync(subscription, cancellationToken);
                continue;
            }

            // For batched mode, we'll deliver on interval (handled elsewhere)
            // For now, we just track that this event is available
            if (subscription.DeliveryMode == DeliveryMode.Batched)
            {
                _logger.LogDebug(
                    "Subscription {SubscriptionId} is Batched mode, event will be delivered in batch",
                    subscription.Id);
                // Still update sequence for catch-up tracking
                subscription.MarkDelivered(sequence);
                await _subscriptionStore.UpdateAsync(subscription, cancellationToken);
                continue;
            }

            // Immediate delivery
            deliveredCount++;

            // Update last delivered sequence
            subscription.MarkDelivered(sequence);

            _logger.LogDebug(
                "Event {EventType} (sequence {Sequence}) delivered to subscription {SubscriptionId}",
                eventTypeName,
                sequence,
                subscription.Id);

            // Check if this is a terminal event
            if (subscription.IsTerminalEvent(eventTypeName))
            {
                subscription.Complete();
                _logger.LogInformation(
                    "Terminal event {EventType} received, subscription {SubscriptionId} completed",
                    eventTypeName,
                    subscription.Id);
            }

            // Save updated subscription
            await _subscriptionStore.UpdateAsync(subscription, cancellationToken);
        }

        return deliveredCount;
    }

    public async Task<int> CatchUpSubscriptionAsync(
        string subscriptionId,
        CancellationToken cancellationToken = default)
    {
        var subscription = await _subscriptionStore.GetByIdAsync(subscriptionId, cancellationToken);
        if (subscription == null)
        {
            _logger.LogWarning(
                "Cannot catch up: subscription {SubscriptionId} not found",
                subscriptionId);
            return 0;
        }

        if (!subscription.CanReceiveEvents)
        {
            _logger.LogDebug(
                "Subscription {SubscriptionId} cannot receive events (status: {Status})",
                subscriptionId,
                subscription.Status);
            return 0;
        }

        // Get missed events
        var missedEvents = await GetPendingEventsAsync(subscriptionId, cancellationToken: cancellationToken);

        if (missedEvents.Count == 0)
        {
            _logger.LogDebug(
                "No missed events for subscription {SubscriptionId}",
                subscriptionId);
            return 0;
        }

        _logger.LogInformation(
            "Catching up subscription {SubscriptionId} with {Count} missed events",
            subscriptionId,
            missedEvents.Count);

        var deliveredCount = missedEvents.Count;

        // Check for terminal events
        foreach (var @event in missedEvents)
        {
            var eventTypeName = @event.GetType().Name;
            if (subscription.IsTerminalEvent(eventTypeName))
            {
                subscription.Complete();
                await _subscriptionStore.UpdateAsync(subscription, cancellationToken);

                _logger.LogInformation(
                    "Terminal event {EventType} received during catch-up, subscription {SubscriptionId} completed",
                    eventTypeName,
                    subscriptionId);

                break; // Stop processing after terminal event
            }
        }

        // Update subscription with latest sequence
        await _subscriptionStore.UpdateAsync(subscription, cancellationToken);

        _logger.LogInformation(
            "Caught up subscription {SubscriptionId} with {Count} events",
            subscriptionId,
            deliveredCount);

        return deliveredCount;
    }

    public async Task<IReadOnlyList<ICorrelatedEvent>> GetPendingEventsAsync(
        string subscriptionId,
        int limit = 100,
        CancellationToken cancellationToken = default)
    {
        var subscription = await _subscriptionStore.GetByIdAsync(subscriptionId, cancellationToken);
        if (subscription == null)
        {
            return Array.Empty<ICorrelatedEvent>();
        }

        // Read events from the stream starting after the last delivered sequence
        var events = await _eventStore.ReadStreamAsync(
            streamName: subscription.CorrelationId, // Use correlation ID as stream identifier
            fromOffset: subscription.LastDeliveredSequence + 1,
            maxCount: limit,
            cancellationToken: cancellationToken);

        // Filter by event types if specified
        var filteredEvents = events
            .Where(e => subscription.ShouldDeliverEventType(e.GetType().Name))
            .ToList();

        return filteredEvents;
    }
}