chore: clean up test node, begin watch-related, cleanup (#149)

core/Entry.ts

@@ -30,7 +30,7 @@ export class Entry<
     return R.read(this, block);
   }
 
-  watch() {
-    return R.watch(this);
+  watch(cb: (message: unknown) => void): any /* TODO */ {
+    return R.watch(this, cb);
   }
 }

core/Header.ts

@@ -13,7 +13,7 @@ export class Header<C extends Chain = Chain> extends NodeBase<"Header"> {
     return R.read(this, block);
   }
 
-  watch() {
-    return R.watch(this);
+  watch(cb: (message: unknown) => void): any /* TODO */ {
+    return R.watch(this, cb);
   }
 }

core/KeyPage.ts

@@ -22,7 +22,7 @@ export class KeyPage<E extends Entry = Entry> extends NodeBase<"KeyPage"> {
     return R.read(this, block);
   }
 
-  watch() {
-    return R.watch(this);
+  watch(cb: (message: unknown) => void): any /* TODO */ {
+    return R.watch(this, cb);
   }
 }

core/Signed.ts

@@ -19,7 +19,7 @@ export class Signed<
     this.chain = call.chain;
   }
 
-  send() {
+  send = async () => {
     return send(this);
-  }
+  };
 }

deno.jsonc

@@ -28,6 +28,7 @@
   "include": ["."],
   "tasks": {
     "run": "deno run -A --no-check=remote",
+    "debug": "deno task run --inspect-brk",
     "download:frame_metadata": "deno task run _tasks/download_frame_metadata.ts",
     "build:wasm": "deno task run _tasks/build_wasm.ts",
     "build:known": "deno task run _tasks/build_known.ts",

examples/transfer.ts

@@ -5,7 +5,7 @@ const chain = C.test.chain(node);
 
 const { alice, bob } = chain.address;
 
-const result = chain
+await chain
   .pallet("Balances")
   .extrinsic("transfer")
   .call({
@@ -15,8 +15,4 @@ const result = chain
   .signed(bob, bob.sign)
   .send();
 
-for await (const event of result) {
-  console.log({ event });
-}
-
 node.close();

examples/watch_events.ts

@@ -0,0 +1,12 @@
+import * as C from "../mod.ts";
+
+const result = await C.westend
+  .pallet("System")
+  .entry("Events")
+  .watch(console.log);
+
+setTimeout(() => {
+  if (typeof result === "function") {
+    result();
+  }
+}, 10000);

frame_metadata/Codec.test.ts

@@ -1,10 +1,8 @@
 import { assertEquals } from "../_deps/asserts.ts";
-import * as path from "../_deps/path.ts";
 import { testPairs } from "../known/mod.ts";
 import { DeriveCodec } from "./Codec.ts";
 import { ChainError } from "./Codec.ts";
 import { Lookup } from "./Lookup.ts";
-import { fromPrefixedHex } from "./Metadata.ts";
 import { Metadata } from "./test-common.ts";
 
 const metadata = await Metadata("polkadot");

frame_metadata/Codec.ts

@@ -10,6 +10,7 @@ export type DeriveCodec = (typeI: number) => $.Codec<unknown>;
  */
 export const $null = $.dummy(null);
 
+// TODO: tuple/array element skip optimization
 export function DeriveCodec(metadata: M.Metadata): DeriveCodec {
   // TODO: don't leak memory!
   const cache: Record<number, $.Codec<unknown> | null> = {};
@@ -150,7 +151,7 @@ export function DeriveCodec(metadata: M.Metadata): DeriveCodec {
       }
       cache[i] = null; // circularity detection
       const ty = metadata.tys[i]!;
-      const $codec = (visitors[ty.type] as any)(ty, false);
+      const $codec = (visitors[ty.type] as any)(ty);
       cache[i] = $codec;
       return $codec;
     },

frame_metadata/mod.ts

@@ -1,5 +1,4 @@
 export * from "./Codec.ts";
-export * from "./Display.ts";
 export * from "./Extrinsic.ts";
 export * from "./Key.ts";
 export * from "./Lookup.ts";

mod.ts

@@ -4,4 +4,4 @@ export * from "./core/mod.ts";
 export * from "./known/mod.ts";
 export * from "./primitives/mod.ts";
 export * as test from "./test-util/mod.ts";
-export { type SubstrateNode, substrateNode, type SubstrateNodeConfig } from "./util/mod.ts";
+export { iter } from "./util/mod.ts";

runtime/watch.ts

@@ -1,7 +1,60 @@
+import { Hashers } from "../bindings/mod.ts";
 import * as core from "../core/mod.ts";
+import * as M from "../frame_metadata/mod.ts";
+import * as rpc from "../rpc/mod.ts";
+import * as U from "../util/mod.ts";
+import { globalContext } from "./Context.ts";
 
 export type WatchTarget = core.Entry | core.KeyPage | core.Metadata | core.Header | core.Block;
 
-export declare function watch<Target extends WatchTarget>(
+export async function watch<Target extends WatchTarget>(
   target: Target,
-): AsyncIterableIterator<unknown>;
+  listener: (notification: rpc.NotifMessage<U.AnyMethods>) => void,
+): Promise<(() => void) | Error> {
+  const chain = await globalContext.register(target.chain.beacon as any);
+  if (chain instanceof Error) {
+    return chain;
+  }
+  const group = await chain.load();
+  switch (target.kind) {
+    case "Entry": {
+      const pallet = group.lookup.getPalletByName(target.pallet.name);
+      const storageEntry = group.lookup.getStorageEntryByPalletAndName(pallet, target.name);
+      const $key = M.$storageKey({
+        deriveCodec: group.deriveCodec,
+        hashers: await Hashers(),
+        pallet,
+        storageEntry,
+      });
+      const key = $key.encode(target.keys.length === 1 ? target.keys[0]! : target.keys);
+      const keyEncoded = U.hex.encode(key) as U.HexString;
+      const $value = group.deriveCodec(storageEntry.value);
+      const outerListener = (message: rpc.NotifMessage<U.AnyMethods>) => {
+        const { result } = message.params;
+        const changes = (result as any).changes;
+        const changesDecoded = changes.map(([key, value]: [any, any]) => {
+          return [
+            $key.decode(U.hex.decode(key.substring(2))),
+            $value.decode(U.hex.decode(value.substring(2))),
+          ];
+        });
+        listener(changesDecoded);
+      };
+      const raw = await chain.rpcClient.subscribe(
+        "state_subscribeStorage",
+        [[keyEncoded]],
+        outerListener,
+      );
+      if (typeof raw === "function") {
+        return async () => {
+          raw();
+          await chain.release();
+        };
+      }
+      return new Error();
+    }
+    default: {
+      throw new Error();
+    }
+  }
+}

test-util/Chain.ts

@@ -1,13 +1,13 @@
 import { Chain } from "../core/Chain.ts";
 import { KnownRpcMethods } from "../known/mod.ts";
 import { ProxyBeacon } from "../rpc/mod.ts";
-import { SubstrateNode } from "../util/mod.ts";
 import { TestAddresses } from "./Addresses.ts";
+import { TestNode } from "./node.ts";
 
 export class TestChain extends Chain<ProxyBeacon<KnownRpcMethods>> {
   override address: TestAddresses<this> = new TestAddresses(this);
 }
 
-export function chain(process: SubstrateNode) {
-  return new TestChain(new ProxyBeacon(process.url));
+export function chain(node: TestNode) {
+  return new TestChain(new ProxyBeacon(node.url));
 }

test-util/node.ts

@@ -1,12 +1,62 @@
-import { assert } from "../_deps/asserts.ts";
-import { SubstrateNode, substrateNode } from "../util/mod.ts";
+import { fail } from "../_deps/asserts.ts";
+import { blue } from "../_deps/colors.ts";
+import { isPortAvailable } from "../util/mod.ts";
 
-export async function node(): Promise<SubstrateNode> {
-  const process = await substrateNode({
-    path: "./node-template",
-    cwd: new URL(".", import.meta.url).pathname,
-    dev: true,
-  });
-  assert(!(process instanceof Error));
-  return process;
+export interface TestNodeConfig {
+  cwd?: string;
+  altRuntime?: "kusama" | "rococo" | "westend";
+}
+
+export interface TestNode {
+  config?: TestNodeConfig;
+  inner: Deno.Process;
+  close(): void;
+  url: string;
+}
+
+let port = 9944;
+export async function node(config?: TestNodeConfig): Promise<TestNode> {
+  while (!isPortAvailable(port)) {
+    port++;
+  }
+  try {
+    const process = Deno.run({
+      cmd: [
+        "polkadot",
+        "--dev",
+        "--ws-port",
+        port.toString(),
+        ...config?.altRuntime ? [`--force-${config.altRuntime}`] : [],
+      ],
+      cwd: config?.cwd,
+      stderr: "piped",
+      stdout: "piped",
+    });
+    // For some reason, logs come in through `stderr`
+    const logs = process.stderr.readable;
+    await (async () => {
+      console.log(blue(`Piping node logs:`));
+      for await (const log of logs) {
+        Deno.stdout.write(log);
+        if (new TextDecoder().decode(log).includes(" Running JSON-RPC WS server")) {
+          console.log(blue("Chain and RPC server initialized"));
+          console.log(blue(`Suspending node logs`));
+          return;
+        }
+      }
+    })();
+    return {
+      config,
+      inner: process,
+      close: () => {
+        process.close();
+      },
+      url: `ws://127.0.0.1:${port}`,
+    };
+  } catch (e) {
+    if (e instanceof Deno.errors.NotFound) {
+      fail("Must have Polkadot installed locally. Visit https://github.com/paritytech/polkadot.");
+    }
+    fail();
+  }
 }

util/isPortAvailable.ts

@@ -0,0 +1,16 @@
+export function isPortAvailable(port: number): boolean {
+  try {
+    const listener = Deno.listen({
+      transport: "tcp",
+      hostname: "127.0.0.1",
+      port,
+    });
+    listener.close();
+    return true;
+  } catch (error) {
+    if (error instanceof Deno.errors.AddrInUse) {
+      return false;
+    }
+    throw error;
+  }
+}