Server-side scenario execution: effects client replaces local engine
- client.Effects: typed effects + completion methods posting scenario callbacks - client.Scenario reduced to TriggerAsync - pending polling via Update() pump: 30s heartbeat + wait-deadline checks - local engine, plan persistence (IPlanStateStore) and resume removed - RealtimeService and realtime transports removed (relay never deployed) - dead IPlanScheduler leftover removed - models regenerated from openapi (ExecutionPlan/BoundaryNode gone, PendingEffect added)
This commit is contained in:
edmand46
@@ -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<ExecutionPlan>
|
||||
Effects = new List<PendingEffect>
|
||||
{
|
||||
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<NotificationSession>();
|
||||
client.Scenario.OnNotification += notifications.Add;
|
||||
var notifications = new List<NotificationEffect>();
|
||||
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<PendingEffect> { effect } });
|
||||
transport.Enqueue(new TriggerScenarioResponse { Effects = new List<PendingEffect> { 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<ExecutionPlan>
|
||||
{
|
||||
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<PendingEffect> { 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<HandleScenarioCallbackRequest>(
|
||||
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<PendingEffect> { 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<ExecutionPlan>
|
||||
Effects = new List<PendingEffect>
|
||||
{
|
||||
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<ExecutionPlan>
|
||||
{
|
||||
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<ExecutionPlan>
|
||||
{
|
||||
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<PendingEffect> { 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<ExecutionPlan>
|
||||
{
|
||||
Plan(
|
||||
"plan",
|
||||
new[] { Node("store", "store") },
|
||||
Array.Empty<PlanEdge>(),
|
||||
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<HandleScenarioCallbackRequest>(
|
||||
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<ExecutionPlan>
|
||||
{
|
||||
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<PendingEffect> { Effect("run-1", "lb", "leaderboard") }
|
||||
});
|
||||
transport.Enqueue(new HandleScenarioCallbackResponse());
|
||||
var client = CreateClient(transport);
|
||||
StoreOfferSession? store = null;
|
||||
var notificationIds = new List<string>();
|
||||
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<HandleScenarioCallbackRequest>(
|
||||
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<ExecutionPlan>
|
||||
{
|
||||
Plan("plan", Node("quest", "quest"))
|
||||
}
|
||||
Effects = new List<PendingEffect> { 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<HandleScenarioCallbackRequest>(
|
||||
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<PendingEffect> { 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<UpdateScenarioCounterRequest>(
|
||||
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<PendingEffect> { 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<ExecutionPlan>
|
||||
{
|
||||
Plan("plan", Node("quest", "quest"))
|
||||
}
|
||||
Effects = new List<PendingEffect> { 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<PendingEffect> { 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<RudderNotFoundException>(failure.Exception);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Expired_Run_Drops_Effects()
|
||||
{
|
||||
var transport = new FakeTransport();
|
||||
transport.Enqueue(new TriggerScenarioResponse
|
||||
{
|
||||
Effects = new List<PendingEffect> { 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<ExecutionPlan>
|
||||
{
|
||||
Plan("plan", Node("future", "future_node"))
|
||||
}
|
||||
Effects = new List<PendingEffect> { 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<PendingEffect>() });
|
||||
transport.Enqueue(new TriggerScenarioResponse
|
||||
{
|
||||
Effects = new List<PendingEffect>
|
||||
{
|
||||
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<PendingEffect> { 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<PendingEffect>() });
|
||||
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<PendingEffect>() });
|
||||
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<PendingEffect> { 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<PendingEffect> { 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<HandleScenarioCallbackRequest>(
|
||||
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<PendingEffect> { 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<HandleScenarioCallbackRequest>(
|
||||
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<PlanEdge>(), runId: runId);
|
||||
|
||||
private static ExecutionPlan Plan(
|
||||
string id,
|
||||
IEnumerable<ExecutionPlanNode> nodes,
|
||||
IEnumerable<PlanEdge> edges,
|
||||
IEnumerable<BoundaryNode>? 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<BoundaryNode>(),
|
||||
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<TResponse>(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<Exception>? Error;
|
||||
public event Action<ArraySegment<byte>>? Received;
|
||||
public Task ConnectAsync(Uri uri, TimeSpan timeout, CancellationToken cancellationToken = default)
|
||||
{
|
||||
IsConnected = true;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
public Task SendAsync(ArraySegment<byte> 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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user