Major backend refactor (#599)

Backend refactor

plugos/lib/dynamodb_kv_primitives.ts

@@ -0,0 +1,73 @@
+import { KV, KvKey } from "../../plug-api/types.ts";
+import { KvPrimitives, KvQueryOptions } from "./kv_primitives.ts";
+import {
+  createClient,
+  DynamoDBClient,
+} from "https://denopkg.com/chiefbiiko/dynamodb@master/mod.ts";
+
+export type AwsOptions = {
+  accessKey: string;
+  secretKey: string;
+  region: string;
+};
+
+const keySeparator = "\0";
+
+const batchReadSize = 100;
+
+/**
+ * Start of an implementation, to be continued at some point
+ */
+
+export class DynamoDBKvPrimitives implements KvPrimitives {
+  client: DynamoDBClient;
+  partitionKey: string;
+  tableName: string;
+
+  constructor(tableName: string, partitionKey: string, options: AwsOptions) {
+    this.tableName = tableName;
+    this.partitionKey = partitionKey;
+    this.client = createClient({
+      credentials: {
+        accessKeyId: options.accessKey,
+        secretAccessKey: options.secretKey,
+      },
+      region: options.region,
+    });
+  }
+
+  batchGet(keys: KvKey[]): Promise<any[]> {
+    const allResults: any[] = [];
+    const promises: Promise<any>[] = [];
+    for (let i = 0; i < keys.length; i += batchReadSize) {
+      const batch = keys.slice(i, i + batchReadSize);
+      promises.push(
+        this.client.batchGetItem(
+          {
+            RequestItems: {
+              [this.tableName]: {
+                Keys: batch.map((key) => ({
+                  pk: this.partitionKey,
+                  sk: key.join(keySeparator),
+                })),
+              },
+            },
+          },
+        ),
+      );
+    }
+    throw new Error("Method not implemented.");
+  }
+  batchSet(entries: KV[]): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
+  batchDelete(keys: KvKey[]): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
+  query(options: KvQueryOptions): AsyncIterableIterator<KV> {
+    throw new Error("Method not implemented.");
+  }
+  close(): void {
+    throw new Error("Method not implemented.");
+  }
+}

plugos/lib/memory_kv_primitives.ts

@@ -43,12 +43,16 @@ export class MemoryKvPrimitives implements KvPrimitives {
   }
 
   async *query(options: KvQueryOptions): AsyncIterableIterator<KV> {
-    const prefix = options.prefix?.join("/");
-    for (const [key, value] of this.store) {
+    const prefix = options.prefix?.join(memoryKeySeparator);
+    const sortedKeys = [...this.store.keys()].sort();
+    for (const key of sortedKeys) {
       if (prefix && !key.startsWith(prefix)) {
         continue;
       }
-      yield { key: key.split(memoryKeySeparator), value };
+      yield {
+        key: key.split(memoryKeySeparator),
+        value: this.store.get(key),
+      };
     }
   }
 

plugos/sandbox.ts

@@ -62,19 +62,23 @@ export class Sandbox<HookT> {
   }
 
   async onMessage(data: ControllerMessage) {
+    if (!this.worker) {
+      console.warn("Received message for terminated worker, ignoring");
+      return;
+    }
     switch (data.type) {
       case "sys":
         try {
           const result = await this.plug.syscall(data.name!, data.args!);
 
-          this.worker!.postMessage({
+          this.worker && this.worker!.postMessage({
             type: "sysr",
             id: data.id,
             result: result,
           } as WorkerMessage);
         } catch (e: any) {
           // console.error("Syscall fail", e);
-          this.worker!.postMessage({
+          this.worker && this.worker!.postMessage({
             type: "sysr",
             id: data.id,
             error: e.message,

plugos/system.ts

@@ -139,7 +139,6 @@ export class System<HookT> extends EventEmitter<SystemEvents<HookT>> {
   }
 
   unload(name: string) {
-    // console.log("Unloading", name);
     const plug = this.plugs.get(name);
     if (!plug) {
       return;