Initial commit

src/scenario/engine/DagWalker.ts

@@ -0,0 +1,74 @@
+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<PlanEdge> {
+  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<BoundaryNode> {
+  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
+  }
+}

src/scenario/engine/IndexedDbPlanStore.ts

@@ -0,0 +1,94 @@
+/**
+ * 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<void>;
+
+  /** Persists the current state to IndexedDB. */
+  save(state: string | null): Promise<void>;
+}
+
+const DB_NAME = 'RudderPlanState';
+const STORE_NAME = 'state';
+const KEY = 'active_runs';
+const DB_VERSION = 1;
+
+export function createIndexedDbPlanStateStore(): PlanStateStore {
+  let dbPromise: Promise<IDBDatabase> | null = null;
+  let cachedState: string | null = null;
+
+  function getDb(): Promise<IDBDatabase> {
+    if (!dbPromise) {
+      dbPromise = new Promise<IDBDatabase>((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<void> {
+      const db = await getDb();
+      return new Promise<void>((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<void> {
+      cachedState = state;
+      const db = await getDb();
+      return new Promise<void>((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);
+      });
+    },
+  };
+}

src/scenario/engine/sessions.ts

@@ -0,0 +1,342 @@
+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<ShopHandle>;
+}
+
+// ---- 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<void>,
+    private readonly onProgress: (runId: string, nodeId: string, counterKey: string, amount: number) => Promise<void>,
+  ) {}
+
+  /** Extracts a typed value from node data. */
+  get<T>(key: string, defaultValue: T): T {
+    const data = this.node.data as Record<string, unknown> | 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<string, unknown> {
+    return (this.node.data as Record<string, unknown>) ?? {};
+  }
+
+  /** Completes the current node with the given output handle. */
+  async complete(handle: string): Promise<void> {
+    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<void> {
+    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<T>(key: string, defaultValue: T): T {
+    return this.context.get(key, defaultValue);
+  }
+
+  get data(): Record<string, unknown> {
+    return this.context.data;
+  }
+
+  get node(): ExecutionPlanNode {
+    return this.context.node;
+  }
+
+  async complete(): Promise<void> {
+    await this.context.complete('output');
+  }
+}
+
+// ---- WaitSession ----
+
+export class WaitSession {
+  constructor(
+    private readonly context: ScenarioNodeContext,
+    public readonly deadlineUtc: Date,
+  ) {}
+
+  get<T>(key: string, defaultValue: T): T {
+    return this.context.get(key, defaultValue);
+  }
+
+  get data(): Record<string, unknown> {
+    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<T>(key: string, defaultValue: T): T {
+    return this.context.get(key, defaultValue);
+  }
+
+  get data(): Record<string, unknown> {
+    return this.context.data;
+  }
+
+  get node(): ExecutionPlanNode {
+    return this.context.node;
+  }
+
+  async getStore(): Promise<ShopHandle> {
+    return this.stores.getBySlug(this.get('storeSlug', ''));
+  }
+
+  async buy(
+    offer: OfferHandle,
+    options?: BuyOptions,
+  ): Promise<PurchaseOfferResponse> {
+    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<void> {
+    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<T>(key: string, defaultValue: T): T {
+    return this.context.get(key, defaultValue);
+  }
+
+  get data(): Record<string, unknown> {
+    return this.context.data;
+  }
+
+  get node(): ExecutionPlanNode {
+    return this.context.node;
+  }
+
+  async end(): Promise<void> {
+    if (this.resolved) return;
+    this.resolved = true;
+    await this.context.complete('onEnd');
+  }
+
+  async rewardClaimed(): Promise<void> {
+    if (this.resolved) return;
+    this.resolved = true;
+    await this.context.complete('onRewardClaimed');
+  }
+}
+
+// ---- 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<Record<string, unknown>> {
+    return this.context.get('objectives', [] as Array<Record<string, unknown>>);
+  }
+
+  get<T>(key: string, defaultValue: T): T {
+    return this.context.get(key, defaultValue);
+  }
+
+  get data(): Record<string, unknown> {
+    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<void> {
+    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<string, unknown> {
+    return this.context.data;
+  }
+
+  get node(): ExecutionPlanNode {
+    return this.context.node;
+  }
+
+  get<T>(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<GetBattlePassProgressResponse> {
+    const { scenarioId, nodeId } = this.ids();
+    return this.battlePass.getProgress(scenarioId, nodeId);
+  }
+
+  /** Credits XP from a configured source. */
+  addXp(source: string, amount: number): Promise<AddBattlePassXpResponse> {
+    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?: string): Promise<ClaimBattlePassRewardResponse> {
+    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<PurchaseBattlePassPremiumResponse> {
+    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<void> {
+    await this.context.complete('onLevelUp');
+  }
+
+  /** Ends the battlepass node via `onComplete`. */
+  async end(): Promise<void> {
+    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('level', 0);
+  }
+
+  get data(): Record<string, unknown> {
+    return this.context.data;
+  }
+
+  get node(): ExecutionPlanNode {
+    return this.context.node;
+  }
+
+  get<T>(key: string, defaultValue: T): T {
+    return this.context.get(key, defaultValue);
+  }
+
+  /** Claims this tier; crosses `onComplete` (server checks level reached). */
+  async claim(): Promise<void> {
+    await this.context.complete('onComplete');
+  }
+}

src/scenario/engine/types.ts

@@ -0,0 +1,63 @@
+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<string, ActiveNodeState>();
+  public readonly completedHandles = new Set<string>();
+  /**
+   * 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;
+}