Hydra State Machine

packages/mesh-hydra/package.json

@@ -24,18 +24,26 @@
     "format": "prettier --check . --ignore-path ../../.gitignore",
     "lint": "eslint",
     "pack": "npm pack --pack-destination=./dist",
-    "test": "jest"
+    "test": "jest --testPathIgnorePatterns=tests/integration",
+    "test:integration": "jest tests/integration/hydra-machine-integration.test.ts --verbose",
+    "test:simple": "jest tests/integration/hydra-machine-simple-integration.test.ts --verbose",
+    "test:all-integration": "jest tests/integration/ --verbose"
   },
   "dependencies": {
     "@meshsdk/common": "1.9.0-beta.71",
     "@meshsdk/core-cst": "1.9.0-beta.71",
-    "axios": "^1.7.2"
+    "axios": "^1.7.2",
+    "xstate": "^5.20.1"
   },
   "devDependencies": {
     "@meshsdk/configs": "*",
     "@swc/core": "^1.10.7",
+    "@types/node-fetch": "^2.6.13",
+    "@types/ws": "^8.18.1",
     "eslint": "^8.57.0",
+    "node-fetch": "^3.3.2",
     "tsup": "^8.0.2",
-    "typescript": "^5.3.3"
+    "typescript": "^5.3.3",
+    "ws": "^8.18.3"
   }
 }

packages/mesh-hydra/src/examples/example.ts

@@ -0,0 +1,26 @@
+import { HydraController } from "../hydra-controller";
+
+const controller = new HydraController();
+
+controller.on("*", (snapshot) => {
+  console.log("New state:", snapshot.value);
+});
+
+// Wait for specific compound state like Connected.Idle
+controller.on("Connected.Idle", () => {
+  console.log("Hydra is now connected and idle, sending Init...");
+  controller.init();
+});
+
+// Connect to the Hydra node
+controller.connect({
+  baseURL: "http://localhost:4001",
+  address: "addr_test1vp5cxztpc6hep9ds7fjgmle3l225tk8ske3rmwr9adu0m6qchmx5z",
+  snapshot: true,
+  history: true,
+});
+
+controller.waitFor("Connected.Initializing.ReadyToCommit").then(() => {
+  console.log("Ready to commit, sending commit data...");
+  controller.commit({});
+});

packages/mesh-hydra/src/examples/usage-example.ts

@@ -0,0 +1,37 @@
+import { createActor } from "xstate";
+import { createHydraMachine } from "../state-management/hydra-machine-refactored";
+import { HTTPClient } from "../utils";
+
+// Example 1: Basic usage (same as before)
+const basicExample = () => {
+  const machine = createHydraMachine();
+  const actor = createActor(machine);
+
+  actor.start();
+  actor.send({ type: "Connect", baseURL: "http://localhost:4001" });
+
+  return actor;
+};
+
+// Example 2: With custom configuration
+const customExample = () => {
+  const machine = createHydraMachine({
+    webSocketFactory: {
+      create: (url) => new WebSocket(url)
+    },
+    httpClientFactory: {
+      create: (baseURL) => {
+        // Could add logging, interceptors, etc.
+        const { HTTPClient } = require("../utils");
+        return new HTTPClient(baseURL);
+      }
+    }
+  });
+
+  const actor = createActor(machine);
+  actor.start();
+
+  return actor;
+};
+
+export { basicExample, customExample };

packages/mesh-hydra/src/hydra-controller.ts

@@ -0,0 +1,199 @@
+import { ActorRefFrom, createActor, StateValue } from "xstate";
+import { machine } from "./state-management/hydra-machine";
+import { Emitter } from "./utils/emitter";
+import { HTTPClient } from "./utils";
+
+type ConnectOptions = {
+  baseURL: string;
+  address?: string;
+  snapshot?: boolean;
+  history?: boolean;
+};
+
+type HydraStateName =
+  | "*"
+  | "Disconnected"
+  | "Connecting"
+  | "Connected.Idle"
+  | "Connected.Initializing.ReadyToCommit"
+  | "Connected.Open"
+  | "Connected.Closed"
+  | "Connected.Final";
+
+type Snapshot = ReturnType<ActorRefFrom<typeof machine>["getSnapshot"]>;
+
+type Events = {
+  "*": (snapshot: Snapshot) => void;
+} & {
+  [K in HydraStateName]: (snapshot: Snapshot) => void;
+};
+
+export class HydraController {
+  private actor = createActor(machine);
+  private emitter = new Emitter<Events>();
+  private _currentSnapshot?: Snapshot;
+  private httpClient?: HTTPClient;
+
+  constructor() {
+    this.actor.subscribe({
+      next: (snapshot) => this.handleState(snapshot),
+      error: (err) => console.error("Hydra error:", err),
+    });
+    this.actor.start();
+  }
+
+  /** Connect to the Hydra head */
+  connect(options: ConnectOptions) {
+    this.actor.send({ type: "Connect", ...options });
+    this.httpClient = new HTTPClient(options.baseURL);
+  }
+
+  /** Protocol commands */
+  init() {
+    this.actor.send({ type: "Init" });
+  }
+  commit(data: unknown = {}) {
+    this.actor.send({ type: "Commit", data });
+  }
+  newTx(tx: string) {
+    this.actor.send({ type: "NewTx", tx });
+  }
+  recover(txHash: string) {
+    this.actor.send({ type: "Recover", txHash });
+  }
+  decommit(tx: string) {
+    this.actor.send({ type: "Decommit", tx });
+  }
+  close() {
+    this.actor.send({ type: "Close" });
+  }
+  contest() {
+    this.actor.send({ type: "Contest" });
+  }
+  fanout() {
+    this.actor.send({ type: "Fanout" });
+  }
+  sideLoadSnapshot(snapshot: unknown) {
+    this.actor.send({ type: "SideLoadSnapshot", snapshot });
+  }
+
+  /** HTTP API methods */
+  async getHeadState() {
+    if (!this.httpClient) throw new Error("Not connected");
+    return await this.httpClient.get("/head");
+  }
+
+  async getPendingDeposits() {
+    if (!this.httpClient) throw new Error("Not connected");
+    return await this.httpClient.get("/commits");
+  }
+
+  async recoverDeposit(txId: string) {
+    if (!this.httpClient) throw new Error("Not connected");
+    return await this.httpClient.delete(`/commits/${txId}`);
+  }
+
+  async getLastSeenSnapshot() {
+    if (!this.httpClient) throw new Error("Not connected");
+    return await this.httpClient.get("/snapshot/last-seen");
+  }
+
+  async getConfirmedUTxO() {
+    if (!this.httpClient) throw new Error("Not connected");
+    return await this.httpClient.get("/snapshot/utxo");
+  }
+
+  async getConfirmedSnapshot() {
+    if (!this.httpClient) throw new Error("Not connected");
+    return await this.httpClient.get("/snapshot");
+  }
+
+  async postSideLoadSnapshot(snapshot: unknown) {
+    if (!this.httpClient) throw new Error("Not connected");
+    return await this.httpClient.post("/snapshot", snapshot);
+  }
+
+  async postDecommit(tx: unknown) {
+    if (!this.httpClient) throw new Error("Not connected");
+    return await this.httpClient.post("/decommit", tx);
+  }
+
+  async getProtocolParameters() {
+    if (!this.httpClient) throw new Error("Not connected");
+    return await this.httpClient.get("/protocol-parameters");
+  }
+
+  async submitCardanoTransaction(tx: unknown) {
+    if (!this.httpClient) throw new Error("Not connected");
+    return await this.httpClient.post("/cardano-transaction", tx);
+  }
+
+  async submitL2Transaction(tx: unknown) {
+    if (!this.httpClient) throw new Error("Not connected");
+    return await this.httpClient.post("/transaction", tx);
+  }
+
+  private handleState(snapshot: Snapshot) {
+    if (
+      JSON.stringify(snapshot.value) ===
+      JSON.stringify(this._currentSnapshot?.value)
+    )
+      return;
+    this._currentSnapshot = snapshot;
+    this.emitter.emit("*", snapshot);
+    this.emitter.emit(_flattenState(snapshot.value), snapshot);
+  }
+
+  on(state: HydraStateName, fn: (s: Snapshot) => void) {
+    return this.emitter.on(state, fn);
+  }
+
+  once(state: HydraStateName, fn: (s: Snapshot) => void) {
+    return this.emitter.once(state, fn);
+  }
+
+  off(state: HydraStateName, fn: (s: Snapshot) => void) {
+    return this.emitter.off(state, fn);
+  }
+
+  waitFor(state: HydraStateName, timeout?: number): Promise<void> {
+    return new Promise((resolve, reject) => {
+      const onMatch = () => {
+        clearTimeout(timer);
+        this.off(state, onMatch);
+        resolve();
+      };
+
+      const timer = timeout
+        ? setTimeout(() => {
+            this.off(state, onMatch);
+            reject(new Error(`Timeout waiting for state "${state}"`));
+          }, timeout)
+        : undefined;
+
+      this.once(state, onMatch);
+    });
+  }
+
+  stop() {
+    this.actor.stop();
+    this.emitter.clear();
+    this._currentSnapshot = undefined;
+    this.httpClient = undefined;
+  }
+
+  get state() {
+    return this._currentSnapshot?.value;
+  }
+
+  get context() {
+    return this._currentSnapshot?.context;
+  }
+}
+
+function _flattenState(value: StateValue): HydraStateName {
+  if (typeof value === "string") return value as HydraStateName;
+  return Object.entries(value)
+    .map(([k, v]) => (v ? `${k}.${_flattenState(v)}` : k))
+    .join(".") as HydraStateName;
+}

packages/mesh-hydra/src/mocks/MockHTTPClient.ts

@@ -0,0 +1,47 @@
+import { HTTPClient } from "../utils";
+
+export class MockHttpClient extends HTTPClient {
+  public static instances: MockHttpClient[] = [];
+  public static nextPostErrors: Error[] = [];
+  public static postCalls: Array<{ endpoint: string; payload: unknown }> = [];
+  public static nextPostResponses: unknown[] = [];
+
+  constructor(baseURL: string) {
+    super(baseURL);
+    MockHttpClient.instances.push(this);
+  }
+
+  async post(endpoint: string, payload: unknown) {
+    MockHttpClient.postCalls.push({ endpoint, payload });
+    if (MockHttpClient.nextPostErrors.length > 0) {
+      throw MockHttpClient.nextPostErrors.shift();
+    }
+    if (MockHttpClient.nextPostResponses.length > 0) {
+      return MockHttpClient.nextPostResponses.shift();
+    }
+    // Default response for /commit endpoint - return a draft transaction
+    if (endpoint === "/commit") {
+      return {
+        type: "TxBabbage",
+        description: "Draft commit tx",
+        cborHex: "84a4...",
+      };
+    }
+    return { status: 200, data: "ok" };
+  }
+
+  async get(endpoint: string) {
+    return { status: 200, data: {} };
+  }
+
+  async delete(endpoint: string) {
+    return { status: 200, data: "ok" };
+  }
+
+  public static reset() {
+    MockHttpClient.instances = [];
+    MockHttpClient.nextPostErrors = [];
+    MockHttpClient.postCalls = [];
+    MockHttpClient.nextPostResponses = [];
+  }
+}

packages/mesh-hydra/src/mocks/MockWebSocket.ts

@@ -0,0 +1,56 @@
+type WebSocketEventType = 'open' | 'message' | 'close' | 'error';
+
+export class MockWebSocket {
+  public readyState: number = WebSocket.CONNECTING;
+  public onopen: ((event: Event) => void) | null = null;
+  public onmessage: ((event: MessageEvent) => void) | null = null;
+  public onclose: ((event: CloseEvent) => void) | null = null;
+  public onerror: ((event: Event) => void) | null = null;
+
+  public sentMessages: string[] = [];
+  public eventLog: { type: WebSocketEventType; data?: unknown }[] = [];
+
+
+  constructor(public url: string) {
+    // Simulate async open
+    setImmediate(() => {
+      this.readyState = WebSocket.OPEN;
+      this.onopen?.(new Event('open'));
+    });
+  }
+
+  send(data: string) {
+    if (this.readyState !== WebSocket.OPEN) {
+      throw new Error("WebSocket is not open");
+    }
+    this.sentMessages.push(data);
+  }
+
+  close(code = 1000, reason = "") {
+    this.readyState = WebSocket.CLOSED;
+    this.onclose?.(new CloseEvent('close', { wasClean: false, code, reason }));
+  }
+
+  // Test helper to simulate an incoming JSON message
+  mockReceive(obj: unknown) {
+    const data = typeof obj === "string" ? obj : JSON.stringify(obj);
+    this.onmessage?.(new MessageEvent('message', { data }));
+  }
+
+  mockError(errorData?: unknown) {
+    const err = new Event('error');
+    this.logEvent('error', errorData ?? err);
+    this.onerror?.(err);
+  }
+
+  mockClose(code = 1006, reason = 'Unexpected closure') {
+    if (this.readyState === WebSocket.CLOSED) return;
+    this.readyState = WebSocket.CLOSED;
+    this.logEvent('close', { code, reason });
+    this.onclose?.(new CloseEvent('close', { wasClean: false, code, reason }));
+  }
+
+  private logEvent(type: WebSocketEventType, data?: unknown) {
+    this.eventLog.push({ type, data });
+  }
+}