Implement Module node for running subgraphs

Author: liamgriffiths

examples/module.ts

@@ -0,0 +1,76 @@
+#!/usr/bin/env -S npx ts-node --transpileOnly
+
+import { Substrate, Box, Module, sb } from "substrate";
+
+async function main() {
+  const SUBSTRATE_API_KEY = process.env["SUBSTRATE_API_KEY"];
+
+  const substrate = new Substrate({ apiKey: SUBSTRATE_API_KEY });
+
+  const x = sb.input({ type: "string", default: "hello" });
+  const y = sb.input({ type: "string" });
+  const z = sb.input({ type: "object", properties: {} });
+
+  const a = new Box({ value: { a: x, z: z, array: [x, x, x] } }, { id: "A" });
+  const b = new Box(
+    { value: { b: sb.interpolate`x=${a.future.value.get("a")}, y=${y}` } },
+    { id: "B" },
+  );
+
+  // publish the module on substrate.run
+  // const publication = await substrate.module.publish({
+  //   name: "my reusable graph",
+  //   nodes: [a, b],
+  //   inputs: { x, y, z },
+  // });
+
+  // update the module on substrate.run
+  // const updated = await substrate.module.publish({
+  //   id: publication.id,
+  //   name: "my reusable graph (edited)",
+  //   nodes: [a, b],
+  //   inputs: { x, y, z },
+  // });
+
+  // using the module from JSON
+  const mod = new Module({
+    module_json: substrate.module.serialize({
+      nodes: [a, b],
+      inputs: { x, y, z },
+    }),
+    inputs: {
+      // when commented will use "hello" because it is defined as the default above
+      // x: 123,
+      y: "yyy",
+      z: {
+        arr: ["123"],
+      },
+    },
+  });
+
+  // using the module from publication/module id
+  // const mod = new Module({
+  //   module_id: publication.id,
+  //   inputs: { y: "yyy", z: { arr: ["123"] } },
+  // });
+
+  // using the module from publication/module uri
+  // const mod = new Module({
+  //   module_uri: publication.uri,
+  //   inputs: { y: "yyy", z: { arr: ["123"] } },
+  // });
+
+  const c = new Box(
+    {
+      value: {
+        "1": mod.future.get("A.value.z.arr[0]"),
+        "2": mod.future.get("B.value.b"),
+      },
+    },
+    { id: "C" },
+  );
+
+  const res = await substrate.run(mod, c);
+  console.log(JSON.stringify(res.json, null, 2));
+}
+main();

examples/types.ts

@@ -0,0 +1,21 @@
+#!/usr/bin/env -S npx ts-node --transpileOnly
+
+
+class Foo<F extends (...args: any[]) => any> {
+  f: F;
+
+  constructor(f: F) {
+    this.f = f;
+  }
+
+  run(...args: Parameters<F>): ReturnType<F> {
+    return this.f(...args);
+  }
+}
+
+// (a: number, b: number) => number
+const add = (a: number, b: number) => a + b;
+
+const foo = new Foo(add);
+// run(a: number, b: number): ReturnType<(a: number, b: number) => number>
+const res = foo.run(1, 2);

package-lock.json

@@ -43,6 +43,7 @@
     "vitest": "^1.0.4"
   },
   "dependencies": {
+    "@types/json-schema": "^7.0.15",
     "@types/node-fetch": "^2.6.11",
     "node-fetch": "2.7.0",
     "pako": "^2.1.0"

package.json

@@ -1,5 +1,6 @@
 import { idGenerator } from "substrate/idGenerator";
 import { Node } from "substrate/Node";
+import { type JSONSchema7 } from "json-schema";
 
 type Accessor = "item" | "attr";
 type TraceOperation = {
@@ -26,6 +27,7 @@ const parsePath = (path: string): TraceProp[] => {
 };
 
 const newFutureId = idGenerator("future");
+const newInputId = idGenerator("input");
 
 abstract class Directive {
   abstract items: any[];
@@ -122,7 +124,7 @@ export class JQ extends Directive {
     rawValue: (val: JQCompatible) => ({ future_id: null, val }),
   };
 
-  override next(...items: TraceProp[]) {
+  override next(..._items: TraceProp[]) {
     return new JQ(this.query, this.target);
   }
 
@@ -315,3 +317,53 @@ export class FutureAnyObject extends Future<Object> {
     return super._result();
   }
 }
+
+export class Input extends Directive {
+  items: any[]; // NOTE: unused field (will remove this from direcitve in a later refactor)
+  name: string | null;
+
+  constructor(items: any[]) {
+    super();
+    this.items = items;
+  }
+
+  override next(...args: any[]) {
+    return new Input(args);
+  }
+
+  override async result(): Promise<any> {
+    return;
+  }
+
+  override toJSON() {
+    return {
+      type: "input",
+      name: this.name,
+    };
+  }
+}
+
+export class FutureInput extends Future<any> {
+  declare _directive: Input;
+  schema: JSONSchema7;
+
+  constructor(schema?: JSONSchema7, id: string = newInputId()) {
+    super(new Input([]), id);
+    this.schema = schema ?? {};
+  }
+}
+
+/**
+ * Specify an `input` future that can be assigned a name when create a new module.
+ * Input types and validation paramters may optionally be described using a JSON Schema object.
+ *
+ * Default values may also be specified here and will be used if user input is not provided for this input.
+ */
+export function input(schema?: FutureInput["schema"]) {
+  // NOTE: using `any` as the return type here for now to ease using
+  // this in general node input args or helper functions.
+  //
+  // Once we ship our Future type reorganization work, we can just
+  // use this as-is (Future<any>)
+  return new FutureInput(schema) as any;
+}

src/Future.ts

@@ -0,0 +1,25 @@
+import { Node, Options } from "substrate/Node";
+import { FutureInput } from "substrate/Future";
+
+export type ModuleInputs = Record<string, FutureInput>;
+
+type ModuleIn =
+  | {
+      module_json: any;
+      inputs: Record<string, any>;
+    }
+  | {
+      module_id: any;
+      inputs: Record<string, any>;
+    }
+  | {
+      module_uri: any;
+      inputs: Record<string, any>;
+    };
+
+export class Module extends Node {
+  constructor(args: ModuleIn, options?: Options) {
+    super(args, options);
+    this.node = "Module";
+  }
+}

src/Module.ts

@@ -8,6 +8,7 @@ import { Future } from "substrate/Future";
 import { getPlatformProperties } from "substrate/Platform";
 import { deflate } from "pako";
 import { randomString } from "substrate/idGenerator";
+import { ModuleInputs } from "./Module";
 
 type Configuration = {
   /**
@@ -291,4 +292,50 @@ export class Substrate {
 
     return headers;
   }
+
+  module = {
+    /**
+     * Returns an object that represents a publishable "module" or code that can be used to construct 
+     * a `Module` node.
+     */
+    serialize: ({ nodes, inputs }: { nodes: Node[]; inputs: ModuleInputs }) => {
+      const inputIdToName = {};
+      const inputNameToSchema = {};
+
+      for (let name in inputs) {
+        let input = inputs[name];
+        // @ts-ignore
+        inputIdToName[input._id] = name;
+        // @ts-ignore
+        inputNameToSchema[name] = input.schema;
+      }
+
+      const dag = Substrate.serialize(...nodes);
+
+      // update variable name bindings in dag using inputs
+      dag.futures = dag.futures.map((future: any) => {
+        if (future.directive.type === "input" && !future.directive.name) {
+          // @ts-ignore
+          future.directive.name = inputIdToName[future.id];
+        }
+        return future;
+      });
+
+      return {
+        dag,
+        inputs: inputNameToSchema,
+        api_version: this.apiVersion,
+      };
+    },
+
+    /**
+     * Publishes a module on substrate.run
+     * A successful response will contain the module id and web uri
+     */
+    publish: async (_publishable: any) => {
+      console.log("not implemented yet");
+      let publication;
+      return publication;
+    },
+  };
 }

src/Substrate.ts

@@ -51,6 +51,8 @@ export {
   DeleteVectors,
 } from "substrate/Nodes";
 
+export { Module } from "substrate/Module";
+
 export { sb } from "substrate/sb";
 export { Substrate };
 import { Substrate } from "substrate/Substrate";

src/index.ts

@@ -1,10 +1,11 @@
-import { FutureAnyObject, FutureString } from "substrate/Future";
+import { FutureAnyObject, FutureString, input } from "substrate/Future";
 import { StreamingResponse } from "substrate/SubstrateStreamingResponse";
 
 export const sb = {
   concat: FutureString.concat,
   jq: FutureAnyObject.jq,
   interpolate: FutureString.interpolate,
+  input,
   streaming: {
     fromSSEResponse: StreamingResponse.fromReponse,
   },