Backporting a bunch of optimizations from db-only branch

plugos/hooks/event.ts

@@ -46,6 +46,7 @@ export class EventHook implements Hook<EventHookT> {
       throw new Error("Event hook is not initialized");
     }
     const responses: any[] = [];
+    const promises: Promise<void>[] = [];
     for (const plug of this.system.loadedPlugs.values()) {
       const manifest = plug.manifest;
       for (
@@ -60,16 +61,19 @@ export class EventHook implements Hook<EventHookT> {
             ) {
               // Only dispatch functions that can run in this environment
               if (await plug.canInvoke(name)) {
-                try {
-                  const result = await plug.invoke(name, args);
-                  if (result !== undefined) {
-                    responses.push(result);
+                // Queue the promise
+                promises.push((async () => {
+                  try {
+                    const result = await plug.invoke(name, args);
+                    if (result !== undefined) {
+                      responses.push(result);
+                    }
+                  } catch (e: any) {
+                    console.error(
+                      `Error dispatching event ${eventName} to plug ${plug.name}: ${e.message}`,
+                    );
                   }
-                } catch (e: any) {
-                  console.error(
-                    `Error dispatching event ${eventName} to plug ${plug.name}: ${e.message}`,
-                  );
-                }
+                })());
               }
             }
           }
@@ -79,13 +83,19 @@ export class EventHook implements Hook<EventHookT> {
     const localListeners = this.localListeners.get(eventName);
     if (localListeners) {
       for (const localListener of localListeners) {
-        const result = await Promise.resolve(localListener(...args));
-        if (result) {
-          responses.push(result);
-        }
+        // Queue the promise
+        promises.push((async () => {
+          const result = await Promise.resolve(localListener(...args));
+          if (result) {
+            responses.push(result);
+          }
+        })());
       }
     }
 
+    // Wait for all promises to resolve
+    await Promise.all(promises);
+
     return responses;
   }
 

plugos/hooks/mq.ts

@@ -3,6 +3,7 @@ import { System } from "../system.ts";
 import { fullQueueName } from "../lib/mq_util.ts";
 import { MQMessage } from "$sb/types.ts";
 import { MessageQueue } from "../lib/mq.ts";
+import { throttle } from "$sb/lib/async.ts";
 
 type MQSubscription = {
   queue: string;
@@ -24,14 +25,14 @@ export class MQHook implements Hook<MQHookT> {
     this.system = system;
     system.on({
       plugLoaded: () => {
-        this.reloadQueues();
+        this.throttledReloadQueues();
       },
       plugUnloaded: () => {
-        this.reloadQueues();
+        this.throttledReloadQueues();
       },
     });
 
-    this.reloadQueues();
+    this.throttledReloadQueues();
   }
 
   stop() {
@@ -40,6 +41,10 @@ export class MQHook implements Hook<MQHookT> {
     this.subscriptions = [];
   }
 
+  throttledReloadQueues = throttle(() => {
+    this.reloadQueues();
+  }, 1000);
+
   reloadQueues() {
     this.stop();
     for (const plug of this.system.loadedPlugs.values()) {

plugos/lib/datastore.test.ts

@@ -7,7 +7,7 @@ import { assertEquals } from "https://deno.land/std@0.165.0/testing/asserts.ts";
 import { PrefixedKvPrimitives } from "./prefixed_kv_primitives.ts";
 
 async function test(db: KvPrimitives) {
-  const datastore = new DataStore(new PrefixedKvPrimitives(db, ["ds"]), {
+  const datastore = new DataStore(new PrefixedKvPrimitives(db, ["ds"]), false, {
     count: (arr: any[]) => arr.length,
   });
   await datastore.set(["user", "peter"], { name: "Peter" });

plugos/lib/datastore.ts

@@ -2,14 +2,18 @@ import { applyQueryNoFilterKV, evalQueryExpression } from "$sb/lib/query.ts";
 import { FunctionMap, KV, KvKey, KvQuery } from "$sb/types.ts";
 import { builtinFunctions } from "$sb/lib/builtin_query_functions.ts";
 import { KvPrimitives } from "./kv_primitives.ts";
+import { LimitedMap } from "../../common/limited_map.ts";
 
 /**
  * This is the data store class you'll actually want to use, wrapping the primitives
  * in a more user-friendly way
  */
 export class DataStore {
+  private cache = new LimitedMap<any>(20);
+
   constructor(
     readonly kv: KvPrimitives,
+    private enableCache = false,
     private functionMap: FunctionMap = builtinFunctions,
   ) {
   }
@@ -50,6 +54,21 @@ export class DataStore {
   }
 
   async query<T = any>(query: KvQuery): Promise<KV<T>[]> {
+    let cacheKey: string | undefined;
+    const cacheSecs = query.cacheSecs;
+    // Should we do caching?
+    if (cacheSecs && this.enableCache) {
+      // Remove the cacheSecs from the query
+      query = { ...query, cacheSecs: undefined };
+      console.log("Going to cache query", query);
+      cacheKey = JSON.stringify(query);
+      const cachedResult = this.cache.get(cacheKey);
+      if (cachedResult) {
+        // Let's use the cached result
+        return cachedResult;
+      }
+    }
+
     const results: KV<T>[] = [];
     let itemCount = 0;
     // Accumulate results
@@ -76,7 +95,12 @@ export class DataStore {
       }
     }
     // Apply order by, limit, and select
-    return applyQueryNoFilterKV(query, results, this.functionMap);
+    const finalResult = applyQueryNoFilterKV(query, results, this.functionMap);
+    if (cacheKey) {
+      // Store in the cache
+      this.cache.set(cacheKey, finalResult, cacheSecs! * 1000);
+    }
+    return finalResult;
   }
 
   async queryDelete(query: KvQuery): Promise<void> {

plugos/lib/mq.datastore.ts

@@ -23,6 +23,9 @@ export class DataStoreMQ implements MessageQueue {
   seq = 0;
 
   async batchSend(queue: string, bodies: any[]): Promise<void> {
+    if (bodies.length === 0) {
+      return;
+    }
     const messages: KV<MQMessage>[] = bodies.map((body) => {
       const id = `${Date.now()}-${String(++this.seq).padStart(6, "0")}`;
       const key = [...queuedPrefix, queue, id];
@@ -54,6 +57,9 @@ export class DataStoreMQ implements MessageQueue {
       prefix: [...queuedPrefix, queue],
       limit: ["number", maxItems],
     });
+    if (messages.length === 0) {
+      return [];
+    }
     // Put them in the processing queue
     await this.ds.batchSet(
       messages.map((m) => ({
@@ -137,6 +143,9 @@ export class DataStoreMQ implements MessageQueue {
   }
 
   async batchAck(queue: string, ids: string[]) {
+    if (ids.length === 0) {
+      return;
+    }
     await this.ds.batchDelete(
       ids.map((id) => [...processingPrefix, queue, id]),
     );
@@ -152,6 +161,9 @@ export class DataStoreMQ implements MessageQueue {
       prefix: processingPrefix,
       filter: ["<", ["attr", "ts"], ["number", now - timeout]],
     });
+    if (messages.length === 0) {
+      return;
+    }
     await this.ds.batchDelete(messages.map((m) => m.key));
     const newMessages: KV<ProcessingMessage>[] = [];
     for (const { value: m } of messages) {