Refactor all the things

plugos/hooks/mq.ts

@@ -1,8 +1,8 @@
 import { Hook, Manifest } from "../types.ts";
 import { System } from "../system.ts";
-import { DexieMQ } from "../lib/mq.dexie.ts";
 import { fullQueueName } from "../lib/mq_util.ts";
-import { Message } from "$sb/types.ts";
+import { MQMessage } from "$sb/types.ts";
+import { MessageQueue } from "../lib/mq.ts";
 
 type MQSubscription = {
   queue: string;
@@ -17,7 +17,7 @@ export type MQHookT = {
 export class MQHook implements Hook<MQHookT> {
   subscriptions: (() => void)[] = [];
 
-  constructor(private system: System<MQHookT>, readonly mq: DexieMQ) {
+  constructor(private system: System<MQHookT>, readonly mq: MessageQueue) {
   }
 
   apply(system: System<MQHookT>): void {
@@ -64,7 +64,7 @@ export class MQHook implements Hook<MQHookT> {
               {
                 batchSize: subscriptionDef.batchSize,
               },
-              async (messages: Message[]) => {
+              async (messages: MQMessage[]) => {
                 try {
                   await plug.invoke(name, [messages]);
                   if (subscriptionDef.autoAck) {

plugos/lib/kv_store.deno_kv.test.ts

@@ -2,8 +2,8 @@ import { assertEquals } from "../../test_deps.ts";
 import { DenoKVStore } from "./kv_store.deno_kv.ts";
 
 Deno.test("Test KV index", async () => {
-  const kv = new DenoKVStore();
-  await kv.init("test.db");
+  const denoKv = await Deno.openKv("test.db");
+  const kv = new DenoKVStore(denoKv);
 
   await kv.set("name", "Peter");
   assertEquals(await kv.get("name"), "Peter");
@@ -52,5 +52,6 @@ Deno.test("Test KV index", async () => {
   await kv.deletePrefix("");
   assertEquals(await kv.queryPrefix(""), []);
 
-  await kv.delete();
+  denoKv.close();
+  await Deno.remove("test.db");
 });

plugos/lib/kv_store.deno_kv.ts

@@ -5,23 +5,7 @@ import { KV, KVStore } from "./kv_store.ts";
 const kvBatchSize = 10;
 
 export class DenoKVStore implements KVStore {
-  kv!: Deno.Kv;
-  path: string | undefined;
-
-  async init(path?: string) {
-    this.path = path;
-    this.kv = await Deno.openKv(path);
-  }
-
-  close() {
-    this.kv.close();
-  }
-
-  async delete() {
-    this.kv.close();
-    if (this.path) {
-      await Deno.remove(this.path);
-    }
+  constructor(private kv: Deno.Kv) {
   }
 
   del(key: string): Promise<void> {

plugos/lib/mq.deno_kv.test.ts

@@ -0,0 +1,20 @@
+import { sleep } from "../../common/async_util.ts";
+import { DenoKvMQ } from "./mq.deno_kv.ts";
+
+Deno.test("Deno MQ", async () => {
+  const denoKv = await Deno.openKv("test.db");
+  const mq = new DenoKvMQ(denoKv);
+  const unsub = mq.subscribe("test", {}, (messages) => {
+    console.log("Received on test", messages);
+  });
+  const unsub2 = mq.subscribe("test2", {}, (messages) => {
+    console.log("Received on test2", messages);
+  });
+  await mq.send("test", "Hello World");
+  await mq.batchSend("test2", ["Hello World 2", "Hello World 3"]);
+
+  // Let's avoid a panic here
+  await sleep(20);
+  denoKv.close();
+  await Deno.remove("test.db");
+});

plugos/lib/mq.deno_kv.ts

@@ -0,0 +1,87 @@
+/// <reference lib="deno.unstable" />
+
+import {
+  MQMessage,
+  MQStats,
+  MQSubscribeOptions,
+} from "../../plug-api/types.ts";
+import { MessageQueue } from "./mq.ts";
+
+type QueuedMessage = [string, MQMessage];
+
+export class DenoKvMQ implements MessageQueue {
+  listeners: Map<string, Set<(messages: MQMessage[]) => void | Promise<void>>> =
+    new Map();
+
+  constructor(private kv: Deno.Kv) {
+    kv.listenQueue(async (message: unknown) => {
+      const [queue, body] = message as QueuedMessage;
+      const listeners = this.listeners.get(queue);
+      if (!listeners) {
+        return;
+      }
+      for (const listener of listeners) {
+        await Promise.resolve(listener([{ id: "_dummyid", queue, body }]));
+      }
+    });
+  }
+
+  // Dummy implementation
+  getQueueStats(_queue: string): Promise<MQStats> {
+    return Promise.resolve({
+      queued: 0,
+      processing: 0,
+      dlq: 0,
+    });
+  }
+
+  // Dummy implementation
+  getAllQueueStats(): Promise<Record<string, MQStats>> {
+    return Promise.resolve({});
+  }
+
+  async batchSend(queue: string, bodies: any[]): Promise<void> {
+    const results = await Promise.all(
+      bodies.map((body) => this.kv.enqueue([queue, body])),
+    );
+    for (const result of results) {
+      if (!result.ok) {
+        throw result;
+      }
+    }
+  }
+  async send(queue: string, body: any): Promise<void> {
+    const result = await this.kv.enqueue([queue, body]);
+    if (!result.ok) {
+      throw result;
+    }
+  }
+  subscribe(
+    queue: string,
+    _options: MQSubscribeOptions,
+    callback: (messages: MQMessage[]) => void | Promise<void>,
+  ): () => void {
+    const listeners = this.listeners.get(queue);
+    if (!listeners) {
+      this.listeners.set(queue, new Set([callback]));
+    } else {
+      listeners.add(callback);
+    }
+
+    return () => {
+      const listeners = this.listeners.get(queue);
+      if (!listeners) {
+        return;
+      }
+      listeners.delete(callback);
+    };
+  }
+  ack(_queue: string, _id: string): Promise<void> {
+    // Doesn't apply to this implementation
+    return Promise.resolve();
+  }
+  batchAck(_queue: string, _ids: string[]): Promise<void> {
+    // Doesn't apply to this implementation
+    return Promise.resolve();
+  }
+}

plugos/lib/mq.dexie.ts

@@ -1,18 +1,14 @@
 import Dexie, { Table } from "dexie";
-import { Message, QueueStats } from "$sb/types.ts";
+import { MQMessage, MQStats, MQSubscribeOptions } from "$sb/types.ts";
+import { MessageQueue } from "./mq.ts";
 
-export type ProcessingMessage = Message & {
+export type ProcessingMessage = MQMessage & {
   ts: number;
 };
 
-export type SubscribeOptions = {
-  batchSize?: number;
-  pollInterval?: number;
-};
-
-export class DexieMQ {
+export class DexieMQ implements MessageQueue {
   db: Dexie;
-  queued: Table<Message, [string, string]>;
+  queued: Table<MQMessage, [string, string]>;
   processing: Table<ProcessingMessage, [string, string]>;
   dlq: Table<ProcessingMessage, [string, string]>;
 
@@ -63,13 +59,15 @@ export class DexieMQ {
     return this.batchSend(queue, [body]);
   }
 
-  poll(queue: string, maxItems: number): Promise<Message[]> {
+  poll(queue: string, maxItems: number): Promise<MQMessage[]> {
     return this.db.transaction(
       "rw",
       [this.queued, this.processing],
       async (tx) => {
         const messages =
-          (await tx.table<Message, [string, string]>("queued").where({ queue })
+          (await tx.table<MQMessage, [string, string]>("queued").where({
+            queue,
+          })
             .sortBy("id")).slice(0, maxItems);
         const ids: [string, string][] = messages.map((m) => [queue, m.id]);
         await tx.table("queued").bulkDelete(ids);
@@ -93,8 +91,8 @@ export class DexieMQ {
    */
   subscribe(
     queue: string,
-    options: SubscribeOptions,
-    callback: (messages: Message[]) => Promise<void> | void,
+    options: MQSubscribeOptions,
+    callback: (messages: MQMessage[]) => Promise<void> | void,
   ): () => void {
     let running = true;
     let timeout: number | undefined;
@@ -219,7 +217,7 @@ export class DexieMQ {
     return this.dlq.clear();
   }
 
-  getQueueStats(queue: string): Promise<QueueStats> {
+  getQueueStats(queue: string): Promise<MQStats> {
     return this.db.transaction(
       "r",
       [this.queued, this.processing, this.dlq],
@@ -237,8 +235,8 @@ export class DexieMQ {
     );
   }
 
-  async getAllQueueStats(): Promise<Record<string, QueueStats>> {
-    const allStatus: Record<string, QueueStats> = {};
+  async getAllQueueStats(): Promise<Record<string, MQStats>> {
+    const allStatus: Record<string, MQStats> = {};
     await this.db.transaction(
       "r",
       [this.queued, this.processing, this.dlq],

plugos/lib/mq.ts

@@ -0,0 +1,16 @@
+import { MQMessage, MQStats, MQSubscribeOptions } from "$sb/types.ts";
+
+export interface MessageQueue {
+  batchSend(queue: string, bodies: any[]): Promise<void>;
+  send(queue: string, body: any): Promise<void>;
+  subscribe(
+    queue: string,
+    options: MQSubscribeOptions,
+    callback: (messages: MQMessage[]) => Promise<void> | void,
+  ): () => void;
+  ack(queue: string, id: string): Promise<void>;
+  batchAck(queue: string, ids: string[]): Promise<void>;
+
+  getQueueStats(queue: string): Promise<MQStats>;
+  getAllQueueStats(): Promise<Record<string, MQStats>>;
+}

plugos/syscalls/mq.dexie.ts

@@ -1,9 +1,9 @@
 import { SysCallMapping } from "../system.ts";
-import { DexieMQ } from "../lib/mq.dexie.ts";
 import { fullQueueName } from "../lib/mq_util.ts";
+import { MessageQueue } from "../lib/mq.ts";
 
 export function mqSyscalls(
-  mq: DexieMQ,
+  mq: MessageQueue,
 ): SysCallMapping {
   return {
     "mq.send": (ctx, queue: string, body: any) => {