Cleanup and progress

plugbox/bin/plugbox-bundle.ts

@@ -6,9 +6,9 @@ import path from "path";
 
 import yargs from "yargs";
 import { hideBin } from "yargs/helpers";
-import {Manifest} from "../types";
+import { Manifest } from "../types";
 
-async function compile(filePath : string, functionName : string, debug: boolean) {
+async function compile(filePath: string, functionName: string, debug: boolean) {
   let outFile = "out.js";
 
   let inFile = filePath;
@@ -47,7 +47,9 @@ export default ${functionName};`
 
 async function bundle(manifestPath: string, sourceMaps: boolean) {
   const rootPath = path.dirname(manifestPath);
-  const manifest = JSON.parse((await readFile(manifestPath)).toString()) as Manifest<any>;
+  const manifest = JSON.parse(
+    (await readFile(manifestPath)).toString()
+  ) as Manifest<any>;
 
   for (let [name, def] of Object.entries(manifest.functions)) {
     let jsFunctionName = "default",
@@ -69,7 +71,10 @@ async function run() {
     .parse();
 
   let generatedManifest = await bundle(args._[0] as string, !!args.debug);
-  await writeFile(args._[1] as string, JSON.stringify(generatedManifest, null, 2));
+  await writeFile(
+    args._[1] as string,
+    JSON.stringify(generatedManifest, null, 2)
+  );
 }
 
 run().catch((e) => {

plugbox/cron.ts

@@ -0,0 +1,11 @@
+import { System } from "./runtime";
+import { CronHook } from "./types";
+import cron from "node-cron";
+
+export function cronSystem(system: System<CronHook>) {
+  let task = cron.schedule("* * * * *", () => {
+
+  });
+  // @ts-ignore
+  task.destroy();
+}

plugbox/endpoint.test.ts

@@ -0,0 +1,47 @@
+import { createSandbox } from "./node_sandbox";
+import { System } from "./runtime";
+import { test, expect } from "@jest/globals";
+import { EndPointDef, EndpointHook, Manifest } from "./types";
+import express from "express";
+import request from "supertest";
+import { exposeSystem } from "./endpoints";
+
+test("Run a plugbox endpoint server", async () => {
+  let system = new System<EndpointHook>();
+  let plug = await system.load(
+    "test",
+    {
+      functions: {
+        testhandler: {
+          code: `(() => {
+          return {
+            default: (req) => {
+              console.log("Req", req);
+              return {status: 200, body: [1, 2, 3], headers: {"Content-type": "application/json"}};
+            }
+          };
+        })()`,
+        },
+      },
+      hooks: {
+        endpoints: [{ method: "GET", path: "/", handler: "testhandler" }],
+      },
+    } as Manifest<EndpointHook>,
+    createSandbox(system)
+  );
+
+  const app = express();
+  const port = 3123;
+  app.use(exposeSystem(system));
+  let server = app.listen(port, () => {
+    console.log(`Listening on port ${port}`);
+  });
+  let resp = await request(app)
+    .get("/_/test/?name=Pete")
+    .expect((resp) => {
+      expect(resp.status).toBe(200);
+      expect(resp.header["content-type"]).toContain("application/json");
+      expect(resp.text).toBe(JSON.stringify([1, 2, 3]));
+    });
+  server.close();
+});

plugbox/endpoints.ts

@@ -0,0 +1,85 @@
+import { System } from "./runtime";
+import { EndpointHook } from "./types";
+import express from "express";
+
+export type EndpointRequest = {
+  method: string;
+  path: string;
+  query: { [key: string]: string };
+  headers: { [key: string]: string };
+  body: any;
+};
+
+export type EndpointResponse = {
+  status: number;
+  headers?: { [key: string]: string };
+  body: any;
+};
+
+const endPointPrefix = "/_";
+
+export function exposeSystem(system: System<EndpointHook>) {
+  return (
+    req: express.Request,
+    res: express.Response,
+    next: express.NextFunction
+  ) => {
+    if (!req.path.startsWith(endPointPrefix)) {
+      return next();
+    }
+    Promise.resolve()
+      .then(async () => {
+        for (const [plugName, plug] of system.loadedPlugs.entries()) {
+          const manifest = plug.manifest;
+          if (!manifest) {
+            continue;
+          }
+          const endpoints = manifest.hooks?.endpoints;
+          if (endpoints) {
+            let prefix = `${endPointPrefix}/${plugName}`;
+            if (!req.path.startsWith(prefix)) {
+              continue;
+            }
+            for (const { path, method, handler } of endpoints) {
+              let prefixedPath = `${prefix}${path}`;
+              if (prefixedPath === req.path && method === req.method) {
+                try {
+                  const response: EndpointResponse = await plug.invoke(
+                    handler,
+                    [
+                      {
+                        path: req.path,
+                        method: req.method,
+                        body: req.body,
+                        query: req.query,
+                        headers: req.headers,
+                      } as EndpointRequest,
+                    ]
+                  );
+                  let resp = res.status(response.status);
+                  if (response.headers) {
+                    for (const [key, value] of Object.entries(
+                      response.headers
+                    )) {
+                      resp = resp.header(key, value);
+                    }
+                  }
+                  resp.send(response.body);
+                  return;
+                } catch (e: any) {
+                  console.error("Error executing function", e);
+                  res.status(500).send(e.message);
+                  return;
+                }
+              }
+            }
+          }
+        }
+        next();
+      })
+      .catch((e) => {
+        console.error(e);
+        next(e);
+      });
+  };
+}

plugbox/iframe_sandbox.ts

@@ -8,6 +8,7 @@ import sandboxHtml from "bundle-text:./iframe_sandbox.html";
 class IFrameWrapper implements WorkerLike {
   private iframe: HTMLIFrameElement;
   onMessage?: (message: any) => Promise<void>;
+  ready: Promise<void>;
 
   constructor() {
     const iframe = document.createElement("iframe", {});
@@ -27,6 +28,12 @@ class IFrameWrapper implements WorkerLike {
       });
     });
     document.body.appendChild(iframe);
+    this.ready = new Promise((resolve) => {
+      iframe.onload = () => {
+        resolve();
+        iframe.onload = null;
+      };
+    });
   }
 
   postMessage(message: any): void {

plugbox/node_sandbox.ts

@@ -6,14 +6,12 @@ import * as fs from "fs";
 import { safeRun } from "./util";
 
 // @ts-ignore
-import workerCode from "bundle-text:./node_worker.ts"
-
-// ParcelJS will simply inline this into the bundle.
-// const workerCode = fs.readFileSync(__dirname + "/node_worker.ts", "utf-8");
+import workerCode from "bundle-text:./node_worker.ts";
 
 class NodeWorkerWrapper implements WorkerLike {
   onMessage?: (message: any) => Promise<void>;
   private worker: Worker;
+  ready: Promise<void>;
 
   constructor(worker: Worker) {
     this.worker = worker;
@@ -22,6 +20,9 @@ class NodeWorkerWrapper implements WorkerLike {
         await this.onMessage!(message);
       });
     });
+    this.ready = new Promise((resolve) => {
+      worker.once("online", resolve);
+    });
   }
 
   postMessage(message: any): void {
@@ -34,6 +35,9 @@ class NodeWorkerWrapper implements WorkerLike {
 }
 
 export function createSandbox(system: System<any>) {
+  let worker = new Worker(workerCode, {
+    eval: true,
+  });
   return new Sandbox(
     system,
     new NodeWorkerWrapper(

plugbox/node_worker.ts

@@ -3,18 +3,18 @@ const { parentPort } = require("worker_threads");
 
 let loadedFunctions = new Map<string, Function>();
 let pendingRequests = new Map<
-    number,
-    {
-      resolve: (result: unknown) => void;
-      reject: (e: any) => void;
-    }
-    >();
+  number,
+  {
+    resolve: (result: unknown) => void;
+    reject: (e: any) => void;
+  }
+>();
 
 let vm = new VM({
   sandbox: {
     console: console,
     self: {
-      syscall: (reqId : number, name : string, args: any[]) => {
+      syscall: (reqId: number, name: string, args: any[]) => {
         return new Promise((resolve, reject) => {
           pendingRequests.set(reqId, { resolve, reject });
           parentPort.postMessage({
@@ -30,17 +30,17 @@ let vm = new VM({
   },
 });
 
-function wrapScript(code : string) {
+function wrapScript(code: string) {
   return `(${code})["default"]`;
 }
 
-function safeRun(fn : any) {
-  fn().catch((e : any) => {
+function safeRun(fn: any) {
+  fn().catch((e: any) => {
     console.error(e);
   });
 }
 
-parentPort.on("message", (data : any) => {
+parentPort.on("message", (data: any) => {
   safeRun(async () => {
     switch (data.type) {
       case "load":
@@ -62,9 +62,10 @@ parentPort.on("message", (data : any) => {
           parentPort.postMessage({
             type: "result",
             id: data.id,
-            result: result,
+            // TOOD: Figure out if this is necessary, because it's expensive
+            result: result && JSON.parse(JSON.stringify(result)),
           });
-        } catch (e : any) {
+        } catch (e: any) {
           // console.log("ERROR", e);
           parentPort.postMessage({
             type: "result",

plugbox/plug_loader.ts

@@ -0,0 +1,71 @@
+import fs, { stat, watch } from "fs/promises";
+import path from "path";
+import { createSandbox } from "./node_sandbox";
+import { System } from "./runtime";
+import { safeRun } from "../server/util";
+
+function extractPlugName(localPath: string): string {
+  const baseName = path.basename(localPath);
+  return baseName.substring(0, baseName.length - ".plug.json".length);
+}
+
+export class DiskPlugLoader<HookT> {
+  private system: System<HookT>;
+  private plugPath: string;
+
+  constructor(system: System<HookT>, plugPath: string) {
+    this.system = system;
+    this.plugPath = plugPath;
+  }
+
+  watcher() {
+    safeRun(async () => {
+      for await (const { filename, eventType } of watch(this.plugPath, {
+        recursive: true,
+      })) {
+        if (!filename.endsWith(".plug.json")) {
+          return;
+        }
+        try {
+          let localPath = path.join(this.plugPath, filename);
+          const plugName = extractPlugName(localPath);
+          try {
+            await fs.stat(localPath);
+          } catch (e) {
+            // Likely removed
+            await this.system.unload(plugName);
+            this.system.emit("plugRemoved", plugName);
+          }
+          const plugDef = await this.loadPlugFromFile(localPath);
+          this.system.emit("plugUpdated", plugName, plugDef);
+        } catch {
+          // ignore, error handled by loadPlug
+        }
+      }
+    });
+  }
+
+  private async loadPlugFromFile(localPath: string) {
+    const plug = await fs.readFile(localPath, "utf8");
+    const plugName = extractPlugName(localPath);
+
+    console.log("Now loading plug", plugName);
+    try {
+      const plugDef = JSON.parse(plug);
+      await this.system.load(plugName, plugDef, createSandbox(this.system));
+      return plugDef;
+    } catch (e) {
+      console.error("Could not parse plugin file", e);
+      throw e;
+    }
+  }
+
+  async loadPlugs() {
+    for (let filename of await fs.readdir(this.plugPath)) {
+      if (filename.endsWith(".plug.json")) {
+        let localPath = path.join(this.plugPath, filename);
+        await this.loadPlugFromFile(localPath);
+      }
+    }
+  }
+}

plugbox/runtime.test.ts

@@ -75,5 +75,5 @@ test("Run a Node sandbox", async () => {
   } catch (e: any) {
     expect(e.message).toBe("#fail");
   }
-  await system.stop();
+  await system.unloadAll();
 });

plugbox/runtime.ts

@@ -4,6 +4,7 @@ import {
   WorkerLike,
   WorkerMessage,
 } from "./types";
+import { EventEmitter } from "../common/event";
 
 interface SysCallMapping {
   [key: string]: (...args: any) => Promise<any> | any;
@@ -31,6 +32,7 @@ export class Sandbox {
   }
 
   async load(name: string, code: string): Promise<void> {
+    await this.worker.ready;
     this.worker.postMessage({
       type: "load",
       name: name,
@@ -119,18 +121,20 @@ export class Plug<HookT> {
       if (!funDef) {
         throw new Error(`Function ${name} not found in manifest`);
       }
-      await this.sandbox.load(name, this.manifest!.functions[name].code!);
+      await this.sandbox.load(name, funDef.code!);
     }
     return await this.sandbox.invoke(name, args);
   }
 
   async dispatchEvent(name: string, data?: any): Promise<any[]> {
+    if (!this.manifest!.hooks?.events) {
+      return [];
+    }
     let functionsToSpawn = this.manifest!.hooks.events[name];
     if (functionsToSpawn) {
       return await Promise.all(
-        functionsToSpawn.map(
-          async (functionToSpawn: string) =>
-            await this.invoke(functionToSpawn, [data])
+        functionsToSpawn.map((functionToSpawn: string) =>
+          this.invoke(functionToSpawn, [data])
         )
       );
     } else {
@@ -143,10 +147,21 @@ export class Plug<HookT> {
   }
 }
 
-export class System<HookT> {
+export type SystemJSON<HookT> = { [key: string]: Manifest<HookT> };
+
+export type SystemEvents<HookT> = {
+  plugUpdated: (name: string, plug: Plug<HookT>) => void;
+  plugRemoved: (name: string) => void;
+};
+
+export class System<HookT> extends EventEmitter<SystemEvents<HookT>> {
   protected plugs = new Map<string, Plug<HookT>>();
   registeredSyscalls: SysCallMapping = {};
 
+  constructor() {
+    super();
+  }
+
   registerSyscalls(...registrationObjects: SysCallMapping[]) {
     for (const registrationObject of registrationObjects) {
       for (let p in registrationObject) {
@@ -171,23 +186,63 @@ export class System<HookT> {
     manifest: Manifest<HookT>,
     sandbox: Sandbox
   ): Promise<Plug<HookT>> {
+    if (this.plugs.has(name)) {
+      await this.unload(name);
+    }
     const plug = new Plug(this, name, sandbox);
     await plug.load(manifest);
     this.plugs.set(name, plug);
     return plug;
   }
 
+  async unload(name: string) {
+    const plug = this.plugs.get(name);
+    if (!plug) {
+      throw Error(`Plug ${name} not found`);
+    }
+    await plug.stop();
+    this.plugs.delete(name);
+  }
+
   async dispatchEvent(name: string, data?: any): Promise<any[]> {
     let promises = [];
     for (let plug of this.plugs.values()) {
-      promises.push(plug.dispatchEvent(name, data));
+      for (let result of await plug.dispatchEvent(name, data)) {
+        promises.push(result);
+      }
     }
     return await Promise.all(promises);
   }
 
-  async stop(): Promise<void[]> {
+  get loadedPlugs(): Map<string, Plug<HookT>> {
+    return this.plugs;
+  }
+
+  toJSON(): SystemJSON<HookT> {
+    let plugJSON: { [key: string]: Manifest<HookT> } = {};
+    for (let [name, plug] of this.plugs) {
+      if (!plug.manifest) {
+        continue;
+      }
+      plugJSON[name] = plug.manifest;
+    }
+    return plugJSON;
+  }
+
+  async replaceAllFromJSON(
+    json: SystemJSON<HookT>,
+    sandboxFactory: () => Sandbox
+  ) {
+    await this.unloadAll();
+    for (let [name, manifest] of Object.entries(json)) {
+      console.log("Loading plug", name);
+      await this.load(name, manifest, sandboxFactory());
+    }
+  }
+
+  async unloadAll(): Promise<void[]> {
     return Promise.all(
-      Array.from(this.plugs.values()).map((plug) => plug.stop())
+      Array.from(this.plugs.keys()).map(this.unload.bind(this))
     );
   }
 }

plugbox/sandbox_worker.ts

@@ -17,9 +17,7 @@ declare global {
 let postMessage = self.postMessage.bind(self);
 
 if (window.parent !== window) {
-  console.log("running in an iframe");
   postMessage = window.parent.postMessage.bind(window.parent);
-  // postMessage({ type: "test" }, "*");
 }
 
 self.syscall = async (id: number, name: string, args: any[]) => {
@@ -43,7 +41,6 @@ return fn["default"].apply(null, arguments);`;
 }
 
 self.addEventListener("message", (event: { data: WorkerMessage }) => {
-  // console.log("Got a message", event.data);
   safeRun(async () => {
     let messageEvent = event;
     let data = messageEvent.data;

plugbox/types.ts

@@ -1,7 +1,3 @@
-export type EventHook = {
-  events: { [key: string]: string[] };
-};
-
 export type WorkerMessageType = "load" | "invoke" | "syscall-response";
 
 export type WorkerMessage = {
@@ -37,7 +33,30 @@ export interface FunctionDef {
   code?: string;
 }
 
+export type EventHook = {
+  events?: { [key: string]: string[] };
+};
+
+export type EndpointHook = {
+  endpoints?: EndPointDef[];
+};
+export type EndPointDef = {
+  method: "GET" | "POST" | "PUT" | "DELETE" | "HEAD" | "OPTIONS";
+  path: string;
+  handler: string; // function name
+};
+
+export type CronHook = {
+  crons?: CronDef[];
+};
+
+export type CronDef = {
+  cron: string;
+  handler: string; // function name
+};
+
 export interface WorkerLike {
+  ready: Promise<void>;
   onMessage?: (message: any) => Promise<void>;
   postMessage(message: any): void;
   terminate(): void;

plugbox/webworker_sandbox.ts

@@ -5,6 +5,7 @@ import { safeRun } from "./util";
 class WebWorkerWrapper implements WorkerLike {
   private worker: Worker;
   onMessage?: (message: any) => Promise<void>;
+  ready: Promise<void>;
 
   constructor(worker: Worker) {
     this.worker = worker;
@@ -15,6 +16,7 @@ class WebWorkerWrapper implements WorkerLike {
         await this.onMessage!(data);
       });
     });
+    this.ready = Promise.resolve();
   }
   postMessage(message: any): void {
     this.worker.postMessage(message);