w4c-workflows-api/Services/Triggers/HandlerStreamConsumer.cs

using System.Text.Json;
using Microsoft.EntityFrameworkCore;
using w4c_workflows.Data;
using w4c_workflows.Models;
using w4c_workflows.Services.Messaging;
using w4c_workflows.Services.Quota;
using w4c_workflows.Services.Runs;

namespace w4c_workflows.Services.Triggers;

/// <summary>
/// Queue-trigger consumer for <c>handler</c>-mode workflows. A handler workflow
/// subscribes to a stream (<c>trigger.stream</c>, default <c>wf:{tenant}:events</c>)
/// and, on each event, starts a run whose input is the event payload.
///
/// v1 consumes each event as its own run; the long-lived-instance semantics
/// (one durable instance, state accumulation across events, checkpoint/resume)
/// are layered on by the run lifecycle engine in step 9, which reuses the same
/// correlation id scheme (<c>handler:{workflowId}:{messageId}</c>).
/// </summary>
public class HandlerStreamConsumer : BackgroundService
{
    /// <summary>How long a fetched handler list is reused before re-querying.</summary>
    private static readonly TimeSpan HandlerCacheTtl = TimeSpan.FromSeconds(2);

    private readonly IServiceScopeFactory _scopeFactory;
    private readonly IEventBus _events;
    private readonly ILogger<HandlerStreamConsumer> _logger;

    private readonly string _consumer;
    private readonly int _pollDelayMs;
    private readonly int _batchSize;
    private readonly TimeSpan _claimIdle;

    // The handler list changes rarely; a short TTL removes a query per 500 ms
    // poll while still picking up new/toggled workflows within seconds.
    private List<HandlerSubscription>? _handlers;
    private DateTime _handlersAt;
    private readonly HashSet<(string TenantId, string Stream)> _ensuredGroups = new();

    public HandlerStreamConsumer(
        IServiceScopeFactory scopeFactory,
        IEventBus events,
        IConfiguration config,
        ILogger<HandlerStreamConsumer> logger)
    {
        _scopeFactory = scopeFactory;
        _events = events;
        _logger = logger;

        _consumer = $"handler-{Environment.MachineName}-{Guid.NewGuid():N}"[..28];
        _pollDelayMs = ParseInt(config["Workflows:HandlerPollDelayMs"], 500);
        _batchSize = ParseInt(config["Workflows:HandlerBatchSize"], 10);
        _claimIdle = TimeSpan.FromSeconds(ParseInt(config["Workflows:HandlerClaimIdleSeconds"], 30));
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("HandlerStreamConsumer started as consumer {Consumer}", _consumer);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                var processed = await ProcessAsync(stoppingToken);
                if (processed == 0)
                    await Task.Delay(_pollDelayMs, stoppingToken);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Handler stream consumer loop error");
                try { await Task.Delay(_pollDelayMs, stoppingToken); } catch (OperationCanceledException) { break; }
            }
        }

        _logger.LogInformation("HandlerStreamConsumer stopped");
    }

    private async Task<int> ProcessAsync(CancellationToken ct)
    {
        using var scope = _scopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<WorkflowsDbContext>();
        var launcher = scope.ServiceProvider.GetRequiredService<IRunLauncher>();

        var handlers = await GetHandlersAsync(db, ct);

        var processed = 0;
        foreach (var workflow in handlers)
        {
            if (ct.IsCancellationRequested)
                return processed;
            if (!workflow.TriggerEnabled)
                continue; // auto-trigger toggled off from the UI

            var spec = TriggerSpec.Parse(workflow.TriggerJson, out _);
            if (spec == null)
                continue;

            var stream = string.IsNullOrWhiteSpace(spec.Stream) ? Streams.DefaultEvents : spec.Stream;

            // The group is immutable once created, so ensure it once per process
            // rather than once per poll per workflow.
            if (_ensuredGroups.Add((workflow.TenantId, stream)))
                await _events.EnsureGroupAsync(workflow.TenantId, stream, ct);

            var messages = await _events.ReadGroupAsync(workflow.TenantId, stream, _consumer, _batchSize, ct);
            if (messages.Count == 0)
                messages = await _events.ClaimPendingAsync(workflow.TenantId, stream, _consumer, _claimIdle, _batchSize, ct);

            foreach (var message in messages)
            {
                ct.ThrowIfCancellationRequested();
                try
                {
                    await DispatchAsync(launcher, workflow, stream, message, ct);
                    processed++;
                }
                catch (WorkflowQuotaExceededException)
                {
                    // No quota left for this tenant. Leave the message UNACKED so it
                    // is redelivered (not dropped) once the quota frees up; stop
                    // draining this workflow's batch to avoid a tight retry loop.
                    _logger.LogWarning(
                        "Handler event for workflow {WorkflowId} (tenant {TenantId}) deferred: execution quota exhausted",
                        workflow.Id, workflow.TenantId);
                    break;
                }
            }
        }

        return processed;
    }

    /// <summary>Minimal handler projection, cached for a short TTL.</summary>
    private sealed record HandlerSubscription(Guid Id, string TenantId, string? TriggerJson, bool TriggerEnabled);

    private async Task<IReadOnlyList<HandlerSubscription>> GetHandlersAsync(WorkflowsDbContext db, CancellationToken ct)
    {
        if (_handlers != null && DateTime.UtcNow - _handlersAt < HandlerCacheTtl)
            return _handlers;

        _handlers = await db.Workflows
            .Where(w => w.Status == WorkflowStatus.Compiled && w.Mode == WorkflowMode.Handler)
            .Select(w => new HandlerSubscription(w.Id, w.TenantId, w.TriggerJson, w.TriggerEnabled))
            .ToListAsync(ct);
        _handlersAt = DateTime.UtcNow;
        return _handlers;
    }

    private async Task DispatchAsync(IRunLauncher launcher, HandlerSubscription workflow, string stream, StreamMessage message, CancellationToken ct)
    {
        var input = SerializeEventInput(message.Fields);
        var correlation = $"handler:{workflow.Id}:{message.Id}";

        await launcher.LaunchAsync(
            new LaunchRequest(workflow.TenantId, workflow.Id, workflow.TriggerJson, input, correlation), ct);
        await _events.AckAsync(workflow.TenantId, stream, message.Id, ct);
    }

    /// <summary>Serializes a stream message's fields as a JSON object (the event payload).</summary>
    internal static string SerializeEventInput(IReadOnlyDictionary<string, string> fields)
        => JsonSerializer.Serialize(fields, JsonSerializerOptions.Web);

    private static int ParseInt(string? text, int fallback)
        => int.TryParse(text, out var value) && value > 0 ? value : fallback;
}