diff --git a/Abstractions/IClock.cs b/Abstractions/IClock.cs index f1b6896..81ae017 100644 --- a/Abstractions/IClock.cs +++ b/Abstractions/IClock.cs @@ -2,7 +2,7 @@ using System; namespace RudderSdk.Core.Abstractions; -/// Time source used by the scenario runtime; override in tests. +/// Time source used by the effects client; override in tests. public interface IClock { /// Current UTC time. diff --git a/Abstractions/IPlanScheduler.cs b/Abstractions/IPlanScheduler.cs deleted file mode 100644 index d1fad85..0000000 --- a/Abstractions/IPlanScheduler.cs +++ /dev/null @@ -1,12 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; - -namespace RudderSdk.Core.Abstractions; - -/// Optional scheduler abstraction for delayed plan work. -public interface IPlanScheduler -{ - /// Completes after the given delay. - Task ScheduleAsync(TimeSpan delay, CancellationToken cancellationToken = default); -} diff --git a/Abstractions/IPlanStateStore.cs b/Abstractions/IPlanStateStore.cs deleted file mode 100644 index 5c26c61..0000000 --- a/Abstractions/IPlanStateStore.cs +++ /dev/null @@ -1,11 +0,0 @@ -namespace RudderSdk.Core.Abstractions; - -/// -/// Persists serialized scenario-run state between app launches. -/// Null by default (no persistence). -/// -public interface IPlanStateStore -{ - /// Serialized state blob, or null when empty. - string? State { get; set; } -} diff --git a/Abstractions/IRealtimeTransport.cs b/Abstractions/IRealtimeTransport.cs deleted file mode 100644 index 0bfb160..0000000 --- a/Abstractions/IRealtimeTransport.cs +++ /dev/null @@ -1,36 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; - -namespace RudderSdk.Core.Abstractions; - -/// -/// Low-level realtime (websocket) transport. There is no default implementation; -/// provide one through (Unity ships its own). -/// -public interface IRealtimeTransport -{ - /// Raised when the connection closes. - event Action Closed; - - /// Raised for every incoming message. - event Action> Received; - - /// Raised on transport errors. - event Action Error; - - /// True while the connection is open. - bool IsConnected { get; } - - /// Opens the connection, giving up after . - Task ConnectAsync(Uri uri, TimeSpan timeout, CancellationToken cancellationToken = default); - - /// Sends one message. - Task SendAsync(ArraySegment data, CancellationToken cancellationToken = default); - - /// Closes the connection. - Task CloseAsync(CancellationToken cancellationToken = default); - - /// Pumps time-dependent logic; call every frame. - void Update(float deltaTime); -} diff --git a/Abstractions/IRealtimeTransportFactory.cs b/Abstractions/IRealtimeTransportFactory.cs deleted file mode 100644 index 9b2e849..0000000 --- a/Abstractions/IRealtimeTransportFactory.cs +++ /dev/null @@ -1,8 +0,0 @@ -namespace RudderSdk.Core.Abstractions; - -/// Creates realtime transports on demand (one per connection). -public interface IRealtimeTransportFactory -{ - /// Creates a new, unconnected transport. - IRealtimeTransport Create(); -} diff --git a/BattlePass/AddBattlePassXpResponse.cs b/BattlePass/AddBattlePassXpResponse.cs index 8607406..3e39dcc 100644 --- a/BattlePass/AddBattlePassXpResponse.cs +++ b/BattlePass/AddBattlePassXpResponse.cs @@ -1,6 +1,5 @@ // Code generated by apigen. DO NOT EDIT. using Newtonsoft.Json; -using RudderSdk.Core.Models; namespace RudderSdk.Core.Models.BattlePass; @@ -15,9 +14,6 @@ public class AddBattlePassXpResponse [JsonProperty("maxLevel")] public bool? MaxLevel { get; set; } - [JsonProperty("plan")] - public ExecutionPlan? Plan { get; set; } - [JsonProperty("xp")] public long? Xp { get; set; } diff --git a/BattlePass/PurchaseBattlePassPremiumResponse.cs b/BattlePass/PurchaseBattlePassPremiumResponse.cs index 7ffbdf6..c64be98 100644 --- a/BattlePass/PurchaseBattlePassPremiumResponse.cs +++ b/BattlePass/PurchaseBattlePassPremiumResponse.cs @@ -1,6 +1,5 @@ // Code generated by apigen. DO NOT EDIT. using Newtonsoft.Json; -using RudderSdk.Core.Models; namespace RudderSdk.Core.Models.BattlePass; @@ -9,9 +8,6 @@ public class PurchaseBattlePassPremiumResponse [JsonProperty("error")] public string? Error { get; set; } - [JsonProperty("plan")] - public ExecutionPlan? Plan { get; set; } - [JsonProperty("success")] public bool? Success { get; set; } diff --git a/Models/BoundaryNode.cs b/Models/BoundaryNode.cs deleted file mode 100644 index 9b5b268..0000000 --- a/Models/BoundaryNode.cs +++ /dev/null @@ -1,30 +0,0 @@ -// Code generated by apigen. DO NOT EDIT. -using Newtonsoft.Json; -using System; - -namespace RudderSdk.Core.Models; - -public class BoundaryNode -{ - [JsonProperty("callbackUrl")] - public string? CallbackUrl { get; set; } - - [JsonProperty("enforcement")] - public string? Enforcement { get; set; } - - [JsonProperty("enteredAt")] - public DateTimeOffset? EnteredAt { get; set; } - - [JsonProperty("nodeId")] - public string? NodeId { get; set; } - - [JsonProperty("sourceHandle")] - public string? SourceHandle { get; set; } - - [JsonProperty("sourceNodeId")] - public string? SourceNodeId { get; set; } - - [JsonProperty("waitDeadline")] - public DateTimeOffset? WaitDeadline { get; set; } - -} diff --git a/Models/ExecutionPlan.cs b/Models/ExecutionPlan.cs deleted file mode 100644 index 5311ab4..0000000 --- a/Models/ExecutionPlan.cs +++ /dev/null @@ -1,37 +0,0 @@ -// Code generated by apigen. DO NOT EDIT. -using Newtonsoft.Json; -using Newtonsoft.Json.Linq; -using System.Collections.Generic; - -namespace RudderSdk.Core.Models; - -public class ExecutionPlan -{ - [JsonProperty("boundaryNodes")] - public List? BoundaryNodes { get; set; } - - [JsonProperty("context")] - public JToken? Context { get; set; } - - [JsonProperty("edges")] - public List? Edges { get; set; } - - [JsonProperty("nodes")] - public List? Nodes { get; set; } - - [JsonProperty("planId")] - public string? PlanId { get; set; } - - [JsonProperty("runId")] - public string? RunId { get; set; } - - [JsonProperty("scenarioId")] - public string? ScenarioId { get; set; } - - [JsonProperty("startNodeId")] - public string? StartNodeId { get; set; } - - [JsonProperty("userId")] - public string? UserId { get; set; } - -} diff --git a/Models/ExecutionPlanNode.cs b/Models/ExecutionPlanNode.cs deleted file mode 100644 index cf40b6b..0000000 --- a/Models/ExecutionPlanNode.cs +++ /dev/null @@ -1,18 +0,0 @@ -// Code generated by apigen. DO NOT EDIT. -using Newtonsoft.Json; -using Newtonsoft.Json.Linq; - -namespace RudderSdk.Core.Models; - -public class ExecutionPlanNode -{ - [JsonProperty("data")] - public JToken? Data { get; set; } - - [JsonProperty("id")] - public string? Id { get; set; } - - [JsonProperty("type")] - public string? Type { get; set; } - -} diff --git a/Models/PlanEdge.cs b/Models/PlanEdge.cs deleted file mode 100644 index 9e897c6..0000000 --- a/Models/PlanEdge.cs +++ /dev/null @@ -1,23 +0,0 @@ -// Code generated by apigen. DO NOT EDIT. -using Newtonsoft.Json; - -namespace RudderSdk.Core.Models; - -public class PlanEdge -{ - [JsonProperty("id")] - public string? Id { get; set; } - - [JsonProperty("source")] - public string? Source { get; set; } - - [JsonProperty("sourceHandle")] - public string? SourceHandle { get; set; } - - [JsonProperty("target")] - public string? Target { get; set; } - - [JsonProperty("targetHandle")] - public string? TargetHandle { get; set; } - -} diff --git a/Rudder.Core.csproj b/Rudder.Core.csproj index 102ac73..26f0580 100644 --- a/Rudder.Core.csproj +++ b/Rudder.Core.csproj @@ -8,7 +8,7 @@ Rudder.Core 0.4.0 Rudder - Rudder LiveOps client SDK for .NET: auth, player, stores, battle pass, quests, leaderboards, inventory, remote config, scenarios, storage and realtime. + Rudder LiveOps client SDK for .NET: auth, player, stores, battle pass, quests, leaderboards, inventory, remote config, scenarios and storage. MIT true diff --git a/RudderClient.cs b/RudderClient.cs index 4e366f5..5269087 100644 --- a/RudderClient.cs +++ b/RudderClient.cs @@ -47,11 +47,11 @@ public sealed class RudderClient /// Global quests. public QuestsService Quests { get; } - /// Scenario runtime: triggers, node sessions, persistence. + /// Scenario trigger. Execution lives on the server. public ScenarioService Scenario { get; } - /// Realtime websocket channel. - public RealtimeService Realtime { get; } + /// Pending scenario effects: subscriptions and completion callbacks. + public EffectsService Effects { get; } internal RudderClientOptions Options { get; } internal IRudderTransport Transport => Options.Transport!; @@ -61,10 +61,7 @@ public sealed class RudderClient /// Project key from . public string ProjectKey => Options.ProjectKey!; - /// Realtime URL from . - public string? RealtimeUrl => Options.RealtimeUrl; - - /// Time source used by the scenario runtime. + /// Time source used by the effects client. public IClock Clock => Options.Clock ?? SystemClock.Instance; /// @@ -90,17 +87,16 @@ public sealed class RudderClient Stores = new StoresService(this); Leaderboards = new LeaderboardsService(this); Inventory = new InventoryService(this); - Scenario = new ScenarioService(this); BattlePass = new BattlePassService(this); Quests = new QuestsService(this); - Realtime = new RealtimeService(this); + Effects = new EffectsService(this); + Scenario = new ScenarioService(this); } /// Pumps time-dependent services; call every frame. public void Update(float deltaTime) { - Scenario.Update(deltaTime); - Realtime.Update(deltaTime); + Effects.Update(deltaTime); } internal Task SendAsync( @@ -151,7 +147,6 @@ public sealed class RudderClient } catch (RudderAuthException) { - // Session rejected — refresh once (single-flight) and retry the call once. if (!await RefreshTokensAsync().ConfigureAwait(false)) throw; @@ -218,7 +213,6 @@ public sealed class RudderClient } catch { - // Transport-level failure during refresh — session is over. } } diff --git a/RudderClientOptions.cs b/RudderClientOptions.cs index 02ccad6..216cff0 100644 --- a/RudderClientOptions.cs +++ b/RudderClientOptions.cs @@ -12,9 +12,6 @@ public sealed class RudderClientOptions /// API base URL, e.g. https://api.example.com. Required. public string? BaseUrl { get; set; } - /// Realtime websocket URL. Required only for . - public string? RealtimeUrl { get; set; } - /// Project key issued in the admin panel. Required. public string? ProjectKey { get; set; } @@ -30,15 +27,6 @@ public sealed class RudderClientOptions /// Diagnostic sink. Null by default (silent). public IRudderLogger? Logger { get; set; } - /// Time source for the scenario runtime; override in tests. + /// Time source for the effects client; override in tests. public IClock? Clock { get; set; } - - /// Scenario-run persistence between app launches. Optional. - public IPlanStateStore? PlanStateStore { get; set; } - - /// Optional scheduler for delayed plan work. - public IPlanScheduler? Scheduler { get; set; } - - /// Realtime transport factory. Required only for . - public IRealtimeTransportFactory? RealtimeTransportFactory { get; set; } } diff --git a/Scenarios/GetScenarioRunRequest.cs b/Scenarios/GetScenarioRunRequest.cs deleted file mode 100644 index db3c0d2..0000000 --- a/Scenarios/GetScenarioRunRequest.cs +++ /dev/null @@ -1,11 +0,0 @@ -// Code generated by apigen. DO NOT EDIT. -using Newtonsoft.Json; - -namespace RudderSdk.Core.Models.Scenarios; - -public class GetScenarioRunRequest -{ - [JsonProperty("runId")] - public string? RunId { get; set; } - -} diff --git a/Scenarios/GetScenarioRunResponse.cs b/Scenarios/GetScenarioRunResponse.cs deleted file mode 100644 index 4643115..0000000 --- a/Scenarios/GetScenarioRunResponse.cs +++ /dev/null @@ -1,18 +0,0 @@ -// Code generated by apigen. DO NOT EDIT. -using Newtonsoft.Json; -using RudderSdk.Core.Models; - -namespace RudderSdk.Core.Models.Scenarios; - -public class GetScenarioRunResponse -{ - [JsonProperty("plan")] - public ExecutionPlan? Plan { get; set; } - - [JsonProperty("runId")] - public string? RunId { get; set; } - - [JsonProperty("status")] - public string? Status { get; set; } - -} diff --git a/Scenarios/HandleScenarioCallbackResponse.cs b/Scenarios/HandleScenarioCallbackResponse.cs index 2b9c47c..7878fa4 100644 --- a/Scenarios/HandleScenarioCallbackResponse.cs +++ b/Scenarios/HandleScenarioCallbackResponse.cs @@ -1,12 +1,12 @@ // Code generated by apigen. DO NOT EDIT. using Newtonsoft.Json; -using RudderSdk.Core.Models; +using Newtonsoft.Json.Linq; namespace RudderSdk.Core.Models.Scenarios; public class HandleScenarioCallbackResponse { - [JsonProperty("plan")] - public ExecutionPlan? Plan { get; set; } + [JsonProperty("effect")] + public JToken? Effect { get; set; } } diff --git a/Scenarios/ListPendingScenarioEffectsResponse.cs b/Scenarios/ListPendingScenarioEffectsResponse.cs new file mode 100644 index 0000000..5998bc3 --- /dev/null +++ b/Scenarios/ListPendingScenarioEffectsResponse.cs @@ -0,0 +1,12 @@ +// Code generated by apigen. DO NOT EDIT. +using Newtonsoft.Json; +using System.Collections.Generic; + +namespace RudderSdk.Core.Models.Scenarios; + +public class ListPendingScenarioEffectsResponse +{ + [JsonProperty("effects")] + public List Effects { get; set; } = null!; + +} diff --git a/Scenarios/PendingEffect.cs b/Scenarios/PendingEffect.cs new file mode 100644 index 0000000..bac2d18 --- /dev/null +++ b/Scenarios/PendingEffect.cs @@ -0,0 +1,28 @@ +// Code generated by apigen. DO NOT EDIT. +using Newtonsoft.Json; +using Newtonsoft.Json.Linq; +using System; + +namespace RudderSdk.Core.Models.Scenarios; + +public class PendingEffect +{ + [JsonProperty("data")] + public JToken Data { get; set; } = null!; + + [JsonProperty("nodeId")] + public string NodeId { get; set; } = null!; + + [JsonProperty("runId")] + public string RunId { get; set; } = null!; + + [JsonProperty("scenarioId")] + public string ScenarioId { get; set; } = null!; + + [JsonProperty("type")] + public string Type { get; set; } = null!; + + [JsonProperty("waitDeadline")] + public DateTimeOffset? WaitDeadline { get; set; } + +} diff --git a/Scenarios/TriggerScenarioResponse.cs b/Scenarios/TriggerScenarioResponse.cs index e36946a..c36b390 100644 --- a/Scenarios/TriggerScenarioResponse.cs +++ b/Scenarios/TriggerScenarioResponse.cs @@ -1,13 +1,12 @@ // Code generated by apigen. DO NOT EDIT. using Newtonsoft.Json; using System.Collections.Generic; -using RudderSdk.Core.Models; namespace RudderSdk.Core.Models.Scenarios; public class TriggerScenarioResponse { - [JsonProperty("plans")] - public List? Plans { get; set; } + [JsonProperty("effects")] + public List Effects { get; set; } = null!; } diff --git a/Scenarios/UpdateScenarioCounterResponse.cs b/Scenarios/UpdateScenarioCounterResponse.cs index 892f9ea..2fa4ec8 100644 --- a/Scenarios/UpdateScenarioCounterResponse.cs +++ b/Scenarios/UpdateScenarioCounterResponse.cs @@ -1,6 +1,6 @@ // Code generated by apigen. DO NOT EDIT. using Newtonsoft.Json; -using RudderSdk.Core.Models; +using Newtonsoft.Json.Linq; namespace RudderSdk.Core.Models.Scenarios; @@ -9,7 +9,7 @@ public class UpdateScenarioCounterResponse [JsonProperty("completed")] public bool? Completed { get; set; } - [JsonProperty("plan")] - public ExecutionPlan? Plan { get; set; } + [JsonProperty("effect")] + public JToken? Effect { get; set; } } diff --git a/Services/BattlePassService.cs b/Services/BattlePassService.cs index 8ef277d..5662b32 100644 --- a/Services/BattlePassService.cs +++ b/Services/BattlePassService.cs @@ -8,7 +8,7 @@ namespace RudderSdk.Core; /// /// Battle pass progress and rewards. Battle pass state is tied to a scenario /// battle pass node, so every call carries the scenario/node ids (and a run id -/// for the mutating calls) — supplies them +/// for the mutating calls) — supplies them /// during scenario runs. /// public sealed class BattlePassService diff --git a/Services/Effects/BattlePassEffect.cs b/Services/Effects/BattlePassEffect.cs new file mode 100644 index 0000000..a7ce69b --- /dev/null +++ b/Services/Effects/BattlePassEffect.cs @@ -0,0 +1,97 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Newtonsoft.Json.Linq; +using RudderSdk.Core.Models.BattlePass; + +namespace RudderSdk.Core; + +/// +/// A scenario battle-pass node. Battle-pass operations are bound to this +/// node's scenario/node/run ids; , +/// and a successful post callbacks. +/// +public sealed class BattlePassEffect +{ + private readonly EffectHandle _handle; + private readonly BattlePassService _battlePass; + + internal BattlePassEffect(EffectHandle handle, BattlePassService battlePass) + { + _handle = handle; + _battlePass = battlePass; + } + + /// Run id. + public string RunId => _handle.RunId; + + /// Scenario id. + public string ScenarioId => _handle.ScenarioId; + + /// Node id. + public string NodeId => _handle.NodeId; + + /// Node data payload. + public JObject Data => _handle.Data; + + /// Reads a typed value from the node data. + public T Get(string key, T defaultValue = default!) => _handle.Get(key, defaultValue); + + /// Reads current progress (xp, level, premium ownership, claimed tiers) for this node. + public Task GetProgressAsync(CancellationToken cancellationToken = default) + => _battlePass.GetProgressAsync(_handle.ScenarioId, _handle.NodeId, cancellationToken); + + /// Credits xp from a configured source. + public Task AddXpAsync(string source, long amount, CancellationToken cancellationToken = default) + => _battlePass.AddXpAsync(new AddBattlePassXpRequest + { + ScenarioId = _handle.ScenarioId, + NodeId = _handle.NodeId, + RunId = _handle.RunId, + Source = source, + Amount = amount + }, cancellationToken); + + /// + /// Claims a tier reward at a reached level. + /// is or . + /// + public Task ClaimRewardAsync(int level, string track, CancellationToken cancellationToken = default) + => _battlePass.ClaimRewardAsync(new ClaimBattlePassRewardRequest + { + ScenarioId = _handle.ScenarioId, + NodeId = _handle.NodeId, + RunId = _handle.RunId, + Level = level, + Track = track + }, cancellationToken); + + /// Purchases the premium track, then posts onPremiumPurchase on success. + public async Task PurchasePremiumAsync(CancellationToken cancellationToken = default) + { + var response = await _battlePass.PurchasePremiumAsync(new PurchaseBattlePassPremiumRequest + { + ScenarioId = _handle.ScenarioId, + NodeId = _handle.NodeId, + RunId = _handle.RunId, + IdempotencyKey = Guid.NewGuid().ToString() + }, cancellationToken).ConfigureAwait(false); + + if (response != null && response.Success == true) + await _handle.CompleteAsync("onPremiumPurchase", cancellationToken).ConfigureAwait(false); + + return response!; + } + + /// Posts the onLevelUp callback. + public Task LevelUpAsync(CancellationToken cancellationToken = default) => _handle.CompleteAsync("onLevelUp", cancellationToken); + + /// Posts the onLevelUp callback (fire-and-forget). + public void LevelUp() => _handle.Complete("onLevelUp"); + + /// Posts the onComplete callback. + public Task EndAsync(CancellationToken cancellationToken = default) => _handle.CompleteAsync("onComplete", cancellationToken); + + /// Posts the onComplete callback (fire-and-forget). + public void End() => _handle.Complete("onComplete"); +} diff --git a/Services/Effects/BattlePassLevelEffect.cs b/Services/Effects/BattlePassLevelEffect.cs new file mode 100644 index 0000000..0bb0507 --- /dev/null +++ b/Services/Effects/BattlePassLevelEffect.cs @@ -0,0 +1,41 @@ +using System.Threading; +using System.Threading.Tasks; +using Newtonsoft.Json.Linq; + +namespace RudderSdk.Core; + +/// +/// A scenario battle-pass-level node — a single claimable tier. +/// posts onComplete, which the server accepts +/// only once the player has reached the node's configured level. +/// +public sealed class BattlePassLevelEffect +{ + private readonly EffectHandle _handle; + + internal BattlePassLevelEffect(EffectHandle handle) => _handle = handle; + + /// Run id. + public string RunId => _handle.RunId; + + /// Scenario id. + public string ScenarioId => _handle.ScenarioId; + + /// Node id. + public string NodeId => _handle.NodeId; + + /// The tier level this node claims. + public int Level => _handle.Get("levelNumber", 0); + + /// Node data payload. + public JObject Data => _handle.Data; + + /// Reads a typed value from the node data. + public T Get(string key, T defaultValue = default!) => _handle.Get(key, defaultValue); + + /// Claims this tier; posts onComplete. + public Task ClaimAsync(CancellationToken cancellationToken = default) => _handle.CompleteAsync("onComplete", cancellationToken); + + /// Claims this tier (fire-and-forget). + public void Claim() => _handle.Complete("onComplete"); +} diff --git a/Services/Effects/LeaderboardEffect.cs b/Services/Effects/LeaderboardEffect.cs new file mode 100644 index 0000000..cee667d --- /dev/null +++ b/Services/Effects/LeaderboardEffect.cs @@ -0,0 +1,61 @@ +using System.Threading; +using System.Threading.Tasks; +using Newtonsoft.Json.Linq; + +namespace RudderSdk.Core; + +/// A scenario leaderboard node. End it with or claim with . +public sealed class LeaderboardEffect +{ + private readonly EffectHandle _handle; + + internal LeaderboardEffect(EffectHandle handle) => _handle = handle; + + /// Run id. + public string RunId => _handle.RunId; + + /// Scenario id. + public string ScenarioId => _handle.ScenarioId; + + /// Node id. + public string NodeId => _handle.NodeId; + + /// Node data payload. + public JObject Data => _handle.Data; + + /// True after the effect was resolved once. + public bool IsResolved { get; private set; } + + /// Reads a typed value from the node data. + public T Get(string key, T defaultValue = default!) => _handle.Get(key, defaultValue); + + /// Posts the onEnd callback. + public Task EndAsync(CancellationToken cancellationToken = default) => ResolveAsync("onEnd", cancellationToken); + + /// Posts the onEnd callback (fire-and-forget). + public void End() => Resolve("onEnd"); + + /// Posts the onClaim callback. The server matches live rank to a place. + public Task ClaimAsync(CancellationToken cancellationToken = default) => ResolveAsync("onClaim", cancellationToken); + + /// Posts the onClaim callback (fire-and-forget). + public void Claim() => Resolve("onClaim"); + + private async Task ResolveAsync(string handle, CancellationToken cancellationToken) + { + if (IsResolved) + return; + + IsResolved = true; + await _handle.CompleteAsync(handle, cancellationToken).ConfigureAwait(false); + } + + private void Resolve(string handle) + { + if (IsResolved) + return; + + IsResolved = true; + _handle.Complete(handle); + } +} diff --git a/Services/Effects/NotificationEffect.cs b/Services/Effects/NotificationEffect.cs new file mode 100644 index 0000000..af986d1 --- /dev/null +++ b/Services/Effects/NotificationEffect.cs @@ -0,0 +1,40 @@ +using System.Threading; +using System.Threading.Tasks; +using Newtonsoft.Json.Linq; + +namespace RudderSdk.Core; + +/// A scenario notification node. posts the output callback. +public sealed class NotificationEffect +{ + private readonly EffectHandle _handle; + + internal NotificationEffect(EffectHandle handle) => _handle = handle; + + /// Run id. + public string RunId => _handle.RunId; + + /// Scenario id. + public string ScenarioId => _handle.ScenarioId; + + /// Node id. + public string NodeId => _handle.NodeId; + + /// Notification title. + public string Title => _handle.Get("title", string.Empty); + + /// Notification message. + public string Message => _handle.Get("message", string.Empty); + + /// Node data payload. + public JObject Data => _handle.Data; + + /// Reads a typed value from the node data. + public T Get(string key, T defaultValue = default!) => _handle.Get(key, defaultValue); + + /// Posts the output callback. + public Task DoneAsync(CancellationToken cancellationToken = default) => _handle.CompleteAsync("output", cancellationToken); + + /// Posts the output callback (fire-and-forget). + public void Done() => _handle.Complete("output"); +} diff --git a/Services/Effects/QuestEffect.cs b/Services/Effects/QuestEffect.cs new file mode 100644 index 0000000..c3ece16 --- /dev/null +++ b/Services/Effects/QuestEffect.cs @@ -0,0 +1,62 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Newtonsoft.Json.Linq; + +namespace RudderSdk.Core; + +/// +/// A scenario quest node. Report objective progress; the server auto-completes +/// the node once every objective is satisfied. +/// +public sealed class QuestEffect +{ + private readonly EffectHandle _handle; + + internal QuestEffect(EffectHandle handle) => _handle = handle; + + /// Run id. + public string RunId => _handle.RunId; + + /// Scenario id. + public string ScenarioId => _handle.ScenarioId; + + /// Node id. + public string NodeId => _handle.NodeId; + + /// Quest name from the node data. + public string Name => _handle.Get("name", string.Empty); + + /// Objective definitions from the node data. + public IReadOnlyList Objectives + { + get + { + if (_handle.Data["objectives"] is not JArray array) + return Array.Empty(); + + var list = new List(array.Count); + foreach (var item in array) + { + if (item is JObject obj) + list.Add(obj); + } + + return list; + } + } + + /// Node data payload. + public JObject Data => _handle.Data; + + /// Reads a typed value from the node data. + public T Get(string key, T defaultValue = default!) => _handle.Get(key, defaultValue); + + /// Reports progress toward an objective via POST /sdk/v1/scenarios/counter. + public Task ReportProgressAsync(string objectiveId, long amount = 1, CancellationToken cancellationToken = default) + => _handle.ReportProgressAsync(objectiveId, amount, cancellationToken); + + /// Reports progress toward an objective (fire-and-forget). + public void ReportProgress(string objectiveId, long amount = 1) => _handle.ReportProgress(objectiveId, amount); +} diff --git a/Services/Effects/ScenarioCompletedEffect.cs b/Services/Effects/ScenarioCompletedEffect.cs new file mode 100644 index 0000000..267275b --- /dev/null +++ b/Services/Effects/ScenarioCompletedEffect.cs @@ -0,0 +1,17 @@ +namespace RudderSdk.Core; + +/// Payload of . +public sealed class ScenarioCompletedEffect +{ + internal ScenarioCompletedEffect(string runId, string scenarioId) + { + RunId = runId; + ScenarioId = scenarioId; + } + + /// Server-issued run id. + public string RunId { get; } + + /// Scenario id. + public string ScenarioId { get; } +} diff --git a/Services/Effects/ScenarioFailedEffect.cs b/Services/Effects/ScenarioFailedEffect.cs new file mode 100644 index 0000000..2c833a7 --- /dev/null +++ b/Services/Effects/ScenarioFailedEffect.cs @@ -0,0 +1,27 @@ +using System; + +namespace RudderSdk.Core; + +/// Payload of . +public sealed class ScenarioFailedEffect +{ + internal ScenarioFailedEffect(string runId, string scenarioId, string nodeId, Exception exception) + { + RunId = runId; + ScenarioId = scenarioId; + NodeId = nodeId; + Exception = exception; + } + + /// Server-issued run id. + public string RunId { get; } + + /// Scenario id. + public string ScenarioId { get; } + + /// Node the failure happened at. + public string NodeId { get; } + + /// The error that failed the run. + public Exception Exception { get; } +} diff --git a/Services/Effects/StoreOfferEffect.cs b/Services/Effects/StoreOfferEffect.cs new file mode 100644 index 0000000..084acc8 --- /dev/null +++ b/Services/Effects/StoreOfferEffect.cs @@ -0,0 +1,74 @@ +using System.Threading; +using System.Threading.Tasks; +using Newtonsoft.Json.Linq; + +namespace RudderSdk.Core; + +/// A scenario store-offer node. Resolve it with or . +public sealed class StoreOfferEffect +{ + private readonly EffectHandle _handle; + + internal StoreOfferEffect(EffectHandle handle) => _handle = handle; + + /// Run id. + public string RunId => _handle.RunId; + + /// Scenario id. + public string ScenarioId => _handle.ScenarioId; + + /// Node id. + public string NodeId => _handle.NodeId; + + /// Store slug from the node data, if present. + public string StoreSlug => _handle.Get("storeSlug", string.Empty); + + /// Optional message from the node data. + public string? Message + { + get + { + var value = _handle.Get("message", null); + return string.IsNullOrEmpty(value) ? null : value; + } + } + + /// Node data payload. + public JObject Data => _handle.Data; + + /// True after the offer was resolved once. + public bool IsResolved { get; private set; } + + /// Reads a typed value from the node data. + public T Get(string key, T defaultValue = default!) => _handle.Get(key, defaultValue); + + /// Posts the onPurchase callback. + public Task PurchaseAsync(CancellationToken cancellationToken = default) => ResolveAsync("onPurchase", cancellationToken); + + /// Posts the onPurchase callback (fire-and-forget). + public void Purchase() => Resolve("onPurchase"); + + /// Posts the onDecline callback. + public Task DeclineAsync(CancellationToken cancellationToken = default) => ResolveAsync("onDecline", cancellationToken); + + /// Posts the onDecline callback (fire-and-forget). + public void Decline() => Resolve("onDecline"); + + private async Task ResolveAsync(string handle, CancellationToken cancellationToken) + { + if (IsResolved) + return; + + IsResolved = true; + await _handle.CompleteAsync(handle, cancellationToken).ConfigureAwait(false); + } + + private void Resolve(string handle) + { + if (IsResolved) + return; + + IsResolved = true; + _handle.Complete(handle); + } +} diff --git a/Services/Effects/WaitEffect.cs b/Services/Effects/WaitEffect.cs new file mode 100644 index 0000000..28c7d57 --- /dev/null +++ b/Services/Effects/WaitEffect.cs @@ -0,0 +1,34 @@ +using System; +using Newtonsoft.Json.Linq; + +namespace RudderSdk.Core; + +/// A scenario wait node. The run resumes when the server advances it at . +public sealed class WaitEffect +{ + private readonly EffectHandle _handle; + + internal WaitEffect(EffectHandle handle, DateTimeOffset deadlineUtc) + { + _handle = handle; + DeadlineUtc = deadlineUtc; + } + + /// Run id. + public string RunId => _handle.RunId; + + /// Scenario id. + public string ScenarioId => _handle.ScenarioId; + + /// Node id. + public string NodeId => _handle.NodeId; + + /// When the wait ends (UTC). + public DateTimeOffset DeadlineUtc { get; } + + /// Node data payload. + public JObject Data => _handle.Data; + + /// Reads a typed value from the node data. + public T Get(string key, T defaultValue = default!) => _handle.Get(key, defaultValue); +} diff --git a/Services/EffectsService.cs b/Services/EffectsService.cs new file mode 100644 index 0000000..60db8eb --- /dev/null +++ b/Services/EffectsService.cs @@ -0,0 +1,410 @@ +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; + +/// +/// Thin effects client. The server owns scenario execution; this service +/// surfaces pending effects and posts callbacks. Drive +/// every frame for the 30s heartbeat and +/// wait-deadline checks. +/// +public sealed class EffectsService +{ + private static readonly TimeSpan HeartbeatInterval = TimeSpan.FromSeconds(30); + + private readonly RudderClient _client; + private readonly object _gate = new(); + private readonly HashSet<(string RunId, string NodeId)> _seen = new(); + private readonly Dictionary<(string RunId, string NodeId), DateTimeOffset> _waitDeadlines = new(); + + private bool _refreshDue; + private bool _inFlight; + private DateTimeOffset _nextHeartbeat; + + /// Raised for a notification effect. + public event Action? OnNotification; + + /// Raised for a store-offer effect. + public event Action? OnStoreOffer; + + /// Raised for a leaderboard effect. + public event Action? OnLeaderboard; + + /// Raised for a wait effect. + public event Action? OnWait; + + /// Raised for a quest effect. + public event Action? OnQuest; + + /// Raised for a battle-pass effect. + public event Action? OnBattlePass; + + /// Raised for a battle-pass-level effect. + public event Action? OnBattlePassLevel; + + /// Raised when a run finishes all its nodes. + public event Action? OnScenarioCompleted; + + /// Raised when a run is dropped after a definitive server rejection or an unsupported effect type. + public event Action? OnScenarioFailed; + + internal EffectsService(RudderClient client) + { + _client = client; + _nextHeartbeat = _client.Clock.UtcNow + HeartbeatInterval; + if (!string.IsNullOrEmpty(_client.TokenStore.GetAccessToken())) + _refreshDue = true; + + _client.Auth.AuthStateChanged += OnAuthStateChanged; + } + + /// Pumps heartbeat and wait-deadline checks; call every frame. + 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? effects) + { + if (effects == null) + return; + + var batch = new List(); + 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.Add(key)) + continue; + + 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( + "POST", + "/sdk/v1/scenarios/callback", + new HandleScenarioCallbackRequest + { + ScenarioId = source.ScenarioId, + 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.ScenarioId)); + else + Ingest(new[] { next }); + } + catch (Exception ex) when (IsDefinitiveRejection(ex)) + { + DropRun(source.RunId, source.ScenarioId, source.NodeId, ex); + } + } + + internal async Task ReportProgressAsync( + PendingEffect source, + string counterKey, + long amount, + CancellationToken cancellationToken = default) + { + try + { + var response = await _client.SendAsync( + "POST", + "/sdk/v1/scenarios/counter", + new UpdateScenarioCounterRequest + { + ScenarioId = source.ScenarioId, + 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.ScenarioId)); + else + Ingest(new[] { next }); + } + catch (Exception ex) when (IsDefinitiveRejection(ex)) + { + DropRun(source.RunId, source.ScenarioId, 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( + "GET", + "/sdk/v1/scenarios/pending", + CancellationToken.None).ConfigureAwait(false); + + Ingest(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.ScenarioId, + effect.NodeId, + new Exception($"Unsupported scenario node type '{effect.Type}'"))); + break; + } + } + + private void DropRun(string runId, string scenarioId, string nodeId, Exception exception) + { + lock (_gate) + { + var toRemove = new List<(string RunId, string NodeId)>(); + foreach (var key in _seen) + { + if (key.RunId == runId) + toRemove.Add(key); + } + + foreach (var key in toRemove) + { + _seen.Remove(key); + _waitDeadlines.Remove(key); + } + } + + Emit(OnScenarioFailed, new ScenarioFailedEffect(runId, scenarioId, nodeId, exception)); + } + + 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(Action? handlers, T effect) + { + if (handlers == null) + return; + + foreach (var subscriber in handlers.GetInvocationList()) + { + try + { + ((Action)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(); + } + + 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); + } +} + +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 ScenarioId => _effect.ScenarioId; + + public string NodeId => _effect.NodeId; + + public JObject Data => _effect.Data as JObject ?? new JObject(); + + public T Get(string key, T defaultValue = default!) + { + if (!Data.TryGetValue(key, out var value) || value == null || value.Type == JTokenType.Null) + return defaultValue; + + try + { + return value.ToObject() ?? 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); +} diff --git a/Services/QuestsService.cs b/Services/QuestsService.cs index a2c124f..9fa6080 100644 --- a/Services/QuestsService.cs +++ b/Services/QuestsService.cs @@ -7,7 +7,7 @@ namespace RudderSdk.Core; /// /// Global quests (list + claim), distinct from scenario quest nodes which -/// advance through . +/// advance through . /// public sealed class QuestsService { diff --git a/Services/RealtimeService.cs b/Services/RealtimeService.cs deleted file mode 100644 index be83d6a..0000000 --- a/Services/RealtimeService.cs +++ /dev/null @@ -1,105 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; -using RudderSdk.Core.Abstractions; - -namespace RudderSdk.Core; - -/// -/// Realtime websocket channel. Requires -/// and -/// . -/// -public sealed class RealtimeService -{ - private static readonly TimeSpan DefaultTimeout = TimeSpan.FromSeconds(10); - - private readonly RudderClient _client; - private RealtimeSession? _session; - - internal RealtimeService(RudderClient client) => _client = client; - - /// The current session, or null when not connected. - public RealtimeSession? Session => _session; - - /// True while a session is connected. - public bool IsConnected => _session?.IsConnected == true; - - /// Connects to the configured realtime URL. - public Task ConnectAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default) - { - if (string.IsNullOrEmpty(_client.RealtimeUrl)) - throw new InvalidOperationException("RealtimeUrl is not configured."); - - return ConnectAsync(new Uri(_client.RealtimeUrl), timeout, cancellationToken); - } - - /// Connects to an explicit realtime URL. - public async Task ConnectAsync(Uri uri, TimeSpan? timeout = null, CancellationToken cancellationToken = default) - { - var factory = _client.Options.RealtimeTransportFactory - ?? throw new InvalidOperationException("RealtimeTransportFactory is not configured."); - - var token = _client.TokenStore.GetAccessToken(); - if (string.IsNullOrEmpty(token)) - throw new InvalidOperationException("LiveOps access token is required for realtime authorization."); - - var transport = factory.Create(); - await transport.ConnectAsync(uri, timeout ?? DefaultTimeout, cancellationToken).ConfigureAwait(false); - _session = new RealtimeSession(transport, token); - return _session; - } - - /// Closes the current session, if any. - public Task DisconnectAsync(CancellationToken cancellationToken = default) - => _session?.DisconnectAsync(cancellationToken) ?? Task.CompletedTask; - - /// Pumps the underlying transport; call every frame. - public void Update(float deltaTime) - { - _session?.Update(deltaTime); - } -} - -/// An open realtime connection. -public sealed class RealtimeSession -{ - private readonly IRealtimeTransport _transport; - - internal RealtimeSession(IRealtimeTransport transport, string accessToken) - { - _transport = transport; - AccessToken = accessToken; - _transport.Closed += () => Closed?.Invoke(); - _transport.Error += ex => Error?.Invoke(ex); - _transport.Received += data => MessageReceived?.Invoke(data); - } - - /// Access token the connection was authorized with. - public string AccessToken { get; } - - /// True while the connection is open. - public bool IsConnected => _transport.IsConnected; - - /// Raised when the connection closes. - public event Action? Closed; - - /// Raised on transport errors. - public event Action? Error; - - /// Raised for every incoming message. - public event Action>? MessageReceived; - - /// Sends one message. - public Task SendAsync(byte[] payload, CancellationToken cancellationToken = default) - => _transport.SendAsync(new ArraySegment(payload ?? Array.Empty()), cancellationToken); - - /// Closes the connection. - public Task DisconnectAsync(CancellationToken cancellationToken = default) - => _transport.CloseAsync(cancellationToken); - - internal void Update(float deltaTime) - { - _transport.Update(deltaTime); - } -} diff --git a/Services/ScenarioService.cs b/Services/ScenarioService.cs index 4ef5e15..9e3c7c9 100644 --- a/Services/ScenarioService.cs +++ b/Services/ScenarioService.cs @@ -1,71 +1,24 @@ -using System; -using System.Collections.Generic; -using System.Linq; using System.Threading; using System.Threading.Tasks; -using Newtonsoft.Json; -using RudderSdk.Core.Abstractions; -using RudderSdk.Core.Models; using RudderSdk.Core.Models.Scenarios; namespace RudderSdk.Core; /// -/// Scenario runtime. starts server-issued plans; -/// active nodes surface as typed sessions through the On* events and are -/// advanced by completing those sessions. +/// Scenario trigger. Execution lives on the server; resulting effects surface +/// through . /// -public sealed partial class ScenarioService +public sealed class ScenarioService { private readonly RudderClient _client; - private readonly Dictionary _runs = new(); - - /// Raised for a notification node. - public event Action? OnNotification; - - /// Raised for a store-offer node. - public event Action? OnStoreOffer; - - /// Raised for a leaderboard node. - public event Action? OnLeaderboard; - - /// Raised for a remote-config-override node, after the patches were applied. - public event Action? OnConfigChanged; - - /// Raised for a wait node. - public event Action? OnWait; - - /// Raised for a quest node. - public event Action? OnQuest; - - /// Raised for a battle-pass node. - public event Action? OnBattlePass; - - /// Raised for a battle-pass-level node. - public event Action? OnBattlePassLevel; - - /// Raised when a run finishes all its nodes. - public event Action? OnScenarioCompleted; - - /// Raised when a run dies on an unrecoverable error. - public event Action? OnScenarioFailed; internal ScenarioService(RudderClient client) => _client = client; - /// True while at least one run is active. - public bool IsRunning => _runs.Count > 0; - - /// First active node id across the runs, or null. - public string? CurrentNodeId => _runs.Values.FirstOrDefault()?.ActiveNodes.Keys.FirstOrDefault(); - - /// Snapshots of the active runs. - public IReadOnlyList ActiveRuns => _runs.Values.Select(ToPlanRun).ToList(); - /// - /// Triggers scenarios by event name and starts the plans the server - /// returns. Returns the runs this call started. + /// Triggers scenarios by event name. Returned pending effects are ingested + /// into . /// - public async Task> TriggerAsync(string eventName, CancellationToken cancellationToken = default) + public async Task TriggerAsync(string eventName, CancellationToken cancellationToken = default) { var response = await _client.SendAsync( "POST", @@ -73,170 +26,6 @@ public sealed partial class ScenarioService new TriggerScenarioRequest { Event = eventName }, cancellationToken).ConfigureAwait(false); - return StartPlans(response?.Plans); - } - - /// - /// Restores persisted runs, reconciles them with the server and re-dispatches - /// active nodes. Call once after startup, after login. - /// - public Task RestoreAsync(CancellationToken cancellationToken = default) - { - return RestoreCoreAsync(cancellationToken); - } - - private async Task RestoreCoreAsync(CancellationToken cancellationToken = default) - { - var store = _client.Options.PlanStateStore; - if (store == null || string.IsNullOrEmpty(store.State)) - return; - - try - { - var state = JsonConvert.DeserializeObject(store.State); - _runs.Clear(); - if (state?.Runs != null) - { - foreach (var savedRun in state.Runs) - { - if (savedRun?.Plan == null) - continue; - - // Reconcile with server - try - { - var response = await _client.SendAsync( - "POST", - "/sdk/v1/scenarios/run", - new GetScenarioRunRequest { RunId = savedRun.RunId }, - cancellationToken); - - if (response?.Status == "unknown_run" || response?.Status == "expired") - continue; - - if (response?.Plan != null) - { - savedRun.Plan = response.Plan; - savedRun.ActiveNodes = null; - savedRun.CompletedHandles = null; - } - } - catch - { - // Network error — keep local state as fallback - } - - var run = RuntimeRun.FromPersisted(savedRun); - _runs[run.RunId] = run; - } - } - - foreach (var run in _runs.Values.ToList()) - { - if (run.ActiveNodes.Count > 0) - { - foreach (var nodeState in run.ActiveNodes.Values.ToList()) - DispatchActiveNode(run, nodeState, restored: true); - } - else - { - // Rebuilt from server — activate start node - var startNode = FindNode(run.Plan, run.Plan.StartNodeId) ?? run.Plan.Nodes[0]; - ActivateNode(run, startNode.Id); - } - } - - Persist(); - } - catch (Exception ex) - { - _client.Options.Logger?.Log(RudderLogLevel.Error, "[Rudder] Failed to restore scenario state. Clearing persisted state. " + ex.Message); - Clear(); - } - } - - /// Drops all runs and the persisted state. - public void Clear() - { - _runs.Clear(); - Persist(); - } - - /// Completes wait nodes whose deadline passed; call every frame. - public void Update(float deltaTime) - { - var now = _client.Clock.UtcNow; - foreach (var run in _runs.Values.ToList()) - { - foreach (var node in run.ActiveNodes.Values.ToList()) - { - if (node.WaitDeadlineUtc.HasValue && now >= node.WaitDeadlineUtc.Value) - _ = CompleteNodeAsync(run.RunId, node.NodeId, "onComplete"); - } - } - } - - /// Completes the first active node with the given handle. - public Task RespondAsync(string handle, CancellationToken cancellationToken = default) - { - var run = _runs.Values.FirstOrDefault(); - var node = run?.ActiveNodes.Values.FirstOrDefault(); - return run == null || node == null - ? Task.CompletedTask - : CompleteNodeAsync(run.RunId, node.NodeId, handle, cancellationToken); - } - - /// Completes the first active node with the given handle (fire-and-forget). - public void Respond(string handle) - { - _ = RespondAsync(handle); - } - - /// Adds progress to a counter of the first active node. - public Task UpdateProgressAsync(string counterKey, long amount, CancellationToken cancellationToken = default) - { - var run = _runs.Values.FirstOrDefault(); - var node = run?.ActiveNodes.Values.FirstOrDefault(); - return run == null || node == null - ? Task.CompletedTask - : UpdateProgressAsync(run.RunId, node.NodeId, counterKey, amount, cancellationToken); - } - - internal async Task UpdateProgressAsync( - string runId, - string nodeId, - string counterKey, - long amount, - CancellationToken cancellationToken = default) - { - if (!_runs.TryGetValue(runId, out var run) || !run.ActiveNodes.ContainsKey(nodeId)) - return; - - try - { - var response = await _client.SendAsync( - "POST", - "/sdk/v1/scenarios/counter", - new UpdateScenarioCounterRequest - { - ScenarioId = run.Plan.ScenarioId, - NodeId = nodeId, - CounterKey = counterKey, - Amount = amount, - RunId = run.RunId - }, - cancellationToken).ConfigureAwait(false); - - // The server reports objective completion; it no longer returns a plan from the - // counter endpoint. On completion, cross the node's onComplete handle (which - // advances the run) — idempotent if the consumer also completes the session. - if (response != null && response.Completed == true) - await CompleteNodeAsync(runId, nodeId, "onComplete", cancellationToken).ConfigureAwait(false); - } - catch (Exception ex) - { - // Counter update failure does not fail the run. - _client.Options.Logger?.Log(RudderLogLevel.Warning, $"[Rudder] Scenario counter update failed at node {nodeId}: {ex.Message}"); - } + _client.Effects.Ingest(response?.Effects); } } diff --git a/Services/Scenarios/Models/PlanRun.cs b/Services/Scenarios/Models/PlanRun.cs deleted file mode 100644 index 3732da6..0000000 --- a/Services/Scenarios/Models/PlanRun.cs +++ /dev/null @@ -1,42 +0,0 @@ -using System.Collections.Generic; -using RudderSdk.Core.Models; - -namespace RudderSdk.Core; - -/// Snapshot of one running scenario plan. -public sealed class PlanRun -{ - internal PlanRun( - string runId, - string planId, - string scenarioId, - string userId, - IReadOnlyList activeNodeIds, - ExecutionPlan plan) - { - RunId = runId; - PlanId = planId; - ScenarioId = scenarioId; - UserId = userId; - ActiveNodeIds = activeNodeIds; - Plan = plan; - } - - /// Server-issued run id. - public string RunId { get; } - - /// Plan id. - public string PlanId { get; } - - /// Scenario id. - public string ScenarioId { get; } - - /// Player the run belongs to. - public string UserId { get; } - - /// Ids of the currently active nodes. - public IReadOnlyList ActiveNodeIds { get; } - - /// The execution plan being run. - public ExecutionPlan Plan { get; } -} diff --git a/Services/Scenarios/Models/ScenarioFailedEvent.cs b/Services/Scenarios/Models/ScenarioFailedEvent.cs deleted file mode 100644 index 5b1a347..0000000 --- a/Services/Scenarios/Models/ScenarioFailedEvent.cs +++ /dev/null @@ -1,23 +0,0 @@ -using System; - -namespace RudderSdk.Core; - -/// Payload of . -public sealed class ScenarioFailedEvent -{ - internal ScenarioFailedEvent(PlanRun run, string nodeId, Exception exception) - { - Run = run; - NodeId = nodeId; - Exception = exception; - } - - /// The failed run. - public PlanRun Run { get; } - - /// Node the failure happened at. - public string NodeId { get; } - - /// The error that failed the run. - public Exception Exception { get; } -} diff --git a/Services/Scenarios/Models/ScenarioNodeContext.cs b/Services/Scenarios/Models/ScenarioNodeContext.cs deleted file mode 100644 index fa3e22c..0000000 --- a/Services/Scenarios/Models/ScenarioNodeContext.cs +++ /dev/null @@ -1,78 +0,0 @@ -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; -using Newtonsoft.Json.Linq; -using RudderSdk.Core.Models; - -namespace RudderSdk.Core; - -/// Context of the scenario node a session was created for. -public class ScenarioNodeContext -{ - private readonly ScenarioService _service; - - internal ScenarioNodeContext(ScenarioService service, PlanRun run, ExecutionPlanNode node) - { - _service = service; - Run = run; - Node = node; - Data = node?.Data as JObject ?? new JObject(); - } - - /// The run this node belongs to. - public PlanRun Run { get; } - - /// The plan node. - public ExecutionPlanNode Node { get; } - - /// Run id. - public string RunId => Run.RunId; - - /// Plan id. - public string PlanId => Run.PlanId; - - /// Scenario id. - public string ScenarioId => Run.ScenarioId; - - /// Node id. - public string NodeId => Node.Id; - - /// Node type. - public string Type => Node.Type; - - /// Node data payload. - public JObject Data { get; } - - /// Reads a typed value from the node data. - public T Get(string key, T defaultValue = default!) - { - if (Data == null || !Data.TryGetValue(key, out var value)) - return defaultValue; - try { return value.ToObject() ?? defaultValue; } - catch { return defaultValue; } - } - - /// Deserializes the whole node data payload. - public T Get() - { - try { return Data == null ? default! : Data.ToObject() ?? default!; } - catch { return default!; } - } - - /// Returns the node data as a plain dictionary. - public Dictionary AsObjectDictionary() - { - return Data?.ToObject>() ?? new Dictionary(); - } - - internal Task CompleteAsync(string handle, CancellationToken cancellationToken = default) - => _service.CompleteNodeAsync(RunId, NodeId, handle, cancellationToken); - - internal void Complete(string handle) - { - _ = CompleteAsync(handle); - } - - internal Task AddProgressAsync(string counterKey, long amount, CancellationToken cancellationToken = default) - => _service.UpdateProgressAsync(RunId, NodeId, counterKey, amount, cancellationToken); -} diff --git a/Services/Scenarios/ScenarioService.Dispatch.cs b/Services/Scenarios/ScenarioService.Dispatch.cs deleted file mode 100644 index 42179b9..0000000 --- a/Services/Scenarios/ScenarioService.Dispatch.cs +++ /dev/null @@ -1,193 +0,0 @@ -using System; -using System.Globalization; -using System.Linq; -using Newtonsoft.Json; -using Newtonsoft.Json.Linq; -using RudderSdk.Core.Abstractions; - -namespace RudderSdk.Core; - -public sealed partial class ScenarioService -{ - private void DispatchActiveNode(RuntimeRun run, ActiveNodeState state, bool restored) - { - var node = FindNode(run.Plan, state.NodeId); - if (node == null) - { - run.ActiveNodes.Remove(state.NodeId); - CheckRunCompleted(run); - return; - } - - var context = new ScenarioNodeContext(this, ToPlanRun(run), node); - switch (node.Type ?? string.Empty) - { - case WaitNode: - DispatchWait(run, state, context); - break; - case RemoteConfigOverrideNode: - DispatchRemoteConfigOverride(run, state, context); - break; - case NotificationNode: - EmitNotification(context); - break; - case StoreNode: - EmitStoreOffer(context); - break; - case QuestNode: - EmitQuest(context); - break; - case LeaderboardNode: - EmitLeaderboard(context); - break; - case BattlePassNode: - EmitBattlePass(context); - break; - case BattlePassLevelNode: - EmitBattlePassLevel(context); - break; - default: - // Unsupported node type — fail the run (surfaced via OnScenarioFailed) - // instead of leaving it stalled on a node no handler will complete. - _client.Options.Logger?.Log(RudderLogLevel.Warning, $"[Rudder] Unsupported scenario node type '{node.Type}' ({node.Id})."); - FailRun(run, state.NodeId, new Exception($"Unsupported scenario node type '{node.Type}'")); - break; - } - } - - private void DispatchWait(RuntimeRun run, ActiveNodeState state, ScenarioNodeContext context) - { - if (!state.WaitDeadlineUtc.HasValue) - { - // Prefer server-provided WaitDeadline from the plan boundary over local calculation. - // The server stamps WaitDeadline on server-enforced wait boundaries (see StampBoundaries). - var boundary = run.Plan.BoundaryNodes?.FirstOrDefault( - b => b.SourceNodeId == state.NodeId && b.WaitDeadline.HasValue - ); - if (boundary?.WaitDeadline is { } parsed) - { - state.WaitDeadlineUtc = parsed; - } - else - { - var delay = GetWaitDelay(context.Data); - state.WaitDeadlineUtc = _client.Clock.UtcNow.Add(delay); - } - Persist(); - } - - var session = new WaitSession(context, state.WaitDeadlineUtc.Value); - OnWait?.Invoke(session); - - if (_client.Clock.UtcNow >= state.WaitDeadlineUtc.Value) - _ = CompleteNodeAsync(run.RunId, state.NodeId, "onComplete"); - } - - private void DispatchRemoteConfigOverride(RuntimeRun run, ActiveNodeState state, ScenarioNodeContext context) - { - var patches = context.Data["patches"] as JArray; - if (patches != null) - { - foreach (var patchToken in patches.OfType()) - { - var key = patchToken.Value("path"); - if (string.IsNullOrEmpty(key)) - continue; - - var valueType = patchToken.Value("valueType") ?? "json"; - var value = SerializeRemoteConfigValue(patchToken["value"], valueType); - _client.RemoteConfig.ApplyOverride(key, value, valueType); - } - } - - var session = new ConfigChangedSession(context); - OnConfigChanged?.Invoke(session); - _ = CompleteNodeAsync(run.RunId, state.NodeId, "output"); - } - - private void EmitNotification(ScenarioNodeContext context) - { - OnNotification?.Invoke(new NotificationSession(context)); - } - - private void EmitStoreOffer(ScenarioNodeContext context) - { - OnStoreOffer?.Invoke(new StoreOfferSession(context)); - } - - private void EmitQuest(ScenarioNodeContext context) - { - OnQuest?.Invoke(new QuestSession(context)); - } - - private void EmitLeaderboard(ScenarioNodeContext context) - { - OnLeaderboard?.Invoke(new LeaderboardSession(context)); - } - - private void EmitBattlePass(ScenarioNodeContext context) - { - OnBattlePass?.Invoke(new BattlePassSession(context, _client.BattlePass)); - } - - private void EmitBattlePassLevel(ScenarioNodeContext context) - { - OnBattlePassLevel?.Invoke(new BattlePassLevelSession(context)); - } - - private static TimeSpan GetWaitDelay(JObject data) - { - var duration = data.Value("duration") ?? 0; - var unit = data.Value("unit") ?? "seconds"; - if (duration <= 0) - return TimeSpan.Zero; - - switch (unit) - { - case "days": - case "day": - case "d": - return TimeSpan.FromDays(duration); - case "hours": - case "hour": - case "hr": - case "h": - return TimeSpan.FromHours(duration); - case "minutes": - case "minute": - case "min": - case "m": - return TimeSpan.FromMinutes(duration); - case "seconds": - case "second": - case "sec": - case "s": - return TimeSpan.FromSeconds(duration); - default: - return TimeSpan.FromSeconds(duration); - } - } - - private static string? SerializeRemoteConfigValue(JToken? token, string valueType) - { - if (token == null || token.Type == JTokenType.Null) - return null; - - switch ((valueType ?? string.Empty).ToLowerInvariant()) - { - case "string": - return token.Type == JTokenType.String ? token.Value() : token.ToString(Formatting.None); - case "bool": - case "boolean": - return token.Value().ToString().ToLowerInvariant(); - case "int": - case "integer": - return token.Value().ToString(CultureInfo.InvariantCulture); - case "float": - case "double": - return token.Value().ToString(CultureInfo.InvariantCulture); - default: - return token.Type == JTokenType.String ? token.Value() : token.ToString(Formatting.None); - } - } -} diff --git a/Services/Scenarios/ScenarioService.NodeTypes.cs b/Services/Scenarios/ScenarioService.NodeTypes.cs deleted file mode 100644 index b1c81e3..0000000 --- a/Services/Scenarios/ScenarioService.NodeTypes.cs +++ /dev/null @@ -1,13 +0,0 @@ -namespace RudderSdk.Core; - -public sealed partial class ScenarioService -{ - private const string NotificationNode = "notification"; - private const string StoreNode = "store"; - private const string WaitNode = "wait"; - private const string RemoteConfigOverrideNode = "remote_config_override"; - private const string QuestNode = "quest"; - private const string LeaderboardNode = "leaderboard"; - private const string BattlePassNode = "battlepass"; - private const string BattlePassLevelNode = "battlepass_level"; -} diff --git a/Services/Scenarios/ScenarioService.Persistence.cs b/Services/Scenarios/ScenarioService.Persistence.cs deleted file mode 100644 index 2b1277a..0000000 --- a/Services/Scenarios/ScenarioService.Persistence.cs +++ /dev/null @@ -1,92 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using Newtonsoft.Json; -using RudderSdk.Core.Models; - -namespace RudderSdk.Core; - -public sealed partial class ScenarioService -{ - private void Persist() - { - var store = _client.Options.PlanStateStore; - if (store == null) - return; - - if (_runs.Count == 0) - { - store.State = null; - return; - } - - var state = new PersistedScenarioState - { - Runs = _runs.Values.Select(run => run.ToPersisted()).ToList() - }; - store.State = JsonConvert.SerializeObject(state); - } - - private sealed class RuntimeRun - { - public RuntimeRun(string runId, ExecutionPlan plan) - { - RunId = runId; - Plan = plan; - } - - public string RunId { get; } - public ExecutionPlan Plan { get; } - public Dictionary ActiveNodes { get; } = new(); - public HashSet CompletedHandles { get; } = new(StringComparer.Ordinal); - - public PersistedRun ToPersisted() - { - return new PersistedRun - { - RunId = RunId, - Plan = Plan, - ActiveNodes = ActiveNodes.Values.ToList(), - CompletedHandles = CompletedHandles.ToList() - }; - } - - public static RuntimeRun FromPersisted(PersistedRun saved) - { - var run = new RuntimeRun(saved.RunId, saved.Plan); - if (saved.ActiveNodes != null) - { - foreach (var node in saved.ActiveNodes) - { - if (!string.IsNullOrEmpty(node?.NodeId)) - run.ActiveNodes[node.NodeId] = node; - } - } - if (saved.CompletedHandles != null) - { - foreach (var handle in saved.CompletedHandles) - run.CompletedHandles.Add(handle); - } - return run; - } - } - - private sealed class PersistedScenarioState - { - public List? Runs { get; set; } - } - - private sealed class PersistedRun - { - public string RunId { get; set; } = string.Empty; - public ExecutionPlan Plan { get; set; } = null!; - public List? ActiveNodes { get; set; } - public List? CompletedHandles { get; set; } - } - - private sealed class ActiveNodeState - { - public string NodeId { get; set; } = string.Empty; - public DateTimeOffset? WaitDeadlineUtc { get; set; } - } -} diff --git a/Services/Scenarios/ScenarioService.Runtime.cs b/Services/Scenarios/ScenarioService.Runtime.cs deleted file mode 100644 index 3847b81..0000000 --- a/Services/Scenarios/ScenarioService.Runtime.cs +++ /dev/null @@ -1,348 +0,0 @@ -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; - -/// -/// 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. -/// -public sealed class TransientBoundaryException : Exception -{ - /// Creates the exception. - public TransientBoundaryException(string message, Exception inner) - : base(message, inner) - { - } - - /// Creates the exception wrapping the transport error. - public TransientBoundaryException(Exception inner) - : base($"Transient boundary error: {inner.Message}", inner) - { - } -} - -public sealed partial class ScenarioService -{ - /// - /// Returns true for exceptions that may succeed on retry - /// (timeout / network failures). - /// - 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 StartPlans(IEnumerable? plans) - { - var started = new List(); - 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; - } - - /// - /// 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. - /// - 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 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( - "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( - "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 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 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); -} diff --git a/Services/Scenarios/Sessions/BattlePassLevelSession.cs b/Services/Scenarios/Sessions/BattlePassLevelSession.cs deleted file mode 100644 index 7694385..0000000 --- a/Services/Scenarios/Sessions/BattlePassLevelSession.cs +++ /dev/null @@ -1,32 +0,0 @@ -using System.Threading; -using System.Threading.Tasks; - -namespace RudderSdk.Core; - -/// -/// Session of a scenario battle-pass-level node — a single claimable tier. -/// crosses onComplete, which the server accepts only -/// once the player has reached the node's configured level. -/// -public sealed class BattlePassLevelSession -{ - internal BattlePassLevelSession(ScenarioNodeContext context) => Context = context; - - /// Underlying node context. - public ScenarioNodeContext Context { get; } - - /// Node id. - public string Id => Context.NodeId; - - /// The tier level this node claims. - public int Level => Context.Get("levelNumber", 0); - - /// Reads a typed value from the node data. - public T Get(string key, T defaultValue = default!) => Context.Get(key, defaultValue); - - /// Claims this tier; crosses onComplete (the server checks the level was reached). - public Task ClaimAsync(CancellationToken cancellationToken = default) => Context.CompleteAsync("onComplete", cancellationToken); - - /// Claims this tier (fire-and-forget). - public void Claim() => Context.Complete("onComplete"); -} diff --git a/Services/Scenarios/Sessions/BattlePassSession.cs b/Services/Scenarios/Sessions/BattlePassSession.cs deleted file mode 100644 index 658ceb7..0000000 --- a/Services/Scenarios/Sessions/BattlePassSession.cs +++ /dev/null @@ -1,95 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; -using RudderSdk.Core.Models.BattlePass; - -namespace RudderSdk.Core; - -/// -/// Session of a scenario battle-pass node. Exposes the battle pass operations -/// bound to this node's scenario/node/run ids, plus explicit boundary crossings -/// the game drives from its UI; the server validates each crossing. -/// -public sealed class BattlePassSession -{ - private readonly BattlePassService _battlePass; - - internal BattlePassSession(ScenarioNodeContext context, BattlePassService battlePass) - { - Context = context; - _battlePass = battlePass; - } - - /// Underlying node context. - public ScenarioNodeContext Context { get; } - - /// Node id. - public string Id => Context.NodeId; - - /// Reads a typed value from the node data. - public T Get(string key, T defaultValue = default!) => Context.Get(key, defaultValue); - - /// Reads current progress (xp, level, premium ownership, claimed tiers) for this node. - public Task GetProgressAsync(CancellationToken cancellationToken = default) - => _battlePass.GetProgressAsync(Context.ScenarioId, Context.NodeId, cancellationToken); - - /// Credits xp from a configured source. - public Task AddXpAsync(string source, long amount, CancellationToken cancellationToken = default) - => _battlePass.AddXpAsync(new AddBattlePassXpRequest - { - ScenarioId = Context.ScenarioId, - NodeId = Context.NodeId, - RunId = Context.RunId, - Source = source, - Amount = amount - }, cancellationToken); - - /// - /// Claims a tier reward at a reached level. - /// is or . - /// - public Task ClaimRewardAsync(int level, string track, CancellationToken cancellationToken = default) - => _battlePass.ClaimRewardAsync(new ClaimBattlePassRewardRequest - { - ScenarioId = Context.ScenarioId, - NodeId = Context.NodeId, - RunId = Context.RunId, - Level = level, - Track = track - }, cancellationToken); - - /// Purchases the premium track, then crosses onPremiumPurchase on success. - public async Task PurchasePremiumAsync(CancellationToken cancellationToken = default) - { - var response = await _battlePass.PurchasePremiumAsync(new PurchaseBattlePassPremiumRequest - { - ScenarioId = Context.ScenarioId, - NodeId = Context.NodeId, - RunId = Context.RunId, - IdempotencyKey = Guid.NewGuid().ToString() - }, cancellationToken).ConfigureAwait(false); - - if (response != null && response.Success == true) - await Context.CompleteAsync("onPremiumPurchase", cancellationToken).ConfigureAwait(false); - - return response!; - } - - /// Advances the run through the onLevelUp handle. - public Task LevelUpAsync(CancellationToken cancellationToken = default) => Context.CompleteAsync("onLevelUp", cancellationToken); - - /// Advances the run through the onLevelUp handle (fire-and-forget). - public void LevelUp() => Context.Complete("onLevelUp"); - - /// Advances the run through the onMaxLevel handle. - public Task MaxLevelAsync(CancellationToken cancellationToken = default) => Context.CompleteAsync("onMaxLevel", cancellationToken); - - /// Advances the run through the onMaxLevel handle (fire-and-forget). - public void MaxLevel() => Context.Complete("onMaxLevel"); - - /// Advances the run through the onComplete handle. - public Task CompleteAsync(CancellationToken cancellationToken = default) => Context.CompleteAsync("onComplete", cancellationToken); - - /// Advances the run through the onComplete handle (fire-and-forget). - public void Complete() => Context.Complete("onComplete"); -} diff --git a/Services/Scenarios/Sessions/ConfigChangedSession.cs b/Services/Scenarios/Sessions/ConfigChangedSession.cs deleted file mode 100644 index 0f6008e..0000000 --- a/Services/Scenarios/Sessions/ConfigChangedSession.cs +++ /dev/null @@ -1,13 +0,0 @@ -namespace RudderSdk.Core; - -/// -/// Session of a scenario remote-config-override node. The patches are already -/// applied to when the event fires. -/// -public sealed class ConfigChangedSession -{ - internal ConfigChangedSession(ScenarioNodeContext context) => Context = context; - - /// Underlying node context. - public ScenarioNodeContext Context { get; } -} diff --git a/Services/Scenarios/Sessions/LeaderboardSession.cs b/Services/Scenarios/Sessions/LeaderboardSession.cs deleted file mode 100644 index 0752d39..0000000 --- a/Services/Scenarios/Sessions/LeaderboardSession.cs +++ /dev/null @@ -1,40 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; - -namespace RudderSdk.Core; - -/// Session of a scenario leaderboard node. -public sealed class LeaderboardSession -{ - internal LeaderboardSession(ScenarioNodeContext context) => Context = context; - - /// Underlying node context. - public ScenarioNodeContext Context { get; } - - /// Node id. - public string Id => Context.NodeId; - - /// Reads a typed value from the node data. - public T Get(string key, T defaultValue = default!) => Context.Get(key, defaultValue); - - /// Advances the run through the onEnd handle. - public Task EndAsync(CancellationToken cancellationToken = default) => Context.CompleteAsync("onEnd", cancellationToken); - - /// Advances the run through the onEnd handle (fire-and-forget). - public void End() => Context.Complete("onEnd"); - - /// Advances the run through the onClaim handle. The server matches live rank to a place. - public Task ClaimAsync(CancellationToken cancellationToken = default) => Context.CompleteAsync("onClaim", cancellationToken); - - /// Advances the run through the onClaim handle (fire-and-forget). - public void Claim() => Context.Complete("onClaim"); - - /// Advances the run through the onClaim handle. - [Obsolete("Use ClaimAsync. Removed in the next SDK version.")] - public Task RewardClaimedAsync(CancellationToken cancellationToken = default) => ClaimAsync(cancellationToken); - - /// Advances the run through the onClaim handle (fire-and-forget). - [Obsolete("Use Claim. Removed in the next SDK version.")] - public void RewardClaimed() => Claim(); -} diff --git a/Services/Scenarios/Sessions/NotificationSession.cs b/Services/Scenarios/Sessions/NotificationSession.cs deleted file mode 100644 index 31b72a0..0000000 --- a/Services/Scenarios/Sessions/NotificationSession.cs +++ /dev/null @@ -1,25 +0,0 @@ -using System.Threading; -using System.Threading.Tasks; - -namespace RudderSdk.Core; - -/// Session of a scenario notification node. -public sealed class NotificationSession -{ - internal NotificationSession(ScenarioNodeContext context) => Context = context; - - /// Underlying node context. - public ScenarioNodeContext Context { get; } - - /// Node id. - public string Id => Context.NodeId; - - /// Reads a typed value from the node data. - public T Get(string key, T defaultValue = default!) => Context.Get(key, defaultValue); - - /// Advances the run through the output handle. - public Task CompleteAsync(CancellationToken cancellationToken = default) => Context.CompleteAsync("output", cancellationToken); - - /// Advances the run through the output handle (fire-and-forget). - public void Complete() => Context.Complete("output"); -} diff --git a/Services/Scenarios/Sessions/QuestSession.cs b/Services/Scenarios/Sessions/QuestSession.cs deleted file mode 100644 index 3785ce6..0000000 --- a/Services/Scenarios/Sessions/QuestSession.cs +++ /dev/null @@ -1,37 +0,0 @@ -using System.Threading; -using System.Threading.Tasks; - -namespace RudderSdk.Core; - -/// Session of a scenario quest node. -public sealed class QuestSession -{ - internal QuestSession(ScenarioNodeContext context) => Context = context; - - /// Underlying node context. - public ScenarioNodeContext Context { get; } - - /// Node id. - public string Id => Context.NodeId; - - /// Reads a typed value from the node data. - public T Get(string key, T defaultValue = default!) => Context.Get(key, defaultValue); - - /// Adds progress to one of the node's counters. - public Task AddProgressAsync(string counterKey, long amount, CancellationToken cancellationToken = default) => Context.AddProgressAsync(counterKey, amount, cancellationToken); - - /// Adds progress to one of the node's counters (fire-and-forget). - public void AddProgress(string counterKey, long amount) => _ = AddProgressAsync(counterKey, amount); - - /// Advances the run through the onComplete handle. - public Task CompleteAsync(CancellationToken cancellationToken = default) => Context.CompleteAsync("onComplete", cancellationToken); - - /// Advances the run through the onComplete handle (fire-and-forget). - public void Complete() => Context.Complete("onComplete"); - - /// Advances the run through the onFail handle. - public Task FailAsync(CancellationToken cancellationToken = default) => Context.CompleteAsync("onFail", cancellationToken); - - /// Advances the run through the onFail handle (fire-and-forget). - public void Fail() => Context.Complete("onFail"); -} diff --git a/Services/Scenarios/Sessions/StoreOfferSession.cs b/Services/Scenarios/Sessions/StoreOfferSession.cs deleted file mode 100644 index b37cc44..0000000 --- a/Services/Scenarios/Sessions/StoreOfferSession.cs +++ /dev/null @@ -1,56 +0,0 @@ -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; - -namespace RudderSdk.Core; - -/// Session of a scenario store-offer node; resolve it with a purchase or a decline. -public sealed class StoreOfferSession -{ - internal StoreOfferSession(ScenarioNodeContext context) - { - Context = context; - Data = context.AsObjectDictionary(); - } - - /// Underlying node context. - public ScenarioNodeContext Context { get; } - - /// Node id. - public string Id => Context.NodeId; - - /// Node data payload. - public IReadOnlyDictionary Data { get; } - - /// True after the session was resolved once. - public bool IsResolved { get; private set; } - - /// Reads a typed value from the node data. - public T Get(string key, T defaultValue = default!) => Context.Get(key, defaultValue); - - /// Resolves the offer as purchased. - public Task PurchaseAsync(CancellationToken cancellationToken = default) => ResolveAsync("onPurchase", cancellationToken); - - /// Resolves the offer as purchased (fire-and-forget). - public void Purchase() => Resolve("onPurchase"); - - /// Resolves the offer as declined. - public Task DeclineAsync(CancellationToken cancellationToken = default) => ResolveAsync("onDecline", cancellationToken); - - /// Resolves the offer as declined (fire-and-forget). - public void Decline() => Resolve("onDecline"); - - private async Task ResolveAsync(string handle, CancellationToken cancellationToken) - { - if (IsResolved) return; - IsResolved = true; - await Context.CompleteAsync(handle, cancellationToken).ConfigureAwait(false); - } - - private void Resolve(string handle) - { - if (IsResolved) return; - IsResolved = true; - Context.Complete(handle); - } -} diff --git a/Services/Scenarios/Sessions/WaitSession.cs b/Services/Scenarios/Sessions/WaitSession.cs deleted file mode 100644 index d1c4cbd..0000000 --- a/Services/Scenarios/Sessions/WaitSession.cs +++ /dev/null @@ -1,19 +0,0 @@ -using System; - -namespace RudderSdk.Core; - -/// Session of a scenario wait node; the run continues automatically at the deadline. -public sealed class WaitSession -{ - internal WaitSession(ScenarioNodeContext context, DateTimeOffset deadlineUtc) - { - Context = context; - DeadlineUtc = deadlineUtc; - } - - /// Underlying node context. - public ScenarioNodeContext Context { get; } - - /// When the wait ends (UTC). - public DateTimeOffset DeadlineUtc { get; } -} diff --git a/tests/Rudder.Core.Tests/RudderClientTests.cs b/tests/Rudder.Core.Tests/RudderClientTests.cs index e074ccc..7c750cc 100644 --- a/tests/Rudder.Core.Tests/RudderClientTests.cs +++ b/tests/Rudder.Core.Tests/RudderClientTests.cs @@ -23,7 +23,7 @@ public sealed class RudderClientTests Assert.NotNull(client.BattlePass); Assert.NotNull(client.Quests); Assert.NotNull(client.Scenario); - Assert.NotNull(client.Realtime); + Assert.NotNull(client.Effects); } [Fact] @@ -85,37 +85,17 @@ public sealed class RudderClientTests Assert.Equal(new[] { RudderAuthState.SignedOut }, states); } - [Fact] - public async Task Realtime_Connect_Returns_Session_From_Injected_Transport() - { - var realtimeTransport = new FakeRealtimeTransport(); - var tokenStore = new FakeTokenStore { AccessToken = "access-token" }; - var client = CreateClient( - tokenStore: tokenStore, - realtimeTransportFactory: new FakeRealtimeTransportFactory(realtimeTransport)); - - var session = await client.Realtime.ConnectAsync(); - - Assert.NotNull(session); - Assert.True(session.IsConnected); - Assert.Equal("access-token", session.AccessToken); - Assert.Equal(new Uri("ws://localhost:8090/api/realtime/ws"), realtimeTransport.Uri); - } - private static RudderClient CreateClient( FakeTransport? transport = null, - FakeTokenStore? tokenStore = null, - IRealtimeTransportFactory? realtimeTransportFactory = null) + FakeTokenStore? tokenStore = null) { return new RudderClient(new RudderClientOptions { BaseUrl = "http://localhost:8082", - RealtimeUrl = "ws://localhost:8090/api/realtime/ws", ProjectKey = "project-key", Transport = transport ?? new FakeTransport(), TokenStore = tokenStore ?? new FakeTokenStore(), - DeviceIdProvider = new FakeDeviceIdProvider(), - RealtimeTransportFactory = realtimeTransportFactory ?? new FakeRealtimeTransportFactory(new FakeRealtimeTransport()) + DeviceIdProvider = new FakeDeviceIdProvider() }); } @@ -169,46 +149,4 @@ public sealed class RudderClientTests { public string DeviceId => "device-id"; } - - private sealed class FakeRealtimeTransportFactory : IRealtimeTransportFactory - { - private readonly IRealtimeTransport _transport; - - public FakeRealtimeTransportFactory(IRealtimeTransport transport) => _transport = transport; - - public IRealtimeTransport Create() => _transport; - } - - private sealed class FakeRealtimeTransport : IRealtimeTransport - { - public Uri? Uri { get; private set; } - public bool IsConnected { get; private set; } - public event Action? Closed; - public event Action? Error; - public event Action>? Received; - - public Task ConnectAsync(Uri uri, TimeSpan timeout, CancellationToken cancellationToken = default) - { - Uri = uri; - IsConnected = true; - return Task.CompletedTask; - } - - public Task SendAsync(ArraySegment payload, CancellationToken cancellationToken = default) - => Task.CompletedTask; - - public Task CloseAsync(CancellationToken cancellationToken = default) - { - IsConnected = false; - Closed?.Invoke(); - return Task.CompletedTask; - } - - public void Update(float deltaTime) - { - } - - public void EmitError(Exception ex) => Error?.Invoke(ex); - public void EmitReceived(ArraySegment data) => Received?.Invoke(data); - } } diff --git a/tests/Rudder.Core.Tests/ScenarioServiceTests.cs b/tests/Rudder.Core.Tests/ScenarioServiceTests.cs index 7c20af1..094e632 100644 --- a/tests/Rudder.Core.Tests/ScenarioServiceTests.cs +++ b/tests/Rudder.Core.Tests/ScenarioServiceTests.cs @@ -2,6 +2,7 @@ using Newtonsoft.Json.Linq; using RudderSdk.Core; using RudderSdk.Core.Abstractions; using RudderSdk.Core.Models; +using RudderSdk.Core.Models.Auth; using RudderSdk.Core.Models.Scenarios; using Xunit; @@ -11,60 +12,98 @@ namespace RudderSdk.Core.Tests; public sealed class ScenarioServiceTests { [Fact] - public async Task TriggerAsync_Starts_All_Returned_Plans() + public async Task TriggerAsync_Emits_Returned_Effects() { var transport = new FakeTransport(); transport.Enqueue(new TriggerScenarioResponse { - Plans = new List + Effects = new List { - Plan("plan-1", Node("n1", "notification")), - Plan("plan-2", Node("n2", "notification")) + Effect("run-1", "n1", "notification"), + Effect("run-2", "n2", "notification") } }); var client = CreateClient(transport); - var notifications = new List(); - client.Scenario.OnNotification += notifications.Add; + var notifications = new List(); + client.Effects.OnNotification += notifications.Add; - var started = await client.Scenario.TriggerAsync("login"); + await client.Scenario.TriggerAsync("login"); Assert.Equal(2, notifications.Count); - Assert.Equal(2, started.Count); - Assert.Equal(2, client.Scenario.ActiveRuns.Count); + Assert.Equal("n1", notifications[0].NodeId); + Assert.Equal("n2", notifications[1].NodeId); } [Fact] - public async Task Completing_Node_Activates_All_Matching_Client_Edges() + public async Task TriggerAsync_Dedups_By_RunId_And_NodeId() + { + var transport = new FakeTransport(); + var effect = Effect("run-1", "n1", "notification"); + transport.Enqueue(new TriggerScenarioResponse { Effects = new List { effect } }); + transport.Enqueue(new TriggerScenarioResponse { Effects = new List { Effect("run-1", "n1", "notification") } }); + var client = CreateClient(transport); + var notifications = 0; + client.Effects.OnNotification += _ => notifications++; + + await client.Scenario.TriggerAsync("login"); + await client.Scenario.TriggerAsync("login"); + + Assert.Equal(1, notifications); + } + + [Fact] + public async Task Completing_Notification_Posts_Output_And_Ingests_Next_Effect() { var transport = new FakeTransport(); transport.Enqueue(new TriggerScenarioResponse { - Plans = new List - { - Plan( - "plan", - new[] { Node("start", "notification"), Node("store", "store"), Node("wait", "wait", new { duration = 1, unit = "minutes" }) }, - new[] - { - Edge("start", "output", "store"), - Edge("start", "output", "wait") - }) - } + Effects = new List { Effect("run-1", "n1", "notification", new { title = "hi", message = "there" }) } + }); + transport.Enqueue(new HandleScenarioCallbackResponse + { + Effect = JObject.FromObject(Effect("run-1", "store", "store")) }); var client = CreateClient(transport); - NotificationSession? notification = null; - var stores = 0; - var waits = 0; - client.Scenario.OnNotification += session => notification = session; - client.Scenario.OnStoreOffer += _ => stores++; - client.Scenario.OnWait += _ => waits++; + NotificationEffect? notification = null; + StoreOfferEffect? store = null; + client.Effects.OnNotification += e => notification = e; + client.Effects.OnStoreOffer += e => store = e; await client.Scenario.TriggerAsync("login"); - await notification!.CompleteAsync(); + Assert.Equal("hi", notification!.Title); + Assert.Equal("there", notification.Message); + await notification.DoneAsync(); - Assert.Equal(1, stores); - Assert.Equal(1, waits); - Assert.Equal(2, client.Scenario.ActiveRuns.Single().ActiveNodeIds.Count); + var callback = Assert.IsType( + transport.Calls.Single(call => call.Path == "/sdk/v1/scenarios/callback").Request); + Assert.Equal("n1", callback.NodeId); + Assert.Equal("output", callback.Handle); + Assert.Equal("run-1", callback.RunId); + Assert.NotNull(store); + Assert.Equal("store", store!.NodeId); + } + + [Fact] + public async Task Completing_Last_Effect_Raises_ScenarioCompleted() + { + var transport = new FakeTransport(); + transport.Enqueue(new TriggerScenarioResponse + { + Effects = new List { Effect("run-1", "n1", "notification") } + }); + transport.Enqueue(new HandleScenarioCallbackResponse()); + var client = CreateClient(transport); + NotificationEffect? notification = null; + ScenarioCompletedEffect? completed = null; + client.Effects.OnNotification += e => notification = e; + client.Effects.OnScenarioCompleted += e => completed = e; + + await client.Scenario.TriggerAsync("login"); + await notification!.DoneAsync(); + + Assert.NotNull(completed); + Assert.Equal("run-1", completed!.RunId); + Assert.Equal("scenario-1", completed.ScenarioId); } [Fact] @@ -73,224 +112,166 @@ public sealed class ScenarioServiceTests var transport = new FakeTransport(); transport.Enqueue(new TriggerScenarioResponse { - Plans = new List + Effects = new List { - Plan( - "plan", - new[] - { - Node("start", "notification"), - Node("store", "store"), - Node("quest", "quest"), - Node("leaderboard", "leaderboard"), - Node("battlepass", "battlepass"), - Node("battlepass-level", "battlepass_level"), - }, - new[] - { - Edge("start", "output", "store"), - Edge("start", "output", "quest"), - Edge("start", "output", "leaderboard"), - Edge("start", "output", "battlepass"), - Edge("start", "output", "battlepass-level"), - }) + Effect("run-1", "n1", "notification"), + Effect("run-2", "n2", "store"), + Effect("run-3", "n3", "quest"), + Effect("run-4", "n4", "leaderboard"), + Effect("run-5", "n5", "battlepass"), + Effect("run-6", "n6", "battlepass_level", new { levelNumber = 3 }), + Effect("run-7", "n7", "wait", waitDeadline: new DateTimeOffset(2026, 5, 30, 10, 30, 0, TimeSpan.Zero)) } }); var client = CreateClient(transport); - NotificationSession? notification = null; + var notification = 0; var store = 0; var quest = 0; var leaderboard = 0; var battlePass = 0; var battlePassLevel = 0; - client.Scenario.OnNotification += session => notification = session; - client.Scenario.OnStoreOffer += _ => store++; - client.Scenario.OnQuest += _ => quest++; - client.Scenario.OnLeaderboard += _ => leaderboard++; - client.Scenario.OnBattlePass += _ => battlePass++; - client.Scenario.OnBattlePassLevel += _ => battlePassLevel++; + var wait = 0; + WaitEffect? waitEffect = null; + BattlePassLevelEffect? levelEffect = null; + client.Effects.OnNotification += _ => notification++; + client.Effects.OnStoreOffer += _ => store++; + client.Effects.OnQuest += _ => quest++; + client.Effects.OnLeaderboard += _ => leaderboard++; + client.Effects.OnBattlePass += _ => battlePass++; + client.Effects.OnBattlePassLevel += e => { battlePassLevel++; levelEffect = e; }; + client.Effects.OnWait += e => { wait++; waitEffect = e; }; await client.Scenario.TriggerAsync("login"); - await notification!.CompleteAsync(); + Assert.Equal(1, notification); Assert.Equal(1, store); Assert.Equal(1, quest); Assert.Equal(1, leaderboard); Assert.Equal(1, battlePass); Assert.Equal(1, battlePassLevel); + Assert.Equal(1, wait); + Assert.Equal(3, levelEffect!.Level); + Assert.Equal(new DateTimeOffset(2026, 5, 30, 10, 30, 0, TimeSpan.Zero), waitEffect!.DeadlineUtc); } [Fact] - public async Task Wait_Persists_Deadline_And_Completes_After_Restore() - { - var clock = new FakeClock(new DateTimeOffset(2026, 5, 30, 10, 0, 0, TimeSpan.Zero)); - var stateStore = new FakePlanStateStore(); - var transport = new FakeTransport(); - transport.Enqueue(new TriggerScenarioResponse - { - Plans = new List - { - Plan( - "plan", - new[] { Node("wait", "wait", new { duration = 30, unit = "minutes" }), Node("done", "notification") }, - new[] { Edge("wait", "onComplete", "done") }) - } - }); - var client = CreateClient(transport, clock: clock, stateStore: stateStore); - - await client.Scenario.TriggerAsync("login"); - Assert.False(string.IsNullOrEmpty(stateStore.State)); - - clock.UtcNow = clock.UtcNow.AddMinutes(31); - var restoredClient = CreateClient(new FakeTransport(), clock: clock, stateStore: stateStore); - var completed = 0; - restoredClient.Scenario.OnNotification += _ => completed++; - - await restoredClient.Scenario.RestoreAsync(); - restoredClient.Update(0); - await Task.Delay(20); - - Assert.Equal(1, completed); - var run = Assert.Single(restoredClient.Scenario.ActiveRuns); - Assert.Equal("done", Assert.Single(run.ActiveNodeIds)); - Assert.Contains("\"done\"", stateStore.State); - } - - [Fact] - public async Task RemoteConfigOverride_Applies_Patches_And_Continues() + public async Task Store_Purchase_And_Decline_Post_Matching_Handles() { var transport = new FakeTransport(); transport.Enqueue(new TriggerScenarioResponse { - Plans = new List - { - Plan( - "plan", - new[] - { - Node("override", "remote_config_override", new - { - patches = new object[] - { - new { path = "difficulty", valueType = "string", value = "hard" }, - new { path = "enemy_count", valueType = "int", value = 12 } - } - }), - Node("done", "notification") - }, - new[] { Edge("override", "output", "done") }) - } + Effects = new List { Effect("run-1", "store", "store") } }); + transport.Enqueue(new HandleScenarioCallbackResponse()); var client = CreateClient(transport); - var notifications = 0; - var configChanges = 0; - client.Scenario.OnNotification += _ => notifications++; - client.Scenario.OnConfigChanged += _ => configChanges++; - - await client.Scenario.TriggerAsync("login"); - - Assert.Equal("hard", client.RemoteConfig.Get("difficulty", "normal")); - Assert.Equal(12, client.RemoteConfig.Get("enemy_count", 0)); - Assert.Equal(1, configChanges); - Assert.Equal(1, notifications); - } - - [Fact] - public async Task Boundary_Callback_Sends_Source_Node_And_Starts_Continuation_Plan() - { - var transport = new FakeTransport(); - transport.Enqueue(new TriggerScenarioResponse - { - Plans = new List - { - Plan( - "plan", - new[] { Node("store", "store") }, - Array.Empty(), - new[] { Boundary("store", "onPurchase", "server-condition") }, - "run-1") - } - }); - transport.Enqueue(new HandleScenarioCallbackResponse - { - Plan = Plan("continuation", Node("done", "notification"), "run-1") - }); - var client = CreateClient(transport); - StoreOfferSession? store = null; - var notifications = 0; - client.Scenario.OnStoreOffer += session => store = session; - client.Scenario.OnNotification += _ => notifications++; + StoreOfferEffect? store = null; + client.Effects.OnStoreOffer += e => store = e; await client.Scenario.TriggerAsync("login"); await store!.PurchaseAsync(); var callback = Assert.IsType( transport.Calls.Single(call => call.Path == "/sdk/v1/scenarios/callback").Request); - Assert.Equal("store", callback.NodeId); Assert.Equal("onPurchase", callback.Handle); - Assert.Equal(1, notifications); + Assert.True(store.IsResolved); + + store.Decline(); + Assert.Equal(1, transport.Calls.Count(call => call.Path == "/sdk/v1/scenarios/callback")); } [Fact] - public async Task Boundary_Wins_Over_Local_Edges() + public async Task Leaderboard_End_Posts_OnEnd() { var transport = new FakeTransport(); transport.Enqueue(new TriggerScenarioResponse { - Plans = new List - { - Plan( - "plan", - new[] { Node("store", "store"), Node("local", "notification") }, - new[] { Edge("store", "onPurchase", "local") }, - new[] { Boundary("store", "onPurchase", "server-condition") }, - "run-1") - } - }); - transport.Enqueue(new HandleScenarioCallbackResponse - { - Plan = Plan("continuation", Node("server", "notification"), "run-1") + Effects = new List { Effect("run-1", "lb", "leaderboard") } }); + transport.Enqueue(new HandleScenarioCallbackResponse()); var client = CreateClient(transport); - StoreOfferSession? store = null; - var notificationIds = new List(); - client.Scenario.OnStoreOffer += session => store = session; - client.Scenario.OnNotification += session => notificationIds.Add(session.Id); + LeaderboardEffect? leaderboard = null; + client.Effects.OnLeaderboard += e => leaderboard = e; await client.Scenario.TriggerAsync("login"); - await store!.PurchaseAsync(); + await leaderboard!.EndAsync(); - Assert.Equal(new[] { "server" }, notificationIds); - Assert.Contains(transport.Calls, call => call.Path == "/sdk/v1/scenarios/callback"); + var callback = Assert.IsType( + transport.Calls.Single(call => call.Path == "/sdk/v1/scenarios/callback").Request); + Assert.Equal("onEnd", callback.Handle); } [Fact] - public async Task Quest_Progress_Completed_Response_Completes_The_Node() + public async Task Leaderboard_Claim_Posts_OnClaim() { var transport = new FakeTransport(); transport.Enqueue(new TriggerScenarioResponse { - Plans = new List - { - Plan("plan", Node("quest", "quest")) - } + Effects = new List { Effect("run-1", "lb", "leaderboard") } }); - transport.Enqueue(new UpdateScenarioCounterResponse { Completed = true }); + transport.Enqueue(new HandleScenarioCallbackResponse()); var client = CreateClient(transport); - QuestSession? quest = null; - var completed = 0; - client.Scenario.OnQuest += session => quest = session; - client.Scenario.OnScenarioCompleted += _ => completed++; + LeaderboardEffect? leaderboard = null; + client.Effects.OnLeaderboard += e => leaderboard = e; await client.Scenario.TriggerAsync("login"); - await quest!.AddProgressAsync("wins", 1); + await leaderboard!.ClaimAsync(); + + var callback = Assert.IsType( + transport.Calls.Single(call => call.Path == "/sdk/v1/scenarios/callback").Request); + Assert.Equal("onClaim", callback.Handle); + } + + [Fact] + public async Task Quest_Progress_Completed_Response_Ingests_Next_Effect() + { + var transport = new FakeTransport(); + transport.Enqueue(new TriggerScenarioResponse + { + Effects = new List { Effect("run-1", "quest", "quest", new { name = "wins" }) } + }); + transport.Enqueue(new UpdateScenarioCounterResponse + { + Completed = true, + Effect = JObject.FromObject(Effect("run-1", "done", "notification")) + }); + var client = CreateClient(transport); + QuestEffect? quest = null; + NotificationEffect? notification = null; + client.Effects.OnQuest += e => quest = e; + client.Effects.OnNotification += e => notification = e; + + await client.Scenario.TriggerAsync("login"); + Assert.Equal("wins", quest!.Name); + await quest.ReportProgressAsync("wins", 1); var counter = Assert.IsType( transport.Calls.Single(call => call.Path == "/sdk/v1/scenarios/counter").Request); Assert.Equal("quest", counter.NodeId); Assert.Equal("wins", counter.CounterKey); + Assert.Equal(1, counter.Amount); + Assert.NotNull(notification); + Assert.Equal("done", notification!.NodeId); + } + + [Fact] + public async Task Quest_Progress_Completed_Without_Next_Raises_ScenarioCompleted() + { + var transport = new FakeTransport(); + transport.Enqueue(new TriggerScenarioResponse + { + Effects = new List { Effect("run-1", "quest", "quest") } + }); + transport.Enqueue(new UpdateScenarioCounterResponse { Completed = true }); + var client = CreateClient(transport); + QuestEffect? quest = null; + var completed = 0; + client.Effects.OnQuest += e => quest = e; + client.Effects.OnScenarioCompleted += _ => completed++; + + await client.Scenario.TriggerAsync("login"); + await quest!.ReportProgressAsync("wins", 1); + Assert.Equal(1, completed); - Assert.Empty(client.Scenario.ActiveRuns); } [Fact] @@ -299,129 +280,253 @@ public sealed class ScenarioServiceTests var transport = new FakeTransport(); transport.Enqueue(new TriggerScenarioResponse { - Plans = new List - { - Plan("plan", Node("quest", "quest")) - } + Effects = new List { Effect("run-1", "quest", "quest") } }); transport.Enqueue(new InvalidOperationException("boom")); var logger = new FakeLogger(); var client = CreateClient(transport, logger: logger); - QuestSession? quest = null; + QuestEffect? quest = null; var failed = 0; - client.Scenario.OnQuest += session => quest = session; - client.Scenario.OnScenarioFailed += _ => failed++; + client.Effects.OnQuest += e => quest = e; + client.Effects.OnScenarioFailed += _ => failed++; await client.Scenario.TriggerAsync("login"); - await quest!.AddProgressAsync("wins", 1); + await quest!.ReportProgressAsync("wins", 1); Assert.Equal(0, failed); - Assert.Single(client.Scenario.ActiveRuns); Assert.Contains(logger.Messages, m => m.Level == RudderLogLevel.Warning && m.Message.Contains("counter update failed")); } + [Fact] + public async Task Unknown_Run_Drops_Effects_And_Raises_Failed() + { + var transport = new FakeTransport(); + transport.Enqueue(new TriggerScenarioResponse + { + Effects = new List { Effect("run-1", "n1", "notification") } + }); + transport.Enqueue(new RudderNotFoundException(404, RudderErrorCodes.UnknownRun, "unknown")); + var client = CreateClient(transport); + NotificationEffect? notification = null; + ScenarioFailedEffect? failure = null; + client.Effects.OnNotification += e => notification = e; + client.Effects.OnScenarioFailed += e => failure = e; + + await client.Scenario.TriggerAsync("login"); + await notification!.DoneAsync(); + + Assert.NotNull(failure); + Assert.Equal("run-1", failure!.RunId); + Assert.Equal("n1", failure.NodeId); + Assert.IsType(failure.Exception); + } + + [Fact] + public async Task Expired_Run_Drops_Effects() + { + var transport = new FakeTransport(); + transport.Enqueue(new TriggerScenarioResponse + { + Effects = new List { Effect("run-1", "n1", "notification") } + }); + transport.Enqueue(new RudderApiException(400, RudderErrorCodes.RunExpired, "expired")); + var client = CreateClient(transport); + NotificationEffect? notification = null; + ScenarioFailedEffect? failure = null; + client.Effects.OnNotification += e => notification = e; + client.Effects.OnScenarioFailed += e => failure = e; + + await client.Scenario.TriggerAsync("login"); + await notification!.DoneAsync(); + + Assert.NotNull(failure); + Assert.Equal("run-1", failure!.RunId); + } + [Fact] public async Task Unknown_Nodes_Fail_The_Run() { var transport = new FakeTransport(); transport.Enqueue(new TriggerScenarioResponse { - Plans = new List - { - Plan("plan", Node("future", "future_node")) - } + Effects = new List { Effect("run-1", "future", "future_node") } }); var logger = new FakeLogger(); var client = CreateClient(transport, logger: logger); - ScenarioFailedEvent? failure = null; - client.Scenario.OnScenarioFailed += e => failure = e; + ScenarioFailedEffect? failure = null; + client.Effects.OnScenarioFailed += e => failure = e; await client.Scenario.TriggerAsync("login"); Assert.NotNull(failure); Assert.Equal("future", failure!.NodeId); - Assert.Empty(client.Scenario.ActiveRuns); Assert.Contains(logger.Messages, m => m.Level == RudderLogLevel.Warning && m.Message.Contains("Unsupported scenario node type 'future_node'")); } + [Fact] + public async Task Wait_Deadline_Refreshes_Pending_Without_Posting_Callback() + { + var clock = new FakeClock(new DateTimeOffset(2026, 5, 30, 10, 0, 0, TimeSpan.Zero)); + var transport = new FakeTransport(); + transport.Enqueue(new ListPendingScenarioEffectsResponse { Effects = new List() }); + transport.Enqueue(new TriggerScenarioResponse + { + Effects = new List + { + Effect("run-1", "wait", "wait", waitDeadline: clock.UtcNow.AddMinutes(30)) + } + }); + var client = CreateClient(transport, clock: clock); + client.Update(0); + await Task.Delay(50); + Assert.Equal(1, transport.Calls.Count(call => call.Path == "/sdk/v1/scenarios/pending")); + + await client.Scenario.TriggerAsync("login"); + client.Update(0); + await Task.Delay(50); + Assert.Equal(1, transport.Calls.Count(call => call.Path == "/sdk/v1/scenarios/pending")); + Assert.DoesNotContain(transport.Calls, call => call.Path == "/sdk/v1/scenarios/callback"); + + transport.Enqueue(new ListPendingScenarioEffectsResponse + { + Effects = new List { Effect("run-1", "done", "notification") } + }); + var notifications = 0; + client.Effects.OnNotification += _ => notifications++; + clock.UtcNow = clock.UtcNow.AddMinutes(31); + client.Update(0); + await Task.Delay(50); + + Assert.Equal(2, transport.Calls.Count(call => call.Path == "/sdk/v1/scenarios/pending")); + Assert.Equal(1, notifications); + Assert.DoesNotContain(transport.Calls, call => call.Path == "/sdk/v1/scenarios/callback"); + } + + [Fact] + public async Task Heartbeat_Refreshes_Pending_After_30_Seconds() + { + var clock = new FakeClock(new DateTimeOffset(2026, 5, 30, 10, 0, 0, TimeSpan.Zero)); + var transport = new FakeTransport(); + transport.Enqueue(new ListPendingScenarioEffectsResponse { Effects = new List() }); + var client = CreateClient(transport, clock: clock); + + client.Update(0); + await Task.Delay(50); + Assert.Equal(1, transport.Calls.Count(call => call.Path == "/sdk/v1/scenarios/pending")); + + clock.UtcNow = clock.UtcNow.AddSeconds(29); + client.Update(0); + await Task.Delay(50); + Assert.Equal(1, transport.Calls.Count(call => call.Path == "/sdk/v1/scenarios/pending")); + + transport.Enqueue(new ListPendingScenarioEffectsResponse { Effects = new List() }); + clock.UtcNow = clock.UtcNow.AddSeconds(1); + client.Update(0); + await Task.Delay(50); + Assert.Equal(2, transport.Calls.Count(call => call.Path == "/sdk/v1/scenarios/pending")); + } + + [Fact] + public async Task Login_Fetches_Pending_On_Next_Update() + { + var transport = new FakeTransport(); + transport.Enqueue(new LoginViaDeviceResponse + { + AccessToken = "access-token", + RefreshToken = "refresh-token" + }); + transport.Enqueue(new ListPendingScenarioEffectsResponse + { + Effects = new List { Effect("run-1", "n1", "notification") } + }); + var tokenStore = new FakeTokenStore(); + var client = CreateClient(transport, tokenStore: tokenStore); + var notifications = 0; + client.Effects.OnNotification += _ => notifications++; + + await client.Auth.LoginWithDeviceAsync("en", "en"); + Assert.Equal(0, notifications); + + client.Update(0); + await Task.Delay(50); + + Assert.Equal(1, notifications); + Assert.Contains(transport.Calls, call => call.Method == "GET" && call.Path == "/sdk/v1/scenarios/pending"); + } + + [Fact] + public async Task BattlePass_LevelUp_Posts_OnLevelUp() + { + var transport = new FakeTransport(); + transport.Enqueue(new TriggerScenarioResponse + { + Effects = new List { Effect("run-1", "bp", "battlepass") } + }); + transport.Enqueue(new HandleScenarioCallbackResponse()); + var client = CreateClient(transport); + BattlePassEffect? battlePass = null; + client.Effects.OnBattlePass += e => battlePass = e; + + await client.Scenario.TriggerAsync("login"); + await battlePass!.LevelUpAsync(); + + var callback = Assert.IsType( + transport.Calls.Single(call => call.Path == "/sdk/v1/scenarios/callback").Request); + Assert.Equal("onLevelUp", callback.Handle); + } + + [Fact] + public async Task BattlePassLevel_Claim_Posts_OnComplete() + { + var transport = new FakeTransport(); + transport.Enqueue(new TriggerScenarioResponse + { + Effects = new List { Effect("run-1", "bpl", "battlepass_level", new { levelNumber = 2 }) } + }); + transport.Enqueue(new HandleScenarioCallbackResponse()); + var client = CreateClient(transport); + BattlePassLevelEffect? level = null; + client.Effects.OnBattlePassLevel += e => level = e; + + await client.Scenario.TriggerAsync("login"); + await level!.ClaimAsync(); + + var callback = Assert.IsType( + transport.Calls.Single(call => call.Path == "/sdk/v1/scenarios/callback").Request); + Assert.Equal("onComplete", callback.Handle); + Assert.Equal(2, level.Level); + } + private static RudderClient CreateClient( FakeTransport transport, FakeClock? clock = null, - FakePlanStateStore? stateStore = null, - IRudderLogger? logger = null) + IRudderLogger? logger = null, + FakeTokenStore? tokenStore = null) { return new RudderClient(new RudderClientOptions { BaseUrl = "http://localhost:8082", - RealtimeUrl = "ws://localhost:8090/api/realtime/ws", ProjectKey = "project-key", Transport = transport, - TokenStore = new FakeTokenStore { AccessToken = "access-token" }, + TokenStore = tokenStore ?? new FakeTokenStore { AccessToken = "access-token" }, DeviceIdProvider = new FakeDeviceIdProvider(), Clock = clock ?? new FakeClock(DateTimeOffset.UtcNow), - PlanStateStore = stateStore ?? new FakePlanStateStore(), - RealtimeTransportFactory = new FakeRealtimeTransportFactory(), Logger = logger }); } - private static ExecutionPlan Plan(string id, ExecutionPlanNode node, string? runId = null) - => Plan(id, new[] { node }, Array.Empty(), runId: runId); - - private static ExecutionPlan Plan( - string id, - IEnumerable nodes, - IEnumerable edges, - IEnumerable? boundaries = null, - string? runId = null) + private static PendingEffect Effect(string runId, string nodeId, string type, object? data = null, DateTimeOffset? waitDeadline = null) { - var nodeList = nodes.ToList(); - return new ExecutionPlan + return new PendingEffect { - PlanId = id, - ScenarioId = "scenario-" + id, - UserId = "user", - StartNodeId = nodeList[0].Id, - Nodes = nodeList, - Edges = edges.ToList(), - BoundaryNodes = boundaries?.ToList() ?? new List(), - RunId = runId!, - Context = new JObject() - }; - } - - private static ExecutionPlanNode Node(string id, string type, object? data = null) - { - return new ExecutionPlanNode - { - Id = id, + RunId = runId, + ScenarioId = "scenario-1", + NodeId = nodeId, Type = type, - Data = data == null ? new JObject() : JObject.FromObject(data) - }; - } - - private static PlanEdge Edge(string source, string handle, string target) - { - return new PlanEdge - { - Id = source + "-" + handle + "-" + target, - Source = source, - SourceHandle = handle, - Target = target, - TargetHandle = "in" - }; - } - - private static BoundaryNode Boundary(string source, string handle, string target) - { - return new BoundaryNode - { - SourceNodeId = source, - SourceHandle = handle, - NodeId = target, - CallbackUrl = "/sdk/v1/scenarios/callback" + Data = data == null ? new JObject() : JObject.FromObject(data), + WaitDeadline = waitDeadline }; } @@ -443,7 +548,11 @@ public sealed class ScenarioServiceTests if (_responses.Count == 0) return Task.FromResult(default(TResponse)!); - return Task.FromResult((TResponse)_responses.Dequeue()); + var next = _responses.Dequeue(); + if (next is Exception ex) + return Task.FromException(ex); + + return Task.FromResult((TResponse)next); } } @@ -492,42 +601,9 @@ public sealed class ScenarioServiceTests public DateTimeOffset UtcNow { get; set; } } - private sealed class FakePlanStateStore : IPlanStateStore - { - public string? State { get; set; } - } - private sealed class FakeLogger : IRudderLogger { public List<(RudderLogLevel Level, string Message)> Messages { get; } = new(); public void Log(RudderLogLevel level, string message) => Messages.Add((level, message)); } - - private sealed class FakeRealtimeTransportFactory : IRealtimeTransportFactory - { - public IRealtimeTransport Create() => new FakeRealtimeTransport(); - } - -#pragma warning disable CS0067 - private sealed class FakeRealtimeTransport : IRealtimeTransport - { - public bool IsConnected { get; private set; } - public event Action? Closed; - public event Action? Error; - public event Action>? Received; - public Task ConnectAsync(Uri uri, TimeSpan timeout, CancellationToken cancellationToken = default) - { - IsConnected = true; - return Task.CompletedTask; - } - public Task SendAsync(ArraySegment payload, CancellationToken cancellationToken = default) => Task.CompletedTask; - public Task CloseAsync(CancellationToken cancellationToken = default) - { - IsConnected = false; - Closed?.Invoke(); - return Task.CompletedTask; - } - public void Update(float deltaTime) { } - } -#pragma warning restore CS0067 }