feat: Unify worker manager

src/typegate/src/runtimes/deno/deno.ts

@@ -18,7 +18,6 @@
 } from "../../typegraph/types.ts";
 import * as ast from "graphql/ast";
 import { InternalAuth } from "../../services/auth/protocols/internal.ts";
-import { DenoMessenger } from "./deno_messenger.ts";
 import type { Task } from "./shared_types.ts";
 import { path } from "compress/deps.ts";
 import { globalConfig as config } from "../../config.ts";
@@ -27,6 +26,7 @@
 import { getInjectionValues } from "../../engine/planner/injection_utils.ts";
 import DynamicInjection from "../../engine/injection/dynamic.ts";
 import { getLogger } from "../../log.ts";
+import { WorkerManager } from "./worker_manager.ts";
 
 const logger = getLogger(import.meta);
 
@@ -37,7 +37,8 @@
   allow: () => "ALLOW" as PolicyResolverOutput,
   deny: () => "DENY" as PolicyResolverOutput,
   pass: () => "PASS" as PolicyResolverOutput,
-  internal_policy: ({ _: { context } }) => context.provider === "internal" ? "ALLOW" : "PASS" as PolicyResolverOutput,
+  internal_policy: ({ _: { context } }) =>
+    context.provider === "internal" ? "ALLOW" : "PASS" as PolicyResolverOutput,
 };
 
 export class DenoRuntime extends Runtime {
@@ -46,8 +47,7 @@
     uuid: string,
     private tg: TypeGraphDS,
     private typegate: Typegate,
-    private w: DenoMessenger,
-    private registry: Map<string, number>,
+    private workerManager: WorkerManager,
     private secrets: Record<string, string>,
   ) {
     super(typegraphName, uuid);
@@ -138,36 +138,24 @@
       }
     }
 
-    const w = new DenoMessenger(
-      name,
-      {
-        ...(args.permissions ?? {}),
-        read: [basePath],
-      } as Deno.PermissionOptionsObject,
-      false,
-      ops,
-      typegate.config.base,
-    );
-
-    if (Deno.env.get("DENO_TESTING") === "true") {
-      w.disableLazyness();
-    }
+    const workerManager = new WorkerManager({
+      timeout_ms: typegate.config.base.timer_max_timeout_ms,
+    });
 
     const rt = new DenoRuntime(
       typegraphName,
       uuid,
       tg,
       typegate,
-      w,
-      registry,
+      workerManager,
       secrets,
     );
 
     return rt;
   }
 
   async deinit(): Promise<void> {
-    await this.w.terminate();
+    // await this.workerManager.deinit();
   }
 
   materialize(
@@ -257,7 +245,10 @@
       const modMat = this.tg.materializers[mat.data.mod as number];
       const entryPoint =
         this.tg.meta.artifacts[modMat.data.entryPoint as string];
-      const op = this.registry.get(entryPoint.hash)!;
+      const depMetas = (modMat.data.deps as string[]).map((dep) =>
+        createArtifactMeta(this.typegraphName, this.tg.meta.artifacts[dep])
+      );
+      const moduleMeta = createArtifactMeta(this.typegraphName, entryPoint);
 
       return async ({
         _: {
@@ -269,33 +260,33 @@
       }) => {
         const token = await InternalAuth.emit(this.typegate.cryptoKeys);
 
-        return await this.w.execute(
-          op,
+        // TODO cache??
+        const entryModulePath = await this.typegate.artifactStore.getLocalPath(
+          moduleMeta,
+          depMetas,
+        );
+
+        return await this.workerManager.callFunction(
+          mat.data.name as string,
+          entryModulePath,
+          entryPoint.path,
+          args,
           {
-            type: "import_func",
-            args,
-            internals: {
-              parent,
-              context,
-              secrets,
-              effect: mat.effect.effect ?? null,
-              meta: {
-                url: `${url.protocol}//${url.host}/${this.typegraphName}`,
-                token,
-              },
-              headers,
+            parent,
+            context,
+            secrets,
+            effect: mat.effect.effect ?? null,
+            meta: {
+              url: `${url.protocol}//${url.host}/${this.typegraphName}`,
+              token,
             },
-            name: mat.data.name as string,
-            verbose,
+            headers,
           },
-          [],
-          pulseCount,
         );
       };
     }
 
     if (mat.name === "function") {
-      const op = this.registry.get(mat.data.script as string)!;
       return async ({
         _: {
           context,
@@ -306,26 +297,29 @@
       }) => {
         const token = await InternalAuth.emit(this.typegate.cryptoKeys);
 
-        return await this.w.execute(
-          op,
+        const modulePath = await this.typegate.artifactStore.getInlineArtifact(
+          this.typegraphName,
+          mat.data.script as string,
+          ".ts",
+          exportInlineFunction("inlineFunction"),
+        );
+
+        return await this.workerManager.callFunction(
+          "inlineFunction",
+          modulePath,
+          "tg",
+          args,
           {
-            type: "func",
-            args,
-            internals: {
-              parent,
-              context,
-              secrets,
-              effect: mat.effect.effect ?? null,
-              meta: {
-                url: `${url.protocol}//${url.host}/${this.typegraphName}`,
-                token,
-              },
-              headers,
+            parent,
+            context,
+            secrets,
+            effect: mat.effect.effect ?? null,
+            meta: {
+              url: `${url.protocol}//${url.host}/${this.typegraphName}`,
+              token,
             },
-            verbose,
+            headers,
           },
-          [],
-          pulseCount,
         );
       };
     }
@@ -365,6 +359,18 @@
   }
 }
 
+function exportInlineFunction(name = "fn", symbol = "_my_lambda") {
+  if (!name.match(/^[a-zA-Z_][a-zA-Z0-9_]*$/)) {
+    throw new Error(`Invalid identifier: ${name}`);
+  }
+  if (!symbol.match(/^[a-zA-Z_][a-zA-Z0-9_]*$/)) {
+    throw new Error(`Invalid identifier: ${symbol}`);
+  }
+  return (code: string) => {
+    return `${code}\nexport const ${name} = ${symbol};`;
+  };
+}
+
 function getInjectionData(d: InjectionData) {
   if ("value" in d) {
     return d.value;

src/typegate/src/runtimes/deno/types.ts

@@ -0,0 +1,21 @@
+// Copyright Metatype OÜ, licensed under the Mozilla Public License Version 2.0.
+// SPDX-License-Identifier: MPL-2.0
+
+import { TaskContext } from "./shared_types.ts";
+
+export type TaskSpec = {
+  modulePath: string;
+  functionName: string;
+};
+
+export type DenoMessage = {
+  type: "CALL";
+  modulePath: string;
+  functionName: string;
+  args: unknown;
+  internals: TaskContext;
+};
+
+export type DenoEvent =
+  | { type: "SUCCESS"; result: unknown }
+  | { type: "FAILURE"; error: string; exception: Error | undefined };