Merge pull request #526 from 2chanhaeng/smq

Add SqliteMessageQueue and related tests

Author: dahlia

.github/workflows/main.yaml

@@ -53,7 +53,7 @@ jobs:
         - 6379:6379
     env:
       AMQP_URL: amqp://guest:guest@localhost:5672
-      DATABASE_URL: postgres://postgres:postgres@localhost:5432/postgres
+      POSTGRES_URL: postgres://postgres:postgres@localhost:5432/postgres
       REDIS_URL: redis://localhost:6379
     steps:
     - uses: actions/checkout@v4
@@ -116,7 +116,7 @@ jobs:
         - 6379:6379
     env:
       AMQP_URL: amqp://guest:guest@localhost:5672
-      DATABASE_URL: postgres://postgres:postgres@localhost:5432/postgres
+      POSTGRES_URL: postgres://postgres:postgres@localhost:5432/postgres
       REDIS_URL: redis://localhost:6379
     steps:
     - uses: actions/checkout@v4
@@ -157,7 +157,7 @@ jobs:
         - 6379:6379
     env:
       AMQP_URL: amqp://guest:guest@localhost:5672
-      DATABASE_URL: postgres://postgres:postgres@localhost:5432/postgres
+      POSTGRES_URL: postgres://postgres:postgres@localhost:5432/postgres
       REDIS_URL: redis://localhost:6379
     steps:
     - uses: actions/checkout@v4

.gitignore

@@ -7,3 +7,4 @@ package-lock.json
 repomix-output.xml
 t.ts
 t2.ts
+plan.md

.hongdown.toml

@@ -8,6 +8,7 @@ exclude = [
   "GEMINI.md",
   "WARP.md",
   "packages/fedify/src/cfworkers/**",
+  "plan.md",
 ]
 
 [heading]

CHANGES.md

@@ -275,6 +275,31 @@ To be released.
 
 [#437]: https://github.com/fedify-dev/fedify/issues/437
 
+### @fedify/sqlite
+
+ -  Added `SqliteMessageQueue` class implementing `MessageQueue` interface
+    using SQLite as the backing store.  This implementation uses polling to
+    check for new messages and is suitable for single-node deployments and
+    development environments.  [[#477], [#526] by ChanHaeng Lee]
+
+     -  Added `SqliteMessageQueue` class.
+     -  Added `SqliteMessageQueueOptions` interface.
+
+[#477]: https://github.com/fedify-dev/fedify/issues/477
+[#526]: https://github.com/fedify-dev/fedify/pull/526
+
+### @fedify/testing
+
+ -  Added `testMessageQueue()` utility function for standardized testing of
+    `MessageQueue` implementations.  This function provides a reusable test
+    harness that covers common message queue operations including `enqueue()`,
+    `enqueue()` with delay, `enqueueMany()`, and multiple listener scenarios.
+    [[#477], [#526] by ChanHaeng Lee]
+
+     -  Added `testMessageQueue()` function.
+     -  Added `waitFor()` helper function.
+     -  Added `getRandomKey()` helper function.
+
 
 Version 1.10.1
 --------------

docs/manual/mq.md

@@ -293,6 +293,115 @@ const federation = createFederation({
 [`AmqpMessageQueue`]: https://jsr.io/@fedify/amqp/doc/mq/~/AmqpMessageQueue
 [RabbitMQ]: https://www.rabbitmq.com/
 
+### `SqliteMessageQueue`
+
+*This API is available since Fedify 2.0.0.*
+
+To use [`SqliteMessageQueue`], you need to install the *@fedify/sqlite* package
+first:
+
+::: code-group
+
+~~~~ bash [Deno]
+deno add jsr:@fedify/sqlite
+~~~~
+
+~~~~ bash [npm]
+npm add @fedify/sqlite
+~~~~
+
+~~~~ bash [pnpm]
+pnpm add @fedify/sqlite
+~~~~
+
+~~~~ bash [Yarn]
+yarn add @fedify/sqlite
+~~~~
+
+~~~~ bash [Bun]
+bun add @fedify/sqlite
+~~~~
+
+:::
+
+[`SqliteMessageQueue`] is a message queue implementation that uses SQLite as
+the backend.  It uses polling to check for new messages and is designed for
+single-node deployments.  It's suitable for development, testing, and
+small-scale production use where simplicity is preferred over high throughput.
+It uses native sqlite modules, [`node:sqlite`] for Node.js and Deno,
+[`bun:sqlite`] for Bun.
+
+Best for
+:   Development and testing.
+
+Pros
+:   Simple, persistent with minimal configuration.
+
+Cons
+:   Limited scalability, not suitable for high-traffic production.
+
+> [!NOTE]
+> `SqliteMessageQueue` uses `DELETE ... RETURNING` to atomically fetch and
+> delete the oldest message that is ready to be processed.  This requires
+> SQLite 3.35.0 or later.
+
+::: code-group
+
+~~~~ typescript twoslash [Deno]
+import type { KvStore } from "@fedify/fedify";
+// ---cut-before---
+import { DatabaseSync } from "node:sqlite";
+import { createFederation } from "@fedify/fedify";
+import { SqliteMessageQueue } from "@fedify/sqlite";
+
+const db = new DatabaseSync(":memory:");
+const federation = createFederation<void>({
+  // ...
+  // ---cut-start---
+    kv: null as unknown as KvStore,
+  // ---cut-end---
+  queue: new SqliteMessageQueue(db),  // [!code highlight]
+});
+~~~~
+
+~~~~ typescript twoslash [Node.js]
+import type { KvStore } from "@fedify/fedify";
+// ---cut-before---
+import { DatabaseSync } from "node:sqlite";
+import { createFederation } from "@fedify/fedify";
+import { SqliteMessageQueue } from "@fedify/sqlite";
+
+const db = new DatabaseSync(":memory:");
+const federation = createFederation<void>({
+  // ...
+  // ---cut-start---
+    kv: null as unknown as KvStore,
+  // ---cut-end---
+  queue: new SqliteMessageQueue(db),  // [!code highlight]
+});
+~~~~
+
+~~~~ typescript [Bun]
+import type { KvStore } from "@fedify/fedify";
+// ---cut-before---
+import { Database } from "bun:sqlite";
+import { createFederation } from "@fedify/fedify";
+import { SqliteMessageQueue } from "@fedify/sqlite";
+
+const db = new Database(":memory:");
+const federation = createFederation<void>({
+  // ...
+  // ---cut-start---
+    kv: null as unknown as KvStore,
+  // ---cut-end---
+  queue: new SqliteMessageQueue(db),  // [!code highlight]
+});
+~~~~
+
+:::
+
+[`SqliteMessageQueue`]: https://jsr.io/@fedify/sqlite/doc/mq/~/SqliteMessageQueue
+
 ### `WorkersMessageQueue` (Cloudflare Workers only)
 
 *This API is available since Fedify 1.6.0.*
@@ -659,6 +768,9 @@ The following implementations do not yet support native retry:
 [`AmqpMessageQueue`]
 :   Native retry support planned for future release.
 
+[`SqliteMessageQueue`]
+:   No native retry support (`~MessageQueue.nativeRetrial` is `false`).
+
 `ParallelMessageQueue` inherits the `~MessageQueue.nativeRetrial` value from
 the wrapped queue.
 

mise.toml

@@ -26,6 +26,17 @@ env = { CI = "true" }  # Prevent pnpm from prompting for confirmation
 depends = ["install"]
 run = "pnpm --filter '@fedify/*' --recursive --parallel build"
 
+[tasks.prepare-each]
+description = "Prepare specific package(s)"
+usage = 'arg "<packages>" help="Package name(s) (without @fedify/ prefix)" var=#true var_min=1'
+env = { CI = "true" }
+run = '''
+for PACKAGE in ${usage_packages}; do
+  echo "Preparing package: $PACKAGE"
+  pnpm --filter "@fedify/$PACKAGE" build
+done
+'''
+
 # Code quality
 [tasks.check]
 description = "Check code formatting, linting, and type checking"
@@ -65,6 +76,24 @@ fi
 description = "Format the codebase"
 run = "deno fmt && hongdown --write"
 
+[tasks.check-each]
+description = "Check code quality for specific package(s)"
+usage = 'arg "<packages>" help="Package name(s) (without @fedify/ prefix)" var=#true var_min=1'
+run = '''
+for PACKAGE in ${usage_packages}; do
+  echo "Checking package: $PACKAGE"
+  PACKAGE_DIR="packages/$PACKAGE"
+  if [ ! -d "$PACKAGE_DIR" ]; then
+    echo "Error: Package directory $PACKAGE_DIR not found"
+    exit 1
+  fi
+
+  deno fmt --check "$PACKAGE_DIR"
+  deno lint "$PACKAGE_DIR"
+  deno check "$PACKAGE_DIR"/**/*.ts
+done
+'''
+
 # Testing
 [tasks."test:deno"]
 description = "Run the test suite using Deno"
@@ -85,6 +114,21 @@ run = "pnpm run --recursive --filter '!{docs}' test:bun"
 description = "Run the test suite across all environments (Deno, Node.js, Bun)"
 depends = ["check", "test:deno", "test:node", "test:bun"]
 
+[tasks.test-each]
+description = "Run tests for a specific package across all environments"
+usage = 'arg "<packages>" help="Package name(s) (without @fedify/ prefix)" var=#true var_min=1'
+run = '''
+mise run prepare-each ${usage_packages}
+mise run check-each ${usage_packages}
+
+for PACKAGE in ${usage_packages}; do
+  echo "Running tests for package: $PACKAGE"
+  deno task --filter "@fedify/$PACKAGE" test
+  pnpm --filter "@fedify/$PACKAGE" test
+  pnpm --filter "@fedify/$PACKAGE" test:bun
+done
+'''
+
 # Documentation
 [tasks.docs]
 description = "Start the documentation development server"

packages/amqp/package.json

@@ -59,6 +59,7 @@
   },
   "devDependencies": {
     "@alinea/suite": "^0.6.3",
+    "@fedify/testing": "workspace:^",
     "@js-temporal/polyfill": "catalog:",
     "@std/assert": "catalog:",
     "@std/async": "catalog:",