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

using Microsoft.EntityFrameworkCore;
using w4c_workflows.Data;
using w4c_workflows.Models;
using w4c_workflows.Services.Runs;

namespace w4c_workflows.Services.Triggers;

/// <summary>
/// Scheduled-trigger engine (cron + interval), modeled on w4c-webapi's
/// <see cref="AlertScheduler"/>. On each tick it loads every tenant's compiled
/// workflows that declare a <c>cron</c> or <c>interval</c> trigger, fires the
/// ones whose schedule has elapsed since the last fire, and records the fire
/// time for dedup. An error in one workflow never breaks the cycle for others.
/// </summary>
public class TriggerScheduler : BackgroundService
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ITriggerState _state;
    private readonly TimeProvider _time;
    private readonly ILogger<TriggerScheduler> _logger;
    private readonly int _pollSeconds;
    private readonly int _maxInFlight;

    public TriggerScheduler(
        IServiceScopeFactory scopeFactory,
        ITriggerState state,
        TimeProvider time,
        IConfiguration config,
        ILogger<TriggerScheduler> logger)
    {
        _scopeFactory = scopeFactory;
        _state = state;
        _time = time;
        _logger = logger;
        _pollSeconds = config.GetValue("Workflows:SchedulerPollSeconds", 30);
        // Backpressure: never let an auto-trigger pile up more in-flight runs
        // than the worker can plausibly clear. When the cap is exceeded the
        // trigger is deferred (lastFire is NOT advanced) so it fires again once
        // the backlog drains — a saturated control plane keeps creating work
        // otherwise, which is how run queues stack into overload.
        _maxInFlight = config.GetValue("Workflows:TriggerMaxInFlight", 5);
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("TriggerScheduler started (poll {Interval}s)", _pollSeconds);
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(_pollSeconds));

        try
        {
            // First pass after a short warm-up so the app can serve initial requests.
            await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
            await EvaluateAllAsync(stoppingToken);

            while (await timer.WaitForNextTickAsync(stoppingToken))
            {
                await EvaluateAllAsync(stoppingToken);
            }
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
        {
            // Graceful shutdown — the host is stopping.
        }
        catch (Exception ex)
        {
            // A transient env/DB error must never take the whole control plane
            // down (HostOptions.BackgroundServiceExceptionBehavior defaults to
            // StopHost). Log and keep the scheduler alive for the next tick.
            _logger.LogError(ex, "TriggerScheduler loop failed; will retry on next tick");
        }
    }

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

        var workflows = await db.Workflows
            .Where(w => w.Status == WorkflowStatus.Compiled && w.TriggerJson != null)
            .ToListAsync(ct);

        foreach (var workflow in workflows)
        {
            if (ct.IsCancellationRequested)
                return;
            if (!workflow.TriggerEnabled)
                continue; // auto-trigger toggled off from the UI; manual runs still work
            await EvaluateWorkflowAsync(workflow, db, launcher, now, ct);
        }
    }

    private async Task EvaluateWorkflowAsync(Workflow workflow, WorkflowsDbContext db, IRunLauncher launcher, DateTimeOffset now, CancellationToken ct)
    {
        var spec = TriggerSpec.Parse(workflow.TriggerJson, out var error);
        if (spec == null)
        {
            _logger.LogWarning("Skipping unparseable trigger for workflow {WorkflowId}: {Error}", workflow.Id, error);
            return;
        }
        if (spec.Type != TriggerType.Cron && spec.Type != TriggerType.Interval)
            return;

        try
        {
            var lastFire = await _state.GetLastFireAsync(workflow.TenantId, workflow.Id, ct);
            if (!TriggerDue.IsDue(spec, lastFire, now))
                return;

            // Backpressure guard: if the tenant already has a full queue of
            // in-flight runs for this workflow, defer the trigger. This is what
            // stops an interval/cron trigger from stacking a never-draining
            // backlog when the worker cannot keep up (single-core / slow tenants).
            if (await IsOverloadedAsync(workflow, db, ct))
            {
                _logger.LogDebug(
                    "Deferring {Type} trigger for workflow {WorkflowId} (tenant {TenantId}): max {Max} in-flight runs reached",
                    spec.Type, workflow.Id, workflow.TenantId, _maxInFlight);
                return;
            }

            var correlation = $"trigger:{workflow.Id}:{now.ToUnixTimeSeconds()}";
            await launcher.LaunchAsync(
                new LaunchRequest(workflow.TenantId, workflow.Id, workflow.TriggerJson, null, correlation), ct);
            await _state.SetLastFireAsync(workflow.TenantId, workflow.Id, now, ct);

            _logger.LogDebug(
                "Trigger {Type} fired for workflow {WorkflowId} (tenant {TenantId})",
                spec.Type, workflow.Id, workflow.TenantId);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Trigger evaluation failed for workflow {WorkflowId} (tenant {TenantId})",
                workflow.Id, workflow.TenantId);
        }
    }

    /// <summary>
    /// Returns true when the tenant already has <see cref="_maxInFlight"/> or more
    /// in-flight runs (pending/running/compensating) for the given workflow. A
    /// deferred trigger leaves <c>lastFire</c> untouched so the same schedule is
    /// re-evaluated next tick once the backlog clears.
    /// </summary>
    private async Task<bool> IsOverloadedAsync(Workflow workflow, WorkflowsDbContext db, CancellationToken ct)
    {
        if (_maxInFlight <= 0)
            return false;

        var inFlight = await db.WorkflowRuns.CountAsync(r =>
            r.WorkflowId == workflow.Id
            && r.TenantId == workflow.TenantId
            && (r.Status == RunStatus.Pending
                || r.Status == RunStatus.Running
                || r.Status == RunStatus.Compensating), ct);

        return inFlight >= _maxInFlight;
    }
}