From cf6a470dc08b4ad9f4b6222ec722940cac3b0fce Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 19:44:46 +0900 Subject: [PATCH 01/16] Add Redis repository package Add @fedify/botkit-redis with a RedisRepository implementation for shared persistent BotKit storage across Deno and Node.js. The repository covers key pairs, messages, followers, follows, followees, quote authorizations, quote authorization references, poll votes, and bot-scoped views. Protect Redis read-modify-write paths with token-checked distributed locks, including lock renewal for long-running async message updates. Add Redis-backed integration and race regression tests, CI Redis service configuration, documentation, build wiring, and changelog notes. Fixes https://github.com/fedify-dev/botkit/issues/12 https://github.com/fedify-dev/botkit/pull/35 Assisted-by: Codex:gpt-5.5 --- .github/workflows/main.yaml | 10 + CHANGES.md | 12 + deno.lock | 59 ++ docs/.vitepress/config.mts | 4 + docs/concepts/repository.md | 60 ++ docs/package.json | 2 + mise.toml | 15 + packages/botkit-redis/README.md | 106 +++ packages/botkit-redis/deno.json | 24 + packages/botkit-redis/package.json | 64 ++ packages/botkit-redis/src/mod.test.ts | 827 +++++++++++++++++++++ packages/botkit-redis/src/mod.ts | 959 +++++++++++++++++++++++++ packages/botkit-redis/tsdown.config.ts | 10 + pnpm-lock.yaml | 110 +++ 14 files changed, 2262 insertions(+) create mode 100644 packages/botkit-redis/README.md create mode 100644 packages/botkit-redis/deno.json create mode 100644 packages/botkit-redis/package.json create mode 100644 packages/botkit-redis/src/mod.test.ts create mode 100644 packages/botkit-redis/src/mod.ts create mode 100644 packages/botkit-redis/tsdown.config.ts diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml index 43b8143..92b2674 100644 --- a/.github/workflows/main.yaml +++ b/.github/workflows/main.yaml @@ -25,11 +25,21 @@ jobs: --health-interval 10s --health-timeout 5s --health-retries 5 + redis: + image: redis:8 + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 10s + --health-timeout 5s + --health-retries 5 permissions: checks: write pull-requests: write env: POSTGRES_URL: postgres://postgres:postgres@localhost:5432/botkit_test + REDIS_URL: redis://localhost:6379/0 steps: - uses: actions/checkout@v4 - uses: jdx/mise-action@v4 diff --git a/CHANGES.md b/CHANGES.md index 968d738..4ea8268 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -188,6 +188,18 @@ To be released. - Upgraded Fedify to 2.3.1, LogTape to 2.2.3, and Postgres.js to 3.4.9. +### @fedify/botkit-redis + + - Added the new *@fedify/botkit-redis* package, which provides + `RedisRepository`, a Redis-backed implementation of BotKit's + `Repository` interface for bots running on Deno or Node.js. It supports + URL-managed and caller-managed Redis clients, configurable key prefixes, + bot-scoped storage, quote authorization storage, reverse lookups, and + poll votes. [[#12], [#35]] + +[#12]: https://github.com/fedify-dev/botkit/issues/12 +[#35]: https://github.com/fedify-dev/botkit/pull/35 + Version 0.4.4 ------------- diff --git a/deno.lock b/deno.lock index 40130cb..c9623e6 100644 --- a/deno.lock +++ b/deno.lock @@ -57,6 +57,7 @@ "npm:mime-db@^1.54.0": "1.54.0", "npm:pkijs@^3.2.5": "3.4.0", "npm:postgres@^3.4.9": "3.4.9", + "npm:redis@^6.1.0": "6.1.0_@opentelemetry+api@1.9.1", "npm:structured-field-values@^2.0.4": "2.0.4", "npm:tsdown@~0.12.8": "0.12.9_rolldown@1.0.0-beta.55", "npm:url-template@^3.1.1": "3.1.1", @@ -546,6 +547,40 @@ "quansync" ] }, + "@redis/bloom@6.1.0_@redis+client@6.1.0__@opentelemetry+api@1.9.1_@opentelemetry+api@1.9.1": { + "integrity": "sha512-Rzascjd9J9bJsM45T/Z9CTg1QY/B63B6YO8QorLVMeXnbBDsKiSCVR/+GQ061hYPk8FpTzWmPY8tAv2sT+JEtQ==", + "dependencies": [ + "@redis/client" + ] + }, + "@redis/client@6.1.0_@opentelemetry+api@1.9.1": { + "integrity": "sha512-7u1LefkezJF0HESlhO7ZFLEPfyY+NejP3SGv+Z4pGaT3oM5GVVLa0u3f4rDLUrcw+SRo8IlX9Y8JAONeDdg1Ag==", + "dependencies": [ + "@opentelemetry/api", + "cluster-key-slot" + ], + "optionalPeers": [ + "@opentelemetry/api" + ] + }, + "@redis/json@6.1.0_@redis+client@6.1.0__@opentelemetry+api@1.9.1_@opentelemetry+api@1.9.1": { + "integrity": "sha512-/GFjQA6bu5pG9ClCJAI5Xx4bNXe7UTpxBBlIupBNTrn1+nY860apGnYJuaSCDV2BmEbTidpa7O2qa28oxKx+rg==", + "dependencies": [ + "@redis/client" + ] + }, + "@redis/search@6.1.0_@redis+client@6.1.0__@opentelemetry+api@1.9.1_@opentelemetry+api@1.9.1": { + "integrity": "sha512-kS5agg+3yZbrdrt8omrew7FLCD8eOm7tarG1CROekPBRe+QGDR9aOpnHIQaYsYi6wPRTH70nQiF06AIjgURefQ==", + "dependencies": [ + "@redis/client" + ] + }, + "@redis/time-series@6.1.0_@redis+client@6.1.0__@opentelemetry+api@1.9.1_@opentelemetry+api@1.9.1": { + "integrity": "sha512-uIDBtV8MmG/xpJsRqbGSO4iX6ryj37MLMP82lRpFvI7ykAVe5GyqgxigEbU+uZNv9kDPNMKw3dvI/S/J1BNBzA==", + "dependencies": [ + "@redis/client" + ] + }, "@rolldown/binding-android-arm64@1.0.0-beta.55": { "integrity": "sha512-5cPpHdO+zp+klznZnIHRO1bMHDq5hS9cqXodEKAaa/dQTPDjnE91OwAsy3o1gT2x4QaY8NzdBXAvutYdaw0WeA==", "os": ["android"], @@ -678,6 +713,9 @@ "readdirp" ] }, + "cluster-key-slot@1.1.2": { + "integrity": "sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==" + }, "commander@2.20.3": { "integrity": "sha512-GpVkmM8vF2vQUkj2LvZmD35JxeJOLCwJ9cUkugyk2nuhbv3+mJvpLYYt+0+USMxE+oj+ey/lJEnhZw75x/OMcQ==" }, @@ -868,6 +906,16 @@ "readdirp@4.1.2": { "integrity": "sha512-GDhwkLfywWL2s6vEjyhri+eXmfH6j1L7JE27WhqLeYzoh/A3DBaYGEj2H/HFZCn/kMfim73FXxEJTw06WtxQwg==" }, + "redis@6.1.0_@opentelemetry+api@1.9.1": { + "integrity": "sha512-0kvUPM8RHP/ZMa0xYaDTcG5e8tIGW6kz6MToVT0V8iOnk6bkXp2jncGRGe2bZEk41lZwiDspUqjZCSk5ohjcKw==", + "dependencies": [ + "@redis/bloom", + "@redis/client", + "@redis/json", + "@redis/search", + "@redis/time-series" + ] + }, "resolve-pkg-maps@1.0.0": { "integrity": "sha512-seS2Tj26TBVOC2NIc2rOe2y2ZO7efxITtLZcGSOnHHNOQ7CkiUBfw0Iw2ck6xkIhPwLhKNLS8BO+hEpngQlqzw==" }, @@ -1056,6 +1104,17 @@ ] } }, + "packages/botkit-redis": { + "dependencies": [ + "jsr:@fedify/vocab@^2.3.1", + "npm:redis@^6.1.0" + ], + "packageJson": { + "dependencies": [ + "npm:redis@^6.1.0" + ] + } + }, "packages/botkit-sqlite": { "dependencies": [ "jsr:@fedify/vocab@^2.3.1" diff --git a/docs/.vitepress/config.mts b/docs/.vitepress/config.mts index 1ed9ca7..0668c9e 100644 --- a/docs/.vitepress/config.mts +++ b/docs/.vitepress/config.mts @@ -62,6 +62,10 @@ const references = { text: "@fedify/botkit-postgres", link: "https://jsr.io/@fedify/botkit-postgres/doc", }, + { + text: "@fedify/botkit-redis", + link: "https://jsr.io/@fedify/botkit-redis/doc", + }, { text: "@fedify/botkit-sqlite", link: "https://jsr.io/@fedify/botkit-sqlite/doc", diff --git a/docs/concepts/repository.md b/docs/concepts/repository.md index 6a73d87..82a65f2 100644 --- a/docs/concepts/repository.md +++ b/docs/concepts/repository.md @@ -164,6 +164,66 @@ properties: [`node:sqlite`]: https://nodejs.org/api/sqlite.html +`RedisRepository` +----------------- + +*This API is available since BotKit 0.5.0.* + +The `RedisRepository` is a repository that stores data in [Redis] using +the [node-redis] client. It is suited for deployments where multiple bot +processes need to share the same repository state, or where you already +operate Redis for other BotKit infrastructure. + +Unlike [`KvRepository`](#kvrepository), `RedisRepository` stores BotKit data +directly in Redis data structures such as strings, sets, and sorted sets. +It supports either an internally owned Redis client created from a connection +URL or an injected client whose lifecycle stays under your control. + +In order to use `RedisRepository`, you need to install the +*@fedify/botkit-redis* package: + +::: code-group + +~~~~ sh [Deno] +deno add jsr:@fedify/botkit-redis +~~~~ + +~~~~ sh [npm] +npm add @fedify/botkit-redis +~~~~ + +~~~~ sh [pnpm] +pnpm add @fedify/botkit-redis +~~~~ + +~~~~ sh [Yarn] +yarn add @fedify/botkit-redis +~~~~ + +::: + +The `RedisRepository` constructor accepts the following properties: + +`url` +: A Redis connection string used to create an internal client. Exactly one + of `url` and `client` must be provided. + +`client` +: An existing node-redis compatible client. When this is provided, the + repository does not own the client and calling `close()` will not shut it + down. + +`prefix` (optional) +: The Redis key prefix used for BotKit data. Defaults to `"botkit"`. + +`clientOptions` (optional) +: Additional node-redis client options. This option is only valid when + `url` is used. + +[Redis]: https://redis.io/ +[node-redis]: https://github.com/redis/node-redis + + `PostgresRepository` -------------------- diff --git a/docs/package.json b/docs/package.json index 67836e3..e5d0de0 100644 --- a/docs/package.json +++ b/docs/package.json @@ -2,6 +2,7 @@ "devDependencies": { "@fedify/botkit": "workspace:", "@fedify/botkit-postgres": "workspace:", + "@fedify/botkit-redis": "workspace:", "@fedify/botkit-sqlite": "workspace:", "@fedify/denokv": "jsr:@fedify/denokv@^2.3.1", "@fedify/fedify": "catalog:", @@ -19,6 +20,7 @@ "markdown-it-footnote": "^4.0.0", "markdown-it-jsr-ref": "^0.5.0", "postgres": "^3.4.9", + "redis": "^6.1.0", "srvx": "^0.11.20", "typescript": "^5.8.3", "vitepress": "^2.0.0-alpha.17", diff --git a/mise.toml b/mise.toml index c4c4f81..e972fde 100644 --- a/mise.toml +++ b/mise.toml @@ -110,6 +110,21 @@ outputs = [ ] run = "pnpm --filter @fedify/botkit-postgres run build" +[tasks."build:botkit-redis"] +description = "Build @fedify/botkit-redis" +depends = ["build:botkit"] +sources = [ + "packages/botkit-redis/deno.json", + "packages/botkit-redis/package.json", + "packages/botkit-redis/tsdown.config.ts", + "packages/botkit-redis/src/**/*", +] +outputs = [ + "packages/botkit-redis/dist/mod.js", + "packages/botkit-redis/dist/mod.d.ts", +] +run = "pnpm --filter @fedify/botkit-redis run build" + [tasks."test:node"] description = "Build packages and run Node.js tests" depends = ["build:*"] diff --git a/packages/botkit-redis/README.md b/packages/botkit-redis/README.md new file mode 100644 index 0000000..f5f9941 --- /dev/null +++ b/packages/botkit-redis/README.md @@ -0,0 +1,106 @@ +@fedify/botkit-redis +==================== + +[![JSR][JSR badge]][JSR] +[![npm][npm badge]][npm] +[![GitHub Actions][GitHub Actions badge]][GitHub Actions] + +This package is a [Redis]-based repository implementation for [BotKit]. It +provides shared persistent storage for bots running on either [Deno] or +[Node.js], using Redis data structures for BotKit repository data. + +[JSR badge]: https://jsr.io/badges/@fedify/botkit-redis +[JSR]: https://jsr.io/@fedify/botkit-redis +[npm badge]: https://img.shields.io/npm/v/@fedify/botkit-redis?logo=npm +[npm]: https://www.npmjs.com/package/@fedify/botkit-redis +[GitHub Actions badge]: https://github.com/fedify-dev/botkit/actions/workflows/main.yaml/badge.svg +[GitHub Actions]: https://github.com/fedify-dev/botkit/actions/workflows/main.yaml +[Redis]: https://redis.io/ +[BotKit]: https://botkit.fedify.dev/ +[Deno]: https://deno.land/ +[Node.js]: https://nodejs.org/ + + +Installation +------------ + +~~~~ sh +deno add jsr:@fedify/botkit-redis +npm add @fedify/botkit-redis +pnpm add @fedify/botkit-redis +yarn add @fedify/botkit-redis +~~~~ + + +Usage +----- + +The `RedisRepository` can be used as a drop-in repository implementation for +BotKit: + +~~~~ typescript +import { createBot, MemoryKvStore } from "@fedify/botkit"; +import { RedisRepository } from "@fedify/botkit-redis"; + +const bot = createBot({ + username: "mybot", + kv: new MemoryKvStore(), + repository: new RedisRepository({ + url: "redis://localhost:6379/0", + prefix: "botkit", + }), +}); +~~~~ + +You can also inject an existing node-redis client. In that case the +repository does not own the client and `close()` will not shut it down: + +~~~~ typescript +import { RedisRepository } from "@fedify/botkit-redis"; +import { createClient } from "redis"; + +const client = createClient({ url: "redis://localhost:6379/0" }); +await client.connect(); + +const repository = new RedisRepository({ + client, + prefix: "botkit", +}); +~~~~ + + +Options +------- + +The `RedisRepository` constructor accepts the following options: + + - **`url`**: A Redis connection string for an internally managed client. + + - **`client`**: An existing node-redis compatible client to use. + + - **`prefix`** (optional): Redis key prefix used for BotKit data. Defaults + to `"botkit"`. + + - **`clientOptions`** (optional): Additional node-redis client options. + This option is only valid together with `url`. + +The options are mutually exclusive: use either `client` or `url`. + + +Features +-------- + + - **Cross-runtime**: Works with both Deno and Node.js using node-redis. + + - **Shared persistent storage**: Suitable for multi-process deployments + backed by Redis. + + - **Key namespacing**: Keeps BotKit data under a configurable Redis prefix. + + - **Full `Repository` API**: Implements BotKit repository storage for key + pairs, messages, followers, follows, followees, quote authorizations, and + poll votes. + + - **Explicit resource ownership**: Repositories created from a URL own their + Redis client, while repositories created from an injected client leave + lifecycle control to the caller. diff --git a/packages/botkit-redis/deno.json b/packages/botkit-redis/deno.json new file mode 100644 index 0000000..09cb43c --- /dev/null +++ b/packages/botkit-redis/deno.json @@ -0,0 +1,24 @@ +{ + "name": "@fedify/botkit-redis", + "version": "0.5.0", + "license": "AGPL-3.0-only", + "exports": { + ".": "./src/mod.ts" + }, + "imports": { + "@fedify/vocab": "jsr:@fedify/vocab@^2.3.1", + "redis": "npm:redis@^6.1.0" + }, + "exclude": [ + "dist", + "junit.xml", + "package.json" + ], + "fmt": { + "exclude": [ + "*.md", + "*.yaml", + "*.yml" + ] + } +} diff --git a/packages/botkit-redis/package.json b/packages/botkit-redis/package.json new file mode 100644 index 0000000..243d816 --- /dev/null +++ b/packages/botkit-redis/package.json @@ -0,0 +1,64 @@ +{ + "name": "@fedify/botkit-redis", + "version": "0.5.0", + "description": "Redis-based repository for BotKit", + "license": "AGPL-3.0-only", + "author": { + "name": "Hong Minhee", + "email": "hong@minhee.org", + "url": "https://hongminhee.org/" + }, + "homepage": "https://botkit.fedify.dev/", + "repository": { + "type": "git", + "url": "git+https://github.com/fedify-dev/botkit.git", + "directory": "packages/botkit-redis" + }, + "bugs": { + "url": "https://github.com/fedify-dev/botkit/issues" + }, + "funding": [ + "https://opencollective.com/fedify", + "https://github.com/sponsors/dahlia" + ], + "engines": { + "deno": ">=2.0.0", + "node": ">=22.0.0" + }, + "type": "module", + "module": "./dist/mod.js", + "types": "./dist/mod.d.ts", + "exports": { + ".": { + "types": "./dist/mod.d.ts", + "import": "./dist/mod.js" + }, + "./package.json": "./package.json" + }, + "sideEffects": true, + "files": [ + "dist", + "LICENSE", + "package.json", + "README.md" + ], + "peerDependencies": { + "@fedify/botkit": "workspace:^" + }, + "dependencies": { + "@fedify/fedify": "catalog:", + "@fedify/vocab": "catalog:", + "@js-temporal/polyfill": "catalog:", + "@logtape/logtape": "catalog:", + "redis": "^6.1.0" + }, + "devDependencies": { + "tsdown": "catalog:" + }, + "scripts": { + "build": "tsdown", + "prepack": "tsdown", + "prepublish": "mise run check && tsdown", + "test": "tsdown && cd src/ && node --test --experimental-transform-types" + } +} diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts new file mode 100644 index 0000000..1ce2417 --- /dev/null +++ b/packages/botkit-redis/src/mod.test.ts @@ -0,0 +1,827 @@ +// BotKit by Fedify: A framework for creating ActivityPub bots +// Copyright (C) 2026 Hong Minhee +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as +// published by the Free Software Foundation, either version 3 of the +// License, or (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . +import { RedisRepository } from "@fedify/botkit-redis"; +import { exportJwk } from "@fedify/fedify/sig"; +import { + Create, + Follow, + Note, + Person, + PUBLIC_COLLECTION, + QuoteAuthorization, +} from "@fedify/vocab"; +import assert from "node:assert/strict"; +import { describe, test } from "node:test"; +import { createClient } from "redis"; + +if (!("Temporal" in globalThis)) { + globalThis.Temporal = (await import("@js-temporal" + "/polyfill")).Temporal; +} + +function getRedisUrl(): string | undefined { + if ("process" in globalThis) return globalThis.process.env.REDIS_URL; + if ("Deno" in globalThis) return globalThis.Deno.env.get("REDIS_URL"); + return undefined; +} + +const redisUrl = getRedisUrl(); + +interface TestRedisClient { + readonly isOpen: boolean; + sendCommand(args: readonly string[]): Promise; +} + +interface DelayedGetState { + count: number; + readonly waiters: (() => void)[]; +} + +interface Deferred { + readonly promise: Promise; + resolve(): void; +} + +async function createKeyPairs(): Promise { + return [ + await crypto.subtle.generateKey( + { + name: "RSASSA-PKCS1-v1_5", + modulusLength: 2048, + publicExponent: new Uint8Array([1, 0, 1]), + hash: "SHA-256", + }, + true, + ["sign", "verify"], + ), + ]; +} + +function createMessage(id: string, content: string, published: string): Create { + return new Create({ + id: new URL(`https://example.com/ap/actor/bot/create/${id}`), + actor: new URL("https://example.com/ap/actor/bot"), + to: new URL("https://example.com/ap/actor/bot/followers"), + cc: PUBLIC_COLLECTION, + object: new Note({ + id: new URL(`https://example.com/ap/actor/bot/note/${id}`), + attribution: new URL("https://example.com/ap/actor/bot"), + content, + published: Temporal.Instant.from(published), + }), + published: Temporal.Instant.from(published), + }); +} + +async function getMessageContent(message: Create): Promise { + const json = await message.toJsonLd({ format: "compact" }); + if (typeof json !== "object" || json == null) { + throw new TypeError("Expected a JSON-LD object."); + } + if (!("object" in json)) { + throw new TypeError("Expected a Create object."); + } + const object = json.object; + if (typeof object !== "object" || object == null) { + throw new TypeError("Expected a nested object."); + } + if (!("content" in object) || typeof object.content !== "string") { + throw new TypeError("Expected string content."); + } + return object.content; +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function cleanupPrefix(url: string, prefix: string): Promise { + const client = createClient({ url }); + await client.connect(); + try { + let cursor = "0"; + for (;;) { + const reply = await client.sendCommand([ + "SCAN", + cursor, + "MATCH", + `${prefix}:*`, + "COUNT", + "1000", + ]); + assert.ok(Array.isArray(reply)); + const keys = reply[1]; + assert.ok(Array.isArray(keys)); + if (keys.length > 0) { + await client.sendCommand(["DEL", ...keys.map(String)]); + } + cursor = String(reply[0]); + if (cursor === "0") break; + } + } finally { + await client.quit(); + } +} + +function createHarness() { + if (redisUrl == null) throw new Error("REDIS_URL is not set."); + const prefix = `botkit_test_${crypto.randomUUID()}`; + const repository = new RedisRepository({ url: redisUrl, prefix }); + return { + prefix, + repository, + async cleanup() { + await repository.close(); + await cleanupPrefix(redisUrl, prefix); + }, + }; +} + +function createDelayedGetClient( + client: TestRedisClient, + key: string, + state: DelayedGetState, +) { + return { + get isOpen(): boolean { + return client.isOpen; + }, + + sendCommand(args: readonly string[]): Promise { + return (async () => { + if (args[0] === "GET" && args[1] === key) { + state.count++; + if (state.count < 2) { + await new Promise((resolve) => { + state.waiters.push(resolve); + setTimeout(resolve, 50); + }); + } else { + for (const resolve of state.waiters) resolve(); + state.waiters.length = 0; + } + } + return await client.sendCommand([...args]); + })(); + }, + }; +} + +function createDeferred(): Deferred { + let resolve = () => {}; + const promise = new Promise((r) => { + resolve = r; + }); + return { promise, resolve }; +} + +function createDelayedDeleteClient( + client: TestRedisClient, + key: string, + observed: Deferred, + release: Deferred, +) { + return { + get isOpen(): boolean { + return client.isOpen; + }, + + sendCommand(args: readonly string[]): Promise { + return (async () => { + if (args[0] === "DEL" && args[1] === key) { + observed.resolve(); + await release.promise; + } + return await client.sendCommand([...args]); + })(); + }, + }; +} + +function createShortLockTtlClient( + client: TestRedisClient, + lockKey: string, + ttlMs: number, +) { + return { + get isOpen(): boolean { + return client.isOpen; + }, + + sendCommand(args: readonly string[]): Promise { + const command = [...args]; + if ( + command[0] === "SET" && command[1] === lockKey && + command[4] === "PX" + ) { + command[5] = ttlMs.toString(); + } + return client.sendCommand(command); + }, + }; +} + +if (redisUrl == null) { + test("RedisRepository integration tests", { skip: true }, () => {}); +} else { + describe("RedisRepository", () => { + test("rejects invalid constructor option combinations", () => { + assert.throws( + () => Reflect.construct(RedisRepository, [{}]), + new TypeError( + "RedisRepositoryOptions must provide exactly one of client or url.", + ), + ); + assert.throws( + () => + Reflect.construct(RedisRepository, [{ + client: { sendCommand: () => Promise.resolve(undefined) }, + url: redisUrl, + }]), + new TypeError( + "RedisRepositoryOptions must provide exactly one of client or url.", + ), + ); + }); + + test("key pairs", async () => { + const { repository, cleanup } = createHarness(); + try { + const keyPairs = await createKeyPairs(); + assert.deepStrictEqual(await repository.getKeyPairs("bot"), undefined); + await repository.setKeyPairs("bot", keyPairs); + assert.deepStrictEqual( + await Promise.all( + (await repository.getKeyPairs("bot"))!.map((pair) => + exportJwk(pair.publicKey) + ), + ), + await Promise.all(keyPairs.map((pair) => exportJwk(pair.publicKey))), + ); + } finally { + await cleanup(); + } + }); + + test("does not close injected clients", async () => { + if (redisUrl == null) throw new Error("REDIS_URL is not set."); + const prefix = `botkit_test_${crypto.randomUUID()}`; + const client = createClient({ url: redisUrl }); + await client.connect(); + const repository = new RedisRepository({ client, prefix }); + try { + assert.ok(client.isOpen); + await repository.close(); + assert.ok(client.isOpen); + } finally { + await client.quit(); + await cleanupPrefix(redisUrl, prefix); + } + }); + + test("messages basic operations and ordering", async () => { + const { repository, cleanup } = createHarness(); + try { + const firstId = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; + const secondId = "01942976-3400-7f34-872e-2cbf0f9eeac4"; + const first = createMessage(firstId, "first", "2025-01-01T00:00:00Z"); + const second = createMessage( + secondId, + "second", + "2025-01-02T00:00:00Z", + ); + + assert.deepStrictEqual(await repository.countMessages("bot"), 0); + await repository.addMessage("bot", firstId, first); + await repository.addMessage("bot", secondId, second); + assert.deepStrictEqual(await repository.countMessages("bot"), 2); + assert.deepStrictEqual( + await repository.getMessage("other", firstId), + undefined, + ); + assert.deepStrictEqual( + await (await repository.getMessage("bot", firstId))?.toJsonLd(), + await first.toJsonLd(), + ); + assert.deepStrictEqual( + (await Array.fromAsync( + repository.getMessages("bot", { order: "oldest" }), + )).length, + 2, + ); + assert.deepStrictEqual( + (await Array.fromAsync(repository.getMessages("bot", { + since: Temporal.Instant.from("2025-01-02T00:00:00Z"), + }))).length, + 1, + ); + assert.deepStrictEqual( + await (await repository.removeMessage("bot", firstId))?.toJsonLd(), + await first.toJsonLd(), + ); + assert.deepStrictEqual(await repository.countMessages("bot"), 1); + } finally { + await cleanup(); + } + }); + + test("serializes concurrent message updates", async () => { + const { repository, cleanup } = createHarness(); + try { + const messageId = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; + const message = createMessage( + messageId, + "base", + "2025-01-01T00:00:00Z", + ); + await repository.addMessage("bot", messageId, message); + + let activeUpdaters = 0; + let maxActiveUpdaters = 0; + const append = (suffix: string) => + repository.updateMessage("bot", messageId, async (existing) => { + assert.ok(existing instanceof Create); + activeUpdaters++; + maxActiveUpdaters = Math.max(maxActiveUpdaters, activeUpdaters); + const content = await getMessageContent(existing); + await delay(50); + activeUpdaters--; + return existing.clone({ + object: new Note({ + content: `${content}${suffix}`, + }), + }); + }); + + assert.deepStrictEqual(await Promise.all([append("A"), append("B")]), [ + true, + true, + ]); + assert.deepStrictEqual(maxActiveUpdaters, 1); + + const updated = await repository.getMessage("bot", messageId); + assert.ok(updated instanceof Create); + const content = await getMessageContent(updated); + assert.ok(content === "baseAB" || content === "baseBA"); + } finally { + await cleanup(); + } + }); + + test("renews Redis locks during long message updates", async () => { + const prefix = `botkit_test_${crypto.randomUUID()}`; + const firstClient = createClient({ url: redisUrl }); + const secondClient = createClient({ url: redisUrl }); + await firstClient.connect(); + await secondClient.connect(); + const messageId = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; + const lockKey = [ + prefix, + "bots", + "bot", + "messages", + messageId, + "lock", + ].join(":"); + const firstRepository = new RedisRepository({ + client: createShortLockTtlClient(firstClient, lockKey, 1_500), + prefix, + }); + const secondRepository = new RedisRepository({ + client: secondClient, + prefix, + }); + try { + const message = createMessage( + messageId, + "base", + "2025-01-01T00:00:00Z", + ); + await firstRepository.addMessage("bot", messageId, message); + + let activeUpdaters = 0; + let maxActiveUpdaters = 0; + const append = ( + repository: RedisRepository, + suffix: string, + waitMs: number, + ) => + repository.updateMessage("bot", messageId, async (existing) => { + assert.ok(existing instanceof Create); + activeUpdaters++; + maxActiveUpdaters = Math.max(maxActiveUpdaters, activeUpdaters); + const content = await getMessageContent(existing); + await delay(waitMs); + activeUpdaters--; + return existing.clone({ + object: new Note({ + content: `${content}${suffix}`, + }), + }); + }); + + assert.deepStrictEqual( + await Promise.all([ + append(firstRepository, "A", 2_200), + append(secondRepository, "B", 50), + ]), + [true, true], + ); + assert.deepStrictEqual(maxActiveUpdaters, 1); + + const updated = await firstRepository.getMessage("bot", messageId); + assert.ok(updated instanceof Create); + const content = await getMessageContent(updated); + assert.ok(content === "baseAB" || content === "baseBA"); + } finally { + await firstRepository.close(); + await secondRepository.close(); + await firstClient.quit(); + await secondClient.quit(); + await cleanupPrefix(redisUrl, prefix); + } + }); + + test("persists data across repository instances", async () => { + if (redisUrl == null) throw new Error("REDIS_URL is not set."); + const prefix = `botkit_test_${crypto.randomUUID()}`; + const firstRepository = new RedisRepository({ url: redisUrl, prefix }); + const secondRepository = new RedisRepository({ url: redisUrl, prefix }); + try { + const messageId = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; + const message = createMessage( + messageId, + "persisted", + "2025-01-01T00:00:00Z", + ); + await firstRepository.addMessage("bot", messageId, message); + assert.deepStrictEqual( + await (await secondRepository.getMessage("bot", messageId)) + ?.toJsonLd(), + await message.toJsonLd(), + ); + } finally { + await firstRepository.close(); + await secondRepository.close(); + await cleanupPrefix(redisUrl, prefix); + } + }); + + test("followers with multiple follow requests", async () => { + const { repository, cleanup } = createHarness(); + try { + const follower = new Person({ + id: new URL("https://example.com/ap/actor/alice"), + preferredUsername: "alice", + }); + const followA = new URL( + "https://example.com/ap/follow/f2fb7255-d3ad-4fef-8f9a-1d0f2c2ec0b4", + ); + const followB = new URL( + "https://example.com/ap/follow/a3d4cc4f-af93-4a9f-a7b3-0b7c0fe4901d", + ); + + await repository.addFollower("bot", followA, follower); + await repository.addFollower("bot", followB, follower); + assert.deepStrictEqual(await repository.countFollowers("bot"), 1); + assert.ok(await repository.hasFollower("bot", follower.id!)); + assert.deepStrictEqual( + await repository.removeFollower("bot", followA, follower.id!), + undefined, + ); + assert.ok(await repository.hasFollower("bot", follower.id!)); + assert.deepStrictEqual( + await (await repository.removeFollower("bot", followB, follower.id!)) + ?.toJsonLd(), + await follower.toJsonLd(), + ); + assert.deepStrictEqual(await repository.countFollowers("bot"), 0); + } finally { + await cleanup(); + } + }); + + test("serializes racing follower replacement and removal", async () => { + const { repository, cleanup } = createHarness(); + try { + const follow = new URL( + "https://example.com/ap/follow/f2fb7255-d3ad-4fef-8f9a-1d0f2c2ec0b4", + ); + const alice = new Person({ + id: new URL("https://example.com/ap/actor/alice"), + preferredUsername: "alice", + }); + const bob = new Person({ + id: new URL("https://example.com/ap/actor/bob"), + preferredUsername: "bob", + }); + + for (let i = 0; i < 20; i++) { + await repository.addFollower("bot", follow, alice); + await Promise.all([ + repository.removeFollower("bot", follow, alice.id!), + repository.addFollower("bot", follow, bob), + ]); + + assert.ok(await repository.hasFollower("bot", bob.id!)); + assert.deepStrictEqual( + await (await repository.removeFollower("bot", follow, bob.id!)) + ?.toJsonLd(), + await bob.toJsonLd(), + ); + assert.deepStrictEqual(await repository.countFollowers("bot"), 0); + } + } finally { + await cleanup(); + } + }); + + test("sent follows, followees, and reverse indexes", async () => { + const { repository, cleanup } = createHarness(); + try { + const followeeId = new URL("https://example.com/ap/actor/john"); + const follow = new Follow({ + id: new URL( + "https://example.com/ap/actor/bot/follow/03a395a2-353a-4894-afdb-2cab31a7b004", + ), + actor: new URL("https://example.com/ap/actor/bot"), + object: followeeId, + }); + const id = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; + + await repository.addSentFollow("bot", id, follow); + assert.deepStrictEqual( + await (await repository.getSentFollow("bot", id))?.toJsonLd(), + await follow.toJsonLd(), + ); + assert.deepStrictEqual( + await (await repository.removeSentFollow("bot", id))?.toJsonLd(), + await follow.toJsonLd(), + ); + + await repository.addFollowee("bot", followeeId, follow); + await repository.addFollowee("other", followeeId, follow); + assert.deepStrictEqual( + await Array.fromAsync(repository.findFollowedBots(followeeId)), + ["bot", "other"], + ); + assert.deepStrictEqual( + await (await repository.removeFollowee("bot", followeeId)) + ?.toJsonLd(), + await follow.toJsonLd(), + ); + assert.deepStrictEqual( + await Array.fromAsync(repository.findFollowedBots(followeeId)), + ["other"], + ); + } finally { + await cleanup(); + } + }); + + test("poll voting", async () => { + const { repository, cleanup } = createHarness(); + try { + const messageId = "01945678-1234-7890-abcd-ef0123456789"; + const voter1 = new URL("https://example.com/ap/actor/alice"); + const voter2 = new URL("https://example.com/ap/actor/bob"); + await repository.vote("bot", messageId, voter1, "option1"); + await repository.vote("bot", messageId, voter1, "option1"); + await repository.vote("bot", messageId, voter2, "option1"); + await repository.vote("bot", messageId, voter1, "option2"); + assert.deepStrictEqual( + await repository.countVoters("bot", messageId), + 2, + ); + assert.deepStrictEqual(await repository.countVotes("bot", messageId), { + option1: 2, + option2: 1, + }); + } finally { + await cleanup(); + } + }); + + test("quote authorizations and references", async () => { + const { repository, cleanup } = createHarness(); + try { + const firstId = "01942976-3400-7f34-872e-2cbf0f9eeac4"; + const secondId = "01942976-3400-7f34-872e-2cbf0f9eeac5"; + const quote = new URL("https://remote.example/notes/quote"); + const target = new URL("https://example.com/ap/note/1"); + const first = new QuoteAuthorization({ + id: new URL( + `https://example.com/ap/actor/bot/quote-authorization/${firstId}`, + ), + attribution: new URL("https://example.com/ap/actor/bot"), + interactingObject: quote, + interactionTarget: target, + }); + const second = new QuoteAuthorization({ + id: new URL( + `https://example.com/ap/actor/bot/quote-authorization/${secondId}`, + ), + attribution: new URL("https://example.com/ap/actor/bot"), + interactingObject: quote, + interactionTarget: target, + }); + + await repository.addQuoteAuthorization("bot", firstId, first); + await repository.addQuoteAuthorization("bot", secondId, second); + assert.deepStrictEqual( + (await repository.findQuoteAuthorization("bot", quote))?.id?.href, + first.id?.href, + ); + assert.deepStrictEqual( + await repository.getQuoteAuthorization("bot", secondId), + undefined, + ); + + const stamp = new URL("https://remote.example/stamps/1"); + const attribution = new URL("https://remote.example/actor/alice"); + await repository.addQuoteAuthorizationReference( + "bot", + stamp, + firstId, + attribution, + ); + await repository.addQuoteAuthorizationReference( + "other", + stamp, + firstId, + ); + assert.deepStrictEqual( + await repository.findQuoteAuthorizationReference("bot", stamp), + firstId, + ); + assert.deepStrictEqual( + await repository.findQuoteAuthorizationReferenceAttribution( + "bot", + stamp, + ), + attribution, + ); + assert.deepStrictEqual( + await Array.fromAsync( + repository.findQuoteAuthorizationReferenceIdentifiers(stamp), + ), + ["bot", "other"], + ); + await repository.removeQuoteAuthorizationReference("bot", stamp); + assert.deepStrictEqual( + await Array.fromAsync( + repository.findQuoteAuthorizationReferenceIdentifiers(stamp), + ), + ["other"], + ); + } finally { + await cleanup(); + } + }); + + test("serializes concurrent quote authorization inserts", async () => { + const prefix = `botkit_test_${crypto.randomUUID()}`; + const firstClient = createClient({ url: redisUrl }); + const secondClient = createClient({ url: redisUrl }); + await firstClient.connect(); + await secondClient.connect(); + const quote = new URL("https://remote.example/notes/raced-quote"); + const state: DelayedGetState = { count: 0, waiters: [] }; + const indexKey = [ + prefix, + "bots", + "bot", + "quoteAuthorizationsByInteractingObject", + encodeURIComponent(quote.href), + ].join(":"); + const firstRepository = new RedisRepository({ + client: createDelayedGetClient(firstClient, indexKey, state), + prefix, + }); + const secondRepository = new RedisRepository({ + client: createDelayedGetClient(secondClient, indexKey, state), + prefix, + }); + try { + const firstId = "01942976-3400-7f34-872e-2cbf0f9eeac6"; + const secondId = "01942976-3400-7f34-872e-2cbf0f9eeac7"; + const target = new URL("https://example.com/ap/note/1"); + const first = new QuoteAuthorization({ + id: new URL( + `https://example.com/ap/actor/bot/quote-authorization/${firstId}`, + ), + attribution: new URL("https://example.com/ap/actor/bot"), + interactingObject: quote, + interactionTarget: target, + }); + const second = new QuoteAuthorization({ + id: new URL( + `https://example.com/ap/actor/bot/quote-authorization/${secondId}`, + ), + attribution: new URL("https://example.com/ap/actor/bot"), + interactingObject: quote, + interactionTarget: target, + }); + + await Promise.all([ + firstRepository.addQuoteAuthorization("bot", firstId, first), + secondRepository.addQuoteAuthorization("bot", secondId, second), + ]); + + const found = await firstRepository.findQuoteAuthorization( + "bot", + quote, + ); + const stored = (await Promise.all([ + firstRepository.getQuoteAuthorization("bot", firstId), + firstRepository.getQuoteAuthorization("bot", secondId), + ])).filter((authorization) => authorization != null); + assert.deepStrictEqual(stored.length, 1); + assert.deepStrictEqual(found?.id?.href, stored[0].id?.href); + } finally { + await firstRepository.close(); + await secondRepository.close(); + await firstClient.quit(); + await secondClient.quit(); + await cleanupPrefix(redisUrl, prefix); + } + }); + + test("preserves quote authorization indexes during stale cleanup", async () => { + const prefix = `botkit_test_${crypto.randomUUID()}`; + const setupClient = createClient({ url: redisUrl }); + const findClient = createClient({ url: redisUrl }); + await setupClient.connect(); + await findClient.connect(); + const quote = new URL("https://remote.example/notes/cleanup-race"); + const indexKey = [ + prefix, + "bots", + "bot", + "quoteAuthorizationsByInteractingObject", + encodeURIComponent(quote.href), + ].join(":"); + const observed = createDeferred(); + const release = createDeferred(); + const findRepository = new RedisRepository({ + client: createDelayedDeleteClient( + findClient, + indexKey, + observed, + release, + ), + prefix, + }); + const addRepository = new RedisRepository({ url: redisUrl, prefix }); + try { + const id = "01942976-3400-7f34-872e-2cbf0f9eeac8"; + const staleId = "01942976-3400-7f34-872e-2cbf0f9eeac9"; + const authorization = new QuoteAuthorization({ + id: new URL( + `https://example.com/ap/actor/bot/quote-authorization/${id}`, + ), + attribution: new URL("https://example.com/ap/actor/bot"), + interactingObject: quote, + interactionTarget: new URL("https://example.com/ap/note/1"), + }); + await setupClient.sendCommand(["SET", indexKey, staleId]); + + const findPromise = findRepository.findQuoteAuthorization("bot", quote); + await observed.promise; + const addPromise = addRepository.addQuoteAuthorization( + "bot", + id, + authorization, + ); + await delay(50); + release.resolve(); + + assert.deepStrictEqual(await findPromise, undefined); + await addPromise; + assert.deepStrictEqual( + (await addRepository.findQuoteAuthorization("bot", quote))?.id?.href, + authorization.id?.href, + ); + } finally { + release.resolve(); + await findRepository.close(); + await addRepository.close(); + await setupClient.quit(); + await findClient.quit(); + await cleanupPrefix(redisUrl, prefix); + } + }); + }); +} diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts new file mode 100644 index 0000000..1ce734a --- /dev/null +++ b/packages/botkit-redis/src/mod.ts @@ -0,0 +1,959 @@ +// BotKit by Fedify: A framework for creating ActivityPub bots +// Copyright (C) 2026 Hong Minhee +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as +// published by the Free Software Foundation, either version 3 of the +// License, or (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . +import { + ActorScopedRepository, + type Repository, + type RepositoryGetFollowersOptions, + type RepositoryGetMessagesOptions, + type Uuid, +} from "@fedify/botkit/repository"; +import { exportJwk, importJwk } from "@fedify/fedify/sig"; +import { + Activity, + type Actor, + Announce, + Create, + Follow, + isActor, + Object, + QuoteAuthorization, +} from "@fedify/vocab"; +import { getLogger } from "@logtape/logtape"; +import { createClient, type RedisClientOptions } from "redis"; + +const logger = getLogger(["botkit", "redis"]); +const redisLockTimeoutMs = 30_000; +const redisLockPollIntervalMs = 20; +const redisLockRenewIntervalMs = 1_000; +const uuidV7Pattern = + /^([0-9a-f]{8})-([0-9a-f]{4})-7[0-9a-f]{3}-[0-9a-f]{4}-[0-9a-f]{12}$/i; + +interface RedisClientLike { + readonly isOpen?: boolean; + connect?(): Promise; + quit?(): Promise; + close?(): Promise | unknown; + sendCommand(args: readonly string[]): Promise; +} + +interface KeyPair { + readonly private: JsonWebKey; + readonly public: JsonWebKey; +} + +interface QuoteAuthorizationReferenceData { + readonly messageId: Uuid; + readonly attribution?: string; +} + +interface RedisRepositoryOptionsBase { + /** + * The Redis key prefix under which all BotKit data is stored. + * @default `"botkit"` + */ + readonly prefix?: string; +} + +interface RedisRepositoryOptionsWithClient extends RedisRepositoryOptionsBase { + /** + * A pre-configured Redis client to use. It must already be connected. + */ + readonly client: RedisClientLike; + + /** + * Disallowed when `client` is provided. + */ + readonly url?: never; + + /** + * Disallowed when `client` is provided. + */ + readonly clientOptions?: never; +} + +interface RedisRepositoryOptionsWithUrl extends RedisRepositoryOptionsBase { + /** + * A Redis connection string to connect with. + */ + readonly url: string | URL; + + /** + * Disallowed when `url` is provided. + */ + readonly client?: never; + + /** + * Additional node-redis client options. + */ + readonly clientOptions?: Omit; +} + +/** + * Options for creating a Redis repository. + * @since 0.5.0 + */ +export type RedisRepositoryOptions = + | RedisRepositoryOptionsWithClient + | RedisRepositoryOptionsWithUrl; + +/** + * A repository for storing bot data using Redis. + * @since 0.5.0 + */ +export class RedisRepository implements Repository, AsyncDisposable { + private readonly client: RedisClientLike; + private readonly ownsClient: boolean; + private readonly ready: Promise; + + /** + * The Redis key prefix under which all BotKit data is stored. + */ + readonly prefix: string; + + /** + * Creates a new Redis repository. + * @param options The options for creating the repository. + */ + constructor(options: RedisRepositoryOptions) { + const hasClient = "client" in options && options.client != null; + const hasUrl = "url" in options && options.url != null; + if (hasClient === hasUrl) { + throw new TypeError( + "RedisRepositoryOptions must provide exactly one of client or url.", + ); + } + this.prefix = options.prefix ?? "botkit"; + if (hasClient) { + this.client = options.client; + this.ownsClient = false; + this.ready = Promise.resolve(); + } else { + const client = createClient({ + ...options.clientOptions, + url: options.url.toString(), + }) as unknown as RedisClientLike; + this.client = client; + this.ownsClient = true; + this.ready = client.connect?.() ?? Promise.resolve(); + } + } + + async [Symbol.asyncDispose](): Promise { + await this.close(); + } + + /** + * Closes the owned Redis connection. + */ + async close(): Promise { + await this.ready; + if (!this.ownsClient) return; + if (this.client.quit != null) { + await this.client.quit(); + } else { + await this.client.close?.(); + } + } + + private key(...segments: readonly string[]): string { + return [ + this.prefix, + ...segments.map((segment) => encodeURIComponent(segment)), + ].join(":"); + } + + private botKey(identifier: string, ...segments: readonly string[]): string { + return this.key("bots", identifier, ...segments); + } + + private async command(args: readonly string[]): Promise { + await this.ready; + try { + return await this.client.sendCommand(args); + } catch (error) { + logger.error("Redis command failed: {command}.", { + command: args[0], + error, + }); + throw error; + } + } + + private async get(key: string): Promise { + const value = await this.command(["GET", key]); + return typeof value === "string" ? value : undefined; + } + + private async set(key: string, value: string): Promise { + await this.command(["SET", key, value]); + } + + private async del(...keys: readonly string[]): Promise { + if (keys.length < 1) return; + await this.command(["DEL", ...keys]); + } + + private async sAdd(key: string, member: string): Promise { + await this.command(["SADD", key, member]); + } + + private async sRem(key: string, member: string): Promise { + await this.command(["SREM", key, member]); + } + + private async sCard(key: string): Promise { + return toNumber(await this.command(["SCARD", key])); + } + + private async zAdd( + key: string, + score: number, + member: string, + nx = false, + ): Promise { + await this.command([ + "ZADD", + key, + ...(nx ? ["NX"] : []), + score.toString(), + member, + ]); + } + + private async zRem(key: string, member: string): Promise { + await this.command(["ZREM", key, member]); + } + + private async zRange(key: string): Promise { + return toStringArray(await this.command(["ZRANGE", key, "0", "-1"])); + } + + private async zRangeByScore( + key: string, + min: string, + max: string, + reverse: boolean, + limit?: number, + ): Promise { + const command = reverse ? "ZREVRANGEBYSCORE" : "ZRANGEBYSCORE"; + const args = [command, key, reverse ? max : min, reverse ? min : max]; + if (limit != null) args.push("LIMIT", "0", limit.toString()); + return toStringArray(await this.command(args)); + } + + private async zCard(key: string): Promise { + return toNumber(await this.command(["ZCARD", key])); + } + + private async nextSequence(key: string): Promise { + return toNumber(await this.command(["INCR", key])); + } + + private async withRedisLock( + lockKey: string, + operation: () => Promise, + ): Promise { + const token = crypto.randomUUID(); + while ( + await this.command([ + "SET", + lockKey, + token, + "NX", + "PX", + redisLockTimeoutMs.toString(), + ]) !== "OK" + ) { + await delay(redisLockPollIntervalMs); + } + const renew = setInterval(() => { + void (async () => { + try { + await this.command([ + "EVAL", + "if redis.call('GET', KEYS[1]) == ARGV[1] then " + + "return redis.call('PEXPIRE', KEYS[1], ARGV[2]) else return 0 end", + "1", + lockKey, + token, + redisLockTimeoutMs.toString(), + ]); + } catch (error) { + logger.warn("Failed to renew Redis lock: {error}", { error }); + } + })(); + }, redisLockRenewIntervalMs); + try { + return await operation(); + } finally { + clearInterval(renew); + try { + await this.command([ + "EVAL", + "if redis.call('GET', KEYS[1]) == ARGV[1] then " + + "return redis.call('DEL', KEYS[1]) else return 0 end", + "1", + lockKey, + token, + ]); + } catch (error) { + logger.warn("Failed to release Redis lock: {error}", { error }); + } + } + } + + async setKeyPairs( + identifier: string, + keyPairs: CryptoKeyPair[], + ): Promise { + const pairs: KeyPair[] = []; + for (const keyPair of keyPairs) { + pairs.push({ + private: await exportJwk(keyPair.privateKey), + public: await exportJwk(keyPair.publicKey), + }); + } + await this.set(this.botKey(identifier, "keyPairs"), JSON.stringify(pairs)); + } + + async getKeyPairs(identifier: string): Promise { + const json = await this.get(this.botKey(identifier, "keyPairs")); + if (json == null) return undefined; + const pairs = JSON.parse(json) as KeyPair[]; + return await Promise.all(pairs.map(async (pair) => ({ + privateKey: await importJwk(pair.private, "private"), + publicKey: await importJwk(pair.public, "public"), + }))); + } + + async addMessage( + identifier: string, + id: Uuid, + activity: Create | Announce, + ): Promise { + await this.set( + this.botKey(identifier, "messages", id), + JSON.stringify(await activity.toJsonLd({ format: "compact" })), + ); + await this.zAdd( + this.botKey(identifier, "messages"), + getMessageScore(id, activity), + id, + ); + } + + async updateMessage( + identifier: string, + id: Uuid, + updater: ( + existing: Create | Announce, + ) => Create | Announce | undefined | Promise, + ): Promise { + const key = this.botKey(identifier, "messages", id); + return await this.withRedisLock( + this.botKey(identifier, "messages", id, "lock"), + async () => { + const json = await this.get(key); + if (json == null) return false; + const activity = await Activity.fromJsonLd(JSON.parse(json)); + if (!(activity instanceof Create || activity instanceof Announce)) { + return false; + } + const updated = await updater(activity); + if (updated == null) return false; + await this.set( + key, + JSON.stringify(await updated.toJsonLd({ format: "compact" })), + ); + await this.zAdd( + this.botKey(identifier, "messages"), + getMessageScore(id, updated), + id, + ); + return true; + }, + ); + } + + async removeMessage( + identifier: string, + id: Uuid, + ): Promise { + const key = this.botKey(identifier, "messages", id); + const json = await this.get(key); + await this.del(key); + await this.zRem(this.botKey(identifier, "messages"), id); + if (json == null) return undefined; + const activity = await Activity.fromJsonLd(JSON.parse(json)); + if (activity instanceof Create || activity instanceof Announce) { + return activity; + } + return undefined; + } + + async *getMessages( + identifier: string, + options: RepositoryGetMessagesOptions = {}, + ): AsyncIterable { + const min = options.since == null + ? "-inf" + : options.since.epochMilliseconds.toString(); + const max = options.until == null + ? "+inf" + : options.until.epochMilliseconds.toString(); + const ids = await this.zRangeByScore( + this.botKey(identifier, "messages"), + min, + max, + options.order == null || options.order === "newest", + options.limit, + ); + for (const id of ids) { + const json = await this.get(this.botKey(identifier, "messages", id)); + if (json == null) continue; + try { + const activity = await Activity.fromJsonLd(JSON.parse(json)); + if (activity instanceof Create || activity instanceof Announce) { + yield activity; + } + } catch { + continue; + } + } + } + + async getMessage( + identifier: string, + id: Uuid, + ): Promise { + const json = await this.get(this.botKey(identifier, "messages", id)); + if (json == null) return undefined; + try { + const activity = await Activity.fromJsonLd(JSON.parse(json)); + if (activity instanceof Create || activity instanceof Announce) { + return activity; + } + } catch { + return undefined; + } + return undefined; + } + + countMessages(identifier: string): Promise { + return this.zCard(this.botKey(identifier, "messages")); + } + + async addFollower( + identifier: string, + followId: URL, + follower: Actor, + ): Promise { + if (follower.id == null) { + throw new TypeError("The follower ID is missing."); + } + const followerId = follower.id.href; + const followIdString = followId.href; + const requestKey = this.botKey( + identifier, + "followRequests", + followIdString, + ); + await this.withRedisLock( + this.botKey(identifier, "followers", "lock"), + async () => { + const previousFollowerId = await this.get(requestKey); + await this.set( + this.botKey(identifier, "followers", followerId), + JSON.stringify(await follower.toJsonLd({ format: "compact" })), + ); + await this.zAdd( + this.botKey(identifier, "followers"), + await this.nextSequence(this.botKey(identifier, "followers", "seq")), + followerId, + true, + ); + await this.sAdd( + this.botKey(identifier, "followerRequests", followerId), + followIdString, + ); + await this.set(requestKey, followerId); + if (previousFollowerId != null && previousFollowerId !== followerId) { + await this.sRem( + this.botKey(identifier, "followerRequests", previousFollowerId), + followIdString, + ); + await this.cleanupFollower(identifier, previousFollowerId); + } + }, + ); + } + + async removeFollower( + identifier: string, + followId: URL, + followerId: URL, + ): Promise { + const followIdString = followId.href; + const requestKey = this.botKey( + identifier, + "followRequests", + followIdString, + ); + return await this.withRedisLock( + this.botKey(identifier, "followers", "lock"), + async () => { + const currentFollowerId = await this.get(requestKey); + if (currentFollowerId !== followerId.href) return undefined; + const followerKey = this.botKey( + identifier, + "followers", + followerId.href, + ); + const json = await this.get(followerKey); + await this.del(requestKey); + await this.sRem( + this.botKey(identifier, "followerRequests", followerId.href), + followIdString, + ); + const removed = await this.cleanupFollower(identifier, followerId.href); + if (!removed || json == null) return undefined; + try { + const actor = await Object.fromJsonLd(JSON.parse(json)); + if (isActor(actor)) return actor; + } catch { + return undefined; + } + return undefined; + }, + ); + } + + private async cleanupFollower( + identifier: string, + followerId: string, + ): Promise { + const requestsKey = this.botKey(identifier, "followerRequests", followerId); + if (await this.sCard(requestsKey) > 0) return false; + await this.del( + requestsKey, + this.botKey(identifier, "followers", followerId), + ); + await this.zRem(this.botKey(identifier, "followers"), followerId); + return true; + } + + async hasFollower(identifier: string, followerId: URL): Promise { + return await this.get( + this.botKey(identifier, "followers", followerId.href), + ) != + null; + } + + async *getFollowers( + identifier: string, + options: RepositoryGetFollowersOptions = {}, + ): AsyncIterable { + const start = options.offset ?? 0; + const stop = options.limit == null ? -1 : start + options.limit - 1; + const ids = toStringArray( + await this.command([ + "ZRANGE", + this.botKey(identifier, "followers"), + start.toString(), + stop.toString(), + ]), + ); + for (const id of ids) { + const json = await this.get(this.botKey(identifier, "followers", id)); + if (json == null) continue; + try { + const actor = await Object.fromJsonLd(JSON.parse(json)); + if (isActor(actor)) yield actor; + } catch { + continue; + } + } + } + + countFollowers(identifier: string): Promise { + return this.zCard(this.botKey(identifier, "followers")); + } + + async addSentFollow( + identifier: string, + id: Uuid, + follow: Follow, + ): Promise { + await this.set( + this.botKey(identifier, "follows", id), + JSON.stringify(await follow.toJsonLd({ format: "compact" })), + ); + } + + async removeSentFollow( + identifier: string, + id: Uuid, + ): Promise { + const follow = await this.getSentFollow(identifier, id); + if (follow == null) return undefined; + await this.del(this.botKey(identifier, "follows", id)); + return follow; + } + + async getSentFollow( + identifier: string, + id: Uuid, + ): Promise { + const json = await this.get(this.botKey(identifier, "follows", id)); + if (json == null) return undefined; + try { + return await Follow.fromJsonLd(JSON.parse(json)); + } catch { + return undefined; + } + } + + async addFollowee( + identifier: string, + followeeId: URL, + follow: Follow, + ): Promise { + await this.set( + this.botKey(identifier, "followees", followeeId.href), + JSON.stringify(await follow.toJsonLd({ format: "compact" })), + ); + await this.zAdd( + this.key("index", "followees", followeeId.href), + 0, + identifier, + ); + } + + async removeFollowee( + identifier: string, + followeeId: URL, + ): Promise { + const follow = await this.getFollowee(identifier, followeeId); + await this.del(this.botKey(identifier, "followees", followeeId.href)); + await this.zRem( + this.key("index", "followees", followeeId.href), + identifier, + ); + return follow; + } + + async getFollowee( + identifier: string, + followeeId: URL, + ): Promise { + const json = await this.get( + this.botKey(identifier, "followees", followeeId.href), + ); + if (json == null) return undefined; + try { + return await Follow.fromJsonLd(JSON.parse(json)); + } catch { + return undefined; + } + } + + async *findFollowedBots(followeeId: URL): AsyncIterable { + yield* await this.zRange(this.key("index", "followees", followeeId.href)); + } + + async addQuoteAuthorization( + identifier: string, + id: Uuid, + authorization: QuoteAuthorization, + ): Promise { + const interactingObject = authorization.interactingObjectId; + if (interactingObject == null) { + throw new TypeError( + "The quote authorization interacting object is missing.", + ); + } + const indexKey = this.botKey( + identifier, + "quoteAuthorizationsByInteractingObject", + interactingObject.href, + ); + await this.withRedisLock( + this.quoteAuthorizationLockKey(indexKey), + async () => { + const existing = await this.get(indexKey); + if (existing != null) { + if ( + await this.getQuoteAuthorization(identifier, existing as Uuid) != + null + ) { + return; + } + await this.del(indexKey); + } + await this.set( + this.botKey(identifier, "quoteAuthorizations", id), + JSON.stringify(await authorization.toJsonLd({ format: "compact" })), + ); + await this.set(indexKey, id); + }, + ); + } + + async getQuoteAuthorization( + identifier: string, + id: Uuid, + ): Promise { + const json = await this.get( + this.botKey(identifier, "quoteAuthorizations", id), + ); + if (json == null) return undefined; + try { + return await QuoteAuthorization.fromJsonLd(JSON.parse(json)); + } catch { + return undefined; + } + } + + async findQuoteAuthorization( + identifier: string, + interactingObject: URL, + ): Promise { + const indexKey = this.botKey( + identifier, + "quoteAuthorizationsByInteractingObject", + interactingObject.href, + ); + return await this.withRedisLock( + this.quoteAuthorizationLockKey(indexKey), + async () => { + const id = await this.get(indexKey); + if (id == null) return undefined; + const authorization = await this.getQuoteAuthorization( + identifier, + id as Uuid, + ); + if (authorization == null && await this.get(indexKey) === id) { + await this.del(indexKey); + } + return authorization; + }, + ); + } + + async removeQuoteAuthorization( + identifier: string, + id: Uuid, + ): Promise { + const key = this.botKey(identifier, "quoteAuthorizations", id); + const json = await this.get(key); + await this.del(key); + if (json == null) return undefined; + let parsed: unknown; + try { + parsed = JSON.parse(json); + } catch { + return undefined; + } + const interactingObject = getRawQuoteAuthorizationInteractingObject(parsed); + if (interactingObject != null) { + const indexKey = this.botKey( + identifier, + "quoteAuthorizationsByInteractingObject", + interactingObject.href, + ); + await this.withRedisLock( + this.quoteAuthorizationLockKey(indexKey), + async () => { + if (await this.get(indexKey) === id) await this.del(indexKey); + }, + ); + } + try { + return await QuoteAuthorization.fromJsonLd(parsed); + } catch { + return undefined; + } + } + + async addQuoteAuthorizationReference( + identifier: string, + authorization: URL, + messageId: Uuid, + attribution?: URL, + ): Promise { + const value: QuoteAuthorizationReferenceData = attribution == null + ? { messageId } + : { messageId, attribution: attribution.href }; + await this.set( + this.botKey(identifier, "quoteAuthorizationRefs", authorization.href), + JSON.stringify(value), + ); + await this.zAdd( + this.key("index", "quoteAuthorizationRefs", authorization.href), + 0, + identifier, + ); + } + + async findQuoteAuthorizationReference( + identifier: string, + authorization: URL, + ): Promise { + const json = await this.get( + this.botKey(identifier, "quoteAuthorizationRefs", authorization.href), + ); + if (json == null) return undefined; + try { + const value = JSON.parse(json) as QuoteAuthorizationReferenceData; + return value.messageId; + } catch { + return undefined; + } + } + + async *findQuoteAuthorizationReferenceIdentifiers( + authorization: URL, + ): AsyncIterable { + yield* await this.zRange( + this.key("index", "quoteAuthorizationRefs", authorization.href), + ); + } + + async findQuoteAuthorizationReferenceAttribution( + identifier: string, + authorization: URL, + ): Promise { + const json = await this.get( + this.botKey(identifier, "quoteAuthorizationRefs", authorization.href), + ); + if (json == null) return undefined; + let value: QuoteAuthorizationReferenceData; + try { + value = JSON.parse(json) as QuoteAuthorizationReferenceData; + } catch { + return undefined; + } + if (value.attribution == null) return undefined; + try { + return new URL(value.attribution); + } catch { + return undefined; + } + } + + async removeQuoteAuthorizationReference( + identifier: string, + authorization: URL, + ): Promise { + await this.del( + this.botKey(identifier, "quoteAuthorizationRefs", authorization.href), + ); + await this.zRem( + this.key("index", "quoteAuthorizationRefs", authorization.href), + identifier, + ); + } + + async vote( + identifier: string, + messageId: Uuid, + voterId: URL, + option: string, + ): Promise { + await this.sAdd( + this.botKey(identifier, "polls", messageId, "voters"), + voterId.href, + ); + await this.sAdd( + this.botKey(identifier, "polls", messageId, "options", option), + voterId.href, + ); + await this.zAdd( + this.botKey(identifier, "polls", messageId, "options"), + 0, + option, + ); + } + + countVoters(identifier: string, messageId: Uuid): Promise { + return this.sCard(this.botKey(identifier, "polls", messageId, "voters")); + } + + async countVotes( + identifier: string, + messageId: Uuid, + ): Promise>> { + const result: Record = {}; + const options = await this.zRange( + this.botKey(identifier, "polls", messageId, "options"), + ); + for (const option of options) { + result[option] = await this.sCard( + this.botKey(identifier, "polls", messageId, "options", option), + ); + } + return result; + } + + forIdentifier(identifier: string): ActorScopedRepository { + return new ActorScopedRepository(this, identifier); + } + + private quoteAuthorizationLockKey(indexKey: string): string { + return `${indexKey}:lock`; + } +} + +function toNumber(value: unknown): number { + if (typeof value === "number") return value; + if (typeof value === "bigint") return Number(value); + if (typeof value === "string") return Number.parseInt(value, 10); + return 0; +} + +function toStringArray(value: unknown): string[] { + if (!Array.isArray(value)) return []; + return value.filter((item): item is string => typeof item === "string"); +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function getMessageScore(id: Uuid, activity: Create | Announce): number { + if (activity.published != null) return activity.published.epochMilliseconds; + return extractTimestamp(id); +} + +function extractTimestamp(uuid: string): number { + const match = uuidV7Pattern.exec(uuid); + if (match == null) return 0; + return Number.parseInt(`${match[1]}${match[2]}`, 16); +} + +function getRawQuoteAuthorizationInteractingObject( + json: unknown, +): URL | undefined { + if (typeof json !== "object" || json == null) return undefined; + if (!("interactingObject" in json)) return undefined; + const interactingObject = json.interactingObject; + if (typeof interactingObject !== "string") return undefined; + try { + return new URL(interactingObject); + } catch { + return undefined; + } +} diff --git a/packages/botkit-redis/tsdown.config.ts b/packages/botkit-redis/tsdown.config.ts new file mode 100644 index 0000000..6a44d92 --- /dev/null +++ b/packages/botkit-redis/tsdown.config.ts @@ -0,0 +1,10 @@ +import { defineConfig } from "tsdown"; + +export default defineConfig({ + entry: "src/mod.ts", + dts: { + sourcemap: true, + }, + format: "esm", + platform: "node", +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index fed9fd3..ba59004 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -41,6 +41,9 @@ importers: '@fedify/botkit-postgres': specifier: 'workspace:' version: link:../packages/botkit-postgres + '@fedify/botkit-redis': + specifier: 'workspace:' + version: link:../packages/botkit-redis '@fedify/botkit-sqlite': specifier: 'workspace:' version: link:../packages/botkit-sqlite @@ -92,6 +95,9 @@ importers: postgres: specifier: ^3.4.9 version: 3.4.9 + redis: + specifier: ^6.1.0 + version: 6.1.0(@opentelemetry/api@1.9.1) srvx: specifier: ^0.11.20 version: 0.11.20 @@ -185,6 +191,31 @@ importers: specifier: ^0.12.8 version: 0.12.8(typescript@5.8.3) + packages/botkit-redis: + dependencies: + '@fedify/botkit': + specifier: workspace:^ + version: link:../botkit + '@fedify/fedify': + specifier: 'catalog:' + version: 2.3.1 + '@fedify/vocab': + specifier: 'catalog:' + version: 2.3.1 + '@js-temporal/polyfill': + specifier: 'catalog:' + version: 0.5.1 + '@logtape/logtape': + specifier: 'catalog:' + version: 2.2.3 + redis: + specifier: ^6.1.0 + version: 6.1.0(@opentelemetry/api@1.9.1) + devDependencies: + tsdown: + specifier: 'catalog:' + version: 0.12.8(typescript@5.8.3) + packages/botkit-sqlite: dependencies: '@fedify/botkit': @@ -639,6 +670,42 @@ packages: resolution: {integrity: sha512-G0OnZbMWEs5LhDyqy2UL17vGhSVHkQIfVojMtEWVenvj0V5S84VBgy86kJIuNsGDp2p7sTKlpSIpBUWdC35OKg==} engines: {node: '>=20.0.0'} + '@redis/bloom@6.1.0': + resolution: {integrity: sha512-Rzascjd9J9bJsM45T/Z9CTg1QY/B63B6YO8QorLVMeXnbBDsKiSCVR/+GQ061hYPk8FpTzWmPY8tAv2sT+JEtQ==} + engines: {node: '>= 20.0.0'} + peerDependencies: + '@redis/client': ^6.1.0 + + '@redis/client@6.1.0': + resolution: {integrity: sha512-7u1LefkezJF0HESlhO7ZFLEPfyY+NejP3SGv+Z4pGaT3oM5GVVLa0u3f4rDLUrcw+SRo8IlX9Y8JAONeDdg1Ag==} + engines: {node: '>= 20.0.0'} + peerDependencies: + '@node-rs/xxhash': ^1.1.0 + '@opentelemetry/api': '>=1 <2' + peerDependenciesMeta: + '@node-rs/xxhash': + optional: true + '@opentelemetry/api': + optional: true + + '@redis/json@6.1.0': + resolution: {integrity: sha512-/GFjQA6bu5pG9ClCJAI5Xx4bNXe7UTpxBBlIupBNTrn1+nY860apGnYJuaSCDV2BmEbTidpa7O2qa28oxKx+rg==} + engines: {node: '>= 20.0.0'} + peerDependencies: + '@redis/client': ^6.1.0 + + '@redis/search@6.1.0': + resolution: {integrity: sha512-kS5agg+3yZbrdrt8omrew7FLCD8eOm7tarG1CROekPBRe+QGDR9aOpnHIQaYsYi6wPRTH70nQiF06AIjgURefQ==} + engines: {node: '>= 20.0.0'} + peerDependencies: + '@redis/client': ^6.1.0 + + '@redis/time-series@6.1.0': + resolution: {integrity: sha512-uIDBtV8MmG/xpJsRqbGSO4iX6ryj37MLMP82lRpFvI7ykAVe5GyqgxigEbU+uZNv9kDPNMKw3dvI/S/J1BNBzA==} + engines: {node: '>= 20.0.0'} + peerDependencies: + '@redis/client': ^6.1.0 + '@rolldown/binding-darwin-arm64@1.0.0-beta.15': resolution: {integrity: sha512-YInZppDBLp5DadbJZGc7xBfDrMCSj3P6i2rPlvOCMlvjBQxJi2kX8Jquh+LufsWUiHD3JsvvH5EuUUc/tF5fkA==} cpu: [arm64] @@ -1131,6 +1198,10 @@ packages: resolution: {integrity: sha512-rwHwUfXL40Chm1r08yrhU3qpUvdVlgkKNeyeGPOxnW8/SyVDvgRaed/Uz54AqWNaTCAThlj6QAs3TZcKI0xDEw==} engines: {node: '>=0.10.0'} + cluster-key-slot@1.1.2: + resolution: {integrity: sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==} + engines: {node: '>=0.10.0'} + color-convert@2.0.1: resolution: {integrity: sha512-RRECPsj7iu/xb5oKYcsFHSppFNnsj/52OVTRKb4zP5onXwVF3zVmmToNcOfGC+CRDpfK/U584fMg38ZHCaElKQ==} engines: {node: '>=7.0.0'} @@ -1665,6 +1736,10 @@ packages: resolution: {integrity: sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==} engines: {node: '>=4'} + redis@6.1.0: + resolution: {integrity: sha512-0kvUPM8RHP/ZMa0xYaDTcG5e8tIGW6kz6MToVT0V8iOnk6bkXp2jncGRGe2bZEk41lZwiDspUqjZCSk5ohjcKw==} + engines: {node: '>= 20.0.0'} + regex-recursion@6.0.2: resolution: {integrity: sha512-0YCaSCq2VRIebiaUviZNs0cBz1kg5kVS2UKUfNIx8YVs1cN3AV7NTctO5FOKBA+UT2BPJIWZauYHPqJODG50cg==} @@ -2451,6 +2526,28 @@ snapshots: dependencies: quansync: 0.2.10 + '@redis/bloom@6.1.0(@redis/client@6.1.0(@opentelemetry/api@1.9.1))': + dependencies: + '@redis/client': 6.1.0(@opentelemetry/api@1.9.1) + + '@redis/client@6.1.0(@opentelemetry/api@1.9.1)': + dependencies: + cluster-key-slot: 1.1.2 + optionalDependencies: + '@opentelemetry/api': 1.9.1 + + '@redis/json@6.1.0(@redis/client@6.1.0(@opentelemetry/api@1.9.1))': + dependencies: + '@redis/client': 6.1.0(@opentelemetry/api@1.9.1) + + '@redis/search@6.1.0(@redis/client@6.1.0(@opentelemetry/api@1.9.1))': + dependencies: + '@redis/client': 6.1.0(@opentelemetry/api@1.9.1) + + '@redis/time-series@6.1.0(@redis/client@6.1.0(@opentelemetry/api@1.9.1))': + dependencies: + '@redis/client': 6.1.0(@opentelemetry/api@1.9.1) + '@rolldown/binding-darwin-arm64@1.0.0-beta.15': optional: true @@ -2891,6 +2988,8 @@ snapshots: cluster-key-slot@1.1.1: {} + cluster-key-slot@1.1.2: {} + color-convert@2.0.1: dependencies: color-name: 1.1.4 @@ -3552,6 +3651,17 @@ snapshots: dependencies: redis-errors: 1.2.0 + redis@6.1.0(@opentelemetry/api@1.9.1): + dependencies: + '@redis/bloom': 6.1.0(@redis/client@6.1.0(@opentelemetry/api@1.9.1)) + '@redis/client': 6.1.0(@opentelemetry/api@1.9.1) + '@redis/json': 6.1.0(@redis/client@6.1.0(@opentelemetry/api@1.9.1)) + '@redis/search': 6.1.0(@redis/client@6.1.0(@opentelemetry/api@1.9.1)) + '@redis/time-series': 6.1.0(@redis/client@6.1.0(@opentelemetry/api@1.9.1)) + transitivePeerDependencies: + - '@node-rs/xxhash' + - '@opentelemetry/api' + regex-recursion@6.0.2: dependencies: regex-utilities: 2.3.0 From a2c25541bf96e98cd934f4c4c4f6aa95b16f5154 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 20:38:26 +0900 Subject: [PATCH 02/16] Harden Redis repository concurrency Address review findings in the Redis repository by making connection startup failures handled, making lock timings configurable, and using a less aggressive default lock renewal interval. Also guard message removal with the same lock as message updates, handle zero-limit follower pagination, and parallelize independent Redis/JWK work. Document the new lock timing options and add the missing throws notes for public methods that validate caller input. The Redis test suite now covers the message delete/update race, zero-limit follower pagination, and lock timing option validation. https://github.com/fedify-dev/botkit/pull/35#discussion_r3535832142 https://github.com/fedify-dev/botkit/pull/35#discussion_r3535832152 https://github.com/fedify-dev/botkit/pull/35#discussion_r3535832156 https://github.com/fedify-dev/botkit/pull/35#discussion_r3535832160 https://github.com/fedify-dev/botkit/pull/35#discussion_r3535832165 https://github.com/fedify-dev/botkit/pull/35#discussion_r3535857266 https://github.com/fedify-dev/botkit/pull/35#discussion_r3535857272 https://github.com/fedify-dev/botkit/pull/35#discussion_r3535859046 Assisted-by: Codex:gpt-5.5 --- docs/concepts/repository.md | 12 +++ packages/botkit-redis/README.md | 9 ++ packages/botkit-redis/src/mod.test.ts | 127 +++++++++++++++++++------- packages/botkit-redis/src/mod.ts | 120 +++++++++++++++++++----- 4 files changed, 211 insertions(+), 57 deletions(-) diff --git a/docs/concepts/repository.md b/docs/concepts/repository.md index 82a65f2..e463af3 100644 --- a/docs/concepts/repository.md +++ b/docs/concepts/repository.md @@ -220,6 +220,18 @@ The `RedisRepository` constructor accepts the following properties: : Additional node-redis client options. This option is only valid when `url` is used. +`lockTimeoutMs` (optional) +: How long a Redis lock can live without renewal, in milliseconds. + Defaults to `30000`. + +`lockPollIntervalMs` (optional) +: How long to wait before retrying a held Redis lock, in milliseconds. + Defaults to `20`. + +`lockRenewIntervalMs` (optional) +: How often to renew a held Redis lock, in milliseconds. Defaults to + `10000`. + [Redis]: https://redis.io/ [node-redis]: https://github.com/redis/node-redis diff --git a/packages/botkit-redis/README.md b/packages/botkit-redis/README.md index f5f9941..4e93c8d 100644 --- a/packages/botkit-redis/README.md +++ b/packages/botkit-redis/README.md @@ -84,6 +84,15 @@ The `RedisRepository` constructor accepts the following options: - **`clientOptions`** (optional): Additional node-redis client options. This option is only valid together with `url`. + - **`lockTimeoutMs`** (optional): How long a Redis lock can live without + renewal, in milliseconds. Defaults to `30000`. + + - **`lockPollIntervalMs`** (optional): How long to wait before retrying a + held Redis lock, in milliseconds. Defaults to `20`. + + - **`lockRenewIntervalMs`** (optional): How often to renew a held Redis lock, + in milliseconds. Defaults to `10000`. + The options are mutually exclusive: use either `client` or `url`. diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index 1ce2417..6c5fb54 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -210,29 +210,6 @@ function createDelayedDeleteClient( }; } -function createShortLockTtlClient( - client: TestRedisClient, - lockKey: string, - ttlMs: number, -) { - return { - get isOpen(): boolean { - return client.isOpen; - }, - - sendCommand(args: readonly string[]): Promise { - const command = [...args]; - if ( - command[0] === "SET" && command[1] === lockKey && - command[4] === "PX" - ) { - command[5] = ttlMs.toString(); - } - return client.sendCommand(command); - }, - }; -} - if (redisUrl == null) { test("RedisRepository integration tests", { skip: true }, () => {}); } else { @@ -254,6 +231,25 @@ if (redisUrl == null) { "RedisRepositoryOptions must provide exactly one of client or url.", ), ); + assert.throws( + () => + new RedisRepository({ + client: { sendCommand: () => Promise.resolve(undefined) }, + lockTimeoutMs: 0, + }), + new RangeError("lockTimeoutMs must be a positive number."), + ); + assert.throws( + () => + new RedisRepository({ + client: { sendCommand: () => Promise.resolve(undefined) }, + lockTimeoutMs: 1_000, + lockRenewIntervalMs: 1_000, + }), + new RangeError( + "lockRenewIntervalMs must be less than lockTimeoutMs.", + ), + ); }); test("key pairs", async () => { @@ -387,21 +383,17 @@ if (redisUrl == null) { await firstClient.connect(); await secondClient.connect(); const messageId = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; - const lockKey = [ - prefix, - "bots", - "bot", - "messages", - messageId, - "lock", - ].join(":"); const firstRepository = new RedisRepository({ - client: createShortLockTtlClient(firstClient, lockKey, 1_500), + client: firstClient, prefix, + lockTimeoutMs: 1_500, + lockRenewIntervalMs: 500, }); const secondRepository = new RedisRepository({ client: secondClient, prefix, + lockTimeoutMs: 1_500, + lockRenewIntervalMs: 500, }); try { const message = createMessage( @@ -454,6 +446,56 @@ if (redisUrl == null) { } }); + test("serializes message removal with concurrent updates", async () => { + const { repository, cleanup } = createHarness(); + try { + const messageId = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; + const message = createMessage( + messageId, + "base", + "2025-01-01T00:00:00Z", + ); + await repository.addMessage("bot", messageId, message); + + let updaterStarted = false; + const update = repository.updateMessage( + "bot", + messageId, + async (existing) => { + assert.ok(existing instanceof Create); + updaterStarted = true; + await delay(50); + const content = await getMessageContent(existing); + return existing.clone({ + object: new Note({ + content: `${content}A`, + }), + }); + }, + ); + while (!updaterStarted) await delay(1); + const removal = repository.removeMessage("bot", messageId); + const [updated, removed] = await Promise.all([update, removal]); + + assert.ok(updated); + assert.deepStrictEqual( + await removed?.toJsonLd(), + await message.clone({ + object: new Note({ + content: "baseA", + }), + }).toJsonLd(), + ); + assert.deepStrictEqual( + await repository.getMessage("bot", messageId), + undefined, + ); + assert.deepStrictEqual(await repository.countMessages("bot"), 0); + } finally { + await cleanup(); + } + }); + test("persists data across repository instances", async () => { if (redisUrl == null) throw new Error("REDIS_URL is not set."); const prefix = `botkit_test_${crypto.randomUUID()}`; @@ -513,6 +555,27 @@ if (redisUrl == null) { } }); + test("returns no followers for zero limits", async () => { + const { repository, cleanup } = createHarness(); + try { + const follower = new Person({ + id: new URL("https://example.com/ap/actor/alice"), + preferredUsername: "alice", + }); + const follow = new URL( + "https://example.com/ap/follow/f2fb7255-d3ad-4fef-8f9a-1d0f2c2ec0b4", + ); + + await repository.addFollower("bot", follow, follower); + assert.deepStrictEqual( + await Array.fromAsync(repository.getFollowers("bot", { limit: 0 })), + [], + ); + } finally { + await cleanup(); + } + }); + test("serializes racing follower replacement and removal", async () => { const { repository, cleanup } = createHarness(); try { diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts index 1ce734a..684da17 100644 --- a/packages/botkit-redis/src/mod.ts +++ b/packages/botkit-redis/src/mod.ts @@ -35,9 +35,9 @@ import { getLogger } from "@logtape/logtape"; import { createClient, type RedisClientOptions } from "redis"; const logger = getLogger(["botkit", "redis"]); -const redisLockTimeoutMs = 30_000; -const redisLockPollIntervalMs = 20; -const redisLockRenewIntervalMs = 1_000; +const defaultRedisLockTimeoutMs = 30_000; +const defaultRedisLockPollIntervalMs = 20; +const defaultRedisLockRenewIntervalMs = 10_000; const uuidV7Pattern = /^([0-9a-f]{8})-([0-9a-f]{4})-7[0-9a-f]{3}-[0-9a-f]{4}-[0-9a-f]{12}$/i; @@ -65,6 +65,24 @@ interface RedisRepositoryOptionsBase { * @default `"botkit"` */ readonly prefix?: string; + + /** + * How long a Redis lock can live without renewal, in milliseconds. + * @default `30000` + */ + readonly lockTimeoutMs?: number; + + /** + * How long to wait before retrying a held Redis lock, in milliseconds. + * @default `20` + */ + readonly lockPollIntervalMs?: number; + + /** + * How often to renew a held Redis lock, in milliseconds. + * @default `10000` + */ + readonly lockRenewIntervalMs?: number; } interface RedisRepositoryOptionsWithClient extends RedisRepositoryOptionsBase { @@ -117,6 +135,9 @@ export class RedisRepository implements Repository, AsyncDisposable { private readonly client: RedisClientLike; private readonly ownsClient: boolean; private readonly ready: Promise; + private readonly lockTimeoutMs: number; + private readonly lockPollIntervalMs: number; + private readonly lockRenewIntervalMs: number; /** * The Redis key prefix under which all BotKit data is stored. @@ -126,6 +147,9 @@ export class RedisRepository implements Repository, AsyncDisposable { /** * Creates a new Redis repository. * @param options The options for creating the repository. + * @throws {TypeError} If exactly one of `client` and `url` is not provided. + * @throws {RangeError} If a lock timing option is not a positive number, or + * if `lockRenewIntervalMs` is not less than `lockTimeoutMs`. */ constructor(options: RedisRepositoryOptions) { const hasClient = "client" in options && options.client != null; @@ -136,6 +160,25 @@ export class RedisRepository implements Repository, AsyncDisposable { ); } this.prefix = options.prefix ?? "botkit"; + this.lockTimeoutMs = options.lockTimeoutMs ?? defaultRedisLockTimeoutMs; + this.lockPollIntervalMs = options.lockPollIntervalMs ?? + defaultRedisLockPollIntervalMs; + this.lockRenewIntervalMs = options.lockRenewIntervalMs ?? + defaultRedisLockRenewIntervalMs; + validatePositiveMilliseconds("lockTimeoutMs", this.lockTimeoutMs); + validatePositiveMilliseconds( + "lockPollIntervalMs", + this.lockPollIntervalMs, + ); + validatePositiveMilliseconds( + "lockRenewIntervalMs", + this.lockRenewIntervalMs, + ); + if (this.lockRenewIntervalMs >= this.lockTimeoutMs) { + throw new RangeError( + "lockRenewIntervalMs must be less than lockTimeoutMs.", + ); + } if (hasClient) { this.client = options.client; this.ownsClient = false; @@ -147,7 +190,9 @@ export class RedisRepository implements Repository, AsyncDisposable { }) as unknown as RedisClientLike; this.client = client; this.ownsClient = true; - this.ready = client.connect?.() ?? Promise.resolve(); + const ready = client.connect?.() ?? Promise.resolve(); + ready.catch(() => {}); + this.ready = ready; } } @@ -159,7 +204,11 @@ export class RedisRepository implements Repository, AsyncDisposable { * Closes the owned Redis connection. */ async close(): Promise { - await this.ready; + try { + await this.ready; + } catch { + return; + } if (!this.ownsClient) return; if (this.client.quit != null) { await this.client.quit(); @@ -274,10 +323,10 @@ export class RedisRepository implements Repository, AsyncDisposable { token, "NX", "PX", - redisLockTimeoutMs.toString(), + this.lockTimeoutMs.toString(), ]) !== "OK" ) { - await delay(redisLockPollIntervalMs); + await delay(this.lockPollIntervalMs); } const renew = setInterval(() => { void (async () => { @@ -289,13 +338,13 @@ export class RedisRepository implements Repository, AsyncDisposable { "1", lockKey, token, - redisLockTimeoutMs.toString(), + this.lockTimeoutMs.toString(), ]); } catch (error) { logger.warn("Failed to renew Redis lock: {error}", { error }); } })(); - }, redisLockRenewIntervalMs); + }, this.lockRenewIntervalMs); try { return await operation(); } finally { @@ -319,13 +368,12 @@ export class RedisRepository implements Repository, AsyncDisposable { identifier: string, keyPairs: CryptoKeyPair[], ): Promise { - const pairs: KeyPair[] = []; - for (const keyPair of keyPairs) { - pairs.push({ + const pairs: KeyPair[] = await Promise.all( + keyPairs.map(async (keyPair) => ({ private: await exportJwk(keyPair.privateKey), public: await exportJwk(keyPair.publicKey), - }); - } + })), + ); await this.set(this.botKey(identifier, "keyPairs"), JSON.stringify(pairs)); } @@ -393,15 +441,20 @@ export class RedisRepository implements Repository, AsyncDisposable { id: Uuid, ): Promise { const key = this.botKey(identifier, "messages", id); - const json = await this.get(key); - await this.del(key); - await this.zRem(this.botKey(identifier, "messages"), id); - if (json == null) return undefined; - const activity = await Activity.fromJsonLd(JSON.parse(json)); - if (activity instanceof Create || activity instanceof Announce) { - return activity; - } - return undefined; + return await this.withRedisLock( + this.botKey(identifier, "messages", id, "lock"), + async () => { + const json = await this.get(key); + await this.del(key); + await this.zRem(this.botKey(identifier, "messages"), id); + if (json == null) return undefined; + const activity = await Activity.fromJsonLd(JSON.parse(json)); + if (activity instanceof Create || activity instanceof Announce) { + return activity; + } + return undefined; + }, + ); } async *getMessages( @@ -456,6 +509,11 @@ export class RedisRepository implements Repository, AsyncDisposable { return this.zCard(this.botKey(identifier, "messages")); } + /** + * Adds a follower for a follow request. + * + * @throws {TypeError} If the follower actor has no ID. + */ async addFollower( identifier: string, followId: URL, @@ -566,6 +624,7 @@ export class RedisRepository implements Repository, AsyncDisposable { identifier: string, options: RepositoryGetFollowersOptions = {}, ): AsyncIterable { + if (options.limit === 0) return; const start = options.offset ?? 0; const stop = options.limit == null ? -1 : start + options.limit - 1; const ids = toStringArray( @@ -674,6 +733,11 @@ export class RedisRepository implements Repository, AsyncDisposable { yield* await this.zRange(this.key("index", "followees", followeeId.href)); } + /** + * Stores a quote authorization. + * + * @throws {TypeError} If the quote authorization has no interacting object. + */ async addQuoteAuthorization( identifier: string, id: Uuid, @@ -900,11 +964,11 @@ export class RedisRepository implements Repository, AsyncDisposable { const options = await this.zRange( this.botKey(identifier, "polls", messageId, "options"), ); - for (const option of options) { + await Promise.all(options.map(async (option) => { result[option] = await this.sCard( this.botKey(identifier, "polls", messageId, "options", option), ); - } + })); return result; } @@ -924,6 +988,12 @@ function toNumber(value: unknown): number { return 0; } +function validatePositiveMilliseconds(name: string, value: number): void { + if (!Number.isFinite(value) || value <= 0) { + throw new RangeError(`${name} must be a positive number.`); + } +} + function toStringArray(value: unknown): string[] { if (!Array.isArray(value)) return []; return value.filter((item): item is string => typeof item === "string"); From f95cd7b5014d720c65c98c11385eb2daf015420f Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 20:52:31 +0900 Subject: [PATCH 03/16] Serialize Redis index writes Protect message creation and followee creation with the same Redis locks used by their removal paths. This keeps value writes and index writes atomic with respect to deletes, preventing phantom message and followee reverse-index entries under concurrent add/remove races. Add regression tests that delay the indexing write while a concurrent remove runs, then verify the record and index are both gone after the serialized operations complete. https://github.com/fedify-dev/botkit/pull/35#discussion_r3536121697 https://github.com/fedify-dev/botkit/pull/35#discussion_r3536121702 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.test.ts | 158 ++++++++++++++++++++++++++ packages/botkit-redis/src/mod.ts | 59 ++++++---- 2 files changed, 195 insertions(+), 22 deletions(-) diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index 6c5fb54..69232fc 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -210,6 +210,29 @@ function createDelayedDeleteClient( }; } +function createDelayedCommandClient( + client: TestRedisClient, + matches: (args: readonly string[]) => boolean, + observed: Deferred, + release: Deferred, +) { + return { + get isOpen(): boolean { + return client.isOpen; + }, + + sendCommand(args: readonly string[]): Promise { + return (async () => { + if (matches(args)) { + observed.resolve(); + await release.promise; + } + return await client.sendCommand([...args]); + })(); + }, + }; +} + if (redisUrl == null) { test("RedisRepository integration tests", { skip: true }, () => {}); } else { @@ -496,6 +519,73 @@ if (redisUrl == null) { } }); + test("serializes message indexing with removal", async () => { + const prefix = `botkit_test_${crypto.randomUUID()}`; + const addClient = createClient({ url: redisUrl }); + const removeClient = createClient({ url: redisUrl }); + await addClient.connect(); + await removeClient.connect(); + const messageId = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; + const indexKey = [ + prefix, + "bots", + "bot", + "messages", + ].join(":"); + const observed = createDeferred(); + const release = createDeferred(); + const addRepository = new RedisRepository({ + client: createDelayedCommandClient( + addClient, + (args) => + args[0] === "ZADD" && args[1] === indexKey && + args.at(-1) === messageId, + observed, + release, + ), + prefix, + }); + const removeRepository = new RedisRepository({ + client: removeClient, + prefix, + }); + try { + const message = createMessage( + messageId, + "base", + "2025-01-01T00:00:00Z", + ); + + const add = addRepository.addMessage("bot", messageId, message); + await observed.promise; + const removal = removeRepository.removeMessage("bot", messageId); + await delay(50); + release.resolve(); + + await add; + assert.deepStrictEqual( + await (await removal)?.toJsonLd(), + await message.toJsonLd(), + ); + assert.deepStrictEqual( + await addRepository.getMessage("bot", messageId), + undefined, + ); + assert.deepStrictEqual(await addRepository.countMessages("bot"), 0); + assert.deepStrictEqual( + await Array.fromAsync(addRepository.getMessages("bot")), + [], + ); + } finally { + release.resolve(); + await addRepository.close(); + await removeRepository.close(); + await addClient.quit(); + await removeClient.quit(); + await cleanupPrefix(redisUrl, prefix); + } + }); + test("persists data across repository instances", async () => { if (redisUrl == null) throw new Error("REDIS_URL is not set."); const prefix = `botkit_test_${crypto.randomUUID()}`; @@ -654,6 +744,74 @@ if (redisUrl == null) { } }); + test("serializes followee indexes with removals", async () => { + const prefix = `botkit_test_${crypto.randomUUID()}`; + const addClient = createClient({ url: redisUrl }); + const removeClient = createClient({ url: redisUrl }); + await addClient.connect(); + await removeClient.connect(); + const followeeId = new URL("https://example.com/ap/actor/john"); + const indexKey = [ + prefix, + "index", + "followees", + encodeURIComponent(followeeId.href), + ].join(":"); + const observed = createDeferred(); + const release = createDeferred(); + const addRepository = new RedisRepository({ + client: createDelayedCommandClient( + addClient, + (args) => + args[0] === "ZADD" && args[1] === indexKey && + args.at(-1) === "bot", + observed, + release, + ), + prefix, + }); + const removeRepository = new RedisRepository({ + client: removeClient, + prefix, + }); + try { + const follow = new Follow({ + id: new URL( + "https://example.com/ap/actor/bot/follow/03a395a2-353a-4894-afdb-2cab31a7b004", + ), + actor: new URL("https://example.com/ap/actor/bot"), + object: followeeId, + }); + + const add = addRepository.addFollowee("bot", followeeId, follow); + await observed.promise; + const removal = removeRepository.removeFollowee("bot", followeeId); + await delay(50); + release.resolve(); + + await add; + assert.deepStrictEqual( + await (await removal)?.toJsonLd(), + await follow.toJsonLd(), + ); + assert.deepStrictEqual( + await addRepository.getFollowee("bot", followeeId), + undefined, + ); + assert.deepStrictEqual( + await Array.fromAsync(addRepository.findFollowedBots(followeeId)), + [], + ); + } finally { + release.resolve(); + await addRepository.close(); + await removeRepository.close(); + await addClient.quit(); + await removeClient.quit(); + await cleanupPrefix(redisUrl, prefix); + } + }); + test("poll voting", async () => { const { repository, cleanup } = createHarness(); try { diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts index 684da17..ea48319 100644 --- a/packages/botkit-redis/src/mod.ts +++ b/packages/botkit-redis/src/mod.ts @@ -392,14 +392,19 @@ export class RedisRepository implements Repository, AsyncDisposable { id: Uuid, activity: Create | Announce, ): Promise { - await this.set( - this.botKey(identifier, "messages", id), - JSON.stringify(await activity.toJsonLd({ format: "compact" })), - ); - await this.zAdd( - this.botKey(identifier, "messages"), - getMessageScore(id, activity), - id, + await this.withRedisLock( + this.botKey(identifier, "messages", id, "lock"), + async () => { + await this.set( + this.botKey(identifier, "messages", id), + JSON.stringify(await activity.toJsonLd({ format: "compact" })), + ); + await this.zAdd( + this.botKey(identifier, "messages"), + getMessageScore(id, activity), + id, + ); + }, ); } @@ -690,14 +695,19 @@ export class RedisRepository implements Repository, AsyncDisposable { followeeId: URL, follow: Follow, ): Promise { - await this.set( - this.botKey(identifier, "followees", followeeId.href), - JSON.stringify(await follow.toJsonLd({ format: "compact" })), - ); - await this.zAdd( - this.key("index", "followees", followeeId.href), - 0, - identifier, + await this.withRedisLock( + this.botKey(identifier, "followees", followeeId.href, "lock"), + async () => { + await this.set( + this.botKey(identifier, "followees", followeeId.href), + JSON.stringify(await follow.toJsonLd({ format: "compact" })), + ); + await this.zAdd( + this.key("index", "followees", followeeId.href), + 0, + identifier, + ); + }, ); } @@ -705,13 +715,18 @@ export class RedisRepository implements Repository, AsyncDisposable { identifier: string, followeeId: URL, ): Promise { - const follow = await this.getFollowee(identifier, followeeId); - await this.del(this.botKey(identifier, "followees", followeeId.href)); - await this.zRem( - this.key("index", "followees", followeeId.href), - identifier, + return await this.withRedisLock( + this.botKey(identifier, "followees", followeeId.href, "lock"), + async () => { + const follow = await this.getFollowee(identifier, followeeId); + await this.del(this.botKey(identifier, "followees", followeeId.href)); + await this.zRem( + this.key("index", "followees", followeeId.href), + identifier, + ); + return follow; + }, ); - return follow; } async getFollowee( From 8eb27fb9ceb3198e736dc8521d1c2493e51bda41 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 21:04:40 +0900 Subject: [PATCH 04/16] Close owned Redis clients reliably Owned node-redis clients can emit error events during connection and reconnection failures. Attach an error listener before connecting so a repository-managed client does not terminate the Node.js process on a Redis outage. Also continue through close() after the initial ready promise rejects. That lets the repository attempt client cleanup and log cleanup failures instead of returning before the owned client has been given a chance to release sockets or timers. https://github.com/fedify-dev/botkit/pull/35#discussion_r3536189416 https://github.com/fedify-dev/botkit/pull/35#discussion_r3536222641 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.test.ts | 13 +++++++++++++ packages/botkit-redis/src/mod.ts | 25 +++++++++++++++++++------ 2 files changed, 32 insertions(+), 6 deletions(-) diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index 69232fc..c7b42a9 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -310,6 +310,19 @@ if (redisUrl == null) { } }); + test("closes owned clients after failed initial connection", async () => { + const repository = new RedisRepository({ + url: new URL("redis://127.0.0.1:1"), + clientOptions: { + socket: { + connectTimeout: 50, + reconnectStrategy: false, + }, + }, + }); + await assert.doesNotReject(repository.close()); + }); + test("messages basic operations and ordering", async () => { const { repository, cleanup } = createHarness(); try { diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts index ea48319..214f93e 100644 --- a/packages/botkit-redis/src/mod.ts +++ b/packages/botkit-redis/src/mod.ts @@ -46,6 +46,7 @@ interface RedisClientLike { connect?(): Promise; quit?(): Promise; close?(): Promise | unknown; + on?(event: "error", listener: (error: unknown) => void): unknown; sendCommand(args: readonly string[]): Promise; } @@ -190,6 +191,9 @@ export class RedisRepository implements Repository, AsyncDisposable { }) as unknown as RedisClientLike; this.client = client; this.ownsClient = true; + client.on?.("error", (error) => { + logger.warn("Owned Redis client emitted an error: {error}", { error }); + }); const ready = client.connect?.() ?? Promise.resolve(); ready.catch(() => {}); this.ready = ready; @@ -206,14 +210,23 @@ export class RedisRepository implements Repository, AsyncDisposable { async close(): Promise { try { await this.ready; - } catch { - return; + } catch (error) { + logger.warn( + "Owned Redis client did not become ready before close: {error}", + { + error, + }, + ); } if (!this.ownsClient) return; - if (this.client.quit != null) { - await this.client.quit(); - } else { - await this.client.close?.(); + try { + if (this.client.quit != null) { + await this.client.quit(); + } else { + await this.client.close?.(); + } + } catch (error) { + logger.warn("Failed to close owned Redis client: {error}", { error }); } } From 4722f4591eaf99fd9445d0edfd6c5906da5f6264 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 21:17:55 +0900 Subject: [PATCH 05/16] Serialize Redis quote references Quote authorization reference updates need the same per-reference lock for both add and remove operations. Without that, a remove could delete a fresh reverse-index member written by a concurrent add, leaving the stored reference unreachable through the identifier index. The Redis lock renewal loop now also stops once the renewal script reports that the lock is no longer owned. That avoids continuing to issue renewal commands after another process has acquired or released the lock. https://github.com/fedify-dev/botkit/pull/35#discussion_r3536283975 https://github.com/fedify-dev/botkit/pull/35#discussion_r3536309500 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.test.ts | 60 +++++++++++++++++++++ packages/botkit-redis/src/mod.ts | 77 ++++++++++++++++++--------- 2 files changed, 111 insertions(+), 26 deletions(-) diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index c7b42a9..82f9cc6 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -925,6 +925,66 @@ if (redisUrl == null) { } }); + test("serializes quote authorization reference indexes with removals", async () => { + const prefix = `botkit_test_${crypto.randomUUID()}`; + const addRepository = new RedisRepository({ url: redisUrl, prefix }); + const removeClient = createClient({ url: redisUrl }); + await removeClient.connect(); + const stamp = new URL("https://remote.example/stamps/serialized"); + const indexKey = `${prefix}:index:quoteAuthorizationRefs:${ + encodeURIComponent(stamp.href) + }`; + const removeObserved = createDeferred(); + const releaseRemove = createDeferred(); + const removeRepository = new RedisRepository({ + client: createDelayedCommandClient( + removeClient, + (args) => + args[0] === "ZREM" && args[1] === indexKey && args[2] === "bot", + removeObserved, + releaseRemove, + ), + prefix, + }); + try { + const oldId = "01942976-3400-7f34-872e-2cbf0f9eeac4"; + const newId = "01942976-3400-7f34-872e-2cbf0f9eeac5"; + await addRepository.addQuoteAuthorizationReference("bot", stamp, oldId); + const removePromise = removeRepository + .removeQuoteAuthorizationReference( + "bot", + stamp, + ); + await removeObserved.promise; + let addFinished = false; + const addPromise = addRepository + .addQuoteAuthorizationReference("bot", stamp, newId) + .then(() => { + addFinished = true; + }); + await delay(50); + assert.ok(!addFinished); + releaseRemove.resolve(); + await Promise.all([removePromise, addPromise]); + assert.deepStrictEqual( + await addRepository.findQuoteAuthorizationReference("bot", stamp), + newId, + ); + assert.deepStrictEqual( + await Array.fromAsync( + addRepository.findQuoteAuthorizationReferenceIdentifiers(stamp), + ), + ["bot"], + ); + } finally { + releaseRemove.resolve(); + await addRepository.close(); + await removeRepository.close(); + await removeClient.quit(); + await cleanupPrefix(redisUrl, prefix); + } + }); + test("serializes concurrent quote authorization inserts", async () => { const prefix = `botkit_test_${crypto.randomUUID()}`; const firstClient = createClient({ url: redisUrl }); diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts index 214f93e..7d236f9 100644 --- a/packages/botkit-redis/src/mod.ts +++ b/packages/botkit-redis/src/mod.ts @@ -344,15 +344,18 @@ export class RedisRepository implements Repository, AsyncDisposable { const renew = setInterval(() => { void (async () => { try { - await this.command([ - "EVAL", - "if redis.call('GET', KEYS[1]) == ARGV[1] then " + - "return redis.call('PEXPIRE', KEYS[1], ARGV[2]) else return 0 end", - "1", - lockKey, - token, - this.lockTimeoutMs.toString(), - ]); + const renewed = toNumber( + await this.command([ + "EVAL", + "if redis.call('GET', KEYS[1]) == ARGV[1] then " + + "return redis.call('PEXPIRE', KEYS[1], ARGV[2]) else return 0 end", + "1", + lockKey, + token, + this.lockTimeoutMs.toString(), + ]), + ); + if (renewed === 0) clearInterval(renew); } catch (error) { logger.warn("Failed to renew Redis lock: {error}", { error }); } @@ -886,17 +889,22 @@ export class RedisRepository implements Repository, AsyncDisposable { messageId: Uuid, attribution?: URL, ): Promise { - const value: QuoteAuthorizationReferenceData = attribution == null - ? { messageId } - : { messageId, attribution: attribution.href }; - await this.set( - this.botKey(identifier, "quoteAuthorizationRefs", authorization.href), - JSON.stringify(value), - ); - await this.zAdd( - this.key("index", "quoteAuthorizationRefs", authorization.href), - 0, - identifier, + await this.withRedisLock( + this.quoteAuthorizationReferenceLockKey(identifier, authorization), + async () => { + const value: QuoteAuthorizationReferenceData = attribution == null + ? { messageId } + : { messageId, attribution: attribution.href }; + await this.set( + this.botKey(identifier, "quoteAuthorizationRefs", authorization.href), + JSON.stringify(value), + ); + await this.zAdd( + this.key("index", "quoteAuthorizationRefs", authorization.href), + 0, + identifier, + ); + }, ); } @@ -950,12 +958,17 @@ export class RedisRepository implements Repository, AsyncDisposable { identifier: string, authorization: URL, ): Promise { - await this.del( - this.botKey(identifier, "quoteAuthorizationRefs", authorization.href), - ); - await this.zRem( - this.key("index", "quoteAuthorizationRefs", authorization.href), - identifier, + await this.withRedisLock( + this.quoteAuthorizationReferenceLockKey(identifier, authorization), + async () => { + await this.del( + this.botKey(identifier, "quoteAuthorizationRefs", authorization.href), + ); + await this.zRem( + this.key("index", "quoteAuthorizationRefs", authorization.href), + identifier, + ); + }, ); } @@ -1007,6 +1020,18 @@ export class RedisRepository implements Repository, AsyncDisposable { private quoteAuthorizationLockKey(indexKey: string): string { return `${indexKey}:lock`; } + + private quoteAuthorizationReferenceLockKey( + identifier: string, + authorization: URL, + ): string { + return this.botKey( + identifier, + "quoteAuthorizationRefs", + authorization.href, + "lock", + ); + } } function toNumber(value: unknown): number { From d5601b119ac0f9fdf267b550741af28357050ae9 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 21:31:32 +0900 Subject: [PATCH 06/16] Harden Redis reconnect and quote removal Owned Redis clients no longer keep a rejected initial connection promise forever. When the eager connection fails, the cached readiness promise is cleared so the next repository command can start a fresh connect attempt. Quote authorization removal now uses the same interacting-object index lock as insertion before deleting the stored authorization or reverse index. A small reverse lookup from authorization ID to interacting object lets removal acquire that lock before inspecting the stored authorization, which prevents a replacement insertion from observing stale state and being dropped. https://github.com/fedify-dev/botkit/pull/35#discussion_r3536362339 https://github.com/fedify-dev/botkit/pull/35#discussion_r3536381349 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.test.ts | 74 ++++++++++++++++ packages/botkit-redis/src/mod.ts | 118 ++++++++++++++++++++------ 2 files changed, 165 insertions(+), 27 deletions(-) diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index 82f9cc6..023c6f9 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -1053,6 +1053,80 @@ if (redisUrl == null) { } }); + test("serializes quote authorization removal with inserts", async () => { + const prefix = `botkit_test_${crypto.randomUUID()}`; + const addRepository = new RedisRepository({ url: redisUrl, prefix }); + const removeClient = createClient({ url: redisUrl }); + await removeClient.connect(); + const quote = new URL("https://remote.example/notes/remove-race"); + const target = new URL("https://example.com/ap/note/remove-race"); + const oldId = "01942976-3400-7f34-872e-2cbf0f9eeac6"; + const newId = "01942976-3400-7f34-872e-2cbf0f9eeac7"; + const oldAuthorization = new QuoteAuthorization({ + id: new URL( + `https://example.com/ap/actor/bot/quote-authorization/${oldId}`, + ), + attribution: new URL("https://example.com/ap/actor/bot"), + interactingObject: quote, + interactionTarget: target, + }); + const newAuthorization = new QuoteAuthorization({ + id: new URL( + `https://example.com/ap/actor/bot/quote-authorization/${newId}`, + ), + attribution: new URL("https://example.com/ap/actor/bot"), + interactingObject: quote, + interactionTarget: target, + }); + const authorizationKey = + `${prefix}:bots:bot:quoteAuthorizations:${oldId}`; + const removeObserved = createDeferred(); + const releaseRemove = createDeferred(); + const removeRepository = new RedisRepository({ + client: createDelayedCommandClient( + removeClient, + (args) => + args[0] === "DEL" && + args.includes(authorizationKey), + removeObserved, + releaseRemove, + ), + prefix, + }); + try { + await addRepository.addQuoteAuthorization( + "bot", + oldId, + oldAuthorization, + ); + const removePromise = removeRepository.removeQuoteAuthorization( + "bot", + oldId, + ); + await removeObserved.promise; + let addFinished = false; + const addPromise = addRepository + .addQuoteAuthorization("bot", newId, newAuthorization) + .then(() => { + addFinished = true; + }); + await delay(50); + assert.ok(!addFinished); + releaseRemove.resolve(); + await Promise.all([removePromise, addPromise]); + assert.deepStrictEqual( + (await addRepository.findQuoteAuthorization("bot", quote))?.id?.href, + newAuthorization.id?.href, + ); + } finally { + releaseRemove.resolve(); + await addRepository.close(); + await removeRepository.close(); + await removeClient.quit(); + await cleanupPrefix(redisUrl, prefix); + } + }); + test("preserves quote authorization indexes during stale cleanup", async () => { const prefix = `botkit_test_${crypto.randomUUID()}`; const setupClient = createClient({ url: redisUrl }); diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts index 7d236f9..63f8203 100644 --- a/packages/botkit-redis/src/mod.ts +++ b/packages/botkit-redis/src/mod.ts @@ -135,7 +135,7 @@ export type RedisRepositoryOptions = export class RedisRepository implements Repository, AsyncDisposable { private readonly client: RedisClientLike; private readonly ownsClient: boolean; - private readonly ready: Promise; + private ready: Promise | undefined; private readonly lockTimeoutMs: number; private readonly lockPollIntervalMs: number; private readonly lockRenewIntervalMs: number; @@ -183,7 +183,6 @@ export class RedisRepository implements Repository, AsyncDisposable { if (hasClient) { this.client = options.client; this.ownsClient = false; - this.ready = Promise.resolve(); } else { const client = createClient({ ...options.clientOptions, @@ -194,9 +193,10 @@ export class RedisRepository implements Repository, AsyncDisposable { client.on?.("error", (error) => { logger.warn("Owned Redis client emitted an error: {error}", { error }); }); - const ready = client.connect?.() ?? Promise.resolve(); - ready.catch(() => {}); - this.ready = ready; + const ready = this.connect(); + ready.catch(() => { + if (this.ready === ready) this.ready = undefined; + }); } } @@ -208,8 +208,9 @@ export class RedisRepository implements Repository, AsyncDisposable { * Closes the owned Redis connection. */ async close(): Promise { + const ready = this.ready; try { - await this.ready; + await ready; } catch (error) { logger.warn( "Owned Redis client did not become ready before close: {error}", @@ -217,6 +218,7 @@ export class RedisRepository implements Repository, AsyncDisposable { error, }, ); + if (this.ready === ready) this.ready = undefined; } if (!this.ownsClient) return; try { @@ -241,8 +243,25 @@ export class RedisRepository implements Repository, AsyncDisposable { return this.key("bots", identifier, ...segments); } + private connect(): Promise { + const ready = this.client.connect?.() ?? Promise.resolve(); + this.ready = ready; + return ready; + } + + private async ensureReady(): Promise { + if (!this.ownsClient) return; + const ready = this.ready ?? this.connect(); + try { + await ready; + } catch (error) { + if (this.ready === ready) this.ready = undefined; + throw error; + } + } + private async command(args: readonly string[]): Promise { - await this.ready; + await this.ensureReady(); try { return await this.client.sendCommand(args); } catch (error) { @@ -797,11 +816,22 @@ export class RedisRepository implements Repository, AsyncDisposable { return; } await this.del(indexKey); + await this.del( + this.botKey( + identifier, + "quoteAuthorizationInteractingObjects", + existing, + ), + ); } await this.set( this.botKey(identifier, "quoteAuthorizations", id), JSON.stringify(await authorization.toJsonLd({ format: "compact" })), ); + await this.set( + this.botKey(identifier, "quoteAuthorizationInteractingObjects", id), + interactingObject.href, + ); await this.set(indexKey, id); }, ); @@ -842,6 +872,9 @@ export class RedisRepository implements Repository, AsyncDisposable { ); if (authorization == null && await this.get(indexKey) === id) { await this.del(indexKey); + await this.del( + this.botKey(identifier, "quoteAuthorizationInteractingObjects", id), + ); } return authorization; }, @@ -853,31 +886,62 @@ export class RedisRepository implements Repository, AsyncDisposable { id: Uuid, ): Promise { const key = this.botKey(identifier, "quoteAuthorizations", id); - const json = await this.get(key); - await this.del(key); - if (json == null) return undefined; - let parsed: unknown; + const interactingObjectKey = this.botKey( + identifier, + "quoteAuthorizationInteractingObjects", + id, + ); + const storedInteractingObject = await this.get(interactingObjectKey) ?? + await this.findStoredQuoteAuthorizationInteractingObject(key); + if (storedInteractingObject == null) return undefined; + const indexKey = this.botKey( + identifier, + "quoteAuthorizationsByInteractingObject", + storedInteractingObject, + ); + const parsed = await this.withRedisLock( + this.quoteAuthorizationLockKey(indexKey), + async () => { + const json = await this.get(key); + if (json == null) { + await this.del(interactingObjectKey); + return undefined; + } + let parsed: unknown; + try { + parsed = JSON.parse(json); + } catch { + return undefined; + } + const interactingObject = getRawQuoteAuthorizationInteractingObject( + parsed, + ); + if (interactingObject?.href !== storedInteractingObject) { + return undefined; + } + await this.del(key, interactingObjectKey); + if (await this.get(indexKey) === id) await this.del(indexKey); + return parsed; + }, + ); + if (parsed == null) return undefined; try { - parsed = JSON.parse(json); + return await QuoteAuthorization.fromJsonLd(parsed); } catch { return undefined; } - const interactingObject = getRawQuoteAuthorizationInteractingObject(parsed); - if (interactingObject != null) { - const indexKey = this.botKey( - identifier, - "quoteAuthorizationsByInteractingObject", - interactingObject.href, - ); - await this.withRedisLock( - this.quoteAuthorizationLockKey(indexKey), - async () => { - if (await this.get(indexKey) === id) await this.del(indexKey); - }, - ); - } + } + + private async findStoredQuoteAuthorizationInteractingObject( + key: string, + ): Promise { + const json = await this.get(key); + if (json == null) return undefined; try { - return await QuoteAuthorization.fromJsonLd(parsed); + const interactingObject = getRawQuoteAuthorizationInteractingObject( + JSON.parse(json), + ); + return interactingObject?.href; } catch { return undefined; } From f1821cff7df97c0b8d9358251f1f7505f0635496 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 21:42:05 +0900 Subject: [PATCH 07/16] Avoid redundant Redis reconnects When an owned Redis client is already open, command readiness no longer tries to call connect() again after the cached ready promise has been cleared. This lets a client that reconnected in the background continue handling commands instead of failing with a duplicate socket-open error. Message removal now treats malformed stored JSON as an already-completed cleanup. The message key and sorted-set index are deleted before parsing, and parse failures return undefined instead of leaking an exception after cleanup has succeeded. https://github.com/fedify-dev/botkit/pull/35#discussion_r3536443778 https://github.com/fedify-dev/botkit/pull/35#discussion_r3536443787 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.test.ts | 37 +++++++++++++++++++++++++++ packages/botkit-redis/src/mod.ts | 11 +++++--- 2 files changed, 45 insertions(+), 3 deletions(-) diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index 023c6f9..c2cdc1e 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -323,6 +323,17 @@ if (redisUrl == null) { await assert.doesNotReject(repository.close()); }); + test("uses already open owned clients without reconnecting", async () => { + const { repository, cleanup } = createHarness(); + try { + assert.deepStrictEqual(await repository.countMessages("bot"), 0); + Reflect.set(repository, "ready", undefined); + await assert.doesNotReject(repository.countMessages("bot")); + } finally { + await cleanup(); + } + }); + test("messages basic operations and ordering", async () => { const { repository, cleanup } = createHarness(); try { @@ -369,6 +380,32 @@ if (redisUrl == null) { } }); + test("removes corrupted messages without throwing", async () => { + const { repository, cleanup, prefix } = createHarness(); + const client = createClient({ url: redisUrl }); + await client.connect(); + try { + const messageId = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; + const key = `${prefix}:bots:bot:messages:${messageId}`; + await client.sendCommand(["SET", key, "{"]); + await client.sendCommand([ + "ZADD", + `${prefix}:bots:bot:messages`, + "0", + messageId, + ]); + assert.deepStrictEqual( + await repository.removeMessage("bot", messageId), + undefined, + ); + assert.deepStrictEqual(await client.sendCommand(["GET", key]), null); + assert.deepStrictEqual(await repository.countMessages("bot"), 0); + } finally { + await client.quit(); + await cleanup(); + } + }); + test("serializes concurrent message updates", async () => { const { repository, cleanup } = createHarness(); try { diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts index 63f8203..5d2617b 100644 --- a/packages/botkit-redis/src/mod.ts +++ b/packages/botkit-redis/src/mod.ts @@ -251,6 +251,7 @@ export class RedisRepository implements Repository, AsyncDisposable { private async ensureReady(): Promise { if (!this.ownsClient) return; + if (this.client.isOpen) return; const ready = this.ready ?? this.connect(); try { await ready; @@ -488,9 +489,13 @@ export class RedisRepository implements Repository, AsyncDisposable { await this.del(key); await this.zRem(this.botKey(identifier, "messages"), id); if (json == null) return undefined; - const activity = await Activity.fromJsonLd(JSON.parse(json)); - if (activity instanceof Create || activity instanceof Announce) { - return activity; + try { + const activity = await Activity.fromJsonLd(JSON.parse(json)); + if (activity instanceof Create || activity instanceof Announce) { + return activity; + } + } catch { + return undefined; } return undefined; }, From 212b334dfb23c3b6a2af90900b4c2e7b9eb21dca Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 22:20:56 +0900 Subject: [PATCH 08/16] Await pending Redis readiness Owned Redis clients may report an open socket before the initial ready promise has completed. Commands now wait for any pending ready promise before using isOpen as a reconnect shortcut, so repositories do not send commands during an incomplete connection handshake. A regression test covers the case where the client is already open but the repository still has an unresolved ready promise, ensuring commands wait for that promise and do not call connect() again. https://github.com/fedify-dev/botkit/pull/35#discussion_r3536750229 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.test.ts | 43 +++++++++++++++++++++++++++ packages/botkit-redis/src/mod.ts | 14 +++++++-- 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index c2cdc1e..05dfbea 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -334,6 +334,49 @@ if (redisUrl == null) { } }); + test("awaits pending readiness before commands", async () => { + const ready = createDeferred(); + let readyResolved = false; + let connectCalled = false; + let commandFinished = false; + const repository = new RedisRepository({ + client: { + get isOpen(): boolean { + return true; + }, + + connect(): Promise { + connectCalled = true; + return Promise.reject(new TypeError("Unexpected reconnect.")); + }, + + sendCommand(args: readonly string[]): Promise { + assert.deepStrictEqual(args[0], "ZCARD"); + assert.ok(readyResolved); + return Promise.resolve(0); + }, + }, + }); + Reflect.set(repository, "ownsClient", true); + Reflect.set( + repository, + "ready", + ready.promise.then(() => { + readyResolved = true; + }), + ); + const countPromise = repository.countMessages("bot").then((count) => { + commandFinished = true; + return count; + }); + await delay(50); + assert.ok(!commandFinished); + assert.ok(!connectCalled); + ready.resolve(); + assert.deepStrictEqual(await countPromise, 0); + assert.ok(!connectCalled); + }); + test("messages basic operations and ordering", async () => { const { repository, cleanup } = createHarness(); try { diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts index 5d2617b..0f5fe87 100644 --- a/packages/botkit-redis/src/mod.ts +++ b/packages/botkit-redis/src/mod.ts @@ -251,8 +251,18 @@ export class RedisRepository implements Repository, AsyncDisposable { private async ensureReady(): Promise { if (!this.ownsClient) return; - if (this.client.isOpen) return; - const ready = this.ready ?? this.connect(); + const ready = this.ready; + if (ready == null) { + if (this.client.isOpen) return; + const nextReady = this.connect(); + try { + await nextReady; + } catch (error) { + if (this.ready === nextReady) this.ready = undefined; + throw error; + } + return; + } try { await ready; } catch (error) { From 7efa3dced13827e83906bf2441cc4eda2fc9926d Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 22:36:55 +0900 Subject: [PATCH 09/16] Harden Redis cleanup edge cases Injected Redis clients now make close() a complete no-op, including avoiding any cached readiness promise. Corrupted stored message JSON during update is handled as a failed update instead of invoking the updater or leaking a parse exception. Follower removal is retryable after a partial delete of the follow request marker: a retry can still remove the follow ID from the follower request set and run follower cleanup. Quote authorization reference attribution lookup also validates parsed JSON before reading object properties. https://github.com/fedify-dev/botkit/pull/35#discussion_r3536802943 https://github.com/fedify-dev/botkit/pull/35#discussion_r3536802957 https://github.com/fedify-dev/botkit/pull/35#discussion_r3536802962 https://github.com/fedify-dev/botkit/pull/35#discussion_r3536831069 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.test.ts | 64 ++++++++++++++++++++++++++- packages/botkit-redis/src/mod.ts | 18 ++++++-- 2 files changed, 76 insertions(+), 6 deletions(-) diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index 05dfbea..e986911 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -300,6 +300,9 @@ if (redisUrl == null) { const client = createClient({ url: redisUrl }); await client.connect(); const repository = new RedisRepository({ client, prefix }); + const rejectedReady = Promise.reject(new TypeError("Unexpected await.")); + rejectedReady.catch(() => {}); + Reflect.set(repository, "ready", rejectedReady); try { assert.ok(client.isOpen); await repository.close(); @@ -449,6 +452,29 @@ if (redisUrl == null) { } }); + test("ignores corrupted messages during updates", async () => { + const { repository, cleanup, prefix } = createHarness(); + const client = createClient({ url: redisUrl }); + await client.connect(); + try { + const messageId = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; + await client.sendCommand([ + "SET", + `${prefix}:bots:bot:messages:${messageId}`, + "{", + ]); + assert.deepStrictEqual( + await repository.updateMessage("bot", messageId, () => { + throw new TypeError("Unexpected updater call."); + }), + false, + ); + } finally { + await client.quit(); + await cleanup(); + } + }); + test("serializes concurrent message updates", async () => { const { repository, cleanup } = createHarness(); try { @@ -705,7 +731,9 @@ if (redisUrl == null) { }); test("followers with multiple follow requests", async () => { - const { repository, cleanup } = createHarness(); + const { repository, cleanup, prefix } = createHarness(); + const client = createClient({ url: redisUrl }); + await client.connect(); try { const follower = new Person({ id: new URL("https://example.com/ap/actor/alice"), @@ -733,7 +761,22 @@ if (redisUrl == null) { await follower.toJsonLd(), ); assert.deepStrictEqual(await repository.countFollowers("bot"), 0); + + await repository.addFollower("bot", followB, follower); + await client.sendCommand([ + "DEL", + `${prefix}:bots:bot:followRequests:${ + encodeURIComponent(followB.href) + }`, + ]); + assert.deepStrictEqual( + await (await repository.removeFollower("bot", followB, follower.id!)) + ?.toJsonLd(), + await follower.toJsonLd(), + ); + assert.deepStrictEqual(await repository.countFollowers("bot"), 0); } finally { + await client.quit(); await cleanup(); } }); @@ -929,7 +972,9 @@ if (redisUrl == null) { }); test("quote authorizations and references", async () => { - const { repository, cleanup } = createHarness(); + const { repository, cleanup, prefix } = createHarness(); + const client = createClient({ url: redisUrl }); + await client.connect(); try { const firstId = "01942976-3400-7f34-872e-2cbf0f9eeac4"; const secondId = "01942976-3400-7f34-872e-2cbf0f9eeac5"; @@ -987,6 +1032,20 @@ if (redisUrl == null) { ), attribution, ); + await client.sendCommand([ + "SET", + `${prefix}:bots:bot:quoteAuthorizationRefs:${ + encodeURIComponent(stamp.href) + }`, + "null", + ]); + assert.deepStrictEqual( + await repository.findQuoteAuthorizationReferenceAttribution( + "bot", + stamp, + ), + undefined, + ); assert.deepStrictEqual( await Array.fromAsync( repository.findQuoteAuthorizationReferenceIdentifiers(stamp), @@ -1001,6 +1060,7 @@ if (redisUrl == null) { ["other"], ); } finally { + await client.quit(); await cleanup(); } }); diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts index 0f5fe87..4be8e1b 100644 --- a/packages/botkit-redis/src/mod.ts +++ b/packages/botkit-redis/src/mod.ts @@ -208,6 +208,7 @@ export class RedisRepository implements Repository, AsyncDisposable { * Closes the owned Redis connection. */ async close(): Promise { + if (!this.ownsClient) return; const ready = this.ready; try { await ready; @@ -220,7 +221,6 @@ export class RedisRepository implements Repository, AsyncDisposable { ); if (this.ready === ready) this.ready = undefined; } - if (!this.ownsClient) return; try { if (this.client.quit != null) { await this.client.quit(); @@ -467,7 +467,12 @@ export class RedisRepository implements Repository, AsyncDisposable { async () => { const json = await this.get(key); if (json == null) return false; - const activity = await Activity.fromJsonLd(JSON.parse(json)); + let activity: Activity; + try { + activity = await Activity.fromJsonLd(JSON.parse(json)); + } catch { + return false; + } if (!(activity instanceof Create || activity instanceof Announce)) { return false; } @@ -629,14 +634,18 @@ export class RedisRepository implements Repository, AsyncDisposable { this.botKey(identifier, "followers", "lock"), async () => { const currentFollowerId = await this.get(requestKey); - if (currentFollowerId !== followerId.href) return undefined; + if ( + currentFollowerId != null && currentFollowerId !== followerId.href + ) { + return undefined; + } const followerKey = this.botKey( identifier, "followers", followerId.href, ); const json = await this.get(followerKey); - await this.del(requestKey); + if (currentFollowerId === followerId.href) await this.del(requestKey); await this.sRem( this.botKey(identifier, "followerRequests", followerId.href), followIdString, @@ -1025,6 +1034,7 @@ export class RedisRepository implements Repository, AsyncDisposable { } catch { return undefined; } + if (typeof value !== "object" || value == null) return undefined; if (value.attribution == null) return undefined; try { return new URL(value.attribution); From 7be16940c405bb98e8d9d391e7204d5dd90d432c Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 22:49:38 +0900 Subject: [PATCH 10/16] Tighten Redis client and cleanup code The owned Redis client is now assigned through the repository's compatible client interface directly, so TypeScript verifies the structural contract without erasing the node-redis type through an unknown double assertion. Stale quote authorization cleanup now deletes the stale forward and reverse index keys with a single Redis DEL command. This keeps the locked cleanup path atomic in behavior while avoiding an unnecessary roundtrip. https://github.com/fedify-dev/botkit/pull/35#discussion_r3536925824 https://github.com/fedify-dev/botkit/pull/35#discussion_r3536939104 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts index 4be8e1b..65b7463 100644 --- a/packages/botkit-redis/src/mod.ts +++ b/packages/botkit-redis/src/mod.ts @@ -184,10 +184,10 @@ export class RedisRepository implements Repository, AsyncDisposable { this.client = options.client; this.ownsClient = false; } else { - const client = createClient({ + const client: RedisClientLike = createClient({ ...options.clientOptions, url: options.url.toString(), - }) as unknown as RedisClientLike; + }); this.client = client; this.ownsClient = true; client.on?.("error", (error) => { @@ -839,8 +839,8 @@ export class RedisRepository implements Repository, AsyncDisposable { ) { return; } - await this.del(indexKey); await this.del( + indexKey, this.botKey( identifier, "quoteAuthorizationInteractingObjects", @@ -895,8 +895,8 @@ export class RedisRepository implements Repository, AsyncDisposable { id as Uuid, ); if (authorization == null && await this.get(indexKey) === id) { - await this.del(indexKey); await this.del( + indexKey, this.botKey(identifier, "quoteAuthorizationInteractingObjects", id), ); } From c536bdbc4f77c64e6d7a19bdb610709d99c4e7a8 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 23:03:20 +0900 Subject: [PATCH 11/16] Batch Redis list item fetches Message and follower listing now fetches the selected Redis payloads with one MGET command instead of issuing one GET per listed identifier. That keeps the sorted-set range lookup behavior unchanged while avoiding sequential network roundtrips on paginated list reads. The Redis repository tests now record commands during those list reads and assert that the batched MGET path is used without falling back to per-item GET commands. https://github.com/fedify-dev/botkit/pull/35#discussion_r3537011099 https://github.com/fedify-dev/botkit/pull/35#discussion_r3537011111 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.test.ts | 104 ++++++++++++++++++++++++++ packages/botkit-redis/src/mod.ts | 14 ++-- 2 files changed, 112 insertions(+), 6 deletions(-) diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index e986911..49826cc 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -233,6 +233,22 @@ function createDelayedCommandClient( }; } +function createRecordingClient( + client: TestRedisClient, + commands: string[][], +) { + return { + get isOpen(): boolean { + return client.isOpen; + }, + + async sendCommand(args: readonly string[]): Promise { + commands.push([...args]); + return await client.sendCommand([...args]); + }, + }; +} + if (redisUrl == null) { test("RedisRepository integration tests", { skip: true }, () => {}); } else { @@ -426,6 +442,47 @@ if (redisUrl == null) { } }); + test("fetches listed messages in one Redis command", async () => { + const prefix = `botkit_test_${crypto.randomUUID()}`; + const client = createClient({ url: redisUrl }); + await client.connect(); + const commands: string[][] = []; + const repository = new RedisRepository({ + client: createRecordingClient(client, commands), + prefix, + }); + try { + const firstId = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; + const secondId = "01942976-3400-7f34-872e-2cbf0f9eeac4"; + await repository.addMessage( + "bot", + firstId, + createMessage(firstId, "first", "2025-01-01T00:00:00Z"), + ); + await repository.addMessage( + "bot", + secondId, + createMessage(secondId, "second", "2025-01-02T00:00:00Z"), + ); + + commands.length = 0; + assert.deepStrictEqual( + (await Array.fromAsync(repository.getMessages("bot"))).length, + 2, + ); + assert.ok(commands.some(([command]) => command === "MGET")); + assert.ok( + !commands.some(([command, key]) => + command === "GET" && key?.includes(":messages:") + ), + ); + } finally { + await repository.close(); + await client.quit(); + await cleanupPrefix(redisUrl, prefix); + } + }); + test("removes corrupted messages without throwing", async () => { const { repository, cleanup, prefix } = createHarness(); const client = createClient({ url: redisUrl }); @@ -781,6 +838,53 @@ if (redisUrl == null) { } }); + test("fetches listed followers in one Redis command", async () => { + const prefix = `botkit_test_${crypto.randomUUID()}`; + const client = createClient({ url: redisUrl }); + await client.connect(); + const commands: string[][] = []; + const repository = new RedisRepository({ + client: createRecordingClient(client, commands), + prefix, + }); + try { + const alice = new Person({ + id: new URL("https://example.com/ap/actor/alice"), + preferredUsername: "alice", + }); + const bob = new Person({ + id: new URL("https://example.com/ap/actor/bob"), + preferredUsername: "bob", + }); + await repository.addFollower( + "bot", + new URL("https://example.com/ap/follow/alice"), + alice, + ); + await repository.addFollower( + "bot", + new URL("https://example.com/ap/follow/bob"), + bob, + ); + + commands.length = 0; + assert.deepStrictEqual( + (await Array.fromAsync(repository.getFollowers("bot"))).length, + 2, + ); + assert.ok(commands.some(([command]) => command === "MGET")); + assert.ok( + !commands.some(([command, key]) => + command === "GET" && key?.includes(":followers:") + ), + ); + } finally { + await repository.close(); + await client.quit(); + await cleanupPrefix(redisUrl, prefix); + } + }); + test("returns no followers for zero limits", async () => { const { repository, cleanup } = createHarness(); try { diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts index 65b7463..0b11244 100644 --- a/packages/botkit-redis/src/mod.ts +++ b/packages/botkit-redis/src/mod.ts @@ -534,9 +534,10 @@ export class RedisRepository implements Repository, AsyncDisposable { options.order == null || options.order === "newest", options.limit, ); - for (const id of ids) { - const json = await this.get(this.botKey(identifier, "messages", id)); - if (json == null) continue; + if (ids.length < 1) return; + const keys = ids.map((id) => this.botKey(identifier, "messages", id)); + const jsons = toStringArray(await this.command(["MGET", ...keys])); + for (const json of jsons) { try { const activity = await Activity.fromJsonLd(JSON.parse(json)); if (activity instanceof Create || activity instanceof Announce) { @@ -699,9 +700,10 @@ export class RedisRepository implements Repository, AsyncDisposable { stop.toString(), ]), ); - for (const id of ids) { - const json = await this.get(this.botKey(identifier, "followers", id)); - if (json == null) continue; + if (ids.length < 1) return; + const keys = ids.map((id) => this.botKey(identifier, "followers", id)); + const jsons = toStringArray(await this.command(["MGET", ...keys])); + for (const json of jsons) { try { const actor = await Object.fromJsonLd(JSON.parse(json)); if (isActor(actor)) yield actor; From c65e94a21f52c955f9c94a7626826103ba134cf9 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 23:04:55 +0900 Subject: [PATCH 12/16] Assert batched follower fetch keys The follower list test now checks that the Redis MGET command receives exactly the follower payload keys. This keeps the regression coverage tied to the batched lookup behavior requested in review, not only to the presence of any MGET command. https://github.com/fedify-dev/botkit/pull/35#discussion_r3537011107 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index 49826cc..14ddceb 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -872,7 +872,10 @@ if (redisUrl == null) { (await Array.fromAsync(repository.getFollowers("bot"))).length, 2, ); - assert.ok(commands.some(([command]) => command === "MGET")); + const mget = commands.find(([command]) => command === "MGET"); + assert.ok(mget != null); + assert.deepStrictEqual(mget.length, 3); + assert.ok(mget.slice(1).every((key) => key.includes(":followers:"))); assert.ok( !commands.some(([command, key]) => command === "GET" && key?.includes(":followers:") From 73cdaa802b430f24a7c9924bc84f1469a8fa8c12 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 23:18:49 +0900 Subject: [PATCH 13/16] Harden Redis repository lifecycle The Redis repository now rejects missing constructor options and marks itself closed before returning from close(), including when the Redis client is injected and must not be closed by the repository. Commands check that state before touching Redis, which prevents accidental reconnects or command execution after disposal. List payload reads now use a shared batched MGET helper. The helper splits large key sets into bounded chunks and resolves those chunks in parallel, so message and follower pagination avoid both sequential GET loops and oversized single commands. Quote authorization reference lookups also validate parsed JSON before reading messageId from it. https://github.com/fedify-dev/botkit/pull/35#discussion_r3537123378 https://github.com/fedify-dev/botkit/pull/35#discussion_r3537123392 https://github.com/fedify-dev/botkit/pull/35#discussion_r3537123418 https://github.com/fedify-dev/botkit/pull/35#discussion_r3537123421 https://github.com/fedify-dev/botkit/pull/35#discussion_r3537123426 https://github.com/fedify-dev/botkit/pull/35#discussion_r3537123432 https://github.com/fedify-dev/botkit/pull/35#discussion_r3537123438 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.test.ts | 48 ++++++++++++++++++--------- packages/botkit-redis/src/mod.ts | 27 +++++++++++++-- 2 files changed, 57 insertions(+), 18 deletions(-) diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index 14ddceb..2eb206f 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -254,6 +254,14 @@ if (redisUrl == null) { } else { describe("RedisRepository", () => { test("rejects invalid constructor option combinations", () => { + assert.throws( + () => Reflect.construct(RedisRepository, [undefined]), + new TypeError("RedisRepositoryOptions must be provided."), + ); + assert.throws( + () => Reflect.construct(RedisRepository, [null]), + new TypeError("RedisRepositoryOptions must be provided."), + ); assert.throws( () => Reflect.construct(RedisRepository, [{}]), new TypeError( @@ -323,6 +331,10 @@ if (redisUrl == null) { assert.ok(client.isOpen); await repository.close(); assert.ok(client.isOpen); + await assert.rejects( + () => repository.countMessages("bot"), + new TypeError("RedisRepository is closed."), + ); } finally { await client.quit(); await cleanupPrefix(redisUrl, prefix); @@ -442,7 +454,7 @@ if (redisUrl == null) { } }); - test("fetches listed messages in one Redis command", async () => { + test("fetches listed messages in batched Redis commands", async () => { const prefix = `botkit_test_${crypto.randomUUID()}`; const client = createClient({ url: redisUrl }); await client.connect(); @@ -452,25 +464,25 @@ if (redisUrl == null) { prefix, }); try { - const firstId = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; - const secondId = "01942976-3400-7f34-872e-2cbf0f9eeac4"; - await repository.addMessage( - "bot", - firstId, - createMessage(firstId, "first", "2025-01-01T00:00:00Z"), - ); - await repository.addMessage( - "bot", - secondId, - createMessage(secondId, "second", "2025-01-02T00:00:00Z"), - ); + for (let i = 0; i < 101; i++) { + const sequence = i.toString(16).padStart(4, "0"); + const id: `${string}-${string}-${string}-${string}-${string}` = + `01941f29-${sequence}-7fe8-ab0a-7b593990a3c0`; + await repository.addMessage( + "bot", + id, + createMessage(id, `message ${i}`, "2025-01-01T00:00:00Z"), + ); + } commands.length = 0; assert.deepStrictEqual( (await Array.fromAsync(repository.getMessages("bot"))).length, - 2, + 101, ); - assert.ok(commands.some(([command]) => command === "MGET")); + const mgets = commands.filter(([command]) => command === "MGET"); + assert.deepStrictEqual(mgets.length, 2); + assert.ok(mgets.every((command) => command.length <= 101)); assert.ok( !commands.some(([command, key]) => command === "GET" && key?.includes(":messages:") @@ -838,7 +850,7 @@ if (redisUrl == null) { } }); - test("fetches listed followers in one Redis command", async () => { + test("fetches listed followers in batched Redis commands", async () => { const prefix = `botkit_test_${crypto.randomUUID()}`; const client = createClient({ url: redisUrl }); await client.connect(); @@ -1146,6 +1158,10 @@ if (redisUrl == null) { }`, "null", ]); + assert.deepStrictEqual( + await repository.findQuoteAuthorizationReference("bot", stamp), + undefined, + ); assert.deepStrictEqual( await repository.findQuoteAuthorizationReferenceAttribution( "bot", diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts index 0b11244..21377c7 100644 --- a/packages/botkit-redis/src/mod.ts +++ b/packages/botkit-redis/src/mod.ts @@ -38,6 +38,7 @@ const logger = getLogger(["botkit", "redis"]); const defaultRedisLockTimeoutMs = 30_000; const defaultRedisLockPollIntervalMs = 20; const defaultRedisLockRenewIntervalMs = 10_000; +const redisMGetBatchSize = 100; const uuidV7Pattern = /^([0-9a-f]{8})-([0-9a-f]{4})-7[0-9a-f]{3}-[0-9a-f]{4}-[0-9a-f]{12}$/i; @@ -139,6 +140,7 @@ export class RedisRepository implements Repository, AsyncDisposable { private readonly lockTimeoutMs: number; private readonly lockPollIntervalMs: number; private readonly lockRenewIntervalMs: number; + private closed = false; /** * The Redis key prefix under which all BotKit data is stored. @@ -153,6 +155,9 @@ export class RedisRepository implements Repository, AsyncDisposable { * if `lockRenewIntervalMs` is not less than `lockTimeoutMs`. */ constructor(options: RedisRepositoryOptions) { + if (options == null) { + throw new TypeError("RedisRepositoryOptions must be provided."); + } const hasClient = "client" in options && options.client != null; const hasUrl = "url" in options && options.url != null; if (hasClient === hasUrl) { @@ -208,6 +213,8 @@ export class RedisRepository implements Repository, AsyncDisposable { * Closes the owned Redis connection. */ async close(): Promise { + if (this.closed) return; + this.closed = true; if (!this.ownsClient) return; const ready = this.ready; try { @@ -250,6 +257,9 @@ export class RedisRepository implements Repository, AsyncDisposable { } private async ensureReady(): Promise { + if (this.closed) { + throw new TypeError("RedisRepository is closed."); + } if (!this.ownsClient) return; const ready = this.ready; if (ready == null) { @@ -289,6 +299,18 @@ export class RedisRepository implements Repository, AsyncDisposable { return typeof value === "string" ? value : undefined; } + private async mGet(keys: readonly string[]): Promise { + if (keys.length < 1) return []; + const chunks: string[][] = []; + for (let i = 0; i < keys.length; i += redisMGetBatchSize) { + chunks.push(keys.slice(i, i + redisMGetBatchSize)); + } + const replies = await Promise.all( + chunks.map((chunk) => this.command(["MGET", ...chunk])), + ); + return replies.flatMap(toStringArray); + } + private async set(key: string, value: string): Promise { await this.command(["SET", key, value]); } @@ -536,7 +558,7 @@ export class RedisRepository implements Repository, AsyncDisposable { ); if (ids.length < 1) return; const keys = ids.map((id) => this.botKey(identifier, "messages", id)); - const jsons = toStringArray(await this.command(["MGET", ...keys])); + const jsons = await this.mGet(keys); for (const json of jsons) { try { const activity = await Activity.fromJsonLd(JSON.parse(json)); @@ -702,7 +724,7 @@ export class RedisRepository implements Repository, AsyncDisposable { ); if (ids.length < 1) return; const keys = ids.map((id) => this.botKey(identifier, "followers", id)); - const jsons = toStringArray(await this.command(["MGET", ...keys])); + const jsons = await this.mGet(keys); for (const json of jsons) { try { const actor = await Object.fromJsonLd(JSON.parse(json)); @@ -1008,6 +1030,7 @@ export class RedisRepository implements Repository, AsyncDisposable { if (json == null) return undefined; try { const value = JSON.parse(json) as QuoteAuthorizationReferenceData; + if (typeof value !== "object" || value == null) return undefined; return value.messageId; } catch { return undefined; From fd9d05f3b8a96bd0b91c3fc4737b5f5aeb502db9 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 23:30:18 +0900 Subject: [PATCH 14/16] Clarify Redis MGET test naming Rename the MGET batch assertion callback variable so it does not reuse command in a context where the surrounding test uses that name for the Redis command verb. https://github.com/fedify-dev/botkit/pull/35#pullrequestreview-4646056541 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index 2eb206f..c1466e2 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -482,7 +482,7 @@ if (redisUrl == null) { ); const mgets = commands.filter(([command]) => command === "MGET"); assert.deepStrictEqual(mgets.length, 2); - assert.ok(mgets.every((command) => command.length <= 101)); + assert.ok(mgets.every((mget) => mget.length <= 101)); assert.ok( !commands.some(([command, key]) => command === "GET" && key?.includes(":messages:") From a5a07c6b90f001232416f887e9dcacc93f0f1bd0 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 23:34:12 +0900 Subject: [PATCH 15/16] Stop Redis lock renewal after close Redis lock renewal now stops as soon as the repository is closed. This prevents background renewal callbacks from issuing commands against a closed repository and repeatedly logging expected closure errors. The regression test closes a repository while an update still holds a lock, then verifies that no renewal warning is emitted after close. https://github.com/fedify-dev/botkit/pull/35#discussion_r3537234426 Assisted-by: Codex:gpt-5.5 --- packages/botkit-redis/src/mod.test.ts | 70 +++++++++++++++++++++++++++ packages/botkit-redis/src/mod.ts | 10 +++- 2 files changed, 79 insertions(+), 1 deletion(-) diff --git a/packages/botkit-redis/src/mod.test.ts b/packages/botkit-redis/src/mod.test.ts index c1466e2..f3aca3e 100644 --- a/packages/botkit-redis/src/mod.test.ts +++ b/packages/botkit-redis/src/mod.test.ts @@ -15,6 +15,7 @@ // along with this program. If not, see . import { RedisRepository } from "@fedify/botkit-redis"; import { exportJwk } from "@fedify/fedify/sig"; +import { configureSync, type LogRecord, resetSync } from "@logtape/logtape"; import { Create, Follow, @@ -657,6 +658,75 @@ if (redisUrl == null) { } }); + test("stops renewing Redis locks after close", async () => { + const prefix = `botkit_test_${crypto.randomUUID()}`; + const client = createClient({ url: redisUrl }); + await client.connect(); + const messageId = "01941f29-7c00-7fe8-ab0a-7b593990a3c0"; + const repository = new RedisRepository({ + client, + prefix, + lockTimeoutMs: 500, + lockRenewIntervalMs: 50, + }); + const records: LogRecord[] = []; + const release = createDeferred(); + let update: Promise | undefined; + try { + await repository.addMessage( + "bot", + messageId, + createMessage(messageId, "base", "2025-01-01T00:00:00Z"), + ); + configureSync({ + sinks: { + capture: (record) => records.push(record), + }, + loggers: [ + { + category: ["botkit", "redis"], + sinks: ["capture"], + lowestLevel: "warning", + }, + { + category: ["logtape", "meta"], + lowestLevel: null, + }, + ], + reset: true, + }); + + let updaterStarted = false; + update = repository.updateMessage( + "bot", + messageId, + async (existing) => { + assert.ok(existing instanceof Create); + updaterStarted = true; + await release.promise; + return existing; + }, + ); + while (!updaterStarted) await delay(1); + + await repository.close(); + await delay(150); + + assert.deepStrictEqual(records, []); + release.resolve(); + await assert.rejects( + update, + new TypeError("RedisRepository is closed."), + ); + } finally { + release.resolve(); + await update?.catch(() => undefined); + resetSync(); + await client.quit(); + await cleanupPrefix(redisUrl, prefix); + } + }); + test("serializes message removal with concurrent updates", async () => { const { repository, cleanup } = createHarness(); try { diff --git a/packages/botkit-redis/src/mod.ts b/packages/botkit-redis/src/mod.ts index 21377c7..a0ff8e0 100644 --- a/packages/botkit-redis/src/mod.ts +++ b/packages/botkit-redis/src/mod.ts @@ -396,6 +396,10 @@ export class RedisRepository implements Repository, AsyncDisposable { const renew = setInterval(() => { void (async () => { try { + if (this.closed) { + clearInterval(renew); + return; + } const renewed = toNumber( await this.command([ "EVAL", @@ -409,7 +413,11 @@ export class RedisRepository implements Repository, AsyncDisposable { ); if (renewed === 0) clearInterval(renew); } catch (error) { - logger.warn("Failed to renew Redis lock: {error}", { error }); + if (this.closed) { + clearInterval(renew); + } else { + logger.warn("Failed to renew Redis lock: {error}", { error }); + } } })(); }, this.lockRenewIntervalMs); From b4dee2de1a3c1b08c57c5438357e05dcdc0550fa Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Tue, 7 Jul 2026 23:34:19 +0900 Subject: [PATCH 16/16] List Redis package imports explicitly Declare the Redis package's direct Fedify and LogTape dependencies in its own Deno import map. This keeps the package self-contained for JSR resolution instead of relying on the workspace root import map. The lockfile now records those package-level dependencies as well. https://github.com/fedify-dev/botkit/pull/35#discussion_r3537234434 Assisted-by: Codex:gpt-5.5 --- deno.lock | 2 ++ packages/botkit-redis/deno.json | 2 ++ 2 files changed, 4 insertions(+) diff --git a/deno.lock b/deno.lock index c9623e6..9908a13 100644 --- a/deno.lock +++ b/deno.lock @@ -1106,7 +1106,9 @@ }, "packages/botkit-redis": { "dependencies": [ + "jsr:@fedify/fedify@^2.3.1", "jsr:@fedify/vocab@^2.3.1", + "jsr:@logtape/logtape@^2.2.3", "npm:redis@^6.1.0" ], "packageJson": { diff --git a/packages/botkit-redis/deno.json b/packages/botkit-redis/deno.json index 09cb43c..18c8d2b 100644 --- a/packages/botkit-redis/deno.json +++ b/packages/botkit-redis/deno.json @@ -6,7 +6,9 @@ ".": "./src/mod.ts" }, "imports": { + "@fedify/fedify": "jsr:@fedify/fedify@^2.3.1", "@fedify/vocab": "jsr:@fedify/vocab@^2.3.1", + "@logtape/logtape": "jsr:@logtape/logtape@^2.2.3", "redis": "npm:redis@^6.1.0" }, "exclude": [