feat: watch blocks effect (#184)

effect/atoms/Metadata.ts

@@ -1,12 +1,12 @@
 import * as M from "../../frame_metadata/mod.ts";
-import { rpc as knownRpc } from "../../known/mod.ts";
+import * as known from "../../known/mod.ts";
 import * as rpc from "../../rpc/mod.ts";
 import * as U from "../../util/mod.ts";
 import { atomFactory } from "../sys/Atom.ts";
 import { Val } from "../sys/mod.ts";
 import { rpcCall } from "./RpcCall.ts";
 
-type ConfigConstraint = knownRpc.Config<string, "state_getMetadata">;
+type ConfigConstraint = known.rpc.Config<string, "state_getMetadata">;
 
 export function metadata<Rest extends [blockHash?: Val<U.HashHexString | undefined>]>(
   config: ConfigConstraint,

effect/examples/ticker.ts

@@ -8,8 +8,8 @@ const root = C.watchEntry(config, "Timestamp", "Now", [], () => {
   let i = 0;
 
   return (m) => {
-    i++;
     console.log({ [i]: m });
+    i++;
   };
 });
 

effect/examples/watch_blocks.ts

@@ -0,0 +1,16 @@
+import * as C from "../../mod.ts";
+import * as U from "../../util/mod.ts";
+
+const root = C.watchBlocks(C.polkadot, (stop) => {
+  let i = 0;
+
+  return (event) => {
+    console.log(event);
+    if (i === 4) {
+      stop();
+    }
+    i++;
+  };
+});
+
+console.log(U.throwIfError(await root.run()));

effect/std/mod.ts

@@ -2,4 +2,5 @@ export * from "./readBlock.ts";
 export * from "./readEntry.ts";
 export * from "./readKeyPage.ts";
 export * from "./submitAndWatchExtrinsic.ts";
+export * from "./watchBlocks.ts";
 export * from "./watchEntry.ts";

effect/std/readBlock.ts

@@ -1,12 +1,9 @@
-import { Codec } from "../../deps/scale.ts";
-import { Extrinsic } from "../../frame_metadata/mod.ts";
-import { rpc as knownRpc } from "../../known/mod.ts";
-import * as rpc from "../../rpc/mod.ts";
+import * as known from "../../known/mod.ts";
 import * as U from "../../util/mod.ts";
 import * as a from "../atoms/mod.ts";
 import * as sys from "../sys/mod.ts";
 
-type Config = knownRpc.Config<string, "state_getMetadata" | "chain_getBlock">;
+type Config = known.rpc.Config<string, "state_getMetadata" | "chain_getBlock">;
 
 export function readBlock<Rest extends [blockHash?: sys.Val<U.HashHexString | undefined>]>(
   config: Config,
@@ -15,22 +12,17 @@ export function readBlock<Rest extends [blockHash?: sys.Val<U.HashHexString | un
   const metadata_ = a.metadata(config, blockHash);
   const $extrinsic = a.$extrinsic(a.deriveCodec(metadata_), metadata_);
   const call = a.rpcCall(config, "chain_getBlock", [blockHash]);
-  const decoded = sys.anon([$extrinsic, call], processChainGetBlock);
+  const decoded = sys.anon([$extrinsic, call], ($extrinsic, call) => {
+    const { block: { extrinsics, header }, justifications } = call.result;
+    return {
+      justifications,
+      block: {
+        header,
+        extrinsics: extrinsics.map((extrinsic) => {
+          return $extrinsic.decode(U.hex.decode(extrinsic));
+        }),
+      },
+    };
+  });
   return a.wrap(decoded, "block");
 }
-
-function processChainGetBlock(
-  $extrinsic: Codec<Extrinsic>,
-  raw: rpc.OkMessage<Config, "chain_getBlock">,
-) {
-  const { block: { extrinsics, header }, justifications } = raw.result;
-  return {
-    justifications,
-    block: {
-      header,
-      extrinsics: extrinsics.map((extrinsic) => {
-        return $extrinsic.decode(U.hex.decode(extrinsic));
-      }),
-    },
-  };
-}

effect/std/readEntry.ts

@@ -1,9 +1,9 @@
-import { rpc as knownRpc } from "../../known/mod.ts";
+import * as known from "../../known/mod.ts";
 import * as U from "../../util/mod.ts";
 import * as a from "../atoms/mod.ts";
 import * as sys from "../sys/mod.ts";
 
-type Config = knownRpc.Config<string, "state_getMetadata" | "state_getStorage">;
+type Config = known.rpc.Config<string, "state_getMetadata" | "state_getStorage">;
 
 export function readEntry<
   PalletName extends sys.Val<string>,

effect/std/readKeyPage.ts

@@ -1,9 +1,9 @@
-import { rpc as knownRpc } from "../../known/mod.ts";
+import * as known from "../../known/mod.ts";
 import * as U from "../../util/mod.ts";
 import * as a from "../atoms/mod.ts";
 import * as sys from "../sys/mod.ts";
 
-type Config = knownRpc.Config<string, "state_getMetadata" | "state_getKeysPaged">;
+type Config = known.rpc.Config<string, "state_getMetadata" | "state_getKeysPaged">;
 
 export function readKeyPage<
   PalletName extends sys.Val<string>,

effect/std/submitAndWatchExtrinsic.ts

@@ -1,14 +1,14 @@
 import { unimplemented } from "../../deps/std/testing/asserts.ts";
 import * as M from "../../frame_metadata/mod.ts";
-import { rpc as knownRpc } from "../../known/mod.ts";
+import * as known from "../../known/mod.ts";
 import * as rpc from "../../rpc/mod.ts";
 import { Ss58 } from "../../ss58/mod.ts";
 import * as U from "../../util/mod.ts";
 import * as a from "../atoms/mod.ts";
 import * as sys from "../sys/mod.ts";
 
 export { type Config as SendAndWatchExtrinsicConfig };
-type Config = knownRpc.Config<
+type Config = known.rpc.Config<
   string,
   | "state_getMetadata"
   | "state_getRuntimeVersion"

effect/std/watchBlocks.ts

@@ -0,0 +1,34 @@
+import { Extrinsic } from "../../frame_metadata/mod.ts";
+import * as known from "../../known/mod.ts";
+import * as U from "../../util/mod.ts";
+import * as a from "../atoms/mod.ts";
+import { readBlock } from "./readBlock.ts";
+
+type Config = known.rpc.Config<
+  string,
+  "state_getMetadata" | "chain_getBlockHash" | "chain_getBlock" | "chain_unsubscribeNewHead",
+  "chain_subscribeNewHeads"
+>;
+
+export function watchBlocks(
+  config: Config,
+  createWatchHandler: U.CreateWatchHandler<known.types.Block<Extrinsic>>,
+) {
+  return a.rpcSubscription(config, "chain_subscribeNewHeads", [], (stop) => {
+    const watchHandler = createWatchHandler(stop);
+    return async (result) => {
+      const blockNum = result.params.result.number;
+      const blockHash = a
+        .rpcCall(config, "chain_getBlockHash", [blockNum])
+        .select("result");
+      const block = await readBlock(config, blockHash as unknown as U.HashHexString).run(); // STOP THIS MADNESS
+      if (block instanceof Error) {
+        // TODO: subscription runtime error channel
+        throw new Error();
+      }
+      watchHandler(block.block);
+    };
+  }, (ok) => {
+    return a.rpcCall(config, "chain_unsubscribeNewHead", [ok.result]);
+  });
+}

effect/std/watchEntry.ts

@@ -1,12 +1,12 @@
-import { type rpc as knownRpc } from "../../known/mod.ts";
+import * as known from "../../known/mod.ts";
 import * as rpc from "../../rpc/mod.ts";
 import * as U from "../../util/mod.ts";
 import * as a from "../atoms/mod.ts";
 import * as sys from "../sys/mod.ts";
 
 export type WatchEntryEvent = [key?: U.HexString, value?: unknown];
 
-type Config = knownRpc.Config<
+type Config = known.rpc.Config<
   string,
   "state_getMetadata" | "state_unsubscribeStorage",
   "state_subscribeStorage"

known/mod.ts

@@ -2,3 +2,4 @@ export * from "./configs.ts";
 export * from "./generated.ts";
 export * from "./metadata.ts";
 export * as rpc from "./rpc.ts";
+export * as types from "./types/mod.ts";

known/rpc.ts

@@ -39,8 +39,9 @@ export type CallMethods = rpc.EnsureMethods<{
     getFinalizedHead(): U.H256String;
   };
   chain: {
-    getBlock(hash?: U.HashHexString): T.Block;
-    getBlockHash(height?: number): U.HashHexString;
+    getBlock(hash?: U.HashHexString): T.Block<U.HexString>;
+    // TODO: confirm that the following params are accurate (seems to work both ways)
+    getBlockHash(height?: U.HexU64String | number): U.HashHexString;
     getFinalisedHead: CallMethods["chain_getFinalizedHead"];
     getFinalizedHead(): U.HashHexString;
     getHead: CallMethods["chain_getBlockHash"];
@@ -205,7 +206,7 @@ export type SubscriptionMethods = rpc.EnsureMethods<{
     subscribeFinalisedHeads: SubscriptionMethods["chain_subscribeFinalizedHeads"];
     subscribeFinalizedHeads(): T.Header; /* TODO: narrow to finalized? */
     subscribeNewHead: SubscriptionMethods["chain_subscribeNewHeads"];
-    subscribeNewHeads(): unknown;
+    subscribeNewHeads(): T.Header;
     subscribeRuntimeVersion: SubscriptionMethods["state_subscribeRuntimeVersion"];
     subscribe_newHead: SubscriptionMethods["chain_subscribeNewHeads"];
   };

known/types/common.ts

@@ -1,9 +1,9 @@
 import * as U from "../../util/mod.ts";
 
-export interface Block {
+export interface Block<Extrinsic> {
   block: {
     header: Header;
-    extrinsics: U.HexString[];
+    extrinsics: Extrinsic[];
   };
   justifications?: [number[], number[]][];
 }