feat: Unify worker manager (#954)

<!--
Pull requests are squashed and merged using:
- their title as the commit message
- their description as the commit body

Having a good title and description is important for the users to get
readable changelog.
-->

<!-- 1. Explain WHAT the change is about -->

- Solves
[MET-667](https://linear.app/metatypedev/issue/MET-667/gate-unify-the-worker-manager-between-workflows-and-runtime)
  - [x] `BaseWorkerManager`
  - [x] Use in Deno runtime
  - [ ] ~Use in Python runtime~ _(followup PR)_
  - [ ] ~Use in Rust runtime~ _(followup PR)_
  - [ ] ~Worker pooling~ _(followup PR)_

<!-- 2. Explain WHY the change cannot be made simpler -->



<!-- 3. Explain HOW users should update their code -->

#### Migration notes

---

- [ ] The change comes with new or modified tests
- [ ] Hard-to-understand functions have explanatory comments
- [ ] End-user documentation is updated to reflect the change


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

## Summary by CodeRabbit

Based on the comprehensive summary, here are the updated release notes:

- **New Features**
- Enhanced worker management system with improved task tracking and
execution.
- Introduced new `WorkerManager` for more robust Deno runtime
operations.
  - Added support for inline artifact generation and management.
- New asynchronous method `getInlineArtifact` in the `ArtifactStore`
class.

- **Improvements**
- Streamlined messaging and event handling across different runtime
components.
  - Improved error reporting and task lifecycle management.
  - Refined type definitions for better type safety.

- **Breaking Changes**
  - Removed `DenoMessenger` and `LazyAsyncMessenger` classes.
  - Restructured workflow event and message handling.
  - Updated task ID generation mechanism.

- **Performance**
  - Optimized worker initialization and task execution.
  - Introduced more efficient task tracking and resource management.

- **Bug Fixes**
  - Improved error handling in worker and runtime environments.
  - Enhanced message communication between workers and main thread.
  - Removed outdated test cases to focus on relevant functionality.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: Natoandro

deno.lock

@@ -18,7 +18,6 @@ import type {
 } 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 { PolicyResolverOutput } from "../../engine/planner/policies.ts";
 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 @@ const predefinedFuncs: Record<string, Resolver<Record<string, unknown>>> = {
   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 @@ export class DenoRuntime extends Runtime {
     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 @@ export class DenoRuntime extends Runtime {
       }
     }
 
-    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 @@ export class DenoRuntime extends Runtime {
       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 @@ export class DenoRuntime extends Runtime {
       }) => {
         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 @@ export class DenoRuntime extends Runtime {
       }) => {
         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 @@ export class DenoRuntime extends Runtime {
   }
 }
 
+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/deno.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 };