feat: RPC client re-write (#341)

Co-authored-by: Harry Solovay <[email protected]>
Co-authored-by: T6 <[email protected]>

_tasks/download_frame_metadata.ts

@@ -10,7 +10,7 @@ await fs.emptyDir(outDir);
 await Promise.all(
   Object.entries({ acala, kusama, moonbeam, polkadot, statemint, subsocial, westend }).map(
     async ([name, config]) => {
-      const client = U.throwIfError(await rpc.stdClient(config));
+      const client = U.throwIfError(await rpc.proxyClient(config));
       try {
         const metadata = await client.call("state_getMetadata", []);
         assert(metadata.result);

codegen.ts

@@ -29,6 +29,7 @@ await codegen({
   metadata: await getMetadata(args.src),
 }).write(args.out);
 
+// Should disallow .scale as input?
 async function getMetadata(src: string): Promise<M.Metadata> {
   if (src.startsWith("ws")) {
     const client = U.throwIfError(await proxyClient(new Config(() => src)));

deps/smoldot.ts

@@ -1,3 +1 @@
-// Smoldot is a peer dependency; we only utilize its types, never its runtime.
-// @deno-types="https://esm.sh/@substrate/[email protected]/dist/index.d.ts"
-export * from "./smoldot_phantom.ts";
+export * from "https://deno.land/x/[email protected]/index-deno.js";

deps/smoldot/client.d.ts

@@ -0,0 +1 @@
+export * from "https://deno.land/x/[email protected]/client.d.ts";

deps/zones.ts

@@ -1 +1 @@
-export * from "https://deno.land/x/[email protected].7/mod.ts";
+export * from "https://deno.land/x/[email protected].8/mod.ts";

effects/blockWatch.ts

@@ -8,7 +8,7 @@ import { run } from "./run.ts";
 
 export function blockWatch(
   config: Config,
-  createWatchHandler: U.CreateWatchHandler<known.SignedBlock>,
+  createWatchHandler: U.CreateListener<known.SignedBlock>,
 ) {
   return rpcSubscription(
     config,

effects/core/rpcClient.ts

@@ -4,6 +4,6 @@ import * as rpc from "../../rpc/mod.ts";
 
 export function rpcClient<C extends Config>(config: C) {
   return Z.call(config, function rpcClientImpl() {
-    return rpc.stdClient(config);
+    return rpc.proxyClient(config);
   });
 }

effects/entryWatch.ts

@@ -21,7 +21,7 @@ export function entryWatch<
   palletName: PalletName,
   entryName: EntryName,
   keys: Keys,
-  createWatchHandler: U.CreateWatchHandler<WatchEntryEvent[]>,
+  createWatchHandler: U.CreateListener<WatchEntryEvent[]>,
 ) {
   const metadata_ = metadata(config);
   const deriveCodec_ = deriveCodec(metadata_);
@@ -37,7 +37,7 @@ export function entryWatch<
     },
   );
   const watchInit = Z.call($entry, function entryWatchInit($entry) {
-    return U.mapCreateWatchHandler(
+    return U.mapCreateListener(
       createWatchHandler,
       (message: rpc.NotifMessage) => {
         return message.params.result.changes.map(([key, val]: any) => {

effects/extrinsic.ts

@@ -131,7 +131,7 @@ export class SignedExtrinsic<
   }
 
   watch<
-    WatchHandler extends U.CreateWatchHandler<NotifMessage<author.TransactionStatus>>,
+    WatchHandler extends U.CreateListener<NotifMessage<author.TransactionStatus>>,
   >(watchHandler: WatchHandler) {
     return rpcSubscription(
       this.config,

effects/rpcCall.ts

@@ -8,11 +8,9 @@ export function rpcCall<MethodName extends Z.$<string>, Params extends Z.Ls$<unk
   methodName: MethodName,
   params: [...Params],
 ) {
-  const client = rpcClient(config);
-  const deps = Z.ls(client, methodName, ...params);
   return Z.call(
-    Z.ls(deps, Z.rc(client, deps)),
-    async function rpcCallImpl([[client, methodName, ...params], rc]) {
+    Z.rc(rpcClient(config), methodName, ...params),
+    async function rpcCallImpl([[client, methodName, ...params], counter]) {
       const result = await client.call(
         methodName,
         params,
@@ -26,7 +24,8 @@ export function rpcCall<MethodName extends Z.$<string>, Params extends Z.Ls$<unk
           },
         });
       }
-      if (rc() == 1) {
+      counter.i--;
+      if (counter.i === 1) {
         const close = await client.close();
         if (close instanceof Error) return close;
       }

effects/rpcSubscription.ts

@@ -9,7 +9,7 @@ import { run } from "./run.ts";
 export function rpcSubscription<
   MethodName extends Z.$<string>,
   Params extends Z.Ls$<unknown[]>,
-  CreateListenerCb extends Z.$<U.CreateWatchHandler<rpc.NotifMessage>>,
+  CreateListenerCb extends Z.$<U.CreateListener<rpc.NotifMessage>>,
 >(
   config: Config,
   methodName: MethodName,
@@ -18,11 +18,11 @@ export function rpcSubscription<
   // TODO: improve cleanup system
   cleanup?: (initOk: rpc.OkMessage) => Z.Effect,
 ) {
-  const client = rpcClient(config);
-  const deps = Z.ls(client, methodName, createListenerCb, ...params);
   return Z.call(
-    Z.ls(deps, Z.rc(client, deps)),
-    async function rpcSubscriptionImpl([[client, methodName, createListenerCb, ...params], rc]) {
+    Z.rc(rpcClient(config), methodName, createListenerCb, ...params),
+    async function rpcSubscriptionImpl(
+      [[client, methodName, createListenerCb, ...params], counter],
+    ) {
       const result = await client.subscribe(
         methodName,
         params,
@@ -38,7 +38,8 @@ export function rpcSubscription<
           },
         });
       }
-      if (rc() == 1) {
+      counter.i--;
+      if (counter.i === 1) {
         const close = await client.close();
         if (close instanceof Error) return close;
       }

examples/.ignore

@@ -1 +1,2 @@
 derived.ts
+watch_blocks_iter.ts

examples/raw_rpc_client/call.ts

@@ -2,7 +2,7 @@ import * as C from "../../mod.ts";
 import * as rpc from "../../rpc/mod.ts";
 import * as U from "../../util/mod.ts";
 
-const client = U.throwIfError(await rpc.stdClient(C.westend));
+const client = U.throwIfError(await rpc.proxyClient(C.westend));
 
 console.log(await client.call("state_getMetadata", []));
 

examples/raw_rpc_client/subscription.ts

@@ -2,7 +2,7 @@ import * as C from "../../mod.ts";
 import * as rpc from "../../rpc/mod.ts";
 import * as U from "../../util/mod.ts";
 
-const client = U.throwIfError(await rpc.stdClient(C.westend));
+const client = U.throwIfError(await rpc.proxyClient(C.westend));
 
 const maybeError = await client.subscribe("chain_subscribeAllHeads", [], (stop) => {
   let i = 1;

known/rpc/Readme.md

@@ -0,0 +1 @@
+# TODO: integrate this dir into RPC re-implementation dir

known/rpc/state.ts

@@ -193,7 +193,7 @@ export type StateRpc = {
   state_getStorageSize(key: StorageKey, at?: Hash): RpcResult<number | null>;
   state_getStorageSizeAt: StateRpc["state_getStorageSize"];
   /** Returns the runtime metadata as an opaque blob. */
-  state_getMetadata(at?: Hash): RpcResult<Hex | null>;
+  state_getMetadata(at?: Hash): RpcResult<Hex>;
   /** Get the runtime version. */
   state_getRuntimeVersion(at?: Hash): RpcResult<RuntimeVersion>;
   chain_getRuntimeVersion: StateRpc["state_getRuntimeVersion"];

mod.ts

@@ -6,11 +6,4 @@ export * from "./effects/mod.ts";
 export * as M from "./frame_metadata/mod.ts";
 export { $era, $null, ChainError, type Era } from "./frame_metadata/mod.ts";
 export { kusama, moonbeam, polkadot, rococo, westend } from "./known/mod.ts"; // TODO: get rid of this!
-export {
-  type CreateWatchHandler,
-  hex,
-  mapCreateWatchHandler,
-  type WatchHandler,
-  type WatchIter,
-  watchIter,
-} from "./util/mod.ts";
+export { hex, type Listener, type WatchIter, watchIter } from "./util/mod.ts";

rpc/Base.ts

@@ -11,7 +11,7 @@ export abstract class Client<
   CloseError extends Error,
 > {
   #nextId = 0;
-  #listenerCbs = new Map<U.WatchHandler<msg.IngressMessage>, true>();
+  #listenerCbs = new Map<U.Listener<msg.IngressMessage>, true>();
 
   /**
    * Construct a new RPC client
@@ -48,7 +48,7 @@ export abstract class Client<
    *
    * @param createListenerCb the factory for the callback to be triggered upon arrival of ingress messages
    */
-  listen = (createListenerCb: U.CreateWatchHandler<msg.IngressMessage>): void => {
+  listen = (createListenerCb: U.CreateListener<msg.IngressMessage>): void => {
     const stopListening = () => {
       this.#listenerCbs.delete(listenerCb);
     };
@@ -144,6 +144,6 @@ export abstract class Client<
   };
 }
 
-export type CreateListenerCb = U.CreateWatchHandler<msg.NotifMessage>;
+export type CreateListenerCb = U.CreateListener<msg.NotifMessage>;
 
 export type SubscriptionCleanup = (initOk: msg.OkMessage) => Promise<void>;

rpc/mod.ts

@@ -2,5 +2,3 @@ export * from "./Base.ts";
 export * from "./common.ts";
 export * from "./messages.ts";
 export * from "./providers/proxy.ts";
-export * from "./providers/smoldot.ts";
-export * from "./providers/std.ts";