enqueueMany() method for batch message operations

Author: dahlia

README.md

@@ -68,6 +68,9 @@ Changelog
 
 To be released.
 
+ -  Added `PostgresMessageQueue.enqueueMany()` method for efficiently enqueuing
+    multiple messages at once.
+
  -  Added some logging using [LogTape] for the sake of debugging.  The following
     categories are used:
 

deno.json

@@ -9,8 +9,8 @@
   },
   "imports": {
     "@deno/dnt": "jsr:@deno/dnt@^0.41.3",
-    "@fedify/fedify": "jsr:@fedify/fedify@^1.0.0",
-    "@logtape/logtape": "jsr:@logtape/logtape@^0.8.0",
+    "@fedify/fedify": "jsr:@fedify/fedify@1.5.0-dev.732",
+    "@logtape/logtape": "jsr:@logtape/logtape@^0.9.0",
     "@std/assert": "jsr:@std/assert@^0.226.0",
     "@std/async": "jsr:@std/async@^1.0.5",
     "postgres": "npm:postgres@^3.4.5"

deno.lock

@@ -56,6 +56,37 @@ Deno.test("PostgresMessageQueue", async (t) => {
     assertGreater(Date.now() - started, 3_000);
   });
 
+  await t.step("enqueueMany()", async () => {
+    while (messages.length > 0) messages.pop();
+    const batchMessages = [
+      "First batch message",
+      { text: "Second batch message" },
+      { text: "Third batch message", priority: "high" },
+    ];
+    await mq.enqueueMany(batchMessages);
+    await waitFor(() => messages.length === batchMessages.length, 15_000);
+    assertEquals(messages, batchMessages);
+  });
+
+  await t.step("enqueueMany() with delay", async () => {
+    while (messages.length > 0) messages.pop();
+    started = Date.now();
+    const delayedBatchMessages = [
+      "Delayed batch 1",
+      "Delayed batch 2",
+    ];
+    await mq.enqueueMany(
+      delayedBatchMessages,
+      { delay: Temporal.Duration.from({ seconds: 2 }) },
+    );
+    await waitFor(
+      () => messages.length === delayedBatchMessages.length,
+      15_000,
+    );
+    assertEquals(messages, delayedBatchMessages);
+    assertGreater(Date.now() - started, 2_000);
+  });
+
   controller.abort();
   await listening;
   await listening2;

src/mq.test.ts

@@ -111,6 +111,39 @@ export class PostgresMessageQueue implements MessageQueue {
     });
   }
 
+  async enqueueMany(
+    // deno-lint-ignore no-explicit-any
+    messages: any[],
+    options?: MessageQueueEnqueueOptions,
+  ): Promise<void> {
+    if (messages.length === 0) return;
+    await this.initialize();
+    const delay = options?.delay ?? Temporal.Duration.from({ seconds: 0 });
+    if (options?.delay) {
+      logger.debug("Enqueuing messages with a delay of {delay}...", {
+        delay,
+        messages,
+      });
+    } else {
+      logger.debug("Enqueuing messages...", { messages });
+    }
+    for (const message of messages) {
+      await this.#sql`
+        INSERT INTO ${this.#sql(this.#tableName)} (message, delay)
+        VALUES (
+          ${this.#json(message)},
+          ${delay.toString()}
+        );
+      `;
+    }
+    logger.debug("Enqueued messages.", { messages });
+    await this.#sql.notify(this.#channelName, delay.toString());
+    logger.debug("Notified the message queue channel {channelName}.", {
+      channelName: this.#channelName,
+      messages,
+    });
+  }
+
   async listen(
     // deno-lint-ignore no-explicit-any
     handler: (message: any) => void | Promise<void>,