From 1491b5e3577e0bd479cd3e0a1c0946ec44df562c Mon Sep 17 00:00:00 2001 From: edmand46 Date: Fri, 4 Sep 2026 14:08:48 +0300 Subject: [PATCH] Server-side scenario execution: thin effects client replaces local engine - engine/ (DagWalker, sessions, IndexedDbPlanStore) deleted; server owns the graph - trigger/callback/counter carry PendingEffect; GET /sdk/v1/scenarios/pending polled (30s jittered heartbeat paused on hidden tab + wait-deadline timers) - client.effects public API unchanged (done/buy/dismiss/end/claim/battlepass/quest) - planStateStore removed; loginEvent kept; reconcile() makes server the source of truth - generated models regenerated; skills/README/CHANGELOG updated --- CHANGELOG.md | 14 + README.md | 7 +- skills/rudder-web-sdk/SKILL.md | 2 +- skills/rudder-web-sdk/reference/battlepass.md | 7 +- .../rudder-web-sdk/reference/remote-config.md | 6 +- skills/rudder-web-sdk/reference/scenarios.md | 67 +- src/client/RudderClient.ts | 53 +- src/client/RudderClientOptions.ts | 18 +- src/effects/EffectsCenter.ts | 99 +- src/generated/api.ts | 13 +- src/generated/battlepass.ts | 4 +- src/generated/common.ts | 36 - src/generated/scenarios.ts | 31 +- src/scenario/ScenarioService.ts | 859 ++++++++---------- src/scenario/engine/DagWalker.ts | 74 -- src/scenario/engine/IndexedDbPlanStore.ts | 94 -- src/scenario/engine/sessions.ts | 347 ------- src/scenario/engine/types.ts | 63 -- test/AuthService.test.ts | 5 +- test/PublicApi.test.ts | 4 +- test/RudderClient.test.ts | 20 +- test/ScenarioService.test.ts | 628 ++++++------- test/e2e-prod/_setup/harness.ts | 7 +- test/e2e-prod/scenarios.e2e.test.ts | 7 +- test/e2e/boundary.e2e.test.ts | 133 +-- test/e2e/offerJourney.e2e.test.ts | 94 +- test/e2e/persistence.e2e.test.ts | 66 +- test/e2e/remoteConfig.e2e.test.ts | 25 +- test/e2e/storage.e2e.test.ts | 2 +- test/helpers/fakeGateway.ts | 129 +-- test/helpers/memoryPlanStore.ts | 26 - test/helpers/plan.ts | 84 +- tsup.config.ts | 2 +- 33 files changed, 976 insertions(+), 2050 deletions(-) delete mode 100644 src/scenario/engine/DagWalker.ts delete mode 100644 src/scenario/engine/IndexedDbPlanStore.ts delete mode 100644 src/scenario/engine/sessions.ts delete mode 100644 src/scenario/engine/types.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 985f801..ebc6a69 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,19 @@ # Changelog +## Unreleased + +- Scenario execution moved server-side. The SDK no longer walks a local DAG + or persists plan state to IndexedDB. +- Removed `RudderClientOptions.runtime.planStateStore` and the + `PlanStateStore` type. +- `GET /sdk/v1/scenarios/pending` is polled on login, on a ~30s heartbeat, + and at each effect `waitDeadline`. Completions POST + `/sdk/v1/scenarios/callback`; the response may carry the next + `PendingEffect`. +- Scenario `remote_config_override` nodes no longer emit `onConfigChanged` + (the server applies the override). The `ConfigChangedEffect` type and + `onConfigChanged` subscription remain. + ## 0.6.0 - `LeaderboardSession.claim()` / effect `claim()` completes the leaderboard diff --git a/README.md b/README.md index 4c52a03..412629a 100644 --- a/README.md +++ b/README.md @@ -115,8 +115,11 @@ The scenario runtime is not exposed directly; scenario nodes surface through - `onWait`, `onQuest`, `onBattlePass`, `onBattlePassLevel` - `onScenarioCompleted`, `onScenarioFailed` -The scenario engine (and its IndexedDB persistence) loads lazily on first -login — a client that only reads domains never pulls it into the page. +The server executes the scenario graph. The SDK is a thin effects client +(trigger, pending poll, callbacks) loaded lazily on first login — a client +that only reads domains never pulls it into the page. `onConfigChanged` +stays on the surface, but scenario `remote_config_override` nodes are +applied server-side and no longer emit that effect. Errors thrown inside effect handlers are reported via the `onEffectError` client option (default: `console.error`). diff --git a/skills/rudder-web-sdk/SKILL.md b/skills/rudder-web-sdk/SKILL.md index 6b4a95b..b4ebf09 100644 --- a/skills/rudder-web-sdk/SKILL.md +++ b/skills/rudder-web-sdk/SKILL.md @@ -36,7 +36,7 @@ const client = new RudderClient({ // requestTimeoutMs?: number — default 10000 // syncIntervalMs?: number — revision poll, default 30000 (±20% jitter) // onEffectError?: (error) => void — default console.error - // runtime?: { planStateStore?, loginEvent? } — advanced, see reference/scenarios.md + // runtime?: { loginEvent? } — advanced, see reference/scenarios.md }); ``` diff --git a/skills/rudder-web-sdk/reference/battlepass.md b/skills/rudder-web-sdk/reference/battlepass.md index 48a5f17..b9d5224 100644 --- a/skills/rudder-web-sdk/reference/battlepass.md +++ b/skills/rudder-web-sdk/reference/battlepass.md @@ -34,7 +34,6 @@ interface AddBattlePassXpResponse { level?: number; leveledUp?: boolean; maxLevel?: boolean; - plan?: ExecutionPlan; // scenario plan continuation, handled by the runtime xp?: number; } @@ -67,7 +66,6 @@ interface PurchaseBattlePassPremiumRequest { } interface PurchaseBattlePassPremiumResponse { error?: string; - plan?: ExecutionPlan; success?: boolean; } ``` @@ -82,6 +80,5 @@ interface PurchaseBattlePassPremiumResponse { in the response and the typed error `code` (`RudderErrorCodes`). - `purchasePremium` charges the player's wallet; idempotent; pass your own `idempotencyKey` for safe retries. -- Mutations return `ExecutionPlan` continuations — when driving battle pass - manually you are responsible for the scenario run state; prefer the - `onBattlePass` effect session which handles this. +- Prefer the `onBattlePass` effect session, which binds `scenarioId` / + `nodeId` / `runId` and posts scenario callbacks for you. diff --git a/skills/rudder-web-sdk/reference/remote-config.md b/skills/rudder-web-sdk/reference/remote-config.md index ee3e395..cfcbb75 100644 --- a/skills/rudder-web-sdk/reference/remote-config.md +++ b/skills/rudder-web-sdk/reference/remote-config.md @@ -56,8 +56,8 @@ already filters inactive configs out and may omit the flag). ## Behavior notes - Warmed at login. -- Scenario `remote_config_override` nodes patch the local snapshot and fire - `client.effects.onConfigChanged({ key })` — the patched value is what - `get()` returns afterwards. +- Scenario `remote_config_override` nodes are applied server-side and do not + reach the client. Reload or wait for the config sync poll to observe the + patched value via `get()`. - Without a `TConfig` type argument, `RemoteConfigShape` defaults to `Record` and `get()` returns `unknown`. diff --git a/skills/rudder-web-sdk/reference/scenarios.md b/skills/rudder-web-sdk/reference/scenarios.md index 34fcc4d..a41d9fb 100644 --- a/skills/rudder-web-sdk/reference/scenarios.md +++ b/skills/rudder-web-sdk/reference/scenarios.md @@ -1,29 +1,28 @@ # Scenarios + effects — `client.effects` Scenarios are server-authored node graphs (configured in the dashboard) that -run per player. The SDK's scenario runtime is deliberately **not** part of the -public client surface: scenario nodes surface as typed effects through -`client.effects` (source: `src/effects/EffectsCenter.ts`), and the engine -itself loads lazily on first login. +run per player. The server executes the graph. The SDK is a thin effects +client: it sends trigger events, turns `PendingEffect` payloads into typed +`client.effects` handlers, posts callbacks when the game completes an +effect, and polls for pending effects. The runtime is deliberately **not** +part of the public client surface (source: `src/effects/EffectsCenter.ts`) +and loads lazily on first login. ## Lifecycle -- The runtime starts after login: scenario state is restored from persistence, - then the login event (`'player_login'` by default) is fired to trigger - event-driven scenarios. -- Runs persist to IndexedDB by default, so waits and pending nodes survive - page reloads. On restore, each run is reconciled with the server - (`unknown_run` / `expired` runs are dropped). -- `logout()` / `dispose()` clear all runs and persisted state. +- After login the SDK fetches `GET /sdk/v1/scenarios/pending` (replacing any + local restore), then fires the login event (`'player_login'` by default). +- A heartbeat polls pending every ~30s (±20% jitter), paused while + `document.hidden`. Each effect with a `waitDeadline` also schedules a + local timer so waits fire on time. +- An effect with an already-active `(runId, nodeId)` is not re-emitted. +- `logout()` / `dispose()` stop the poll, cancel wait timers, and drop + active effects. There is no local plan persistence. ### Runtime options (`RudderClientOptions.runtime`) ```ts runtime?: { - // Scenario plan persistence. undefined = default IndexedDB store (created - // lazily); null = disable persistence entirely; or pass a PlanStateStore. - planStateStore?: PlanStateStore | null; - // Scenario event fired automatically after login. // undefined = 'player_login'; null = fire nothing. loginEvent?: string | null; @@ -60,6 +59,11 @@ interface Effects { } ``` +`onConfigChanged` remains on the public surface but scenario +`remote_config_override` nodes no longer reach the client — the server +applies them. Patched values show up through `client.remoteConfig` after +sync/reload. + ### `NotificationEffect` — a notification node became active ```ts @@ -81,19 +85,15 @@ interface Effects { ### `LeaderboardEffect` — a leaderboard node became active ```ts -{ end(): Promise; rewardClaimed(): Promise } +{ end(): Promise; claim(): Promise; rewardClaimed(): Promise } ``` -### `ConfigChangedEffect` — a `remote_config_override` node patched config - -```ts -{ readonly key: string } // client.remoteConfig.get(key) already returns the override -``` +`rewardClaimed()` is a deprecated alias for `claim()`. ### `WaitEffect` — a wait node became active ```ts -{ readonly deadlineUtc: Date } // run resumes automatically at the deadline +{ readonly deadlineUtc: Date } // SDK polls pending at the deadline; the server advances the run ``` ### `QuestEffect` — a scenario quest node became active @@ -106,8 +106,8 @@ interface Effects { } ``` -The node auto-completes server-side once every objective is satisfied; the SDK -crosses the `onComplete` boundary itself when the server reports completion. +The node auto-completes server-side once every objective is satisfied; the +counter response may carry the next `PendingEffect`. ### `BattlePassEffect` — a battle pass node became active @@ -137,7 +137,7 @@ already bound — prefer them over calling the service manually. interface ScenarioCompletedEffect { readonly runId: string; readonly scenarioId: string } interface ScenarioFailedEffect { readonly runId: string; readonly scenarioId: string; readonly nodeId: string; - readonly error: Error; // transport gave up, or unsupported node type + readonly error: Error; } ``` @@ -146,15 +146,20 @@ as console warnings. ## Supported node types -`wait`, `remote_config_override`, `notification`, `store`, `leaderboard`, -`quest`, `battlepass`, `battlepass_level`. Any other node type fails the run -(surfaced via `onScenarioFailed`). +`wait`, `notification`, `store`, `leaderboard`, `quest`, `battlepass`, +`battlepass_level`. Any other node type fails the run (surfaced via +`onScenarioFailed`). `remote_config_override` is applied server-side and is +not delivered to the client. ## Reliability notes -- Node completion is idempotent client-side (completed handles are tracked). -- Transient boundary failures (network error, 5xx) leave the node active for - retry on reconnect; terminal server errors fail the run. +- Completion methods POST `/sdk/v1/scenarios/callback` with + `{scenarioId, runId, nodeId, handle}` using the handles `output`, + `onPurchase`, `onDecline`, `onEnd`, `onClaim`, `onComplete`, `onLevelUp`, + `onPremiumPurchase`. +- Transient callback failures (network error, 5xx) leave the effect active + for retry; `unknown_run` / `run_expired` drop that run's effects and fire + `onScenarioFailed`. - Server scenario errors use the typed codes `run_expired`, `run_not_active`, `node_not_active`, `scenario_not_active`, `unknown_run`, `early_completion`, `objectives_incomplete`, `level_not_reached`, diff --git a/src/client/RudderClient.ts b/src/client/RudderClient.ts index 0ab680b..036c1e8 100644 --- a/src/client/RudderClient.ts +++ b/src/client/RudderClient.ts @@ -9,8 +9,8 @@ * client (which keeps the module graph acyclic). Cross-domain wiring (the * inventory↔catalog link, purchase invalidation) is done here at construction. * - * The scenario runtime (engine + plan persistence) is loaded lazily on first - * use — a client that never logs in never pays for the scenario engine. + * The scenario effects client is loaded lazily on first use — a client that + * never logs in never pays for the scenario machinery. */ import type { @@ -26,7 +26,6 @@ import { LeaderboardsService } from '../leaderboards/LeaderboardsService.js'; import { BattlePassService } from '../battlepass/BattlePassService.js'; import { QuestsService } from '../quests/QuestsService.js'; import type { ScenarioService } from '../scenario/ScenarioService.js'; -import type { PlanStateStore } from '../scenario/engine/IndexedDbPlanStore.js'; import { EffectsCenter, type Effects } from '../effects/EffectsCenter.js'; import { SyncEngine } from '../state/SyncEngine.js'; import type { RemoteConfigShape } from '../state/RemoteConfigState.js'; @@ -38,7 +37,6 @@ import { StorageDomain } from '../domains/StorageDomain.js'; import { ProjectStorageDomain } from '../domains/ProjectStorageDomain.js'; import { StoresDomain } from '../domains/StoresDomain.js'; -/** A domain the client can bulk-invalidate (re-login) or reset (logout). */ type ManagedDomain = { invalidate(): void; reset(): void }; export class RudderClient { @@ -50,7 +48,6 @@ export class RudderClient public readonly quests: QuestsService; public readonly effects: Effects; - // Observable domains: subscribe via `onChange`, read `.data`, or `reload()`. public readonly player: PlayerDomain; public readonly catalog: CatalogDomain; public readonly inventory: InventoryDomain; @@ -60,8 +57,8 @@ export class RudderClient public readonly stores: StoresDomain; private readonly effectsCenter: EffectsCenter; - private scenarioRuntime: ScenarioService | null = null; - private scenarioRuntimePromise: Promise> | null = null; + private scenarioRuntime: ScenarioService | null = null; + private scenarioRuntimePromise: Promise | null = null; private readonly syncEngine: SyncEngine; private readonly loginEvent: string | null; private readonly ctx: RudderContext; @@ -137,11 +134,6 @@ export class RudderClient ); } - /** - * Tears down all background work: stops the sync poll (and its - * visibilitychange listener), cancels scenario wait timers, and drops cached - * state. Call when disposing the client (e.g. on unmount / HMR) to avoid leaks. - */ dispose(): void { this.syncEngine.stop(); this.scenarioRuntime?.clear(); @@ -157,8 +149,6 @@ export class RudderClient this.invalidateAll(); } this.runtimeStarted = true; - // These four warms are independent — fetch them concurrently so login - // doesn't pay a serial round-trip per entity. await Promise.all([ this.remoteConfig.load(), this.player.load(), @@ -166,51 +156,34 @@ export class RudderClient this.catalog.load(), ]); const runtime = await runtimePromise; - await runtime.restore(); + await runtime.start(); if (this.loginEvent) { await this.sendRuntimeEvent(runtime, this.loginEvent); } this.syncEngine.start(); } - /** - * Loads the scenario engine and plan persistence on first use and caches the - * runtime. Dynamic imports keep the engine out of the initial module graph, - * so a thin client never pulls the scenario machinery. - */ - private ensureScenarioRuntime(): Promise> { + private ensureScenarioRuntime(): Promise { this.scenarioRuntimePromise ??= this.createScenarioRuntime(); return this.scenarioRuntimePromise; } - private async createScenarioRuntime(): Promise> { + private async createScenarioRuntime(): Promise { const { ScenarioService } = await import('../scenario/ScenarioService.js'); - const configured = this.options.runtime?.planStateStore; - let planStore: PlanStateStore | null; - if (configured !== undefined) { - planStore = configured; - } else { - const { createIndexedDbPlanStateStore } = await import( - '../scenario/engine/IndexedDbPlanStore.js' - ); - planStore = createIndexedDbPlanStateStore(); - } - this.scenarioRuntime = new ScenarioService( + this.scenarioRuntime = new ScenarioService( this.ctx, { player: this.player, inventory: this.inventory, - config: this.remoteConfig, stores: this.stores, }, this.battlePass, - planStore, ); return this.scenarioRuntime; } private async sendRuntimeEvent( - runtime: ScenarioService, + runtime: ScenarioService, event: string, ): Promise { try { @@ -240,12 +213,10 @@ export class RudderClient ]; } - /** Re-login: refetch every domain that is in use. */ private invalidateAll(): void { for (const domain of this.managedDomains) domain.invalidate(); } - /** Logout: drop all cached data. */ private resetAll(): void { for (const domain of this.managedDomains) domain.reset(); } @@ -254,14 +225,14 @@ export class RudderClient /** * Test/internal access to the scenario runtime, which is deliberately not part * of the public client surface (it is driven through `client.effects`). - * Lazily loads the scenario engine on first call. + * Lazily loads the scenario effects client on first call. * * @internal */ export function getScenarioRuntime( client: RudderClient, -): Promise> { +): Promise { return ( - client as unknown as { ensureScenarioRuntime(): Promise> } + client as unknown as { ensureScenarioRuntime(): Promise } ).ensureScenarioRuntime(); } diff --git a/src/client/RudderClientOptions.ts b/src/client/RudderClientOptions.ts index 00608a0..0e93693 100644 --- a/src/client/RudderClientOptions.ts +++ b/src/client/RudderClientOptions.ts @@ -1,22 +1,6 @@ import type { TokenStore } from '../token/TokenStore.js'; -import type { PlanStateStore } from '../scenario/engine/IndexedDbPlanStore.js'; -/** - * Advanced runtime knobs. Mainly for tests and non-browser hosts; browser - * consumers can ignore these and take the defaults. - */ export interface RudderRuntimeOptions { - /** - * Scenario plan persistence backend. `undefined` uses the default - * IndexedDB store (created lazily with the scenario runtime); pass `null` - * to disable persistence entirely. - */ - planStateStore?: PlanStateStore | null; - - /** - * Scenario event fired automatically after login. `undefined` uses - * `'player_login'`; pass `null` to fire nothing. - */ loginEvent?: string | null; } @@ -46,7 +30,7 @@ export interface RudderClientOptions { */ onEffectError?: (error: unknown) => void; - /** Advanced runtime knobs (plan persistence, login event). */ + /** Advanced runtime knobs (login event). */ runtime?: RudderRuntimeOptions; } diff --git a/src/effects/EffectsCenter.ts b/src/effects/EffectsCenter.ts index 034853c..d38e2f9 100644 --- a/src/effects/EffectsCenter.ts +++ b/src/effects/EffectsCenter.ts @@ -1,15 +1,5 @@ import type { PurchaseOfferResponse } from '../generated/stores.js'; import type { BuyOptions, OfferHandle, ShopHandle } from '../state/shops.js'; -import type { - BattlePassLevelSession, - BattlePassSession, - LeaderboardSession, - NotificationSession, - QuestSession, - StoreSession, - WaitSession, -} from '../scenario/engine/sessions.js'; -import type { PlanRun, ScenarioRunFailedEvent } from '../scenario/engine/types.js'; import type { AddBattlePassXpResponse, ClaimBattlePassRewardResponse, @@ -44,22 +34,15 @@ export interface ConfigChangedEffect { readonly key: string; } -/** A scenario wait node became active; the run resumes at `deadlineUtc`. */ export interface WaitEffect { readonly deadlineUtc: Date; } -/** A scenario run reached a terminal end successfully. */ export interface ScenarioCompletedEffect { readonly runId: string; readonly scenarioId: string; } -/** - * A scenario run failed at a node (transport gave up, or the node type is not - * supported by this SDK). Surfaced so the game can react instead of the failure - * being swallowed into a console warning. - */ export interface ScenarioFailedEffect { readonly runId: string; readonly scenarioId: string; @@ -67,17 +50,12 @@ export interface ScenarioFailedEffect { readonly error: Error; } -/** - * A scenario quest node became active. Report objective progress; the node - * auto-completes server-side once every objective is satisfied. - */ export interface QuestEffect { readonly name: string; readonly objectives: ReadonlyArray>; reportProgress(objectiveId: string, amount?: number): Promise; } -/** A scenario battle pass node became active. */ export interface BattlePassEffect { getProgress(): Promise; addXp(source: string, amount: number): Promise; @@ -87,7 +65,6 @@ export interface BattlePassEffect { end(): Promise; } -/** A scenario battlepass_level node became active (a single claimable tier). */ export interface BattlePassLevelEffect { readonly level: number; claim(): Promise; @@ -166,36 +143,20 @@ export class EffectsCenter implements Effects { } /** @internal */ - emitNotification(session: NotificationSession): void { - this.emit(this.notificationHandlers, { - title: session.title, - message: session.message, - done: () => session.complete(), - }); + emitNotification(effect: NotificationEffect): void { + this.emit(this.notificationHandlers, effect); } /** @internal */ - emitStoreOffer(session: StoreSession): void { - session.getStore() - .then((store) => { - this.emit(this.storeOfferHandlers, { - store, - offers: store.offers, - message: session.get('message', undefined as string | undefined), - buy: (offer, options) => session.buy(offer, options), - dismiss: () => session.decline(), - }); - }) + emitStoreOffer(effect: StoreOfferEffect | Promise): void { + Promise.resolve(effect) + .then((value) => this.emit(this.storeOfferHandlers, value)) .catch((error) => this.onError(error)); } /** @internal */ - emitLeaderboard(session: LeaderboardSession): void { - this.emit(this.leaderboardHandlers, { - end: () => session.end(), - claim: () => session.claim(), - rewardClaimed: () => session.claim(), - }); + emitLeaderboard(effect: LeaderboardEffect): void { + this.emit(this.leaderboardHandlers, effect); } /** @internal */ @@ -204,55 +165,33 @@ export class EffectsCenter implements Effects { } /** @internal */ - emitWait(session: WaitSession): void { - this.emit(this.waitHandlers, { deadlineUtc: session.deadlineUtc }); + emitWait(effect: WaitEffect): void { + this.emit(this.waitHandlers, effect); } /** @internal */ - emitScenarioCompleted(run: PlanRun): void { - this.emit(this.scenarioCompletedHandlers, { - runId: run.runId, - scenarioId: run.scenarioId, - }); + emitScenarioCompleted(effect: ScenarioCompletedEffect): void { + this.emit(this.scenarioCompletedHandlers, effect); } /** @internal */ - emitScenarioFailed(event: ScenarioRunFailedEvent): void { - this.emit(this.scenarioFailedHandlers, { - runId: event.run.runId, - scenarioId: event.run.scenarioId, - nodeId: event.nodeId, - error: event.error, - }); + emitScenarioFailed(effect: ScenarioFailedEffect): void { + this.emit(this.scenarioFailedHandlers, effect); } /** @internal */ - emitQuest(session: QuestSession): void { - this.emit(this.questHandlers, { - name: session.name, - objectives: session.objectives, - reportProgress: (objectiveId, amount) => session.reportProgress(objectiveId, amount), - }); + emitQuest(effect: QuestEffect): void { + this.emit(this.questHandlers, effect); } /** @internal */ - emitBattlePass(session: BattlePassSession): void { - this.emit(this.battlePassHandlers, { - getProgress: () => session.getProgress(), - addXp: (source, amount) => session.addXp(source, amount), - claimReward: (level, track) => session.claimReward(level, track), - purchasePremium: () => session.purchasePremium(), - levelUp: () => session.levelUp(), - end: () => session.end(), - }); + emitBattlePass(effect: BattlePassEffect): void { + this.emit(this.battlePassHandlers, effect); } /** @internal */ - emitBattlePassLevel(session: BattlePassLevelSession): void { - this.emit(this.battlePassLevelHandlers, { - level: session.level, - claim: () => session.claim(), - }); + emitBattlePassLevel(effect: BattlePassLevelEffect): void { + this.emit(this.battlePassLevelHandlers, effect); } private addHandler( diff --git a/src/generated/api.ts b/src/generated/api.ts index 5b530e2..afd3736 100644 --- a/src/generated/api.ts +++ b/src/generated/api.ts @@ -9,7 +9,7 @@ import type { PlayerProfile } from './player.js'; import type { GetProjectStorageResponse, UpdateProjectStorageRequest } from './project-storage.js'; import type { ClaimQuestRequest, ClaimQuestResponse, ListQuestsResponse, ReportQuestProgressRequest, ReportQuestProgressResponse } from './quests.js'; import type { ListRemoteConfigsResponse, RemoteConfig } from './remote-config.js'; -import type { GetScenarioRunRequest, GetScenarioRunResponse, HandleScenarioCallbackRequest, HandleScenarioCallbackResponse, TriggerScenarioRequest, TriggerScenarioResponse, UpdateScenarioCounterRequest, UpdateScenarioCounterResponse } from './scenarios.js'; +import type { HandleScenarioCallbackRequest, HandleScenarioCallbackResponse, ListPendingScenarioEffectsResponse, TriggerScenarioRequest, TriggerScenarioResponse, UpdateScenarioCounterRequest, UpdateScenarioCounterResponse } from './scenarios.js'; import type { GetStorageResponse, UpdateStorageRequest } from './storage.js'; import type { ListStoresResponse, PurchaseOfferRequest, PurchaseOfferResponse, Store } from './stores.js'; @@ -100,12 +100,6 @@ export const api = { ): Promise<{ [key: string]: number }> => t.request<{ [key: string]: number }>('GET', '/sdk/v1/sync'), - getScenarioRun: ( - t: Transport, - body: GetScenarioRunRequest, - ): Promise => - t.request('POST', '/sdk/v1/scenarios/run', body), - getStorage: ( t: Transport, query?: { types?: string; limit?: number; cursor?: string }, @@ -129,6 +123,11 @@ export const api = { ): Promise => t.request('GET', '/sdk/v1/catalog'), + listPendingScenarioEffects: ( + t: Transport, + ): Promise => + t.request('GET', '/sdk/v1/scenarios/pending'), + listQuests: ( t: Transport, ): Promise => diff --git a/src/generated/battlepass.ts b/src/generated/battlepass.ts index 51917a2..f712268 100644 --- a/src/generated/battlepass.ts +++ b/src/generated/battlepass.ts @@ -1,6 +1,6 @@ // Code generated by apigen. DO NOT EDIT. -import type { ExecutionPlan, Reward } from './common.js'; +import type { Reward } from './common.js'; export interface AddBattlePassXpRequest { "amount"?: number; @@ -14,7 +14,6 @@ export interface AddBattlePassXpResponse { "level"?: number; "leveledUp"?: boolean; "maxLevel"?: boolean; - "plan"?: ExecutionPlan; "xp"?: number; } @@ -59,7 +58,6 @@ export interface PurchaseBattlePassPremiumRequest { export interface PurchaseBattlePassPremiumResponse { "error"?: string; - "plan"?: ExecutionPlan; "success"?: boolean; } diff --git a/src/generated/common.ts b/src/generated/common.ts index f18d0f0..d137ede 100644 --- a/src/generated/common.ts +++ b/src/generated/common.ts @@ -1,15 +1,5 @@ // Code generated by apigen. DO NOT EDIT. -export interface BoundaryNode { - "callbackUrl"?: string; - "enforcement"?: "client" | "server"; - "enteredAt"?: string; - "nodeId"?: string; - "sourceHandle"?: string; - "sourceNodeId"?: string; - "waitDeadline"?: string; -} - export interface ErrorResponse { "code"?: "early_completion" | "forbidden" | "level_not_reached" | "node_not_active" | "objectives_incomplete" | "run_expired" | "run_not_active" | "scenario_not_active" | "unknown_run"; "error"?: string; @@ -18,32 +8,6 @@ export interface ErrorResponse { "requestId"?: string; } -export interface ExecutionPlan { - "boundaryNodes"?: BoundaryNode[]; - "context"?: { [key: string]: unknown }; - "edges"?: PlanEdge[]; - "nodes"?: ExecutionPlanNode[]; - "planId"?: string; - "runId"?: string; - "scenarioId"?: string; - "startNodeId"?: string; - "userId"?: string; -} - -export interface ExecutionPlanNode { - "data"?: { [key: string]: unknown }; - "id"?: string; - "type"?: string; -} - -export interface PlanEdge { - "id"?: string; - "source"?: string; - "sourceHandle"?: string; - "target"?: string; - "targetHandle"?: string; -} - export interface Reward { "amount"?: number; "currency"?: string; diff --git a/src/generated/scenarios.ts b/src/generated/scenarios.ts index e886378..cd78abf 100644 --- a/src/generated/scenarios.ts +++ b/src/generated/scenarios.ts @@ -1,17 +1,5 @@ // Code generated by apigen. DO NOT EDIT. -import type { ExecutionPlan } from './common.js'; - -export interface GetScenarioRunRequest { - "runId"?: string; -} - -export interface GetScenarioRunResponse { - "plan"?: ExecutionPlan; - "runId"?: string; - "status"?: "active" | "completed" | "expired"; -} - export interface HandleScenarioCallbackRequest { "handle"?: string; "nodeId"?: string; @@ -20,7 +8,20 @@ export interface HandleScenarioCallbackRequest { } export interface HandleScenarioCallbackResponse { - "plan"?: ExecutionPlan; + "effect"?: PendingEffect; +} + +export interface ListPendingScenarioEffectsResponse { + "effects": PendingEffect[]; +} + +export interface PendingEffect { + "data": { [key: string]: unknown }; + "nodeId": string; + "runId": string; + "scenarioId": string; + "type": string; + "waitDeadline"?: string; } export interface TriggerScenarioRequest { @@ -28,7 +29,7 @@ export interface TriggerScenarioRequest { } export interface TriggerScenarioResponse { - "plans"?: ExecutionPlan[]; + "effects": PendingEffect[]; } export interface UpdateScenarioCounterRequest { @@ -41,6 +42,6 @@ export interface UpdateScenarioCounterRequest { export interface UpdateScenarioCounterResponse { "completed"?: boolean; - "plan"?: ExecutionPlan; + "effect"?: PendingEffect; } diff --git a/src/scenario/ScenarioService.ts b/src/scenario/ScenarioService.ts index 3e06457..3d5d16e 100644 --- a/src/scenario/ScenarioService.ts +++ b/src/scenario/ScenarioService.ts @@ -1,603 +1,452 @@ import type { RudderContext } from '../core/context.js'; -import type { RemoteConfigShape, RemoteConfigState } from '../state/RemoteConfigState.js'; -import type { ShopHandle } from '../state/shops.js'; +import type { ShopHandle, OfferHandle, BuyOptions } from '../state/shops.js'; +import type { PurchaseOfferResponse } from '../generated/stores.js'; import { api } from '../generated/api.js'; -import type { TriggerScenarioResponse } from '../generated/scenarios.js'; -import type { ExecutionPlan } from '../generated/common.js'; -import type { PlanStateStore } from './engine/IndexedDbPlanStore.js'; - -/** - * The subset of domains the scenario runtime drives: it invalidates the player - * and inventory after server mutations, patches remote config for override - * nodes, and resolves stores for store nodes. - */ -export interface ScenarioDomains { - readonly player: { invalidate(): void }; - readonly inventory: { invalidate(): void }; - readonly config: RemoteConfigState; - readonly stores: { getBySlug(slug: string): Promise }; -} +import type { PendingEffect, TriggerScenarioResponse } from '../generated/scenarios.js'; +import { RudderErrorCodes } from '../generated/errors.js'; import { RudderNetworkError, RudderHttpError, } from '../client/RudderError.js'; - -import { - findNode, - matchingEdges, - matchingBoundaryNodes, - completedHandleKey, - durationToMs, -} from './engine/DagWalker.js'; -import { RuntimeRun, PlanRun, type ActiveNodeState, type ScenarioRunFailedEvent } from './engine/types.js'; -import { - ScenarioNodeContext, - NotificationSession, - WaitSession, - StoreSession, - LeaderboardSession, - QuestSession, - BattlePassSession, - BattlePassLevelSession, -} from './engine/sessions.js'; import type { BattlePassService } from '../battlepass/BattlePassService.js'; -/** - * Error thrown when a boundary HTTP call fails with a transient error - * (network failure or server 5xx). The caller should NOT advance the run; - * the node stays active and the handle stays pending for retry on reconnect. - */ -class TransientBoundaryError extends Error { - constructor(public readonly inner: unknown) { - super('Transient boundary error'); - this.name = 'TransientBoundaryError'; - } +export interface ScenarioDomains { + readonly player: { invalidate(): void }; + readonly inventory: { invalidate(): void }; + readonly stores: { getBySlug(slug: string): Promise }; +} + +const DEFAULT_PENDING_INTERVAL_MS = 30_000; +const JITTER_RATIO = 0.2; + +type ActiveRecord = { + runId: string; + nodeId: string; + scenarioId: string; + storePurchased?: boolean; + lastPurchase?: PurchaseOfferResponse; +}; + +function effectKey(runId: string, nodeId: string): string { + return `${runId}:${nodeId}`; } -/** Returns true for network / 5xx errors that may succeed on retry. */ function isTransientHttpError(err: unknown): boolean { if (err instanceof RudderNetworkError) return true; if (err instanceof RudderHttpError && err.status >= 500) return true; return false; } -function isRankNotEligible(err: unknown): boolean { - return err instanceof RudderHttpError && err.code === 'rank_not_eligible'; +function isDroppedRunError(err: unknown): boolean { + if (!(err instanceof RudderHttpError)) return false; + return ( + err.code === RudderErrorCodes.unknownRun || err.code === RudderErrorCodes.runExpired + ); } +function asString(value: unknown, fallback = ''): string { + return typeof value === 'string' ? value : fallback; +} -export class ScenarioService { - private readonly runs = new Map(); +function asObjectives(value: unknown): Array> { + if (!Array.isArray(value)) return []; + return value.filter( + (item): item is Record => + item !== null && typeof item === 'object' && !Array.isArray(item), + ); +} + +function asLevel(value: unknown): number { + return typeof value === 'number' && Number.isFinite(value) ? value : 0; +} + +export class ScenarioService { + private readonly active = new Map(); private readonly waitTimers = new Map>(); + private readonly droppedRuns = new Set(); + private readonly completedRuns = new Set(); + private heartbeatTimer: ReturnType | undefined; + private running = false; + private polling = false; + private pollQueued = false; constructor( private readonly ctx: RudderContext, private readonly domains: ScenarioDomains, private readonly battlePass: BattlePassService, - private readonly planStore: PlanStateStore | null, + private readonly intervalMs: number = DEFAULT_PENDING_INTERVAL_MS, ) {} - // ---- Public Events (subscribe to receive node dispatches) ---- - - onNotification?: (session: NotificationSession) => void; - onWait?: (session: WaitSession) => void; - onStore?: (session: StoreSession) => void; - onLeaderboard?: (session: LeaderboardSession) => void; - onRunCompleted?: (run: PlanRun) => void; - onRunFailed?: (event: ScenarioRunFailedEvent) => void; - onCompleted?: () => void; - get isRunning(): boolean { - return this.runs.size > 0; + return this.active.size > 0; } - get activeRuns(): readonly PlanRun[] { - return [...this.runs.values()].map((r) => this.toPlanRun(r)); - } - - // ---- Public API ---- - - /** Triggers scenarios by event name. Returns all started plans. */ async send(eventName: string): Promise { const response = await api.triggerScenario(this.ctx, { event: eventName }); this.domains.player.invalidate(); this.domains.inventory.invalidate(); - this.startPlans(response.plans ?? []); + this.ingest(response.effects ?? []); return response; } - /** Completes the current active node with the given handle. */ - async respond(handle: string): Promise { - for (const run of this.runs.values()) { - for (const nodeId of run.activeNodes.keys()) { - await this.completeNodeAsync(run.runId, nodeId, handle); - return; - } + async start(): Promise { + if (this.running) return; + this.running = true; + if (typeof document !== 'undefined') { + document.addEventListener('visibilitychange', this.onVisibilityChange); } + await this.pollPending(); + this.scheduleHeartbeat(); } - /** - * Restores persisted scenario state from IndexedDB. - * Call once after constructing the client. - */ - async restore(): Promise { - const store = this.planStore; - if (!store) return; - try { - await store.load(); - const json = store.state; - if (!json) return; - const persisted = JSON.parse(json) as { runs: Array<{ - runId: string; - plan: ExecutionPlan; - activeNodes: ActiveNodeState[]; - completedHandles: string[]; - }> }; - for (const saved of persisted.runs ?? []) { - let plan = saved.plan; - let activeNodes = saved.activeNodes; - let completedHandles = saved.completedHandles; - - try { - const response = await api.getScenarioRun(this.ctx, { runId: saved.runId }); - const status: string | undefined = response?.status; - - if (status === 'unknown_run' || status === 'expired') { - continue; - } - - if (response?.plan) { - plan = response.plan; - activeNodes = []; - completedHandles = []; - } - } catch { - // Network/server error — fall back to persisted local state. - } - - const run = new RuntimeRun(saved.runId, plan); - for (const handle of completedHandles ?? []) { - run.completedHandles.add(handle); - } - this.runs.set(run.runId, run); - if (activeNodes.length > 0) { - for (const nodeState of activeNodes) { - run.activeNodes.set(nodeState.nodeId, nodeState); - this.dispatchActiveNode(run, nodeState, true); - } - } else if (plan?.nodes?.length) { - const startNode = findNode(plan, plan.startNodeId) ?? plan.nodes[0]; - if (startNode?.id) { - this.activateNode(run, startNode.id); - } - } - } - await this.persist(); - } catch { - // Corrupted state — clear and continue. - this.clear(); - } - } - - /** Clears all active runs and persisted state. */ clear(): void { + this.running = false; + this.pollQueued = false; + if (this.heartbeatTimer) { + clearTimeout(this.heartbeatTimer); + this.heartbeatTimer = undefined; + } for (const timer of this.waitTimers.values()) { clearTimeout(timer); } this.waitTimers.clear(); - this.runs.clear(); - this.persist().catch(() => {}); - } - - // ---- Internal: Plan lifecycle ---- - - private startPlans(plans: ExecutionPlan[]): void { - for (const plan of plans ?? []) { - this.startPlan(plan); + this.active.clear(); + this.droppedRuns.clear(); + this.completedRuns.clear(); + if (typeof document !== 'undefined') { + document.removeEventListener('visibilitychange', this.onVisibilityChange); } } - private startPlan(plan: ExecutionPlan): void { - if (!plan.nodes?.length) return; - if (plan.boundaryNodes?.length && !plan.runId) { - throw new Error('ExecutionPlan has boundaryNodes but missing runId'); + private readonly onVisibilityChange = (): void => { + if (this.running && typeof document !== 'undefined' && !document.hidden) { + void this.heartbeatTick(); } - // Dedup: server returned a plan with a runId already active — skip without - // restarting wait timers or re-dispatching node sessions. - if (plan.runId && this.runs.has(plan.runId)) return; - const startNode = findNode(plan, plan.startNodeId) ?? plan.nodes[0]; - if (!startNode?.id) return; - const runId = plan.runId ?? crypto.randomUUID().replace(/-/g, ''); - const run = new RuntimeRun(runId, plan); - this.runs.set(run.runId, run); - this.activateNode(run, startNode.id); - this.persist().catch(() => {}); + }; + + private scheduleHeartbeat(): void { + if (!this.running) return; + if (this.heartbeatTimer) clearTimeout(this.heartbeatTimer); + const jitter = 1 + (Math.random() * 2 - 1) * JITTER_RATIO; + this.heartbeatTimer = setTimeout(() => { + void this.heartbeatTick(); + }, this.intervalMs * jitter); } - private activateNode( - run: RuntimeRun, - nodeId: string, - restoredState?: ActiveNodeState, - ): void { - const node = findNode(run.plan, nodeId); - if (!node?.id) return; - const state = restoredState ?? { nodeId }; - run.activeNodes.set(nodeId, state); - this.dispatchActiveNode(run, state, restoredState !== undefined); + private async heartbeatTick(): Promise { + if (!this.running) return; + if (typeof document !== 'undefined' && document.hidden) { + this.scheduleHeartbeat(); + return; + } + await this.pollPending(); + this.scheduleHeartbeat(); } - // ---- Internal: Node Dispatch ---- + private async pollPending(): Promise { + if (this.polling) { + this.pollQueued = true; + return; + } + this.polling = true; + try { + do { + this.pollQueued = false; + try { + const response = await api.listPendingScenarioEffects(this.ctx); + this.reconcile(response.effects ?? []); + } catch { + } + } while (this.pollQueued); + } finally { + this.polling = false; + } + } - private dispatchActiveNode( - run: RuntimeRun, - state: ActiveNodeState, - restored: boolean, - ): void { - const node = findNode(run.plan, state.nodeId); - if (!node) { - run.activeNodes.delete(state.nodeId); - this.checkRunCompleted(run); + private reconcile(effects: PendingEffect[]): void { + const incoming = effects.filter((effect) => !this.droppedRuns.has(effect.runId)); + const incomingKeys = new Set( + incoming.map((effect) => effectKey(effect.runId, effect.nodeId)), + ); + const before = new Set(this.active.keys()); + + this.ingest(incoming); + + for (const [key, record] of [...this.active.entries()]) { + if (incomingKeys.has(key)) continue; + this.deactivate(key); + this.emitCompletedIfIdle(record.runId, record.scenarioId); + } + + const after = new Set(this.active.keys()); + if (before.size !== after.size || [...before].some((key) => !after.has(key))) { + this.domains.player.invalidate(); + this.domains.inventory.invalidate(); + } + } + + private ingest(effects: PendingEffect[]): void { + for (const effect of effects) { + this.ingestOne(effect); + } + } + + private ingestOne(pending: PendingEffect): void { + if (this.droppedRuns.has(pending.runId)) return; + const key = effectKey(pending.runId, pending.nodeId); + if (this.active.has(key)) { + if (pending.waitDeadline && !this.waitTimers.has(key)) { + this.scheduleWait(key, pending.waitDeadline); + } return; } - const ctx = new ScenarioNodeContext( - this.toPlanRun(run), - node, - (rid, nid, h) => this.completeNodeInternal(rid, nid, h, true), - (rid, nid, ck, amt) => this.updateProgressInternal(rid, nid, ck, amt), - ); + const record: ActiveRecord = { + runId: pending.runId, + nodeId: pending.nodeId, + scenarioId: pending.scenarioId, + }; + this.active.set(key, record); + this.completedRuns.delete(pending.runId); + if (pending.waitDeadline) { + this.scheduleWait(key, pending.waitDeadline); + } + this.dispatch(pending, record); + } - switch (node.type) { - case 'wait': - this.dispatchWait(run, state, ctx); - break; - case 'remote_config_override': - this.dispatchRemoteConfigOverride(run, state, ctx); - break; + private dispatch(pending: PendingEffect, record: ActiveRecord): void { + switch (pending.type) { case 'notification': - { - const session = new NotificationSession(ctx); - this.ctx.effects.emitNotification(session); - this.onNotification?.(session); - } + this.ctx.effects.emitNotification({ + title: asString(pending.data.title), + message: asString(pending.data.message), + done: () => this.complete(record, 'output'), + }); break; case 'store': - { - const session = new StoreSession(ctx, this.domains.stores); - this.ctx.effects.emitStoreOffer(session); - this.onStore?.(session); - } + this.dispatchStore(pending, record); break; - case 'leaderboard': - { - const session = new LeaderboardSession(ctx); - this.ctx.effects.emitLeaderboard(session); - this.onLeaderboard?.(session); - } + case 'wait': + this.ctx.effects.emitWait({ + deadlineUtc: pending.waitDeadline + ? new Date(pending.waitDeadline) + : new Date(), + }); break; + case 'leaderboard': { + const end = () => this.complete(record, 'onEnd'); + const claim = () => this.complete(record, 'onClaim'); + this.ctx.effects.emitLeaderboard({ + end, + claim, + rewardClaimed: claim, + }); + break; + } case 'quest': - this.ctx.effects.emitQuest(new QuestSession(ctx)); + this.ctx.effects.emitQuest({ + name: asString(pending.data.name), + objectives: asObjectives(pending.data.objectives), + reportProgress: (objectiveId, amount) => + this.reportQuestProgress(record, objectiveId, amount), + }); break; case 'battlepass': - this.ctx.effects.emitBattlePass(new BattlePassSession(ctx, this.battlePass)); + this.dispatchBattlePass(record); break; case 'battlepass_level': - this.ctx.effects.emitBattlePassLevel(new BattlePassLevelSession(ctx)); + this.ctx.effects.emitBattlePassLevel({ + level: asLevel(pending.data.levelNumber), + claim: () => this.complete(record, 'onComplete'), + }); break; default: - // Unsupported node type — fail the run (surfaced via onScenarioFailed) - // instead of leaving it stalled on a node no handler will complete. console.warn( - `[Rudder] Unsupported scenario node type '${node.type}' (${node.id})`, + `[Rudder] Unsupported scenario node type '${pending.type}' (${pending.nodeId})`, ); - this.failRun( - run, - state.nodeId, - new Error(`Unsupported scenario node type '${node.type}'`), + this.dropRun( + record.runId, + record.nodeId, + new Error(`Unsupported scenario node type '${pending.type}'`), ); break; } } - private dispatchWait( - run: RuntimeRun, - state: ActiveNodeState, - ctx: ScenarioNodeContext, - ): void { - // Prefer server-provided waitDeadline from the plan boundary over local calculation. - // The server stamps waitDeadline on server-enforced wait boundaries (see StampBoundaries). - if (!state.waitDeadlineUtc) { - const boundary = (run.plan.boundaryNodes ?? []).find( - b => b.sourceNodeId === state.nodeId && b.waitDeadline, - ); - if (boundary?.waitDeadline) { - state.waitDeadlineUtc = boundary.waitDeadline; - } else { - const data = ctx.data; - const duration = (data.duration as number) ?? 0; - const unit = (data.unit as string) ?? 'seconds'; - const ms = durationToMs(duration, unit); - state.waitDeadlineUtc = new Date(Date.now() + ms).toISOString(); - } - this.persist().catch(() => {}); - } - - const deadline = new Date(state.waitDeadlineUtc).getTime(); - const session = new WaitSession(ctx, new Date(deadline)); - this.onWait?.(session); - this.ctx.effects.emitWait(session); - - const remaining = deadline - Date.now(); - if (remaining <= 0) { - this.completeNode(run.runId, state.nodeId, 'onComplete'); - } else { - const timerKey = `${run.runId}:${state.nodeId}`; - clearTimeout(this.waitTimers.get(timerKey)); - this.waitTimers.set( - timerKey, - setTimeout(() => { - this.completeNode(run.runId, state.nodeId, 'onComplete'); - this.waitTimers.delete(timerKey); - }, remaining), - ); - } + private dispatchStore(pending: PendingEffect, record: ActiveRecord): void { + const slug = asString(pending.data.storeSlug); + this.ctx.effects.emitStoreOffer( + this.domains.stores.getBySlug(slug).then((store) => ({ + store, + offers: store.offers, + message: typeof pending.data.message === 'string' ? pending.data.message : undefined, + buy: (offer, options) => this.buyStore(record, offer, options), + dismiss: () => this.dismissStore(record), + })), + ); } - private dispatchRemoteConfigOverride( - run: RuntimeRun, - state: ActiveNodeState, - ctx: ScenarioNodeContext, - ): void { - const patches = ctx.data.patches as Array<{ - path?: string; - valueType?: string; - value?: string; - }> | undefined; - - if (patches) { - for (const patch of patches) { - if (patch.path) { - this.domains.config.applyOverride( - patch.path, - patch.value ?? '', - patch.valueType ?? 'json', - ); - this.ctx.effects.emitConfigChanged(patch.path); - } - } - } - - this.completeNode(run.runId, state.nodeId, 'output'); - } - - // ---- Internal: Node completion ---- - - /** Synchronous fire-and-forget completion. */ - private completeNode(runId: string, nodeId: string, handle: string): void { - this.completeNodeInternal(runId, nodeId, handle, true).catch(() => {}); - } - - /** Async completion — called by sessions. */ - async completeNodeAsync( - runId: string, - nodeId: string, - handle: string, - ): Promise { - return this.completeNodeInternal(runId, nodeId, handle, true); - } - - private async completeNodeInternal( - runId: string, - nodeId: string, - handle: string, - continueOnBoundary: boolean, - ): Promise { - const run = this.runs.get(runId); - if (!run) return; - - const key = completedHandleKey(nodeId, handle); - if (run.completedHandles.has(key)) return; // idempotent - - // Mark this transition as in flight so that a transiently-empty - // activeNodes (e.g. between this node and an auto-completing successor) - // does not cause the run to be reported complete prematurely. - run.pendingTransitions++; - - try { - let boundaryContinued = false; - if (continueOnBoundary) { - try { - boundaryContinued = await this.continueBoundary(run, nodeId, handle); - } catch (err: unknown) { - if (err instanceof TransientBoundaryError) { - // Transient error — don't advance the run. - // Node stays active, handle stays pending for retry on reconnect. - run.pendingTransitions--; - await this.persist(); - return; - } - throw err; // terminal — handled by outer catch - } - } - - // Only now — after the server has confirmed — mark the handle and node. - run.completedHandles.add(key); - run.activeNodes.delete(nodeId); - await this.persist(); - - if (!boundaryContinued) { - for (const edge of matchingEdges(run.plan, nodeId, handle)) { - if (edge.target) { - this.activateNode(run, edge.target); - } - } - } - - run.pendingTransitions--; - this.checkRunCompleted(run); - await this.persist(); - } catch (err) { - run.pendingTransitions--; - if (isRankNotEligible(err)) { - await this.persist(); - throw err; - } - this.failRun( - run, - nodeId, - err instanceof Error ? err : new Error(String(err)), - ); - } - } - private async continueBoundary( - run: RuntimeRun, - nodeId: string, - handle: string, - ): Promise { - const boundaries = [...matchingBoundaryNodes(run.plan, nodeId, handle)]; - if (boundaries.length === 0) return false; - - for (const boundary of boundaries) { - try { - const response = await api.handleScenarioCallback(this.ctx, { - scenarioId: run.plan.scenarioId, - nodeId: boundary.sourceNodeId, - handle: boundary.sourceHandle, - runId: run.runId, + private dispatchBattlePass(record: ActiveRecord): void { + const { scenarioId, nodeId, runId } = record; + this.ctx.effects.emitBattlePass({ + getProgress: () => this.battlePass.getProgress(scenarioId, nodeId), + addXp: (source, amount) => + this.battlePass.addXp({ scenarioId, nodeId, source, amount, runId }), + claimReward: (level, track) => + this.battlePass.claimReward({ scenarioId, nodeId, level, track, runId }), + purchasePremium: async () => { + const response = await this.battlePass.purchasePremium({ + scenarioId, + nodeId, + idempotencyKey: crypto.randomUUID(), + runId, }); - this.domains.player.invalidate(); - this.domains.inventory.invalidate(); - if (response?.plan) { - this.startPlan(response.plan); + if (response.success) { + await this.complete(record, 'onPremiumPurchase'); } - } catch (err: unknown) { - // Boundary call failed — try to reconcile with server. - let reconciled = false; - try { - const reconcile = await api.getScenarioRun(this.ctx, { runId: run.runId }); - const reconcileStatus: string | undefined = reconcile?.status; - - if (reconcileStatus === 'unknown_run' || reconcileStatus === 'expired') { - this.runs.delete(run.runId); - await this.persist(); - reconciled = true; - } else if (reconcile?.plan) { - const newRun = new RuntimeRun(run.runId, reconcile.plan); - this.runs.set(run.runId, newRun); - const startNode = findNode(reconcile.plan, reconcile.plan.startNodeId) ?? reconcile.plan.nodes?.[0]; - if (startNode?.id) { - this.activateNode(newRun, startNode.id); - } - reconciled = true; - } - } catch { - // Reconciliation also failed. - } - - if (!reconciled) { - // Reconcile did not resolve — distinguish transient from terminal. - if (isTransientHttpError(err)) { - throw new TransientBoundaryError(err); - } - // Terminal error (typed error code from the server) — let the caller failRun. - throw err; - } - // If reconciled, the boundary was handled (run corrected or removed). - // Fall through to continue to the next boundary. - } - } - return true; + return response; + }, + levelUp: () => this.complete(record, 'onLevelUp'), + end: () => this.complete(record, 'onComplete'), + }); } - // ---- Internal: Counter progress ---- + private async buyStore( + record: ActiveRecord, + offer: OfferHandle, + options?: BuyOptions, + ): Promise { + const key = effectKey(record.runId, record.nodeId); + if (!this.active.has(key)) { + return { success: false, error: 'store session already resolved' }; + } + if (!record.storePurchased) { + const purchase = await offer.buy(options); + if (!purchase.success) return purchase; + record.storePurchased = true; + record.lastPurchase = purchase; + } + await this.complete(record, 'onPurchase'); + return record.lastPurchase ?? { success: true }; + } - async updateProgressInternal( - runId: string, - nodeId: string, - counterKey: string, - amount: number, + private async dismissStore(record: ActiveRecord): Promise { + const key = effectKey(record.runId, record.nodeId); + if (!this.active.has(key)) return; + if (record.storePurchased) return; + await this.complete(record, 'onDecline'); + } + + private async reportQuestProgress( + record: ActiveRecord, + objectiveId: string, + amount = 1, ): Promise { - const run = this.runs.get(runId); try { const response = await api.updateScenarioCounter(this.ctx, { - scenarioId: run?.plan.scenarioId ?? '', - nodeId, - counterKey, + scenarioId: record.scenarioId, + nodeId: record.nodeId, + counterKey: objectiveId, amount, - runId, + runId: record.runId, }); - // The server reports objective completion; it no longer returns a plan from the - // counter endpoint. On completion, cross the quest's onComplete boundary (which - // grants rewards + advances the run) — idempotent if the consumer also calls complete(). - if (response?.completed) { - await this.completeNodeInternal(runId, nodeId, 'onComplete', true); + if (!response?.completed) return; + this.domains.player.invalidate(); + this.domains.inventory.invalidate(); + const key = effectKey(record.runId, record.nodeId); + if (this.active.has(key)) { + this.deactivate(key); + } + if (response.effect) { + this.ingest([response.effect]); + } else { + this.emitCompletedIfIdle(record.runId, record.scenarioId); } } catch { - // Counter update failure does not fail the run. } } - // ---- Internal: Run lifecycle ---- - - private checkRunCompleted(run: RuntimeRun): void { - if (run.activeNodes.size > 0) return; - // A node transition is still settling (e.g. an auto-completing - // remote_config_override node is about to activate its successor). - // Wait for it to finish before declaring the run complete. - if (run.pendingTransitions > 0) return; - if (!this.runs.has(run.runId)) return; // already completed/removed - this.runs.delete(run.runId); - const planRun = this.toPlanRun(run); - this.onRunCompleted?.(planRun); - this.onCompleted?.(); - this.ctx.effects.emitScenarioCompleted(planRun); - this.persist().catch(() => {}); - } - - private failRun( - run: RuntimeRun, - nodeId: string, - error: Error, - ): void { - console.warn( - `[Rudder] Scenario run ${run.runId} failed at node ${nodeId}: ${error.message}`, - ); - this.runs.delete(run.runId); - const event: ScenarioRunFailedEvent = { run: this.toPlanRun(run), nodeId, error }; - this.onRunFailed?.(event); - this.ctx.effects.emitScenarioFailed(event); - this.persist().catch(() => {}); - } - - // ---- Internal: Helpers ---- - - private toPlanRun(run: RuntimeRun): PlanRun { - return new PlanRun( - run.runId, - run.plan.planId ?? '', - run.plan.scenarioId ?? '', - run.plan.userId ?? '', - [...run.activeNodes.keys()], - run.plan, - ); - } - - private async persist(): Promise { - const store = this.planStore; - if (!store) return; - const state = JSON.stringify({ - runs: [...this.runs.values()].map((run) => ({ - runId: run.runId, - plan: run.plan, - activeNodes: [...run.activeNodes.values()], - completedHandles: [...run.completedHandles], - })), - }); - store.state = state; + private async complete(record: ActiveRecord, handle: string): Promise { + const key = effectKey(record.runId, record.nodeId); + if (!this.active.has(key)) return; + if (this.droppedRuns.has(record.runId)) return; try { - await store.save(state); - } catch { - // Best-effort persistence. + const response = await api.handleScenarioCallback(this.ctx, { + scenarioId: record.scenarioId, + nodeId: record.nodeId, + handle, + runId: record.runId, + }); + this.domains.player.invalidate(); + this.domains.inventory.invalidate(); + if (!this.active.has(key)) return; + this.deactivate(key); + if (response?.effect) { + this.ingest([response.effect]); + } else { + this.emitCompletedIfIdle(record.runId, record.scenarioId); + } + } catch (err) { + if (isTransientHttpError(err)) return; + if (isDroppedRunError(err)) { + const error = err instanceof Error ? err : new Error(String(err)); + this.dropRun(record.runId, record.nodeId, error); + throw err; + } + throw err; } } + + private scheduleWait(key: string, waitDeadline: string): void { + const deadline = Date.parse(waitDeadline); + if (Number.isNaN(deadline)) return; + const remaining = Math.max(0, deadline - Date.now()); + const existing = this.waitTimers.get(key); + if (existing) clearTimeout(existing); + this.waitTimers.set( + key, + setTimeout(() => { + this.waitTimers.delete(key); + void this.pollPending(); + }, remaining), + ); + } + + private deactivate(key: string): void { + const timer = this.waitTimers.get(key); + if (timer) { + clearTimeout(timer); + this.waitTimers.delete(key); + } + this.active.delete(key); + } + + private dropRun(runId: string, nodeId: string, error: Error): void { + this.droppedRuns.add(runId); + let scenarioId = ''; + for (const [key, record] of [...this.active.entries()]) { + if (record.runId !== runId) continue; + if (!scenarioId) scenarioId = record.scenarioId; + this.deactivate(key); + } + console.warn( + `[Rudder] Scenario run ${runId} failed at node ${nodeId}: ${error.message}`, + ); + this.ctx.effects.emitScenarioFailed({ + runId, + scenarioId, + nodeId, + error, + }); + } + + private emitCompletedIfIdle(runId: string, scenarioId: string): void { + if (this.droppedRuns.has(runId)) return; + if (this.completedRuns.has(runId)) return; + for (const record of this.active.values()) { + if (record.runId === runId) return; + } + this.completedRuns.add(runId); + this.ctx.effects.emitScenarioCompleted({ runId, scenarioId }); + } } diff --git a/src/scenario/engine/DagWalker.ts b/src/scenario/engine/DagWalker.ts deleted file mode 100644 index 22ed93d..0000000 --- a/src/scenario/engine/DagWalker.ts +++ /dev/null @@ -1,74 +0,0 @@ -import type { ExecutionPlan, ExecutionPlanNode, PlanEdge, BoundaryNode } from '../../generated/common.js'; - -/** Finds a node by ID within a plan (linear scan of the nodes array). */ -export function findNode( - plan: ExecutionPlan, - nodeId: string | undefined, -): ExecutionPlanNode | undefined { - if (!nodeId || !plan.nodes) return undefined; - return plan.nodes.find((n) => n.id === nodeId); -} - -/** Yields all edges whose source node and sourceHandle match. */ -export function* matchingEdges( - plan: ExecutionPlan, - sourceNodeId: string, - sourceHandle: string, -): Generator { - for (const edge of plan.edges ?? []) { - if ( - edge.source === sourceNodeId && - (edge.sourceHandle ?? '') === (sourceHandle ?? '') - ) { - yield edge; - } - } -} - -/** Yields all boundary nodes matching the completed node + handle. */ -export function* matchingBoundaryNodes( - plan: ExecutionPlan, - sourceNodeId: string, - sourceHandle: string, -): Generator { - for (const boundary of plan.boundaryNodes ?? []) { - if ( - boundary.sourceNodeId === sourceNodeId && - (boundary.sourceHandle ?? '') === (sourceHandle ?? '') - ) { - yield boundary; - } - } -} - -/** Returns a deduplication key for completed node+handle pairs. */ -export function completedHandleKey(nodeId: string, handle: string): string { - return `${nodeId}:${handle ?? ''}`; -} - -/** Converts wait duration+unit to milliseconds. Unit required; defaults to seconds when unit is unrecognized. */ -export function durationToMs(duration: number, unit: string): number { - switch ((unit ?? 'seconds').toLowerCase()) { - case 'days': - case 'day': - case 'd': - return duration * 86_400_000; - case 'hours': - case 'hour': - case 'hr': - case 'h': - return duration * 3_600_000; - case 'minutes': - case 'minute': - case 'min': - case 'm': - return duration * 60_000; - case 'seconds': - case 'second': - case 'sec': - case 's': - return duration * 1000; - default: - return duration * 1000; // default: seconds - } -} diff --git a/src/scenario/engine/IndexedDbPlanStore.ts b/src/scenario/engine/IndexedDbPlanStore.ts deleted file mode 100644 index aa15a6f..0000000 --- a/src/scenario/engine/IndexedDbPlanStore.ts +++ /dev/null @@ -1,94 +0,0 @@ -/** - * IndexedDB-backed plan state store for persisting scenario execution state. - * - * The C# SDK uses a synchronous `IPlanStateStore { string State { get; set; } }`. - * IndexedDB is inherently async, so we provide an async `load()` / `save()` API - * alongside a synchronous `state` property for immediate reads after load. - */ - -export interface PlanStateStore { - /** Current serialized state (available after `load()` or `save()`). */ - state: string | null; - - /** Loads persisted state from IndexedDB. Call once on SDK initialization. */ - load(): Promise; - - /** Persists the current state to IndexedDB. */ - save(state: string | null): Promise; -} - -const DB_NAME = 'RudderPlanState'; -const STORE_NAME = 'state'; -const KEY = 'active_runs'; -const DB_VERSION = 1; - -export function createIndexedDbPlanStateStore(): PlanStateStore { - let dbPromise: Promise | null = null; - let cachedState: string | null = null; - - function getDb(): Promise { - if (!dbPromise) { - dbPromise = new Promise((resolve, reject) => { - const request = indexedDB.open(DB_NAME, DB_VERSION); - request.onupgradeneeded = () => { - if (!request.result.objectStoreNames.contains(STORE_NAME)) { - request.result.createObjectStore(STORE_NAME); - } - }; - request.onsuccess = () => resolve(request.result); - request.onerror = () => reject(request.error); - }); - } - return dbPromise; - } - - return { - get state(): string | null { - return cachedState; - }, - - set state(value: string | null) { - cachedState = value; - // Best-effort async write — does not block the setter. - getDb() - .then((db) => { - const tx = db.transaction(STORE_NAME, 'readwrite'); - if (value === null) { - tx.objectStore(STORE_NAME).delete(KEY); - } else { - tx.objectStore(STORE_NAME).put(value, KEY); - } - }) - .catch(() => { - // IndexedDB write failed — state is still in memory. - }); - }, - - async load(): Promise { - const db = await getDb(); - return new Promise((resolve, reject) => { - const tx = db.transaction(STORE_NAME, 'readonly'); - const req = tx.objectStore(STORE_NAME).get(KEY); - req.onsuccess = () => { - cachedState = (req.result as string) ?? null; - resolve(); - }; - req.onerror = () => reject(req.error); - }); - }, - - async save(state: string | null): Promise { - cachedState = state; - const db = await getDb(); - return new Promise((resolve, reject) => { - const tx = db.transaction(STORE_NAME, 'readwrite'); - const req = - state === null - ? tx.objectStore(STORE_NAME).delete(KEY) - : tx.objectStore(STORE_NAME).put(state, KEY); - req.onsuccess = () => resolve(); - req.onerror = () => reject(req.error); - }); - }, - }; -} diff --git a/src/scenario/engine/sessions.ts b/src/scenario/engine/sessions.ts deleted file mode 100644 index 466f664..0000000 --- a/src/scenario/engine/sessions.ts +++ /dev/null @@ -1,347 +0,0 @@ -import type { ExecutionPlanNode } from '../../generated/common.js'; -import type { PlanRun } from './types.js'; -import type { BuyOptions, OfferHandle, ShopHandle } from '../../state/shops.js'; -import type { PurchaseOfferResponse } from '../../generated/stores.js'; -import type { BattlePassService } from '../../battlepass/BattlePassService.js'; -import type { - AddBattlePassXpResponse, - ClaimBattlePassRewardResponse, - GetBattlePassProgressResponse, - PurchaseBattlePassPremiumResponse, -} from '../../generated/battlepass.js'; - -/** Resolves a store handle by slug — satisfied by the stores domain. */ -interface StoreResolver { - getBySlug(slug: string): Promise; -} - -// ---- ScenarioNodeContext ---- - -/** - * Context object passed to each session. - * Wraps the current run, plan node, and raw node data. - * The session calls `complete(handle)` to advance the DAG. - */ -export class ScenarioNodeContext { - constructor( - public readonly run: PlanRun, - public readonly node: ExecutionPlanNode, - private readonly onComplete: (runId: string, nodeId: string, handle: string) => Promise, - private readonly onProgress: (runId: string, nodeId: string, counterKey: string, amount: number) => Promise, - ) {} - - /** Extracts a typed value from node data. */ - get(key: string, defaultValue: T): T { - const data = this.node.data as Record | undefined; - const value = data?.[key]; - if (value === undefined || value === null) return defaultValue; - return value as unknown as T; - } - - /** Full node data as a typed record. */ - get data(): Record { - return (this.node.data as Record) ?? {}; - } - - /** Completes the current node with the given output handle. */ - async complete(handle: string): Promise { - await this.onComplete(this.run.runId, this.node.id!, handle); - } - - /** Updates counter progress for server-managed scenario nodes. */ - async updateProgress(counterKey: string, amount: number): Promise { - await this.onProgress(this.run.runId, this.node.id!, counterKey, amount); - } -} - -// ---- NotificationSession ---- - -export class NotificationSession { - public readonly title: string; - public readonly message: string; - - constructor(private readonly context: ScenarioNodeContext) { - this.title = context.get('title', ''); - this.message = context.get('message', ''); - } - - get(key: string, defaultValue: T): T { - return this.context.get(key, defaultValue); - } - - get data(): Record { - return this.context.data; - } - - get node(): ExecutionPlanNode { - return this.context.node; - } - - async complete(): Promise { - await this.context.complete('output'); - } -} - -// ---- WaitSession ---- - -export class WaitSession { - constructor( - private readonly context: ScenarioNodeContext, - public readonly deadlineUtc: Date, - ) {} - - get(key: string, defaultValue: T): T { - return this.context.get(key, defaultValue); - } - - get data(): Record { - return this.context.data; - } - - get node(): ExecutionPlanNode { - return this.context.node; - } -} - -// ---- StoreSession ---- - -export class StoreSession { - private resolved = false; - - constructor( - private readonly context: ScenarioNodeContext, - private readonly stores: StoreResolver, - ) {} - - get isResolved(): boolean { - return this.resolved; - } - - get(key: string, defaultValue: T): T { - return this.context.get(key, defaultValue); - } - - get data(): Record { - return this.context.data; - } - - get node(): ExecutionPlanNode { - return this.context.node; - } - - async getStore(): Promise { - return this.stores.getBySlug(this.get('storeSlug', '')); - } - - async buy( - offer: OfferHandle, - options?: BuyOptions, - ): Promise { - if (this.resolved) { - return { success: false, error: 'store session already resolved' }; - } - const purchase = await offer.buy(options); - if (purchase.success) { - this.resolved = true; - await this.context.complete('onPurchase'); - } - return purchase; - } - - async decline(): Promise { - if (this.resolved) return; - this.resolved = true; - await this.context.complete('onDecline'); - } -} - -// ---- LeaderboardSession ---- - -export class LeaderboardSession { - private resolved = false; - - constructor(private readonly context: ScenarioNodeContext) {} - - get isResolved(): boolean { - return this.resolved; - } - - get(key: string, defaultValue: T): T { - return this.context.get(key, defaultValue); - } - - get data(): Record { - return this.context.data; - } - - get node(): ExecutionPlanNode { - return this.context.node; - } - - async end(): Promise { - if (this.resolved) return; - await this.context.complete('onEnd'); - this.resolved = true; - } - - async claim(): Promise { - if (this.resolved) return; - await this.context.complete('onClaim'); - this.resolved = true; - } - - /** @deprecated Use {@link LeaderboardSession.claim}. Removed in the next SDK version. */ - async rewardClaimed(): Promise { - return this.claim(); - } -} - -// ---- QuestSession ---- - -/** - * A scenario quest node. The game reports objective progress; the server - * auto-completes the node (crossing `onComplete`, which grants rewards and - * advances the run) once every objective is satisfied. - */ -export class QuestSession { - constructor(private readonly context: ScenarioNodeContext) {} - - get name(): string { - return this.context.get('name', ''); - } - - get objectives(): Array> { - return this.context.get('objectives', [] as Array>); - } - - get(key: string, defaultValue: T): T { - return this.context.get(key, defaultValue); - } - - get data(): Record { - return this.context.data; - } - - get node(): ExecutionPlanNode { - return this.context.node; - } - - /** - * Reports progress toward an objective. When the reported counter completes - * every objective, the server signals completion and the node crosses - * `onComplete` automatically. - */ - async reportProgress(objectiveId: string, amount = 1): Promise { - await this.context.updateProgress(objectiveId, amount); - } -} - -// ---- BattlePassSession ---- - -/** - * A scenario battlepass node. Exposes the battlepass operations (xp, premium, - * progress) bound to this node's scenario/node/run ids, plus explicit boundary - * crossings (`onLevelUp`, `onPremiumPurchase`) the game drives from its UI. The - * server validates each crossing (e.g. onLevelUp requires the level be reached). - */ -export class BattlePassSession { - constructor( - private readonly context: ScenarioNodeContext, - private readonly battlePass: BattlePassService, - ) {} - - get data(): Record { - return this.context.data; - } - - get node(): ExecutionPlanNode { - return this.context.node; - } - - get(key: string, defaultValue: T): T { - return this.context.get(key, defaultValue); - } - - private ids(): { scenarioId: string; nodeId: string; runId: string } { - return { - scenarioId: this.context.run.scenarioId, - nodeId: this.context.node.id!, - runId: this.context.run.runId, - }; - } - - /** Current xp/level/premium/claimed-tiers for this node. */ - getProgress(): Promise { - const { scenarioId, nodeId } = this.ids(); - return this.battlePass.getProgress(scenarioId, nodeId); - } - - /** Credits XP from a configured source. */ - addXp(source: string, amount: number): Promise { - const { scenarioId, nodeId, runId } = this.ids(); - return this.battlePass.addXp({ scenarioId, nodeId, source, amount, runId }); - } - - /** Claims a tier reward at a reached level. */ - claimReward(level: number, track?: 'free' | 'premium'): Promise { - const { scenarioId, nodeId, runId } = this.ids(); - return this.battlePass.claimReward({ scenarioId, nodeId, level, track, runId }); - } - - /** Purchases premium, then crosses `onPremiumPurchase` on success. */ - async purchasePremium(): Promise { - const { scenarioId, nodeId, runId } = this.ids(); - const response = await this.battlePass.purchasePremium({ - scenarioId, - nodeId, - idempotencyKey: crypto.randomUUID(), - runId, - }); - if (response.success) { - await this.context.complete('onPremiumPurchase'); - } - return response; - } - - /** Crosses `onLevelUp` (server validates the node's level was reached). */ - async levelUp(): Promise { - await this.context.complete('onLevelUp'); - } - - /** Ends the battlepass node via `onComplete`. */ - async end(): Promise { - await this.context.complete('onComplete'); - } -} - -// ---- BattlePassLevelSession ---- - -/** - * A scenario battlepass_level node — a single claimable tier. `claim()` crosses - * `onComplete`, which the server accepts only once the player has reached the - * node's configured level. - */ -export class BattlePassLevelSession { - constructor(private readonly context: ScenarioNodeContext) {} - - get level(): number { - return this.context.get('levelNumber', 0); - } - - get data(): Record { - return this.context.data; - } - - get node(): ExecutionPlanNode { - return this.context.node; - } - - get(key: string, defaultValue: T): T { - return this.context.get(key, defaultValue); - } - - /** Claims this tier; crosses `onComplete` (server checks level reached). */ - async claim(): Promise { - await this.context.complete('onComplete'); - } -} diff --git a/src/scenario/engine/types.ts b/src/scenario/engine/types.ts deleted file mode 100644 index c8b340b..0000000 --- a/src/scenario/engine/types.ts +++ /dev/null @@ -1,63 +0,0 @@ -import type { ExecutionPlan } from '../../generated/common.js'; - -/** - * Active node state within a running scenario plan. - * For wait nodes, this includes the deadline. - */ -export interface ActiveNodeState { - nodeId: string; - waitDeadlineUtc?: string; // ISO 8601 string, set for wait nodes -} - -/** - * A running scenario plan instance — tracks active nodes and completed handles. - */ -export class RuntimeRun { - public readonly activeNodes = new Map(); - public readonly completedHandles = new Set(); - /** - * Number of node transitions currently in flight for this run. - * A transition begins when a node starts completing and ends once its - * successors have been activated. While > 0 the run must not be considered - * complete, even if `activeNodes` is transiently empty (e.g. an - * auto-completing remote_config_override node between two client nodes). - */ - public pendingTransitions = 0; - - constructor( - public readonly runId: string, - public readonly plan: ExecutionPlan, - ) {} -} - -/** Serialization format for a persisted run. */ -export interface PersistedRun { - runId: string; - plan: ExecutionPlan; - activeNodes: ActiveNodeState[]; - completedHandles: string[]; -} - -/** Top-level serialization format for all active runs. */ -export interface PersistedState { - runs: PersistedRun[]; -} - -/** Read-only snapshot of a running plan, surfaced in events. */ -export class PlanRun { - constructor( - public readonly runId: string, - public readonly planId: string, - public readonly scenarioId: string, - public readonly userId: string, - public readonly activeNodeIds: readonly string[], - public readonly plan: ExecutionPlan, - ) {} -} - -/** Failure event payload. */ -export interface ScenarioRunFailedEvent { - run: PlanRun; - nodeId: string; - error: Error; -} diff --git a/test/AuthService.test.ts b/test/AuthService.test.ts index bfc24af..7f590e8 100644 --- a/test/AuthService.test.ts +++ b/test/AuthService.test.ts @@ -30,7 +30,7 @@ describe('AuthService', () => { const response = await client.auth.loginWithDevice({ region: 'us', language: 'en' }); - expect(fetchMock).toHaveBeenCalledTimes(8); + expect(fetchMock).toHaveBeenCalledTimes(9); const [url, init] = fetchMock.mock.calls[0]; expect(url).toContain('/sdk/v1/authorization/device'); const body = JSON.parse(init.body); @@ -46,10 +46,9 @@ describe('AuthService', () => { '/sdk/v1/player/information', '/sdk/v1/stores', '/sdk/v1/catalog', + '/sdk/v1/scenarios/pending', '/sdk/v1/scenarios/trigger', - // The login event invalidates the warmed profile → refetch. '/sdk/v1/player/information', - // Sync engine baseline poll, fired right after login. '/sdk/v1/sync', ]); }); diff --git a/test/PublicApi.test.ts b/test/PublicApi.test.ts index b8060e5..1114676 100644 --- a/test/PublicApi.test.ts +++ b/test/PublicApi.test.ts @@ -64,7 +64,7 @@ describe('public API surface', () => { const client = new RudderClient({ baseUrl: 'https://api.test.rudder.build', projectKey: 'test-project', - runtime: { planStateStore: null, loginEvent: null }, + runtime: { loginEvent: null }, }); expect(client.effects).toBeDefined(); @@ -75,7 +75,7 @@ describe('public API surface', () => { const client = new RudderClient({ baseUrl: 'https://api.test.rudder.build', projectKey: 'test-project', - runtime: { planStateStore: null, loginEvent: null }, + runtime: { loginEvent: null }, }); expect(client.remoteConfig).toBeDefined(); diff --git a/test/RudderClient.test.ts b/test/RudderClient.test.ts index 754c09d..153bb89 100644 --- a/test/RudderClient.test.ts +++ b/test/RudderClient.test.ts @@ -148,7 +148,7 @@ describe('lazy scenario runtime', () => { const client = new RudderClient({ baseUrl: 'https://api.example.com', projectKey: 'proj_123', - runtime: { planStateStore: null, loginEvent: null }, + runtime: { loginEvent: null }, }); const internals = client as unknown as { scenarioRuntime: unknown; @@ -163,7 +163,7 @@ describe('lazy scenario runtime', () => { const client = new RudderClient({ baseUrl: 'https://api.example.com', projectKey: 'proj_123', - runtime: { planStateStore: null }, // default loginEvent: player_login + runtime: {}, }); const notifications: string[] = []; @@ -182,18 +182,18 @@ describe('lazy scenario runtime', () => { } if (path === '/sdk/v1/scenarios/trigger') { return Promise.resolve(new Response(JSON.stringify({ - plans: [{ - planId: 'plan-1', - scenarioId: 'scenario-1', - userId: 'user-1', - startNodeId: 'start', + effects: [{ runId: 'run-1', - nodes: [{ id: 'start', type: 'notification', data: { message: 'Welcome!' } }], - edges: [], - boundaryNodes: [], + scenarioId: 'scenario-1', + nodeId: 'start', + type: 'notification', + data: { message: 'Welcome!' }, }], }), { status: 200 })); } + if (path === '/sdk/v1/scenarios/pending') { + return Promise.resolve(new Response(JSON.stringify({ effects: [] }), { status: 200 })); + } if (path === '/sdk/v1/remote-configs') { return Promise.resolve(new Response(JSON.stringify({ configs: {} }), { status: 200 })); } diff --git a/test/ScenarioService.test.ts b/test/ScenarioService.test.ts index fb6f473..1d6bb6a 100644 --- a/test/ScenarioService.test.ts +++ b/test/ScenarioService.test.ts @@ -2,43 +2,34 @@ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; import { ScenarioService } from '../src/scenario/ScenarioService.js'; import { getScenarioRuntime, RudderClient } from '../src/client/RudderClient.js'; import { createFakeTokenStore } from './helpers/FakeTokenStore.js'; -import type { ExecutionPlan, ExecutionPlanNode, PlanEdge } from '../src/generated/common.js'; -import { NotificationSession, StoreSession, WaitSession, LeaderboardSession } from '../src/scenario/engine/sessions.js'; - -/** Builds a simple execution plan with the given nodes and edges. */ -function makePlan(overrides?: Partial): ExecutionPlan { - return { - planId: 'plan-1', - scenarioId: 'scenario-1', - userId: 'user-1', - startNodeId: 'start', - runId: 'run-1', - nodes: [], - edges: [], - boundaryNodes: [], - context: undefined, - ...overrides, - }; -} - -function makeNode(id: string, type: string, data?: Record): ExecutionPlanNode { - return { id, type, data: data as ExecutionPlanNode['data'] }; -} - -function makeEdge(source: string, sourceHandle: string, target: string): PlanEdge { - return { id: `edge-${source}-${target}`, source, sourceHandle, target }; -} +import { effect } from './helpers/plan.js'; +import { RudderHttpError } from '../src/client/RudderError.js'; +import type { NotificationEffect, StoreOfferEffect, WaitEffect } from '../src/index.js'; +import type { PendingEffect } from '../src/generated/scenarios.js'; async function createClientWithScenario(): Promise<{ client: RudderClient; scenarios: ScenarioService }> { const client = new RudderClient({ baseUrl: 'https://api.test.rudder.build', projectKey: 'test-key', tokenStore: createFakeTokenStore(), - runtime: { planStateStore: null, loginEvent: null }, // Disable IndexedDB for tests + runtime: { loginEvent: null }, }); return { client, scenarios: await getScenarioRuntime(client) }; } +function jsonResponse(body: unknown, status = 200): Response { + return new Response(JSON.stringify(body), { status }); +} + +function stubFetch(handler: (path: string, method: string, body: unknown) => Response | Promise): void { + vi.stubGlobal('fetch', vi.fn().mockImplementation((url: string, init?: RequestInit) => { + const parsed = new URL(url); + const method = (init?.method ?? 'GET').toUpperCase(); + const body = init?.body ? JSON.parse(init.body as string) : undefined; + return Promise.resolve(handler(parsed.pathname, method, body)); + })); +} + describe('ScenarioService', () => { beforeEach(() => { vi.stubGlobal('crypto', { @@ -46,119 +37,100 @@ describe('ScenarioService', () => { }); }); + afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); + }); + describe('send (trigger)', () => { - it('calls POST /sdk/v1/scenarios/trigger and starts plans', async () => { + it('calls POST /sdk/v1/scenarios/trigger and emits effects', async () => { const { client, scenarios } = await createClientWithScenario(); - - const plan = makePlan({ - nodes: [makeNode('start', 'notification')], - }); - - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - const onNotification = vi.fn(); - scenarios.onNotification = onNotification; + client.effects.onNotification(onNotification); + + stubFetch((path, method) => { + if (method === 'POST' && path === '/sdk/v1/scenarios/trigger') { + return jsonResponse({ + effects: [effect('notification', { title: 'Hi', message: 'Welcome' })], + }); + } + return jsonResponse({}); + }); const response = await scenarios.send('level_complete'); - expect(response.plans).toHaveLength(1); + expect(response.effects).toHaveLength(1); expect(scenarios.isRunning).toBe(true); expect(onNotification).toHaveBeenCalledOnce(); - expect(onNotification.mock.calls[0][0]).toBeInstanceOf(NotificationSession); + expect(onNotification.mock.calls[0][0].title).toBe('Hi'); + expect(onNotification.mock.calls[0][0].message).toBe('Welcome'); }); }); - describe('node dispatch', () => { + describe('effect emission', () => { it('notification node fires onNotification', async () => { - const { scenarios } = await createClientWithScenario(); + const { client, scenarios } = await createClientWithScenario(); const onNotif = vi.fn(); - scenarios.onNotification = onNotif; - - const plan = makePlan({ - nodes: [makeNode('start', 'notification')], - }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - + client.effects.onNotification(onNotif); + stubFetch(() => jsonResponse({ effects: [effect('notification')] })); await scenarios.send('test'); expect(onNotif).toHaveBeenCalledOnce(); }); - it('store node fires onStore', async () => { - const { scenarios } = await createClientWithScenario(); + it('store node fires onStoreOffer', async () => { + const { client, scenarios } = await createClientWithScenario(); const onStore = vi.fn(); - scenarios.onStore = onStore; - - const plan = makePlan({ - nodes: [makeNode('start', 'store')], + client.effects.onStoreOffer(onStore); + stubFetch((path) => { + if (path === '/sdk/v1/scenarios/trigger') { + return jsonResponse({ + effects: [effect('store', { storeSlug: 'starter', message: 'Buy!' })], + }); + } + if (path === '/sdk/v1/stores/starter') { + return jsonResponse({ slug: 'starter', name: 'Starter', offers: [{ id: 'pack_1', name: 'Pack' }] }); + } + return jsonResponse({}); }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - await scenarios.send('test'); - expect(onStore).toHaveBeenCalledOnce(); - expect(onStore.mock.calls[0][0]).toBeInstanceOf(StoreSession); + await vi.waitFor(() => expect(onStore).toHaveBeenCalledOnce()); + const offer = onStore.mock.calls[0][0] as StoreOfferEffect; + expect(offer.message).toBe('Buy!'); + expect(offer.store.slug).toBe('starter'); }); it('wait node fires onWait', async () => { - const { scenarios } = await createClientWithScenario(); + const { client, scenarios } = await createClientWithScenario(); const onWait = vi.fn(); - scenarios.onWait = onWait; - - const plan = makePlan({ - nodes: [makeNode('start', 'wait', { duration: 5, unit: 'minutes' })], - }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - + client.effects.onWait(onWait); + const deadline = new Date(Date.now() + 60_000).toISOString(); + stubFetch(() => jsonResponse({ + effects: [effect('wait', {}, { waitDeadline: deadline })], + })); await scenarios.send('test'); expect(onWait).toHaveBeenCalledOnce(); - expect(onWait.mock.calls[0][0]).toBeInstanceOf(WaitSession); + expect(onWait.mock.calls[0][0].deadlineUtc).toBeInstanceOf(Date); }); it('leaderboard node fires onLeaderboard', async () => { - const { scenarios } = await createClientWithScenario(); + const { client, scenarios } = await createClientWithScenario(); const onLb = vi.fn(); - scenarios.onLeaderboard = onLb; - - const plan = makePlan({ - nodes: [makeNode('start', 'leaderboard')], - }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - + client.effects.onLeaderboard(onLb); + stubFetch(() => jsonResponse({ effects: [effect('leaderboard')] })); await scenarios.send('test'); expect(onLb).toHaveBeenCalledOnce(); - expect(onLb.mock.calls[0][0]).toBeInstanceOf(LeaderboardSession); }); - it('quest node dispatches onQuest instead of stalling', async () => { + it('quest node dispatches onQuest', async () => { const { client, scenarios } = await createClientWithScenario(); const onQuest = vi.fn(); client.effects.onQuest(onQuest); - - const plan = makePlan({ - nodes: [ - makeNode('start', 'quest', { - name: 'Daily', - objectives: [{ objectiveId: 'kills', target: 10 }], - }), - ], - }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - + stubFetch(() => jsonResponse({ + effects: [effect('quest', { name: 'Daily', objectives: [{ objectiveId: 'kills', target: 10 }] })], + })); await scenarios.send('test'); expect(onQuest).toHaveBeenCalledOnce(); expect(onQuest.mock.calls[0][0].name).toBe('Daily'); - // Active and waiting for progress — not stalled, not failed. expect(scenarios.isRunning).toBe(true); }); @@ -166,14 +138,7 @@ describe('ScenarioService', () => { const { client, scenarios } = await createClientWithScenario(); const onBp = vi.fn(); client.effects.onBattlePass(onBp); - - const plan = makePlan({ - nodes: [makeNode('start', 'battlepass', { premiumPrice: 100 })], - }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - + stubFetch(() => jsonResponse({ effects: [effect('battlepass', { premiumPrice: 100 })] })); await scenarios.send('test'); expect(onBp).toHaveBeenCalledOnce(); expect(scenarios.isRunning).toBe(true); @@ -183,346 +148,269 @@ describe('ScenarioService', () => { const { client, scenarios } = await createClientWithScenario(); const onLevel = vi.fn(); client.effects.onBattlePassLevel(onLevel); - - const plan = makePlan({ - nodes: [makeNode('start', 'battlepass_level', { levelNumber: 3 })], - }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - + stubFetch(() => jsonResponse({ + effects: [effect('battlepass_level', { levelNumber: 3 })], + })); await scenarios.send('test'); expect(onLevel).toHaveBeenCalledOnce(); expect(onLevel.mock.calls[0][0].level).toBe(3); }); - it('remote_config_override node applies patches internally and auto-completes', async () => { - const { client, scenarios } = await createClientWithScenario(); - - const plan = makePlan({ - nodes: [ - makeNode('start', 'remote_config_override', { - patches: [ - { path: 'difficulty', valueType: 'string', value: 'hard' }, - ], - }), - ], - }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - - await scenarios.send('test'); - expect(client.remoteConfig.get('difficulty', '')).toBe('hard'); - // RemoteConfigOverride auto-completes asynchronously — wait for the promise. - await new Promise((r) => setTimeout(r, 50)); - expect(scenarios.isRunning).toBe(false); - }); - it('unsupported node type fails the run and surfaces onScenarioFailed', async () => { const { client, scenarios } = await createClientWithScenario(); const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); const onFailed = vi.fn(); client.effects.onScenarioFailed(onFailed); - - const plan = makePlan({ - nodes: [makeNode('start', 'unknown_type')], - }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - + stubFetch(() => jsonResponse({ effects: [effect('unknown_type')] })); await scenarios.send('test'); expect(warn).toHaveBeenCalledWith( expect.stringContaining('Unsupported scenario node type'), ); - // The run must NOT stall: it fails and the game is notified. expect(onFailed).toHaveBeenCalledOnce(); expect(scenarios.isRunning).toBe(false); warn.mockRestore(); }); }); - describe('DAG traversal', () => { - it('completing a node follows matching edges', async () => { - const { scenarios } = await createClientWithScenario(); - + describe('callback continuation', () => { + it('completing an effect posts callback and emits the next effect', async () => { + const { client, scenarios } = await createClientWithScenario(); const onNotif = vi.fn(); - scenarios.onNotification = onNotif; + client.effects.onNotification(onNotif); + const first = effect('notification', { message: 'one' }, { nodeId: 'n1' }); + const second = effect('notification', { message: 'two' }, { nodeId: 'n2' }); - // start(notif) → complete("output") → node2(notif) - const plan = makePlan({ - nodes: [ - makeNode('start', 'notification'), - makeNode('node2', 'notification'), - ], - edges: [makeEdge('start', 'output', 'node2')], - }); - - let callCount = 0; - vi.stubGlobal('fetch', vi.fn().mockImplementation(() => { - callCount++; - if (callCount === 1) { - return Promise.resolve(new Response(JSON.stringify({ plans: [plan] }), { status: 200 })); + stubFetch((path, method) => { + if (method === 'POST' && path === '/sdk/v1/scenarios/trigger') { + return jsonResponse({ effects: [first] }); } - return Promise.resolve(new Response(JSON.stringify({}), { status: 200 })); - })); + if (method === 'POST' && path === '/sdk/v1/scenarios/callback') { + return jsonResponse({ effect: second }); + } + return jsonResponse({}); + }); await scenarios.send('test'); - expect(onNotif).toHaveBeenCalledOnce(); // start node dispatched - - // Complete the notification node with handle "output" - const session = onNotif.mock.calls[0][0] as NotificationSession; - await session.complete(); + expect(onNotif).toHaveBeenCalledOnce(); + const session = onNotif.mock.calls[0][0] as NotificationEffect; + await session.done(); expect(onNotif).toHaveBeenCalledTimes(2); - }); + expect(onNotif.mock.calls[1][0].message).toBe('two'); - it('completing a node with already-completed handle is idempotent', async () => { - const { scenarios } = await createClientWithScenario(); - - const onNotif = vi.fn(); - scenarios.onNotification = onNotif; - - const plan = makePlan({ - nodes: [ - makeNode('start', 'notification'), - makeNode('node2', 'notification'), - ], - edges: [makeEdge('start', 'output', 'node2')], + const fetchMock = vi.mocked(fetch); + const callbackCall = fetchMock.mock.calls.find((call) => { + const url = String(call[0]); + return url.includes('/sdk/v1/scenarios/callback'); + }); + expect(JSON.parse(callbackCall![1]!.body as string)).toEqual({ + scenarioId: 'scenario-1', + nodeId: 'n1', + handle: 'output', + runId: 'run-1', }); - - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - - await scenarios.send('test'); - expect(scenarios.activeRuns).toHaveLength(1); - - const session = onNotif.mock.calls[0][0] as NotificationSession; - await session.complete(); - - // node2 should now be active - expect(scenarios.activeRuns).toHaveLength(1); - - // Complete node2 - const session2 = onNotif.mock.calls[1][0] as NotificationSession; - await session2.complete(); - - // Run should be completed - expect(scenarios.isRunning).toBe(false); }); - it('run completion fires onRunCompleted and onCompleted', async () => { - const { scenarios } = await createClientWithScenario(); - + it('completing the last effect emits onScenarioCompleted', async () => { + const { client, scenarios } = await createClientWithScenario(); const onCompleted = vi.fn(); - const onRunCompleted = vi.fn(); - scenarios.onCompleted = onCompleted; - scenarios.onRunCompleted = onRunCompleted; - - const plan = makePlan({ - nodes: [makeNode('start', 'notification')], + const onNotif = vi.fn(); + client.effects.onScenarioCompleted(onCompleted); + client.effects.onNotification(onNotif); + stubFetch((path) => { + if (path === '/sdk/v1/scenarios/trigger') { + return jsonResponse({ effects: [effect('notification')] }); + } + return jsonResponse({}); }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - await scenarios.send('test'); - const session = (scenarios as unknown as { onNotification?: (s: NotificationSession) => void }).onNotification?.( - // Use the mock calls to get the session - vi.mocked(vi.fn()).mock.calls[0]?.[0] as NotificationSession, - ); - // We need to get the session from the mock spy - // Actually let's trigger via respond() - scenarios.respond('output'); - // Fire-and-forget — wait a tick - await new Promise((r) => setTimeout(r, 10)); - - expect(onRunCompleted).toHaveBeenCalled(); - expect(onCompleted).toHaveBeenCalled(); + await (onNotif.mock.calls[0][0] as NotificationEffect).done(); + expect(onCompleted).toHaveBeenCalledOnce(); + expect(onCompleted.mock.calls[0][0]).toEqual({ + runId: 'run-1', + scenarioId: 'scenario-1', + }); + expect(scenarios.isRunning).toBe(false); }); }); - describe('wait nodes', () => { - it('sets a deadline and fires onWait', async () => { - const { scenarios } = await createClientWithScenario(); - const onWait = vi.fn(); - scenarios.onWait = onWait; + describe('pending dedup', () => { + it('does not re-emit an effect with the same runId and nodeId', async () => { + const { client, scenarios } = await createClientWithScenario(); + const onNotification = vi.fn(); + client.effects.onNotification(onNotification); + const pending = effect('notification', { message: 'once' }); + stubFetch(() => jsonResponse({ effects: [pending] })); + await scenarios.send('first-event'); + await scenarios.send('second-event'); + expect(onNotification).toHaveBeenCalledTimes(1); + expect(scenarios.isRunning).toBe(true); + }); + }); - const plan = makePlan({ - nodes: [makeNode('start', 'wait', { duration: 10, unit: 'minutes' })], + describe('waitDeadline timer', () => { + it('polls pending at waitDeadline and emits the next effect', async () => { + vi.useFakeTimers(); + const { client, scenarios } = await createClientWithScenario(); + const onWait = vi.fn(); + const onNotif = vi.fn(); + client.effects.onWait(onWait); + client.effects.onNotification(onNotif); + + const deadline = new Date(Date.now() + 5_000).toISOString(); + const wait = effect('wait', {}, { waitDeadline: deadline, nodeId: 'wait-1' }); + const next = effect('notification', { message: 'after wait' }, { nodeId: 'n2' }); + let pending: PendingEffect[] = [wait]; + + stubFetch((path, method) => { + if (method === 'POST' && path === '/sdk/v1/scenarios/trigger') { + return jsonResponse({ effects: [wait] }); + } + if (method === 'GET' && path === '/sdk/v1/scenarios/pending') { + return jsonResponse({ effects: pending }); + } + return jsonResponse({}); }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); await scenarios.send('test'); expect(onWait).toHaveBeenCalledOnce(); - const session = onWait.mock.calls[0][0] as WaitSession; - expect(session.deadlineUtc).toBeInstanceOf(Date); - // Deadline should be ~10 minutes from now. - const diff = session.deadlineUtc.getTime() - Date.now(); - expect(diff).toBeGreaterThan(9 * 60 * 1000); - expect(diff).toBeLessThan(11 * 60 * 1000); + expect(onNotif).not.toHaveBeenCalled(); + + pending = [next]; + await vi.advanceTimersByTimeAsync(5_000); + expect(onNotif).toHaveBeenCalledOnce(); + expect(onNotif.mock.calls[0][0].message).toBe('after wait'); + const session = onWait.mock.calls[0][0] as WaitEffect; + expect(session.deadlineUtc.toISOString()).toBe(deadline); + }); + }); + + describe('expired-run drop', () => { + it('drops the run on unknown_run and does not retry it', async () => { + const { client, scenarios } = await createClientWithScenario(); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const onFailed = vi.fn(); + const onNotif = vi.fn(); + client.effects.onScenarioFailed(onFailed); + client.effects.onNotification(onNotif); + const pending = effect('notification', { message: 'hello' }); + + stubFetch((path) => { + if (path === '/sdk/v1/scenarios/trigger') { + return jsonResponse({ effects: [pending] }); + } + if (path === '/sdk/v1/scenarios/callback') { + return jsonResponse({ code: 'unknown_run', error: 'gone' }, 410); + } + if (path === '/sdk/v1/scenarios/pending') { + return jsonResponse({ effects: [pending] }); + } + return jsonResponse({}); + }); + + await scenarios.send('test'); + expect(onNotif).toHaveBeenCalledOnce(); + await expect((onNotif.mock.calls[0][0] as NotificationEffect).done()) + .rejects.toBeInstanceOf(RudderHttpError); + expect(onFailed).toHaveBeenCalledOnce(); + expect(onFailed.mock.calls[0][0].runId).toBe('run-1'); + expect(scenarios.isRunning).toBe(false); + + await scenarios.start(); + expect(onNotif).toHaveBeenCalledTimes(1); + warn.mockRestore(); + client.dispose(); }); - it('completes immediately if deadline has already passed', async () => { - const { scenarios } = await createClientWithScenario(); - const onWait = vi.fn(); - const onCompleted = vi.fn(); - scenarios.onWait = onWait; - scenarios.onCompleted = onCompleted; + it('leaves the effect active on a transient callback error', async () => { + const { client, scenarios } = await createClientWithScenario(); + const onFailed = vi.fn(); + const onNotif = vi.fn(); + client.effects.onScenarioFailed(onFailed); + client.effects.onNotification(onNotif); - // Duration of 0 should result in an immediate completion. - const plan = makePlan({ - nodes: [makeNode('start', 'wait', { duration: 0, unit: 'seconds' })], + stubFetch((path) => { + if (path === '/sdk/v1/scenarios/trigger') { + return jsonResponse({ effects: [effect('notification')] }); + } + if (path === '/sdk/v1/scenarios/callback') { + return jsonResponse({ error: 'boom' }, 500); + } + return jsonResponse({}); }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); await scenarios.send('test'); - expect(onWait).toHaveBeenCalledOnce(); - // Wait for the setTimeout(0) to fire. - await new Promise((r) => setTimeout(r, 50)); - expect(onCompleted).toHaveBeenCalled(); + await (onNotif.mock.calls[0][0] as NotificationEffect).done(); + expect(onFailed).not.toHaveBeenCalled(); + expect(scenarios.isRunning).toBe(true); }); }); describe('store session', () => { it('buy() purchases the selected offer and completes with onPurchase', async () => { - const { scenarios } = await createClientWithScenario(); + const { client, scenarios } = await createClientWithScenario(); const onStore = vi.fn(); - scenarios.onStore = onStore; + client.effects.onStoreOffer(onStore); - const plan = makePlan({ - nodes: [makeNode('start', 'store', { storeSlug: 'starter' })], - }); - vi.stubGlobal('fetch', vi.fn().mockImplementation((url: string, init?: RequestInit) => { - const parsed = new URL(url); - if (init?.method === 'POST' && parsed.pathname === '/sdk/v1/scenarios/trigger') { - return Promise.resolve(new Response(JSON.stringify({ plans: [plan] }), { status: 200 })); + stubFetch((path, method) => { + if (method === 'POST' && path === '/sdk/v1/scenarios/trigger') { + return jsonResponse({ effects: [effect('store', { storeSlug: 'starter' })] }); } - if (init?.method === 'GET' && parsed.pathname === '/sdk/v1/stores/starter') { - return Promise.resolve(new Response(JSON.stringify({ + if (method === 'GET' && path === '/sdk/v1/stores/starter') { + return jsonResponse({ name: 'Starter', slug: 'starter', offers: [{ id: 'pack_1', name: 'Starter Pack' }], - }), { status: 200 })); + }); } - if (init?.method === 'POST' && parsed.pathname === '/sdk/v1/stores/starter/offers/pack_1/purchase') { - return Promise.resolve(new Response(JSON.stringify({ success: true, purchaseId: 'purchase-1' }), { status: 200 })); + if (method === 'POST' && path === '/sdk/v1/stores/starter/offers/pack_1/purchase') { + return jsonResponse({ success: true, purchaseId: 'purchase-1' }); } - return Promise.resolve(new Response(JSON.stringify({}), { status: 200 })); - })); + return jsonResponse({}); + }); await scenarios.send('test'); - const session = onStore.mock.calls[0][0] as StoreSession; - - const onCompleted = vi.fn(); - scenarios.onCompleted = onCompleted; - - const store = await session.getStore(); - const purchase = await session.buy(store.offers[0]); + await vi.waitFor(() => expect(onStore).toHaveBeenCalledOnce()); + const session = onStore.mock.calls[0][0] as StoreOfferEffect; + const purchase = await session.buy(session.offers[0]); expect(purchase.success).toBe(true); - expect(session.isResolved).toBe(true); - // Second call should be a no-op. - const duplicate = await session.buy(store.offers[0]); + const duplicate = await session.buy(session.offers[0]); expect(duplicate.success).toBe(false); }); - it('decline() completes with onDecline', async () => { - const { scenarios } = await createClientWithScenario(); + it('dismiss() completes with onDecline', async () => { + const { client, scenarios } = await createClientWithScenario(); const onStore = vi.fn(); - scenarios.onStore = onStore; - - const plan = makePlan({ - nodes: [makeNode('start', 'store')], + client.effects.onStoreOffer(onStore); + stubFetch((path) => { + if (path === '/sdk/v1/scenarios/trigger') { + return jsonResponse({ effects: [effect('store', { storeSlug: 'starter' })] }); + } + if (path === '/sdk/v1/stores/starter') { + return jsonResponse({ slug: 'starter', offers: [{ id: 'pack_1' }] }); + } + return jsonResponse({}); }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - await scenarios.send('test'); - const session = onStore.mock.calls[0][0] as StoreSession; - await session.decline(); - expect(session.isResolved).toBe(true); + await vi.waitFor(() => expect(onStore).toHaveBeenCalledOnce()); + await (onStore.mock.calls[0][0] as StoreOfferEffect).dismiss(); + const fetchMock = vi.mocked(fetch); + const callbackCall = fetchMock.mock.calls.find((call) => + String(call[0]).includes('/sdk/v1/scenarios/callback'), + ); + expect(JSON.parse(callbackCall![1]!.body as string).handle).toBe('onDecline'); }); }); - describe('respond / clear', () => { - it('respond completes the first active node', async () => { - const { scenarios } = await createClientWithScenario(); - const onNotif = vi.fn(); - scenarios.onNotification = onNotif; - - const plan = makePlan({ - nodes: [ - makeNode('start', 'notification'), - makeNode('node2', 'notification'), - ], - edges: [makeEdge('start', 'output', 'node2')], - }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - - await scenarios.send('test'); - - // respond() completes the first active node with the given handle. - scenarios.respond('output'); - - // Wait for async completion. - await new Promise((r) => setTimeout(r, 50)); - - // node2 should now be active (second notification dispatched). - expect(scenarios.isRunning).toBe(true); - }); - + describe('clear', () => { it('clear removes all runs', async () => { - const { scenarios } = await createClientWithScenario(); - - const plan = makePlan({ - nodes: [makeNode('start', 'notification')], - }); - vi.stubGlobal('fetch', vi.fn().mockResolvedValue( - new Response(JSON.stringify({ plans: [plan] }), { status: 200 }), - )); - + const { client, scenarios } = await createClientWithScenario(); + client.effects.onNotification(vi.fn()); + stubFetch(() => jsonResponse({ effects: [effect('notification')] })); await scenarios.send('test'); expect(scenarios.isRunning).toBe(true); - scenarios.clear(); expect(scenarios.isRunning).toBe(false); }); }); - - describe('runId filtering', () => { - it('rejects plans with duplicate runId', async () => { - const { scenarios } = await createClientWithScenario(); - - const onNotification = vi.fn(); - scenarios.onNotification = onNotification; - - const plan = makePlan({ - runId: 'run-1', - nodes: [makeNode('start', 'notification')], - }); - - vi.stubGlobal('fetch', vi.fn().mockImplementation(() => - Promise.resolve(new Response(JSON.stringify({ plans: [plan] }), { status: 200 })), - )); - - // First send creates a run - await scenarios.send('first-event'); - expect(scenarios.activeRuns).toHaveLength(1); - expect(onNotification).toHaveBeenCalledTimes(1); - - // Second send with same runId is rejected by startPlan dedup - await scenarios.send('second-event'); - expect(scenarios.activeRuns).toHaveLength(1); - expect(onNotification).toHaveBeenCalledTimes(1); - }); - }); }); diff --git a/test/e2e-prod/_setup/harness.ts b/test/e2e-prod/_setup/harness.ts index 2e5e593..1b1494a 100644 --- a/test/e2e-prod/_setup/harness.ts +++ b/test/e2e-prod/_setup/harness.ts @@ -9,11 +9,6 @@ import { readArtifact } from './artifact.js'; import type { SeedArtifact } from './types.js'; type RuntimeOptions = { - planStateStore?: { - state: string | null; - load(): Promise; - save(state: string | null): Promise; - } | null; loginEvent?: string | null; }; @@ -52,7 +47,7 @@ export function createInMemoryTokenStore(): TokenStore { /** A client with no auth yet (in-memory tokens, no IndexedDB persistence). */ export function makeProdClient( artifact: SeedArtifact = loadArtifact(), - runtime: RuntimeOptions = { planStateStore: null, loginEvent: null }, + runtime: RuntimeOptions = { loginEvent: null }, ): RudderClient { return new RudderClient({ baseUrl: artifact.baseUrl, diff --git a/test/e2e-prod/scenarios.e2e.test.ts b/test/e2e-prod/scenarios.e2e.test.ts index 173294f..7da4051 100644 --- a/test/e2e-prod/scenarios.e2e.test.ts +++ b/test/e2e-prod/scenarios.e2e.test.ts @@ -16,7 +16,6 @@ describe('e2e-prod: scenarios', () => { let runCompleted = false; const client = makeProdClient(artifact, { - planStateStore: null, loginEvent: artifact.scenario.event, }); @@ -25,9 +24,9 @@ describe('e2e-prod: scenarios', () => { storeOffer = effect; }); const runtime = await getScenarioRuntime(client); - runtime.onRunCompleted = () => { + client.effects.onScenarioCompleted(() => { runCompleted = true; - }; + }); await client.auth.loginWithDevice({ region: 'en', language: 'en' }); // The boundary buy uses the paid offer — fund the wallet so the purchase succeeds. @@ -45,8 +44,8 @@ describe('e2e-prod: scenarios', () => { await vi.waitFor(() => expect(notifications.length).toBe(1), { timeout: 10_000 }); await notifications[0].done(); - // remote_config_override applies, then the store offer surfaces. await vi.waitFor(() => expect(storeOffer).toBeDefined(), { timeout: 10_000 }); + await client.remoteConfig.reload(); expect(client.remoteConfig.get('spawn_rate', 0)).toBeCloseTo( artifact.remoteConfigs.spawnRateOverride, ); diff --git a/test/e2e/boundary.e2e.test.ts b/test/e2e/boundary.e2e.test.ts index 8b005dd..7dc7c7f 100644 --- a/test/e2e/boundary.e2e.test.ts +++ b/test/e2e/boundary.e2e.test.ts @@ -3,44 +3,17 @@ import { getScenarioRuntime, RudderClient } from '../../src/client/RudderClient. import { createFakeTokenStore } from '../helpers/FakeTokenStore.js'; import { createFakeGateway, type FakeGateway } from '../helpers/fakeGateway.js'; import { stubDeterministicUuid } from '../helpers/memoryPlanStore.js'; -import { plan } from '../helpers/plan.js'; +import { effect } from '../helpers/plan.js'; import type { StoreOfferEffect } from '../../src/index.js'; -import type { PlanRun, ScenarioRunFailedEvent } from '../../src/scenario/engine/types.js'; function makeClient(): RudderClient { return new RudderClient({ baseUrl: 'https://api.test.rudder.build', projectKey: 'test-project', tokenStore: createFakeTokenStore(), - runtime: { planStateStore: null }, }); } -/** - * An offer whose `onPurchase` continuation lives server-side (boundary node): - * the client doesn't hold the post-purchase DAG — it must call back so the - * server can validate the transaction and decide what happens next. - * - * `onDecline`, by contrast, is a plain local edge handled entirely on-device. - */ -function offerWithBoundary() { - return plan('offer_flow') - .runId('offer_flow-run') - .node('offer', 'store', { storeSlug: 'starter', message: 'Buy the starter pack!' }) - .node('consolation', 'notification', { message: 'Maybe next time!' }) - .edge('offer', 'onDecline', 'consolation') - .boundary('offer', 'onPurchase') // server owns what comes after a purchase - .build(); -} - -/** What the server returns from the callback — a fresh plan = a new run. */ -function rewardContinuation() { - return plan('reward_flow') - .runId('reward_flow-run') - .node('reward', 'notification', { message: 'Reward granted: 500 gems!' }) - .build(); -} - describe('E2E: boundary nodes (server-side continuation)', () => { let gateway: FakeGateway; let client: RudderClient; @@ -61,99 +34,93 @@ describe('E2E: boundary nodes (server-side continuation)', () => { afterEach(() => vi.unstubAllGlobals()); it('purchase crosses the boundary → callback fires and continues the scenario', async () => { - gateway.onEvent('player_login', offerWithBoundary()); - gateway.onCallback('offer', 'onPurchase', rewardContinuation()); + const offerEffect = effect('store', { storeSlug: 'starter', message: 'Buy the starter pack!' }, { + scenarioId: 'offer_flow', + runId: 'offer_flow-run', + nodeId: 'offer', + }); + const reward = effect('notification', { message: 'Reward granted: 500 gems!' }, { + scenarioId: 'offer_flow', + runId: 'offer_flow-run', + nodeId: 'reward', + }); + gateway.onEvent('player_login', offerEffect); + gateway.onCallback('offer', 'onPurchase', reward); const messages: string[] = []; - const completed: PlanRun[] = []; + const completed: string[] = []; let offer: StoreOfferEffect | undefined; - const runtime = await getScenarioRuntime(client); - client.effects.onStoreOffer((effect) => { offer = effect; }); - client.effects.onNotification((effect) => { messages.push(effect.message); }); - runtime.onRunCompleted = (r) => completed.push(r); + client.effects.onStoreOffer((e) => { offer = e; }); + client.effects.onNotification((e) => { messages.push(e.message); }); + client.effects.onScenarioCompleted((e) => { completed.push(e.scenarioId); }); const token = (await client.auth.loginWithDevice()).accessToken; await vi.waitFor(() => expect(offer).toBeDefined()); await offer!.buy(offer!.offers[0]); - // The callback hit the gateway with the boundary's source node/handle + token. const callback = gateway.recorded.find((r) => r.path === '/sdk/v1/scenarios/callback'); - expect(callback?.body).toEqual({ scenarioId: 'offer_flow', nodeId: 'offer', handle: 'onPurchase', runId: 'offer_flow-run' }); + expect(callback?.body).toEqual({ + scenarioId: 'offer_flow', + nodeId: 'offer', + handle: 'onPurchase', + runId: 'offer_flow-run', + }); expect(callback?.authToken).toBe(token); - // The server-supplied continuation ran (as a new run), not the local edge. expect(messages).toEqual(['Reward granted: 500 gems!']); expect(messages).not.toContain('Maybe next time!'); - - // The original run finished; the continuation ('reward_flow') is now active. - expect(completed.map((r) => r.scenarioId)).toContain('offer_flow'); - expect(runtime.activeRuns.map((r) => r.scenarioId)).toEqual(['reward_flow']); + expect(completed).toEqual([]); }); - it('decline stays local → no callback, local edge is followed', async () => { - gateway.onEvent('player_login', offerWithBoundary()); + it('decline posts callback and continues with the server-supplied next effect', async () => { + const offerEffect = effect('store', { storeSlug: 'starter', message: 'Buy the starter pack!' }, { + scenarioId: 'offer_flow', + runId: 'offer_flow-run', + nodeId: 'offer', + }); + const consolation = effect('notification', { message: 'Maybe next time!' }, { + scenarioId: 'offer_flow', + runId: 'offer_flow-run', + nodeId: 'consolation', + }); + gateway.onEvent('player_login', offerEffect); + gateway.onCallback('offer', 'onDecline', consolation); const messages: string[] = []; let offer: StoreOfferEffect | undefined; - client.effects.onStoreOffer((effect) => { offer = effect; }); - client.effects.onNotification((effect) => { messages.push(effect.message); }); + client.effects.onStoreOffer((e) => { offer = e; }); + client.effects.onNotification((e) => { messages.push(e.message); }); await client.auth.loginWithDevice(); await vi.waitFor(() => expect(offer).toBeDefined()); await offer!.dismiss(); - expect(gateway.recorded.some((r) => r.path === '/sdk/v1/scenarios/callback')).toBe(false); + expect(gateway.recorded.some((r) => r.path === '/sdk/v1/scenarios/callback')).toBe(true); expect(messages).toEqual(['Maybe next time!']); }); - it('a boundary takes precedence over a local edge on the SAME handle', async () => { - // `offer` has BOTH a boundary AND a local edge on onPurchase. - const conflicting = plan('offer_flow') - .runId('offer_flow-conflict') - .node('offer', 'store', { storeSlug: 'starter', message: 'Buy!' }) - .node('local_next', 'notification', { message: 'LOCAL branch' }) - .edge('offer', 'onPurchase', 'local_next') - .boundary('offer', 'onPurchase') - .build(); - gateway.onEvent('player_login', conflicting); - gateway.onCallback('offer', 'onPurchase', rewardContinuation()); - - const messages: string[] = []; - let offer: StoreOfferEffect | undefined; - client.effects.onStoreOffer((effect) => { offer = effect; }); - client.effects.onNotification((effect) => { messages.push(effect.message); }); - - await client.auth.loginWithDevice(); - await vi.waitFor(() => expect(offer).toBeDefined()); - await offer!.buy(offer!.offers[0]); - - // Only the server continuation ran; the local edge was suppressed. - expect(messages).toEqual(['Reward granted: 500 gems!']); - expect(messages).not.toContain('LOCAL branch'); - }); - it('a failing callback does not fail or crash the run', async () => { - gateway.onEvent('player_login', offerWithBoundary()); + const offerEffect = effect('store', { storeSlug: 'starter' }, { + scenarioId: 'offer_flow', + runId: 'offer_flow-run', + nodeId: 'offer', + }); + gateway.onEvent('player_login', offerEffect); gateway.onCallbackError('offer', 'onPurchase', 500); - const failures: ScenarioRunFailedEvent[] = []; - const completed: PlanRun[] = []; + const failures: unknown[] = []; let offer: StoreOfferEffect | undefined; const runtime = await getScenarioRuntime(client); - client.effects.onStoreOffer((effect) => { offer = effect; }); - runtime.onRunFailed = (e) => failures.push(e); - runtime.onRunCompleted = (r) => completed.push(r); + client.effects.onStoreOffer((e) => { offer = e; }); + client.effects.onScenarioFailed((e) => { failures.push(e); }); await client.auth.loginWithDevice(); await vi.waitFor(() => expect(offer).toBeDefined()); - // Must not throw despite the 500 from the callback endpoint. await expect(offer!.buy(offer!.offers[0])).resolves.toMatchObject({ success: true }); - expect(failures).toHaveLength(0); // 500 is transient — no terminal failure - // The run stays pending because the boundary call failed transiently. - // On reconnection the consumer can retry the purchase completion. + expect(failures).toHaveLength(0); expect(runtime.isRunning).toBe(true); }); }); diff --git a/test/e2e/offerJourney.e2e.test.ts b/test/e2e/offerJourney.e2e.test.ts index 5aecaef..38fb045 100644 --- a/test/e2e/offerJourney.e2e.test.ts +++ b/test/e2e/offerJourney.e2e.test.ts @@ -3,7 +3,7 @@ import { RudderClient } from '../../src/client/RudderClient.js'; import { createFakeTokenStore } from '../helpers/FakeTokenStore.js'; import { createFakeGateway, type FakeGateway } from '../helpers/fakeGateway.js'; import { stubDeterministicUuid } from '../helpers/memoryPlanStore.js'; -import { plan } from '../helpers/plan.js'; +import { effect } from '../helpers/plan.js'; import type { NotificationEffect, StoreOfferEffect } from '../../src/index.js'; const MINUTE = 60_000; @@ -13,28 +13,38 @@ function makeClient(): RudderClient { baseUrl: 'https://api.test.rudder.build', projectKey: 'test-project', tokenStore: createFakeTokenStore(), - runtime: { planStateStore: null }, // exercised separately in the persistence test }); } -/** - * The offer scenario both branches share: - * - * login → wait 1m → offer ──onDecline──→ wait 1m → "still available" - * └─onPurchase─→ "thanks for your purchase" - */ -function offerScenario() { - return plan('offer_flow') - .node('wait_intro', 'wait', { duration: 1, unit: 'minutes' }) - .node('offer', 'store', { storeSlug: 'starter', offerId: 'pack_1', message: 'Limited starter pack!' }) - .node('wait_reminder', 'wait', { duration: 1, unit: 'minutes' }) - .node('reminder', 'notification', { message: 'Your offer is still available!' }) - .node('thanks', 'notification', { message: 'Thanks for your purchase!' }) - .edge('wait_intro', 'onComplete', 'offer') - .edge('offer', 'onDecline', 'wait_reminder') - .edge('wait_reminder', 'onComplete', 'reminder') - .edge('offer', 'onPurchase', 'thanks') - .build(); +function offerScenario(now: number) { + const waitIntro = effect('wait', {}, { + scenarioId: 'offer_flow', + runId: 'offer_flow-run', + nodeId: 'wait_intro', + waitDeadline: new Date(now + MINUTE).toISOString(), + }); + const offer = effect('store', { storeSlug: 'starter', offerId: 'pack_1', message: 'Limited starter pack!' }, { + scenarioId: 'offer_flow', + runId: 'offer_flow-run', + nodeId: 'offer', + }); + const waitReminder = effect('wait', {}, { + scenarioId: 'offer_flow', + runId: 'offer_flow-run', + nodeId: 'wait_reminder', + waitDeadline: new Date(now + 2 * MINUTE).toISOString(), + }); + const reminder = effect('notification', { message: 'Your offer is still available!' }, { + scenarioId: 'offer_flow', + runId: 'offer_flow-run', + nodeId: 'reminder', + }); + const thanks = effect('notification', { message: 'Thanks for your purchase!' }, { + scenarioId: 'offer_flow', + runId: 'offer_flow-run', + nodeId: 'thanks', + }); + return { waitIntro, offer, waitReminder, reminder, thanks }; } describe('E2E: player offer journey', () => { @@ -66,43 +76,41 @@ describe('E2E: player offer journey', () => { expect(res.accessToken).toBeTruthy(); expect(client.options.tokenStore.getAccessToken()).toBe(res.accessToken); - // The login request carried the project key + a device id. const login = gateway.recorded.find((r) => r.path === '/sdk/v1/authorization/device'); expect(login?.body).toMatchObject({ key: 'test-project', region: 'eu', language: 'en' }); expect((login?.body as { deviceId?: string }).deviceId).toBeTruthy(); - // The post-login runtime trigger carries the bearer token. const trigger = gateway.recorded.find((r) => r.path === '/sdk/v1/scenarios/trigger'); expect(trigger?.authToken).toBe(res.accessToken); }); it('offer appears after 1 min, player declines, reminder fires 1 min later', async () => { - gateway.onEvent('player_login', offerScenario()); + const now = Date.now(); + const nodes = offerScenario(now); + gateway.onEvent('player_login', nodes.waitIntro); + gateway.onCallback('wait_intro', 'onComplete', nodes.offer); + gateway.onCallback('offer', 'onDecline', nodes.waitReminder); + gateway.onCallback('wait_reminder', 'onComplete', nodes.reminder); const notifications: NotificationEffect[] = []; let offer: StoreOfferEffect | undefined; - client.effects.onStoreOffer((effect) => { offer = effect; }); - client.effects.onNotification((effect) => { notifications.push(effect); }); + client.effects.onStoreOffer((e) => { offer = e; }); + client.effects.onNotification((e) => { notifications.push(e); }); await client.auth.loginWithDevice(); - // Immediately after login the player is just waiting — no offer yet. expect(offer).toBeUndefined(); - // The offer must NOT appear before the full minute has elapsed... await vi.advanceTimersByTimeAsync(MINUTE - 1_000); expect(offer).toBeUndefined(); - // ...and surfaces exactly when the minute is up. await vi.advanceTimersByTimeAsync(1_000); await vi.waitFor(() => expect(offer).toBeDefined()); expect(offer!.message).toBe('Limited starter pack!'); - // Player declines → the DAG moves to the reminder wait, no notification yet. await offer!.dismiss(); expect(notifications).toHaveLength(0); - // The reminder also honours the full minute, not a moment sooner. await vi.advanceTimersByTimeAsync(MINUTE - 1_000); expect(notifications).toHaveLength(0); await vi.advanceTimersByTimeAsync(1_000); @@ -111,38 +119,46 @@ describe('E2E: player offer journey', () => { }); it('offer appears, player buys → "thanks" branch, and the purchase transacts', async () => { - gateway.onEvent('player_login', offerScenario()); + const now = Date.now(); + const nodes = offerScenario(now); + gateway.onEvent('player_login', nodes.waitIntro); + gateway.onCallback('wait_intro', 'onComplete', nodes.offer); + gateway.onCallback('offer', 'onPurchase', nodes.thanks); let offer: StoreOfferEffect | undefined; const notifications: NotificationEffect[] = []; - client.effects.onStoreOffer((effect) => { offer = effect; }); - client.effects.onNotification((effect) => { notifications.push(effect); }); + client.effects.onStoreOffer((e) => { offer = e; }); + client.effects.onNotification((e) => { notifications.push(e); }); const accessToken = (await client.auth.loginWithDevice()).accessToken; await vi.advanceTimersByTimeAsync(MINUTE); await vi.waitFor(() => expect(offer).toBeDefined()); - // The game buys a selected offer; the session advances only after success. const selectedOffer = offer!.offers[0]; const purchase = await offer!.buy(selectedOffer, { idempotencyKey: 'idem-key-123' }); expect(purchase.success).toBe(true); - // The purchase hit the gateway with idempotency key + bearer token. expect(gateway.purchases).toEqual([ { storeSlug: 'starter', offerId: 'pack_1', idempotencyKey: 'idem-key-123', authToken: accessToken }, ]); - // The scenario took the onPurchase branch → "thanks", and the run finished. expect(notifications).toHaveLength(1); expect(notifications[0].message).toBe('Thanks for your purchase!'); }); it('declining does NOT take the purchase branch (handles are isolated)', async () => { - gateway.onEvent('player_login', offerScenario()); + const now = Date.now(); + const nodes = offerScenario(now); + gateway.onEvent('player_login', nodes.waitIntro); + gateway.onCallback('wait_intro', 'onComplete', nodes.offer); + gateway.onCallback('offer', 'onDecline', nodes.waitReminder); + gateway.onCallback('wait_reminder', 'onComplete', nodes.reminder); + gateway.onCallback('offer', 'onPurchase', nodes.thanks); + const messages: string[] = []; let offer: StoreOfferEffect | undefined; - client.effects.onStoreOffer((effect) => { offer = effect; }); - client.effects.onNotification((effect) => { messages.push(effect.message); }); + client.effects.onStoreOffer((e) => { offer = e; }); + client.effects.onNotification((e) => { messages.push(e.message); }); await client.auth.loginWithDevice(); await vi.advanceTimersByTimeAsync(MINUTE); diff --git a/test/e2e/persistence.e2e.test.ts b/test/e2e/persistence.e2e.test.ts index 2644103..3f4b10c 100644 --- a/test/e2e/persistence.e2e.test.ts +++ b/test/e2e/persistence.e2e.test.ts @@ -2,39 +2,44 @@ import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; import { getScenarioRuntime, RudderClient } from '../../src/client/RudderClient.js'; import { createFakeTokenStore } from '../helpers/FakeTokenStore.js'; import { createFakeGateway, type FakeGateway } from '../helpers/fakeGateway.js'; -import { createMemoryPlanStore, stubDeterministicUuid } from '../helpers/memoryPlanStore.js'; -import { plan } from '../helpers/plan.js'; -import type { StoreSession, WaitSession } from '../../src/scenario/engine/sessions.js'; +import { stubDeterministicUuid } from '../helpers/memoryPlanStore.js'; +import { effect } from '../helpers/plan.js'; +import type { StoreOfferEffect, WaitEffect } from '../../src/index.js'; const MINUTE = 60_000; -describe('E2E: scenario state survives a reload', () => { +describe('E2E: pending effects resume after a new login', () => { let gateway: FakeGateway; - // Shared backing store mimics IndexedDB persisting across page loads. - const backing = { value: null as string | null }; - function clientWithSharedStore(): RudderClient { + function makeClient(): RudderClient { return new RudderClient({ baseUrl: 'https://api.test.rudder.build', projectKey: 'test-project', tokenStore: createFakeTokenStore(), - runtime: { planStateStore: createMemoryPlanStore(backing), loginEvent: null }, + runtime: { loginEvent: null }, }); } beforeEach(() => { stubDeterministicUuid(); vi.useFakeTimers(); - backing.value = null; - gateway = createFakeGateway(); - gateway.onEvent( - 'session_start', - plan('offer_flow') - .node('wait_intro', 'wait', { duration: 1, unit: 'minutes' }) - .node('offer', 'store', { message: 'Limited starter pack!' }) - .edge('wait_intro', 'onComplete', 'offer') - .build(), - ); + gateway = createFakeGateway({ + stores: [{ slug: 'starter', name: 'Starter', offers: [{ id: 'pack_1', name: 'Pack' }] }], + }); + const now = Date.now(); + const wait = effect('wait', {}, { + scenarioId: 'offer_flow', + runId: 'offer_flow-run', + nodeId: 'wait_intro', + waitDeadline: new Date(now + MINUTE).toISOString(), + }); + const offer = effect('store', { message: 'Limited starter pack!', storeSlug: 'starter' }, { + scenarioId: 'offer_flow', + runId: 'offer_flow-run', + nodeId: 'offer', + }); + gateway.onEvent('session_start', wait); + gateway.onCallback('wait_intro', 'onComplete', offer); gateway.install(); }); @@ -43,33 +48,26 @@ describe('E2E: scenario state survives a reload', () => { vi.unstubAllGlobals(); }); - it('a wait started before reload resumes and fires the offer after reload', async () => { - // --- Session 1: trigger the scenario, then "close the tab" mid-wait --- - const client1 = clientWithSharedStore(); + it('a wait started before reload resumes from GET pending after login', async () => { + const client1 = makeClient(); await client1.auth.loginWithDevice(); const runtime1 = await getScenarioRuntime(client1); await runtime1.send('session_start'); expect(runtime1.isRunning).toBe(true); - expect(backing.value).toBeTruthy(); // run was persisted - // --- Session 2: fresh client, same persisted state (page reload) --- - const client2 = clientWithSharedStore(); - const runtime2 = await getScenarioRuntime(client2); - const resumedWaits: WaitSession[] = []; - let offer: StoreSession | undefined; - runtime2.onWait = (s) => resumedWaits.push(s); - runtime2.onStore = (s) => { offer = s; }; + const client2 = makeClient(); + const resumedWaits: WaitEffect[] = []; + let offer: StoreOfferEffect | undefined; + client2.effects.onWait((s) => { resumedWaits.push(s); }); + client2.effects.onStoreOffer((s) => { offer = s; }); await client2.auth.loginWithDevice(); - // The wait node was rehydrated and re-dispatched. - expect(runtime2.isRunning).toBe(true); expect(resumedWaits).toHaveLength(1); expect(offer).toBeUndefined(); - // The remaining wait time still elapses → the offer surfaces on the new client. await vi.advanceTimersByTimeAsync(MINUTE); - expect(offer).toBeDefined(); - expect(offer!.get('message', '')).toBe('Limited starter pack!'); + await vi.waitFor(() => expect(offer).toBeDefined()); + expect(offer!.message).toBe('Limited starter pack!'); }); }); diff --git a/test/e2e/remoteConfig.e2e.test.ts b/test/e2e/remoteConfig.e2e.test.ts index f664845..f4a0657 100644 --- a/test/e2e/remoteConfig.e2e.test.ts +++ b/test/e2e/remoteConfig.e2e.test.ts @@ -3,7 +3,6 @@ import { RudderClient } from '../../src/client/RudderClient.js'; import { createFakeTokenStore } from '../helpers/FakeTokenStore.js'; import { createFakeGateway, type FakeGateway } from '../helpers/fakeGateway.js'; import { stubDeterministicUuid } from '../helpers/memoryPlanStore.js'; -import { plan } from '../helpers/plan.js'; import type { RemoteConfig } from '../../src/generated/remote-config.js'; interface GameRemoteConfig extends Record { @@ -25,7 +24,7 @@ function makeClient(): RudderClient { baseUrl: 'https://api.test.rudder.build', projectKey: 'test-project', tokenStore: createFakeTokenStore(), - runtime: { planStateStore: null, loginEvent: null }, + runtime: { loginEvent: null }, }); } @@ -76,26 +75,4 @@ describe('E2E: remote config', () => { expect(client.remoteConfig.status).toBe('idle'); expect(client.remoteConfig.get('max_energy', 99)).toBe(99); }); - - it('a remote_config_override scenario node patches the live cache', async () => { - client = new RudderClient({ - baseUrl: 'https://api.test.rudder.build', - projectKey: 'test-project', - tokenStore: createFakeTokenStore(), - runtime: { planStateStore: null }, - }); - gateway.onEvent( - 'player_login', - plan('boost') - .node('override', 'remote_config_override', { - patches: [{ path: 'drop_rate', valueType: 'float', value: '0.9' }], - }) - .build(), - ); - - await client.auth.loginWithDevice(); - - // The override is reflected immediately, without a reload. - expect(client.remoteConfig.get('drop_rate', 0)).toBeCloseTo(0.9); - }); }); diff --git a/test/e2e/storage.e2e.test.ts b/test/e2e/storage.e2e.test.ts index 3ceb6d3..dca74f9 100644 --- a/test/e2e/storage.e2e.test.ts +++ b/test/e2e/storage.e2e.test.ts @@ -8,7 +8,7 @@ function makeClient(): RudderClient { baseUrl: 'https://api.test.rudder.build', projectKey: 'test-project', tokenStore: createFakeTokenStore(), - runtime: { planStateStore: null, loginEvent: null }, + runtime: { loginEvent: null }, }); } diff --git a/test/helpers/fakeGateway.ts b/test/helpers/fakeGateway.ts index 08fa685..7079587 100644 --- a/test/helpers/fakeGateway.ts +++ b/test/helpers/fakeGateway.ts @@ -1,5 +1,5 @@ import { vi } from 'vitest'; -import type { ExecutionPlan } from '../../src/generated/common.js'; +import type { PendingEffect } from '../../src/generated/scenarios.js'; import type { RemoteConfig } from '../../src/generated/remote-config.js'; import type { Store } from '../../src/generated/stores.js'; import type { StorageItem } from '../../src/generated/storage.js'; @@ -8,18 +8,6 @@ import type { ListQuestsResponse, } from '../../src/generated/quests.js'; -/** - * In-process fake of the LiveOps API gateway. - * - * Stubs the global `fetch` and routes requests the same way the real gateway - * does — so tests exercise the *real* SDK transport, auth-header injection, - * scenario engine and DAG traversal, just against deterministic in-memory state - * instead of a live backend. - * - * Seed state up front (scenarios per event, remote configs, stores), then drive - * the SDK through a journey and assert on `recorded` requests / server state. - */ - export interface RecordedRequest { method: string; path: string; @@ -29,36 +17,28 @@ export interface RecordedRequest { } export interface FakeGatewayState { - /** Plans returned by POST /scenarios/trigger, keyed by event name. */ - scenarios: Map; - /** Plans returned by POST /scenarios/callback, keyed by `nodeId:handle`. */ - callbacks: Map; - /** HTTP status to fail a callback with, keyed by `nodeId:handle`. */ - callbackErrors: Map; - /** Optional response for POST /scenarios/counter. */ - counterPlan?: ExecutionPlan; + scenarios: Map; + callbacks: Map; + callbackErrors: Map; + pendingEffects: PendingEffect[]; + counterCompleted: boolean; + counterEffect?: PendingEffect; remoteConfigs: Record; stores: Store[]; quests: ListQuestsResponse; questClaims: Map; - /** Player KV storage, keyed by item id. */ storage: Map; } export interface FakeGateway { state: FakeGatewayState; recorded: RecordedRequest[]; - /** Purchases received, in order. */ purchases: Array<{ storeSlug: string; offerId: string; idempotencyKey: string; authToken: string | null }>; - /** Quest claim requests received, in order. */ questClaims: Array<{ questId: string; authToken: string | null }>; install(): void; - /** Convenience: register the plans returned for a trigger event. */ - onEvent(event: string, ...plans: ExecutionPlan[]): void; - /** Convenience: register the plan returned when a boundary handle calls back. */ - onCallback(nodeId: string, handle: string, plan: ExecutionPlan): void; - /** Convenience: make a boundary callback fail with an HTTP status (default 500). */ - onCallbackError(nodeId: string, handle: string, status?: number): void; + onEvent(event: string, ...effects: PendingEffect[]): void; + onCallback(nodeId: string, handle: string, next: PendingEffect | null): void; + onCallbackError(nodeId: string, handle: string, status?: number, code?: string): void; } function json(body: unknown, status = 200): Response { @@ -70,12 +50,19 @@ function json(body: unknown, status = 200): Response { const noContent = (): Response => new Response(null, { status: 204 }); +function replaceRun(pending: PendingEffect[], next: PendingEffect[]): PendingEffect[] { + const runIds = new Set(next.map((effect) => effect.runId)); + return [...pending.filter((effect) => !runIds.has(effect.runId)), ...next]; +} + export function createFakeGateway(seed?: Partial): FakeGateway { const state: FakeGatewayState = { scenarios: seed?.scenarios ?? new Map(), callbacks: seed?.callbacks ?? new Map(), callbackErrors: seed?.callbackErrors ?? new Map(), - counterPlan: seed?.counterPlan, + pendingEffects: seed?.pendingEffects ?? [], + counterCompleted: seed?.counterCompleted ?? false, + counterEffect: seed?.counterEffect, remoteConfigs: seed?.remoteConfigs ?? {}, stores: seed?.stores ?? [], quests: seed?.quests ?? { quests: [] }, @@ -87,10 +74,32 @@ export function createFakeGateway(seed?: Partial): FakeGateway const purchases: FakeGateway['purchases'] = []; const questClaims: FakeGateway['questClaims'] = []; + function advanceExpiredWaits(): PendingEffect[] { + const now = Date.now(); + const next: PendingEffect[] = []; + for (const effect of state.pendingEffects) { + if ( + effect.type === 'wait' && + effect.waitDeadline && + Date.parse(effect.waitDeadline) <= now + ) { + const continued = state.callbacks.get(`${effect.nodeId}:onComplete`); + if (continued === undefined) { + next.push(effect); + } else if (continued !== null) { + next.push(continued); + } + } else { + next.push(effect); + } + } + state.pendingEffects = next; + return next; + } + async function handle(req: RecordedRequest): Promise { const { method, path, query, body } = req; - // --- Auth --- if (method === 'POST' && path === '/sdk/v1/authorization/device') { const b = body as { deviceId?: string }; return json({ @@ -99,39 +108,56 @@ export function createFakeGateway(seed?: Partial): FakeGateway }); } - // --- Player --- if (method === 'GET' && path === '/sdk/v1/player/information') { return json({ player: null, wallets: [] }); } - // --- Sync (baseline revision poll after login) --- if (method === 'GET' && path === '/sdk/v1/sync') { return json({}); } - // --- Catalog --- if (method === 'GET' && path === '/sdk/v1/catalog') { return json({ items: [] }); } - // --- Scenarios --- if (method === 'POST' && path === '/sdk/v1/scenarios/trigger') { const event = (body as { event?: string }).event ?? ''; - return json({ plans: state.scenarios.get(event) ?? [] }); + const effects = state.scenarios.get(event) ?? []; + state.pendingEffects = replaceRun(state.pendingEffects, effects); + return json({ effects }); + } + if (method === 'GET' && path === '/sdk/v1/scenarios/pending') { + return json({ effects: advanceExpiredWaits() }); } if (method === 'POST' && path === '/sdk/v1/scenarios/callback') { - const b = body as { nodeId?: string; handle?: string }; + const b = body as { nodeId?: string; handle?: string; runId?: string }; const key = `${b.nodeId}:${b.handle}`; - const errStatus = state.callbackErrors.get(key); - if (errStatus) return json({ error: 'callback failed' }, errStatus); - const plan = state.callbacks.get(key); - return json(plan ? { plan } : {}); + const err = state.callbackErrors.get(key); + if (err) { + return json({ code: err.code ?? 'callback failed', error: 'callback failed' }, err.status); + } + const next = state.callbacks.get(key); + state.pendingEffects = state.pendingEffects.filter( + (effect) => !(effect.nodeId === b.nodeId && effect.runId === (b.runId ?? effect.runId)), + ); + if (next) { + state.pendingEffects = replaceRun(state.pendingEffects, [next]); + return json({ effect: next }); + } + return json({}); } if (method === 'POST' && path === '/sdk/v1/scenarios/counter') { - return json(state.counterPlan ? { plan: state.counterPlan, completed: false } : { completed: false }); + if (state.counterCompleted && state.counterEffect) { + state.pendingEffects = replaceRun(state.pendingEffects, [state.counterEffect]); + } else if (state.counterCompleted) { + const b = body as { nodeId?: string; runId?: string }; + state.pendingEffects = state.pendingEffects.filter( + (effect) => !(effect.nodeId === b.nodeId && effect.runId === (b.runId ?? effect.runId)), + ); + } + return json({ completed: state.counterCompleted, effect: state.counterEffect }); } - // --- Remote config --- if (method === 'GET' && path === '/sdk/v1/remote-configs') { return json({ configs: state.remoteConfigs }); } @@ -141,7 +167,6 @@ export function createFakeGateway(seed?: Partial): FakeGateway return cfg ? json(cfg) : json({ error: 'not found' }, 404); } - // --- Stores --- if (method === 'GET' && path === '/sdk/v1/stores') { return json({ stores: state.stores, total: state.stores.length }); } @@ -163,7 +188,6 @@ export function createFakeGateway(seed?: Partial): FakeGateway return store ? json(store) : json({ error: 'not found' }, 404); } - // --- Quests --- if (method === 'POST' && path === '/sdk/v1/quests/list') { return json(state.quests); } @@ -174,7 +198,6 @@ export function createFakeGateway(seed?: Partial): FakeGateway return json(state.questClaims.get(questId) ?? { success: false, error: 'not found' }); } - // --- Storage --- if (path === '/sdk/v1/storage') { if (method === 'GET') { const typeFilter = query.get('types'); @@ -228,14 +251,14 @@ export function createFakeGateway(seed?: Partial): FakeGateway purchases, questClaims, install, - onEvent(event, ...plans) { - state.scenarios.set(event, plans); + onEvent(event, ...effects) { + state.scenarios.set(event, effects); }, - onCallback(nodeId, handle, plan) { - state.callbacks.set(`${nodeId}:${handle}`, plan); + onCallback(nodeId, handle, next) { + state.callbacks.set(`${nodeId}:${handle}`, next); }, - onCallbackError(nodeId, handle, status = 500) { - state.callbackErrors.set(`${nodeId}:${handle}`, status); + onCallbackError(nodeId, handle, status = 500, code) { + state.callbackErrors.set(`${nodeId}:${handle}`, { status, code }); }, }; } diff --git a/test/helpers/memoryPlanStore.ts b/test/helpers/memoryPlanStore.ts index 78f2654..4d78366 100644 --- a/test/helpers/memoryPlanStore.ts +++ b/test/helpers/memoryPlanStore.ts @@ -1,29 +1,3 @@ -import type { PlanStateStore } from '../../src/scenario/engine/IndexedDbPlanStore.js'; - -/** - * In-memory PlanStateStore that survives across RudderClient instances — lets a - * test simulate a page reload: drive scenario A on one client, construct a fresh - * client sharing the same `backing`, call `scenarios.restore()`, and assert the - * run resumed mid-DAG. - */ -export function createMemoryPlanStore(backing: { value: string | null } = { value: null }): PlanStateStore { - return { - get state() { - return backing.value; - }, - set state(v: string | null) { - backing.value = v; - }, - async load() { - /* state already in `backing` */ - }, - async save(s: string | null) { - backing.value = s; - }, - }; -} - -/** Deterministic, collision-free crypto.randomUUID() stub for scenario run IDs. */ export function stubDeterministicUuid(): void { let n = 0; const existing = (globalThis as { crypto?: Crypto }).crypto ?? ({} as Crypto); diff --git a/test/helpers/plan.ts b/test/helpers/plan.ts index 566051d..00b86ff 100644 --- a/test/helpers/plan.ts +++ b/test/helpers/plan.ts @@ -1,72 +1,16 @@ -import type { ExecutionPlan, ExecutionPlanNode, PlanEdge, BoundaryNode } from '../../src/generated/common.js'; +import type { PendingEffect } from '../../src/generated/scenarios.js'; -/** - * Small fluent builder for ExecutionPlans, so scenario journeys read like the - * DAGs they model rather than walls of object literals. - * - * plan('offer_flow') - * .node('wait1', 'wait', { duration: 1, unit: 'minutes' }) - * .node('offer', 'store', { storeSlug: 'starter', offerId: 'pack_1' }) - * .edge('wait1', 'onComplete', 'offer') - * .build(); - * - * The first node added becomes the start node unless `.start(id)` is called. - */ -export class PlanBuilder { - private readonly nodes: ExecutionPlanNode[] = []; - private readonly edges: PlanEdge[] = []; - private readonly boundaryNodes: BoundaryNode[] = []; - private startNodeId?: string; - private runIdValue?: string; - - constructor( - private readonly scenarioId: string, - private readonly opts: { planId?: string; userId?: string } = {}, - ) {} - - /** Sets a specific run ID (default: auto-derived from scenarioId). */ - runId(id: string): this { - this.runIdValue = id; - return this; - } - - node(id: string, type: string, data?: Record): this { - this.nodes.push({ id, type, data: data as ExecutionPlanNode['data'] }); - if (!this.startNodeId) this.startNodeId = id; - return this; - } - - edge(source: string, sourceHandle: string, target: string): this { - this.edges.push({ id: `e-${source}-${sourceHandle}-${target}`, source, sourceHandle, target }); - return this; - } - - /** Registers a server-side boundary node (handle that calls back to the gateway). */ - boundary(sourceNodeId: string, sourceHandle: string, nodeId = `b-${sourceNodeId}-${sourceHandle}`): this { - this.boundaryNodes.push({ sourceNodeId, sourceHandle, nodeId }); - return this; - } - - start(id: string): this { - this.startNodeId = id; - return this; - } - - build(): ExecutionPlan { - return { - planId: this.opts.planId ?? `${this.scenarioId}-plan`, - scenarioId: this.scenarioId, - userId: this.opts.userId ?? 'player-1', - startNodeId: this.startNodeId, - runId: this.runIdValue ?? `${this.scenarioId}-run`, - nodes: this.nodes, - edges: this.edges, - boundaryNodes: this.boundaryNodes, - context: undefined, - }; - } -} - -export function plan(scenarioId: string, opts?: { planId?: string; userId?: string }): PlanBuilder { - return new PlanBuilder(scenarioId, opts); +export function effect( + type: string, + data: Record = {}, + overrides: Partial = {}, +): PendingEffect { + return { + runId: 'run-1', + scenarioId: 'scenario-1', + nodeId: `${type}-1`, + type, + data, + ...overrides, + }; } diff --git a/tsup.config.ts b/tsup.config.ts index ad4cc80..920e8ba 100644 --- a/tsup.config.ts +++ b/tsup.config.ts @@ -6,7 +6,7 @@ export default defineConfig({ target: 'es2022', dts: true, clean: true, - // Code-split the lazily-imported scenario engine out of the initial chunk. + // Code-split the lazily-imported scenario client out of the initial chunk. splitting: true, treeshake: true, platform: 'browser',