rudder-csharp-sdk/Services/Scenarios/ScenarioService.Runtime.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using RudderSdk.Core.Abstractions;
using RudderSdk.Core.Models;
using RudderSdk.Core.Models.Scenarios;

namespace RudderSdk.Core;

/// <summary>
/// Thrown when a boundary HTTP call fails with a transient error
/// (network failure, timeout, or server 5xx). The caller should NOT advance
/// the run; the node stays active and the handle stays pending for retry.
/// </summary>
public sealed class TransientBoundaryException : Exception
{
    /// <summary>Creates the exception.</summary>
    public TransientBoundaryException(string message, Exception inner)
        : base(message, inner)
    {
    }

    /// <summary>Creates the exception wrapping the transport error.</summary>
    public TransientBoundaryException(Exception inner)
        : base($"Transient boundary error: {inner.Message}", inner)
    {
    }
}

public sealed partial class ScenarioService
{
    /// <summary>
    /// Returns true for exceptions that may succeed on retry
    /// (timeout / network failures).
    /// </summary>
    private static bool IsTransientException(Exception ex)
    {
        return ex is OperationCanceledException || ex is RudderNetworkException;
    }

    private static bool IsRankNotEligible(Exception ex)
    {
        return ex is RudderApiException api && api.Code == "rank_not_eligible";
    }

    private IReadOnlyList<PlanRun> StartPlans(IEnumerable<ExecutionPlan>? plans)
    {
        var started = new List<PlanRun>();
        if (plans == null)
            return started;

        foreach (var plan in plans)
        {
            var run = StartPlan(plan);
            if (run != null)
                started.Add(ToPlanRun(run));
        }

        return started;
    }

    private RuntimeRun? StartPlan(ExecutionPlan? plan)
    {
        if (plan?.Nodes == null || plan.Nodes.Count == 0)
            return null;

        // Dedup: server returned same runId — skip without restarting.
        if (!string.IsNullOrEmpty(plan.RunId) && _runs.ContainsKey(plan.RunId))
            return null;

        if (plan.BoundaryNodes?.Count > 0 && string.IsNullOrEmpty(plan.RunId))
            throw new InvalidOperationException("ExecutionPlan has boundaryNodes but missing RunId.");

        var runId = plan.RunId ?? Guid.NewGuid().ToString("N");
        var startNode = FindNode(plan, plan.StartNodeId) ?? plan.Nodes[0];
        var run = new RuntimeRun(runId, plan);
        _runs[run.RunId] = run;
        ActivateNode(run, startNode.Id);
        Persist();
        return run;
    }

    /// <summary>
    /// Replaces an existing run's plan with a server-provided continuation (same RunId):
    /// the previous segment is done, the new segment's start node becomes active.
    /// Idempotent: if the continuation was already applied, nothing is re-dispatched.
    /// </summary>
    private void ReplaceRun(ExecutionPlan? plan, string? fallbackRunId = null)
    {
        if (plan?.Nodes == null || plan.Nodes.Count == 0)
            return;

        var runId = plan.RunId ?? fallbackRunId;
        if (string.IsNullOrEmpty(runId))
            return;

        var startNode = FindNode(plan, plan.StartNodeId) ?? plan.Nodes[0];
        if (startNode == null)
            return;

        if (_runs.TryGetValue(runId, out var existing) && existing.ActiveNodes.ContainsKey(startNode.Id))
            return; // continuation already applied (idempotent callback echo)

        var run = new RuntimeRun(runId, plan);
        _runs[runId] = run;
        ActivateNode(run, startNode.Id);
        Persist();
    }

    private void ActivateNode(RuntimeRun run, string nodeId, ActiveNodeState? restoredState = null)
    {
        var node = FindNode(run.Plan, nodeId);
        if (node == null)
            return;

        var state = restoredState ?? new ActiveNodeState { NodeId = nodeId };
        run.ActiveNodes[nodeId] = state;
        DispatchActiveNode(run, state, restored: restoredState != null);
    }

    internal Task CompleteNodeAsync(
        string runId,
        string nodeId,
        string handle,
        CancellationToken cancellationToken = default)
    {
        return CompleteNodeAsync(runId, nodeId, handle, continueOnBoundary: true, cancellationToken);
    }

    private async Task CompleteNodeAsync(
        string runId,
        string nodeId,
        string handle,
        bool continueOnBoundary,
        CancellationToken cancellationToken = default)
    {
        if (!_runs.TryGetValue(runId, out var run))
            return;

        var key = CompletedHandleKey(nodeId, handle);
        if (run.CompletedHandles.Contains(key))
            return; // idempotent

        try
        {
            var continuedOnBoundary = false;
            if (continueOnBoundary)
            {
                try
                {
                    continuedOnBoundary = await ContinueBoundaryAsync(
                        run, nodeId, handle, cancellationToken).ConfigureAwait(false);
                }
                catch (TransientBoundaryException)
                {
                    // Transient error — don't advance the run.
                    // Node stays active, handle stays pending for retry on reconnect.
                    return;
                }
            }

            if (continuedOnBoundary)
            {
                // The boundary may have replaced or removed the run (continuation plan
                // or reconcile). The transferred run owns the state — touching the stale
                // object here would complete or delete the new run.
                if (!_runs.TryGetValue(runId, out var currentRun) || !ReferenceEquals(currentRun, run))
                    return;
            }

            // Only now — after the server has confirmed — mark the handle and node.
            run.CompletedHandles.Add(key);
            run.ActiveNodes.Remove(nodeId);
            Persist();

            if (!continuedOnBoundary)
            {
                foreach (var edge in MatchingEdges(run.Plan, nodeId, handle))
                    ActivateNode(run, edge.Target);
            }

            CheckRunCompleted(run);
            Persist();
        }
        catch (Exception ex)
        {
            if (IsRankNotEligible(ex))
                throw;
            FailRun(run, nodeId, ex);
        }
    }

    private async Task<bool> ContinueBoundaryAsync(
        RuntimeRun run,
        string nodeId,
        string handle,
        CancellationToken cancellationToken)
    {
        var boundaries = MatchingBoundaryNodes(run.Plan, nodeId, handle).ToList();
        if (boundaries.Count == 0)
            return false;

        foreach (var boundary in boundaries)
        {
            try
            {
                var response = await _client.SendAsync<HandleScenarioCallbackRequest, HandleScenarioCallbackResponse>(
                    "POST",
                    "/sdk/v1/scenarios/callback",
                    new HandleScenarioCallbackRequest
                    {
                        ScenarioId = run.Plan.ScenarioId,
                        NodeId = boundary.SourceNodeId,
                        Handle = boundary.SourceHandle,
                        RunId = run.RunId
                    },
                    cancellationToken).ConfigureAwait(false);

                if (response?.Plan != null)
                {
                    // Continuation of the current run (server keeps the RunId) replaces
                    // the run's plan; fresh/terminal plans start as new runs.
                    if (_runs.ContainsKey(response.Plan.RunId ?? string.Empty))
                        ReplaceRun(response.Plan);
                    else
                        StartPlan(response.Plan);
                }
            }
            catch (Exception ex)
            {
                // Boundary call failed — try to reconcile with server.
                var reconciled = false;
                try
                {
                    var reconcile = await _client.SendAsync<GetScenarioRunRequest, GetScenarioRunResponse>(
                        "POST",
                        "/sdk/v1/scenarios/run",
                        new GetScenarioRunRequest { RunId = run.RunId },
                        cancellationToken).ConfigureAwait(false);

                    if (reconcile?.Status == "unknown_run" || reconcile?.Status == "expired")
                    {
                        _runs.Remove(run.RunId);
                        Persist();
                        reconciled = true;
                    }

                    if (reconcile?.Plan != null)
                    {
                        ReplaceRun(reconcile.Plan, run.RunId);
                        reconciled = true;
                    }
                }
                catch
                {
                    // Reconciliation also failed.
                }

                if (!reconciled)
                {
                    // Reconcile did not resolve — distinguish transient from terminal.
                    if (IsTransientException(ex))
                        throw new TransientBoundaryException(ex);
                    // Terminal error — let the caller fail the run.
                    throw;
                }
                // If reconciled, the boundary was handled (run corrected or removed).
                // Fall through to continue to the next boundary.
            }
        }

        return true;
    }

    private void CheckRunCompleted(RuntimeRun run)
    {
        if (run.ActiveNodes.Count > 0)
            return;

        _runs.Remove(run.RunId);
        Persist();
        OnScenarioCompleted?.Invoke(ToPlanRun(run));
    }

    private void FailRun(RuntimeRun run, string nodeId, Exception ex)
    {
        _client.Options.Logger?.Log(RudderLogLevel.Error, $"[Rudder] Scenario run {run.RunId} failed at node {nodeId}: {ex.Message}");
        _runs.Remove(run.RunId);
        Persist();
        OnScenarioFailed?.Invoke(new ScenarioFailedEvent(ToPlanRun(run), nodeId, ex));
    }

    private static PlanRun ToPlanRun(RuntimeRun run)
    {
        return new PlanRun(
            run.RunId,
            run.Plan.PlanId,
            run.Plan.ScenarioId,
            run.Plan.UserId,
            run.ActiveNodes.Keys.ToList(),
            run.Plan);
    }

    private static ExecutionPlanNode? FindNode(ExecutionPlan? plan, string? nodeId)
    {
        if (plan?.Nodes == null || plan.Nodes.Count == 0)
            return null;

        if (!string.IsNullOrEmpty(nodeId))
        {
            foreach (var node in plan.Nodes)
            {
                if (node.Id == nodeId)
                    return node;
            }
        }

        return null;
    }

    private static IEnumerable<PlanEdge> MatchingEdges(ExecutionPlan plan, string sourceNodeId, string sourceHandle)
    {
        if (plan?.Edges == null)
            yield break;

        foreach (var edge in plan.Edges)
        {
            if (edge.Source == sourceNodeId && string.Equals(edge.SourceHandle ?? string.Empty, sourceHandle ?? string.Empty, StringComparison.Ordinal))
                yield return edge;
        }
    }

    private static IEnumerable<BoundaryNode> MatchingBoundaryNodes(ExecutionPlan plan, string sourceNodeId, string sourceHandle)
    {
        if (plan?.BoundaryNodes == null)
            yield break;

        foreach (var boundary in plan.BoundaryNodes)
        {
            if (boundary.SourceNodeId == sourceNodeId && string.Equals(boundary.SourceHandle ?? string.Empty, sourceHandle ?? string.Empty, StringComparison.Ordinal))
                yield return boundary;
        }
    }

    private static string CompletedHandleKey(string nodeId, string? handle) => nodeId + ":" + (handle ?? string.Empty);
}