rudder-csharp-sdk/Services/EffectsService.cs

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

namespace RudderSdk.Core;

/// <summary>
/// Thin effects client. The server owns scenario execution; this service
/// surfaces pending effects and posts callbacks. Drive
/// <see cref="RudderClient.Update"/> every frame for the 30s heartbeat and
/// wait-deadline checks.
/// </summary>
public sealed class EffectsService
{
    private static readonly TimeSpan HeartbeatInterval = TimeSpan.FromSeconds(30);

    private readonly RudderClient _client;
    private readonly object _gate = new();
    private readonly Dictionary<(string RunId, string NodeId), string> _seen = new();
    private readonly Dictionary<(string RunId, string NodeId), DateTimeOffset> _waitDeadlines = new();

    private bool _refreshDue;
    private bool _inFlight;
    private DateTimeOffset _nextHeartbeat;

    /// <summary>Raised for a notification effect.</summary>
    public event Action<NotificationEffect>? OnNotification;

    /// <summary>Raised for a store-offer effect.</summary>
    public event Action<StoreOfferEffect>? OnStoreOffer;

    /// <summary>Raised for a leaderboard effect.</summary>
    public event Action<LeaderboardEffect>? OnLeaderboard;

    /// <summary>Raised for a wait effect.</summary>
    public event Action<WaitEffect>? OnWait;

    /// <summary>Raised for a quest effect.</summary>
    public event Action<QuestEffect>? OnQuest;

    /// <summary>Raised for a battle-pass effect.</summary>
    public event Action<BattlePassEffect>? OnBattlePass;

    /// <summary>Raised for a battle-pass-level effect.</summary>
    public event Action<BattlePassLevelEffect>? OnBattlePassLevel;

    /// <summary>Raised when a run finishes all its nodes.</summary>
    public event Action<ScenarioCompletedEffect>? OnScenarioCompleted;

    /// <summary>Raised when a run is dropped after a definitive server rejection or an unsupported effect type.</summary>
    public event Action<ScenarioFailedEffect>? OnScenarioFailed;

    internal EffectsService(RudderClient client)
    {
        _client = client;
        _nextHeartbeat = _client.Clock.UtcNow + HeartbeatInterval;
        if (!string.IsNullOrEmpty(_client.TokenStore.GetAccessToken()))
            _refreshDue = true;

        _client.Auth.AuthStateChanged += OnAuthStateChanged;
    }

    /// <summary>Pumps heartbeat and wait-deadline checks; call every frame.</summary>
    public void Update(float deltaTime)
    {
        if (string.IsNullOrEmpty(_client.TokenStore.GetAccessToken()))
            return;

        var now = _client.Clock.UtcNow;
        lock (_gate)
        {
            if (_inFlight)
                return;
            if (!_refreshDue && now < _nextHeartbeat && !HasDueWaitUnlocked(now))
                return;
            _refreshDue = false;
            _inFlight = true;
        }

        _ = RefreshPendingAsync();
    }

    internal void Ingest(IEnumerable<PendingEffect>? effects)
    {
        if (effects == null)
            return;

        var batch = new List<PendingEffect>();
        lock (_gate)
        {
            foreach (var effect in effects)
            {
                if (effect == null || string.IsNullOrEmpty(effect.RunId) || string.IsNullOrEmpty(effect.NodeId))
                    continue;

                var key = (effect.RunId, effect.NodeId);
                if (_seen.ContainsKey(key))
                    continue;
                _seen[key] = effect.ScenarioSlug;

                if (string.Equals(effect.Type, EffectTypes.Wait, StringComparison.Ordinal))
                {
                    var deadline = effect.WaitDeadline ?? _client.Clock.UtcNow;
                    _waitDeadlines[key] = deadline;
                }

                batch.Add(effect);
            }
        }

        foreach (var effect in batch)
            Dispatch(effect);
    }

    internal async Task CompleteAsync(PendingEffect source, string handle, CancellationToken cancellationToken = default)
    {
        try
        {
            var response = await _client.SendAsync<HandleScenarioCallbackRequest, HandleScenarioCallbackResponse>(
                "POST",
                "/sdk/v1/scenarios/callback",
                new HandleScenarioCallbackRequest
                {
                    ScenarioSlug = source.ScenarioSlug,
                    NodeId = source.NodeId,
                    Handle = handle,
                    RunId = source.RunId
                },
                cancellationToken).ConfigureAwait(false);

            ForgetWait(source.RunId, source.NodeId);
            var next = ReadEffect(response?.Effect);
            if (next == null)
                Emit(OnScenarioCompleted, new ScenarioCompletedEffect(source.RunId, source.ScenarioSlug));
            else
                Ingest(new[] { next });
        }
        catch (Exception ex) when (IsDefinitiveRejection(ex))
        {
            DropRun(source.RunId, source.ScenarioSlug, source.NodeId, ex);
        }
    }

    internal async Task ReportProgressAsync(
        PendingEffect source,
        string counterKey,
        long amount,
        CancellationToken cancellationToken = default)
    {
        try
        {
            var response = await _client.SendAsync<UpdateScenarioCounterRequest, UpdateScenarioCounterResponse>(
                "POST",
                "/sdk/v1/scenarios/counter",
                new UpdateScenarioCounterRequest
                {
                    ScenarioSlug = source.ScenarioSlug,
                    NodeId = source.NodeId,
                    CounterKey = counterKey,
                    Amount = amount,
                    RunId = source.RunId
                },
                cancellationToken).ConfigureAwait(false);

            if (response == null || response.Completed != true)
                return;

            ForgetWait(source.RunId, source.NodeId);
            var next = ReadEffect(response.Effect);
            if (next == null)
                Emit(OnScenarioCompleted, new ScenarioCompletedEffect(source.RunId, source.ScenarioSlug));
            else
                Ingest(new[] { next });
        }
        catch (Exception ex) when (IsDefinitiveRejection(ex))
        {
            DropRun(source.RunId, source.ScenarioSlug, source.NodeId, ex);
        }
        catch (Exception ex)
        {
            _client.Options.Logger?.Log(
                RudderLogLevel.Warning,
                $"[Rudder] Scenario counter update failed at node {source.NodeId}: {ex.Message}");
        }
    }

    private void OnAuthStateChanged(RudderAuthState state)
    {
        if (state == RudderAuthState.SignedIn)
        {
            lock (_gate)
                _refreshDue = true;
            return;
        }

        lock (_gate)
        {
            _seen.Clear();
            _waitDeadlines.Clear();
            _refreshDue = false;
            _nextHeartbeat = _client.Clock.UtcNow + HeartbeatInterval;
        }
    }

    private async Task RefreshPendingAsync()
    {
        try
        {
            var response = await _client.SendAsync<ListPendingScenarioEffectsResponse>(
                "GET",
                "/sdk/v1/scenarios/pending",
                CancellationToken.None).ConfigureAwait(false);

            if (response != null)
            {
                Ingest(response.Effects);
                Reconcile(response.Effects);
            }

            lock (_gate)
                _nextHeartbeat = _client.Clock.UtcNow + HeartbeatInterval;
        }
        catch (Exception ex)
        {
            _client.Options.Logger?.Log(
                RudderLogLevel.Warning,
                "[Rudder] Failed to refresh pending scenario effects. " + ex.Message);
            lock (_gate)
                _nextHeartbeat = _client.Clock.UtcNow + HeartbeatInterval;
        }
        finally
        {
            lock (_gate)
                _inFlight = false;
        }
    }

    private void Dispatch(PendingEffect effect)
    {
        var handle = new EffectHandle(this, effect);
        switch (effect.Type ?? string.Empty)
        {
            case EffectTypes.Notification:
                Emit(OnNotification, new NotificationEffect(handle));
                break;
            case EffectTypes.Store:
                Emit(OnStoreOffer, new StoreOfferEffect(handle));
                break;
            case EffectTypes.Leaderboard:
                Emit(OnLeaderboard, new LeaderboardEffect(handle));
                break;
            case EffectTypes.Wait:
                Emit(OnWait, new WaitEffect(handle, effect.WaitDeadline ?? _client.Clock.UtcNow));
                break;
            case EffectTypes.Quest:
                Emit(OnQuest, new QuestEffect(handle));
                break;
            case EffectTypes.BattlePass:
                Emit(OnBattlePass, new BattlePassEffect(handle, _client.BattlePass));
                break;
            case EffectTypes.BattlePassLevel:
                Emit(OnBattlePassLevel, new BattlePassLevelEffect(handle));
                break;
            default:
                _client.Options.Logger?.Log(
                    RudderLogLevel.Warning,
                    $"[Rudder] Unsupported scenario node type '{effect.Type}' ({effect.NodeId}).");
                Emit(
                    OnScenarioFailed,
                    new ScenarioFailedEffect(
                        effect.RunId,
                        effect.ScenarioSlug,
                        effect.NodeId,
                        new Exception($"Unsupported scenario node type '{effect.Type}'")));
                break;
        }
    }

    private void DropRun(string runId, string scenarioSlug, string nodeId, Exception exception)
    {
        lock (_gate)
        {
            var toRemove = new List<(string RunId, string NodeId)>();
            foreach (var key in _seen.Keys)
            {
                if (key.RunId == runId)
                    toRemove.Add(key);
            }

            foreach (var key in toRemove)
            {
                _seen.Remove(key);
                _waitDeadlines.Remove(key);
            }
        }

        Emit(OnScenarioFailed, new ScenarioFailedEffect(runId, scenarioSlug, nodeId, exception));
    }

    private void Reconcile(IEnumerable<PendingEffect>? effects)
    {
        var incoming = new HashSet<(string RunId, string NodeId)>();
        if (effects != null)
        {
            foreach (var effect in effects)
            {
                if (effect == null || string.IsNullOrEmpty(effect.RunId) || string.IsNullOrEmpty(effect.NodeId))
                    continue;

                incoming.Add((effect.RunId, effect.NodeId));
            }
        }

        var finished = new Dictionary<string, string>();
        lock (_gate)
        {
            var stale = new List<(string RunId, string NodeId)>();
            foreach (var key in _seen.Keys)
            {
                if (!incoming.Contains(key))
                    stale.Add(key);
            }

            foreach (var key in stale)
            {
                finished[key.RunId] = _seen[key];
                _seen.Remove(key);
                _waitDeadlines.Remove(key);
            }

            foreach (var key in _seen.Keys)
                finished.Remove(key.RunId);
        }

        foreach (var entry in finished)
            Emit(OnScenarioCompleted, new ScenarioCompletedEffect(entry.Key, entry.Value));
    }

    private void ForgetWait(string runId, string nodeId)
    {
        lock (_gate)
            _waitDeadlines.Remove((runId, nodeId));
    }

    private bool HasDueWaitUnlocked(DateTimeOffset now)
    {
        foreach (var deadline in _waitDeadlines.Values)
        {
            if (now >= deadline)
                return true;
        }

        return false;
    }

    private void Emit<T>(Action<T>? handlers, T effect)
    {
        if (handlers == null)
            return;

        foreach (var subscriber in handlers.GetInvocationList())
        {
            try
            {
                ((Action<T>)subscriber).Invoke(effect);
            }
            catch (Exception ex)
            {
                _client.Options.Logger?.Log(
                    RudderLogLevel.Error,
                    "[Rudder] Effect handler failed. " + ex.Message);
            }
        }
    }

    private static PendingEffect? ReadEffect(JToken? token)
    {
        if (token == null || token.Type == JTokenType.Null)
            return null;

        return token.ToObject<PendingEffect>();
    }

    private static bool IsDefinitiveRejection(Exception ex)
    {
        if (ex is RudderNotFoundException)
            return true;

        return ex is RudderApiException api
            && (api.Code == RudderErrorCodes.UnknownRun
                || api.Code == RudderErrorCodes.RunExpired
                || api.Code == RudderErrorCodes.RunNotActive);
    }
}

internal static class EffectTypes
{
    public const string Notification = "notification";
    public const string Store = "store";
    public const string Leaderboard = "leaderboard";
    public const string Wait = "wait";
    public const string Quest = "quest";
    public const string BattlePass = "battlepass";
    public const string BattlePassLevel = "battlepass_level";
}

internal sealed class EffectHandle
{
    private readonly EffectsService _service;
    private readonly PendingEffect _effect;

    public EffectHandle(EffectsService service, PendingEffect effect)
    {
        _service = service;
        _effect = effect;
    }

    public PendingEffect Source => _effect;

    public string RunId => _effect.RunId;

    public string ScenarioSlug => _effect.ScenarioSlug;

    public string NodeId => _effect.NodeId;

    public JObject Data => _effect.Data as JObject ?? new JObject();

    public T Get<T>(string key, T defaultValue = default!)
    {
        if (!Data.TryGetValue(key, out var value) || value == null || value.Type == JTokenType.Null)
            return defaultValue;

        try
        {
            return value.ToObject<T>() ?? defaultValue;
        }
        catch
        {
            return defaultValue;
        }
    }

    public Task CompleteAsync(string handle, CancellationToken cancellationToken = default)
        => _service.CompleteAsync(_effect, handle, cancellationToken);

    public void Complete(string handle) => _ = CompleteAsync(handle);

    public Task ReportProgressAsync(string counterKey, long amount, CancellationToken cancellationToken = default)
        => _service.ReportProgressAsync(_effect, counterKey, amount, cancellationToken);

    public void ReportProgress(string counterKey, long amount) => _ = ReportProgressAsync(counterKey, amount);
}