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
This commit is contained in:
edmand46
@@ -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);
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user