diff --git a/CHANGES.md b/CHANGES.md index c0a64fe..8589bd4 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -6,6 +6,29 @@ Version 0.4.4 To be released. +### @fedify/botkit + + - Fixed `MemoryRepository`, `KvRepository`, and `MemoryCachedRepository` so + removing one of multiple active follow requests for the same actor no + longer deletes the follower too early or fires a premature unfollow event, + and reassigning a follow request no longer leaves stale followers behind. + [[#25], [#26]] + +[#25]: https://github.com/fedify-dev/botkit/issues/25 +[#26]: https://github.com/fedify-dev/botkit/pull/26 + +### @fedify/botkit-sqlite + + - Fixed `SqliteRepository` so removing one of multiple active follow requests + for the same actor no longer fails with a foreign key error, and reassigning + a follow request no longer leaves stale followers behind. [[#25]] + +### @fedify/botkit-postgres + + - Fixed `PostgresRepository` so removing one of multiple active follow + requests for the same actor no longer reports a follower removal until the + last active follow request is removed. [[#25]] + Version 0.4.3 ------------- diff --git a/deno.lock b/deno.lock index d92b3c8..fbcd638 100644 --- a/deno.lock +++ b/deno.lock @@ -7,11 +7,15 @@ "jsr:@fedify/markdown-it-hashtag@0.3": "0.3.0", "jsr:@fedify/markdown-it-mention@0.3": "0.3.0", "jsr:@fedify/vocab-runtime@^2.1.15": "2.2.4", + "jsr:@fedify/vocab-runtime@^2.1.17": "2.2.4", "jsr:@fedify/vocab-runtime@^2.1.2": "2.2.4", "jsr:@fedify/vocab-runtime@^2.2.4": "2.2.4", + "jsr:@fedify/vocab-runtime@~2.1.15": "2.1.17", "jsr:@fedify/vocab@^2.1.15": "2.2.4", "jsr:@fedify/vocab@^2.1.2": "2.2.4", + "jsr:@fedify/vocab@~2.1.15": "2.1.17", "jsr:@fedify/webfinger@^2.1.15": "2.2.4", + "jsr:@fedify/webfinger@^2.1.17": "2.2.4", "jsr:@fedify/webfinger@^2.2.4": "2.2.4", "jsr:@hongminhee/x-forwarded-fetch@0.2": "0.2.0", "jsr:@hono/hono@^4.8.2": "4.12.23", @@ -32,6 +36,8 @@ "npm:@logtape/logtape@^1.3.5": "1.3.7", "npm:@multiformats/base-x@^4.0.1": "4.0.1", "npm:@opentelemetry/api@^1.9.0": "1.9.1", + "npm:@opentelemetry/core@^2.5.0": "2.6.1_@opentelemetry+api@1.9.1", + "npm:@opentelemetry/sdk-trace-base@^2.5.0": "2.6.1_@opentelemetry+api@1.9.1", "npm:@opentelemetry/semantic-conventions@^1.39.0": "1.40.0", "npm:@types/markdown-it@^14.1.1": "14.1.2", "npm:asn1js@^3.0.6": "3.0.7", @@ -81,6 +87,8 @@ "jsr:@fedify/webfinger@^2.1.15", "jsr:@logtape/logtape@^2.0.5", "npm:@opentelemetry/api", + "npm:@opentelemetry/core", + "npm:@opentelemetry/sdk-trace-base", "npm:@opentelemetry/semantic-conventions", "npm:byte-encodings", "npm:es-toolkit@^1.43.0", @@ -104,6 +112,16 @@ "npm:@types/markdown-it" ] }, + "@fedify/vocab@2.1.17": { + "integrity": "4202c6aa2f76679ee6da5ad97db436ab7692f15a2ef42f6e1ff02f2eba2d983a", + "dependencies": [ + "jsr:@fedify/vocab-runtime@^2.1.17", + "jsr:@fedify/webfinger@^2.1.17", + "jsr:@logtape/logtape@^2.0.5", + "npm:@opentelemetry/api", + "npm:es-toolkit@^1.43.0" + ] + }, "@fedify/vocab@2.2.4": { "integrity": "de2fa46b141e2cc94c6ddd066826fcdbfe1b0831c8535e994601813d6d00d9cc", "dependencies": [ @@ -114,6 +132,18 @@ "npm:es-toolkit@^1.43.0" ] }, + "@fedify/vocab-runtime@2.1.17": { + "integrity": "6fc564fe42746a7cae6e985b7fc1d0b8743abda4d7e22f18b9895ddb2335e0f2", + "dependencies": [ + "jsr:@logtape/logtape@^2.0.5", + "npm:@multiformats/base-x", + "npm:@opentelemetry/api", + "npm:asn1js", + "npm:byte-encodings", + "npm:jsonld", + "npm:pkijs" + ] + }, "@fedify/vocab-runtime@2.2.4": { "integrity": "0bc68243d7cc455459435743f73f4d9876cb006a5690f1b9542b74cb503f2226", "dependencies": [ diff --git a/packages/botkit-postgres/src/mod.test.ts b/packages/botkit-postgres/src/mod.test.ts index 3922974..291060e 100644 --- a/packages/botkit-postgres/src/mod.test.ts +++ b/packages/botkit-postgres/src/mod.test.ts @@ -872,8 +872,8 @@ if (postgresUrl == null) { assert.deepStrictEqual(await repo.countFollowers(), 2); assert.ok(await repo.hasFollower(followerA.id!)); assert.deepStrictEqual( - await (await repo.removeFollower(followA, followerA.id!))?.toJsonLd(), - await followerA.toJsonLd(), + await repo.removeFollower(followA, followerA.id!), + undefined, ); assert.ok(await repo.hasFollower(followerA.id!)); assert.deepStrictEqual(await repo.countFollowers(), 2); diff --git a/packages/botkit-postgres/src/mod.ts b/packages/botkit-postgres/src/mod.ts index 7006ecc..c7b4fa5 100644 --- a/packages/botkit-postgres/src/mod.ts +++ b/packages/botkit-postgres/src/mod.ts @@ -539,8 +539,8 @@ export class PostgresRepository implements Repository, AsyncDisposable { WHERE follow_request_id = $1`, [followId.href], ); - await this.cleanupFollower(sql, followerId.href); - return await parseActor(row.actor_json); + const removed = await this.cleanupFollower(sql, followerId.href); + return removed ? await parseActor(row.actor_json) : undefined; }); } @@ -758,9 +758,9 @@ export class PostgresRepository implements Repository, AsyncDisposable { private async cleanupFollower( sql: Queryable, followerId: string, - ): Promise { + ): Promise { await this.lockFollower(sql, followerId); - await this.query( + const rows = await this.query<{ readonly follower_id: string }>( sql, `DELETE FROM ${this.table("followers")} WHERE follower_id = $1 @@ -768,9 +768,11 @@ export class PostgresRepository implements Repository, AsyncDisposable { SELECT 1 FROM ${this.table("follow_requests")} WHERE follower_id = $1 - )`, + ) + RETURNING follower_id`, [followerId], ); + return rows.length > 0; } private async ensureReady(): Promise { diff --git a/packages/botkit-sqlite/src/mod.test.ts b/packages/botkit-sqlite/src/mod.test.ts index c18b5e6..98cf1cd 100644 --- a/packages/botkit-sqlite/src/mod.test.ts +++ b/packages/botkit-sqlite/src/mod.test.ts @@ -167,6 +167,95 @@ describe("SqliteRepository", () => { } }); + test("followers with multiple follow requests", async () => { + const repo = createSqliteRepository(); + 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 repo.addFollower(followA, follower); + await repo.addFollower(followB, follower); + assert.deepStrictEqual(await repo.countFollowers(), 1); + assert.ok(await repo.hasFollower(follower.id!)); + + assert.deepStrictEqual( + await repo.removeFollower(followA, follower.id!), + undefined, + ); + assert.deepStrictEqual(await repo.countFollowers(), 1); + assert.ok(await repo.hasFollower(follower.id!)); + + assert.deepStrictEqual( + await (await repo.removeFollower(followB, follower.id!))?.toJsonLd(), + await follower.toJsonLd(), + ); + assert.deepStrictEqual(await repo.countFollowers(), 0); + assert.deepStrictEqual(await repo.hasFollower(follower.id!), false); + } finally { + repo.close(); + } + }); + + test("followers with reassigned follow requests", async () => { + const repo = createSqliteRepository(); + try { + const oldFollower = new Person({ + id: new URL("https://example.com/ap/actor/alice"), + preferredUsername: "alice", + }); + const newFollower = new Person({ + id: new URL("https://example.com/ap/actor/bob"), + preferredUsername: "bob", + }); + 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 repo.addFollower(followA, oldFollower); + await repo.addFollower(followB, oldFollower); + await repo.addFollower(followA, newFollower); + + assert.deepStrictEqual(await repo.countFollowers(), 2); + assert.ok(await repo.hasFollower(oldFollower.id!)); + assert.ok(await repo.hasFollower(newFollower.id!)); + + assert.deepStrictEqual( + await (await repo.removeFollower(followB, oldFollower.id!)) + ?.toJsonLd(), + await oldFollower.toJsonLd(), + ); + assert.deepStrictEqual(await repo.countFollowers(), 1); + assert.deepStrictEqual(await repo.hasFollower(oldFollower.id!), false); + assert.ok(await repo.hasFollower(newFollower.id!)); + + await repo.addFollower(followA, oldFollower); + assert.deepStrictEqual(await repo.countFollowers(), 1); + assert.ok(await repo.hasFollower(oldFollower.id!)); + assert.deepStrictEqual(await repo.hasFollower(newFollower.id!), false); + assert.deepStrictEqual( + await Promise.all( + (await Array.fromAsync(repo.getFollowers())).map((follower) => + follower.toJsonLd() + ), + ), + [await oldFollower.toJsonLd()], + ); + } finally { + repo.close(); + } + }); + test("poll voting", async () => { const repo = createSqliteRepository(); try { diff --git a/packages/botkit-sqlite/src/mod.ts b/packages/botkit-sqlite/src/mod.ts index 8423c18..6804968 100644 --- a/packages/botkit-sqlite/src/mod.ts +++ b/packages/botkit-sqlite/src/mod.ts @@ -130,6 +130,11 @@ export class SqliteRepository implements Repository, Disposable { ) `); + this.db.exec(` + CREATE INDEX IF NOT EXISTS idx_follow_requests_follower + ON follow_requests(follower_id) + `); + // Sent follows table this.db.exec(` CREATE TABLE IF NOT EXISTS sent_follows ( @@ -374,19 +379,35 @@ export class SqliteRepository implements Repository, Disposable { ); const insertFollowerStmt = this.db.prepare(` - INSERT OR REPLACE INTO followers (follower_id, actor_json) + INSERT INTO followers (follower_id, actor_json) VALUES (?, ?) + ON CONFLICT(follower_id) + DO UPDATE SET actor_json = excluded.actor_json `); const insertRequestStmt = this.db.prepare(` - INSERT OR REPLACE INTO follow_requests (follow_request_id, follower_id) + INSERT INTO follow_requests (follow_request_id, follower_id) VALUES (?, ?) + ON CONFLICT(follow_request_id) + DO UPDATE SET follower_id = excluded.follower_id `); - this.db.exec("BEGIN TRANSACTION"); + const previousFollowerStmt = this.db.prepare(` + SELECT follower_id FROM follow_requests WHERE follow_request_id = ? + `); + + this.db.exec("BEGIN IMMEDIATE TRANSACTION"); try { + const previousRow = previousFollowerStmt.get(followRequestId.href) as + | { follower_id: string } + | undefined; insertFollowerStmt.run(follower.id.href, followerJson); insertRequestStmt.run(followRequestId.href, follower.id.href); + if ( + previousRow != null && previousRow.follower_id !== follower.id.href + ) { + this.cleanupFollower(previousRow.follower_id); + } this.db.exec("COMMIT"); } catch (error) { this.db.exec("ROLLBACK"); @@ -398,40 +419,46 @@ export class SqliteRepository implements Repository, Disposable { followRequestId: URL, actorId: URL, ): Promise { - // Check if the follow request exists and matches the actor const checkStmt = this.db.prepare(` - SELECT fr.follower_id, f.actor_json - FROM follow_requests fr - JOIN followers f ON fr.follower_id = f.follower_id + SELECT fr.follower_id, f.actor_json + FROM follow_requests fr + JOIN followers f ON fr.follower_id = f.follower_id WHERE fr.follow_request_id = ? AND fr.follower_id = ? `); - const row = checkStmt.get(followRequestId.href, actorId.href) as { - follower_id: string; - actor_json: string; - } | undefined; - - if (!row) return undefined; - - // Remove the follower and follow request const deleteRequestStmt = this.db.prepare(` - DELETE FROM follow_requests WHERE follow_request_id = ? - `); - - const deleteFollowerStmt = this.db.prepare(` - DELETE FROM followers WHERE follower_id = ? + DELETE FROM follow_requests + WHERE follow_request_id = ? AND follower_id = ? `); - this.db.exec("BEGIN TRANSACTION"); + let row: { follower_id: string; actor_json: string } | undefined; + let removed = false; + this.db.exec("BEGIN IMMEDIATE TRANSACTION"); try { - deleteRequestStmt.run(followRequestId.href); - deleteFollowerStmt.run(actorId.href); + row = checkStmt.get(followRequestId.href, actorId.href) as + | { follower_id: string; actor_json: string } + | undefined; + if (row == null) { + this.db.exec("COMMIT"); + return undefined; + } + const deleteResult = deleteRequestStmt.run( + followRequestId.href, + actorId.href, + ) as { readonly changes: number }; + if (deleteResult.changes < 1) { + this.db.exec("COMMIT"); + return undefined; + } + removed = this.cleanupFollower(actorId.href); this.db.exec("COMMIT"); } catch (error) { this.db.exec("ROLLBACK"); throw error; } + if (!removed || row == null) return undefined; + try { const actorData = JSON.parse(row.actor_json); const actor = await Object.fromJsonLd(actorData); @@ -446,6 +473,21 @@ export class SqliteRepository implements Repository, Disposable { return undefined; } + private cleanupFollower(followerId: string): boolean { + const deleteFollowerStmt = this.db.prepare(` + DELETE FROM followers + WHERE follower_id = ? + AND NOT EXISTS ( + SELECT 1 FROM follow_requests WHERE follower_id = ? + ) + `); + + const result = deleteFollowerStmt.run(followerId, followerId) as { + readonly changes: number; + }; + return result.changes > 0; + } + hasFollower(followerId: URL): Promise { const stmt = this.db.prepare(` SELECT 1 FROM followers WHERE follower_id = ? diff --git a/packages/botkit/package.json b/packages/botkit/package.json index 0e5f2da..8a0daa4 100644 --- a/packages/botkit/package.json +++ b/packages/botkit/package.json @@ -76,7 +76,7 @@ }, "./package.json": "./package.json" }, - "sideEffects": false, + "sideEffects": true, "files": [ "dist", "LICENSE", diff --git a/packages/botkit/src/bot-impl.test.ts b/packages/botkit/src/bot-impl.test.ts index 752abb3..cc4aef1 100644 --- a/packages/botkit/src/bot-impl.test.ts +++ b/packages/botkit/src/bot-impl.test.ts @@ -1482,6 +1482,53 @@ test("BotImpl.onUnfollowed()", async (t) => { }); }); +test("BotImpl.onUnfollowed() with multiple follow requests", async () => { + const repository = new MemoryRepository(); + const bot = new BotImpl({ + kv: new MemoryKvStore(), + repository, + username: "bot", + }); + const unfollowed: [Session, Actor][] = []; + bot.onUnfollow = (session, actor) => void (unfollowed.push([session, actor])); + const ctx = createMockInboxContext(bot, "https://example.com", "bot"); + const follower = new Person({ + id: new URL("https://example.com/ap/actor/john"), + preferredUsername: "john", + }); + const followA = new URL("https://example.com/ap/actor/john/follows/bot"); + const followB = new URL( + "https://example.com/ap/actor/john/follows/bot-again", + ); + + await repository.addFollower(followA, follower); + await repository.addFollower(followB, follower); + + await bot.onUnfollowed( + ctx, + new Undo({ + actor: follower.id, + object: new Follow({ id: followA }), + }), + ); + assert.deepStrictEqual(await repository.countFollowers(), 1); + assert.ok(await repository.hasFollower(follower.id!)); + assert.deepStrictEqual(unfollowed, []); + + await bot.onUnfollowed( + ctx, + new Undo({ + actor: follower.id, + object: new Follow({ id: followB }), + }), + ); + assert.deepStrictEqual(await repository.countFollowers(), 0); + assert.deepStrictEqual(unfollowed.length, 1); + const [session, actor] = unfollowed[0] as [Session, Actor]; + assert.deepStrictEqual(session.bot, bot); + assert.deepStrictEqual(actor.id, follower.id); +}); + test("BotImpl.onFollowAccepted()", async (t) => { const repository = new MemoryRepository(); const bot = new BotImpl({ diff --git a/packages/botkit/src/repository-load.test.ts b/packages/botkit/src/repository-load.test.ts new file mode 100644 index 0000000..2f764f5 --- /dev/null +++ b/packages/botkit/src/repository-load.test.ts @@ -0,0 +1,50 @@ +// BotKit by Fedify: A framework for creating ActivityPub bots +// Copyright (C) 2025–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 assert from "node:assert"; +import { test } from "node:test"; + +test("repository module loads before Temporal is registered", async () => { + if (typeof Deno === "undefined") return; + const repositoryUrl = new URL( + `./repository.ts?no-temporal=${crypto.randomUUID()}`, + import.meta.url, + ).href; + const workerCode = ` + delete (globalThis as Record).Temporal; + try { + await import(${JSON.stringify(repositoryUrl)}); + self.postMessage({ ok: true }); + } catch (error) { + self.postMessage({ + ok: false, + message: error instanceof Error ? error.message : String(error), + }); + } + `; + const worker = new Worker( + `data:application/typescript,${encodeURIComponent(workerCode)}`, + { type: "module" }, + ); + const result = await new Promise((resolve, reject) => { + worker.addEventListener("message", (event: MessageEvent) => { + resolve(event.data); + }, { once: true }); + worker.addEventListener("error", reject, { once: true }); + }); + worker.terminate(); + + assert.deepStrictEqual(result, { ok: true }); +}); diff --git a/packages/botkit/src/repository.test.ts b/packages/botkit/src/repository.test.ts index 0376526..3008a2b 100644 --- a/packages/botkit/src/repository.test.ts +++ b/packages/botkit/src/repository.test.ts @@ -13,7 +13,13 @@ // // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -import { MemoryKvStore } from "@fedify/fedify/federation"; +import { + type KvKey, + type KvStore, + type KvStoreListEntry, + type KvStoreSetOptions, + MemoryKvStore, +} from "@fedify/fedify/federation"; import { importJwk } from "@fedify/fedify/sig"; import { Create, Follow, Note, Person, PUBLIC_COLLECTION } from "@fedify/vocab"; import assert from "node:assert"; @@ -43,6 +49,109 @@ const factories: Record Repository> = { MemoryCachedRepository: createMemoryCachedRepository, }; +class RecordingMemoryKvStore extends MemoryKvStore { + readonly lockOptions: KvStoreSetOptions[] = []; + readonly lockReleaseOptions: KvStoreSetOptions[] = []; + readonly undefinedLockReleases: KvKey[] = []; + releasedLockAcquisitions = 0; + + override set( + key: KvKey, + value: unknown, + options?: KvStoreSetOptions, + ): Promise { + if (key.includes("lock")) this.lockOptions.push(options ?? {}); + return super.set(key, value, options); + } + + override cas( + key: KvKey, + expectedValue: unknown, + newValue: unknown, + options?: KvStoreSetOptions, + ): Promise { + if (key.includes("lock")) { + if (newValue === undefined) { + this.undefinedLockReleases.push(key); + } else if ( + typeof newValue === "object" && newValue != null && + "released" in newValue + ) { + this.lockReleaseOptions.push(options ?? {}); + } else if ( + typeof expectedValue === "object" && expectedValue != null && + "released" in expectedValue + ) { + this.releasedLockAcquisitions++; + } else { + this.lockOptions.push(options ?? {}); + } + } + return super.cas(key, expectedValue, newValue, options); + } +} + +class RecordingListMemoryKvStore extends MemoryKvStore { + listCalls = 0; + + override list(prefix?: KvKey): AsyncIterable { + this.listCalls++; + return super.list(prefix); + } +} + +class NonCasMemoryKvStore implements KvStore { + readonly #kv = new MemoryKvStore(); + + get(key: KvKey): Promise { + return this.#kv.get(key); + } + + set( + key: KvKey, + value: unknown, + options?: KvStoreSetOptions, + ): Promise { + return this.#kv.set(key, value, options); + } + + delete(key: KvKey): Promise { + return this.#kv.delete(key); + } + + list(prefix?: KvKey): AsyncIterable { + return this.#kv.list(prefix); + } +} + +class FailingReleaseMemoryKvStore extends MemoryKvStore { + override set( + key: KvKey, + value: unknown, + options?: KvStoreSetOptions, + ): Promise { + if (!key.includes("lock")) { + throw new TypeError("Write failed."); + } + return super.set(key, value, options); + } + + override cas( + key: KvKey, + expectedValue: unknown, + newValue: unknown, + options?: KvStoreSetOptions, + ): Promise { + if ( + key.includes("lock") && typeof newValue === "object" && + newValue != null && "released" in newValue + ) { + throw new TypeError("Release failed."); + } + return super.cas(key, expectedValue, newValue, options); + } +} + const keyPairs: CryptoKeyPair[] = [ { publicKey: await importJwk({ @@ -97,6 +206,274 @@ const keyPairs: CryptoKeyPair[] = [ }, ]; +test("KvRepository uses expiring follower locks", async () => { + const kv = new RecordingMemoryKvStore(); + const repo = new KvRepository(kv); + const follower = new Person({ + id: new URL("https://example.com/ap/actor/lock-test"), + preferredUsername: "lock-test", + }); + + await repo.addFollower( + new URL("https://example.com/ap/follow/lock-test"), + follower, + ); + + assert.ok(kv.lockOptions.length > 0); + assert.ok(kv.lockOptions.every((options) => options.ttl != null)); + assert.ok(kv.lockReleaseOptions.length > 0); + assert.ok(kv.lockReleaseOptions.every((options) => options.ttl != null)); + assert.deepStrictEqual(kv.undefinedLockReleases, []); +}); + +test("KvRepository reacquires released follower locks", async () => { + const kv = new RecordingMemoryKvStore(); + const repo = new KvRepository(kv); + const follower = new Person({ + id: new URL("https://example.com/ap/actor/released-lock"), + preferredUsername: "released-lock", + }); + + await repo.addFollower( + new URL("https://example.com/ap/follow/released-lock/1"), + follower, + ); + await repo.addFollower( + new URL("https://example.com/ap/follow/released-lock/2"), + follower, + ); + + assert.ok(kv.releasedLockAcquisitions > 0); +}); + +test("KvRepository deletes stale follower indexes", async () => { + const kv = new MemoryKvStore(); + const repo = new KvRepository(kv); + const follower = new Person({ + id: new URL("https://example.com/ap/actor/index-test"), + preferredUsername: "index-test", + }); + const follow = new URL("https://example.com/ap/follow/index-test"); + const indexKey: KvKey = [ + "_botkit", + "followRequests", + "followers", + follower.id!.href, + ]; + + await repo.addFollower(follow, follower); + assert.deepStrictEqual(await kv.get(indexKey), [follow.href]); + + await repo.removeFollower(follow, follower.id!); + assert.deepStrictEqual(await kv.get(indexKey), undefined); +}); + +test("KvRepository rebuilds missing follower indexes", async () => { + const kv = new MemoryKvStore(); + const repo = new KvRepository(kv); + const follower = new Person({ + id: new URL("https://example.com/ap/actor/missing-index"), + preferredUsername: "missing-index", + }); + const firstFollow = new URL( + "https://example.com/ap/follow/missing-index/1", + ); + const secondFollow = new URL( + "https://example.com/ap/follow/missing-index/2", + ); + const indexKey: KvKey = [ + "_botkit", + "followRequests", + "followers", + follower.id!.href, + ]; + + await repo.addFollower(firstFollow, follower); + await repo.addFollower(secondFollow, follower); + await kv.delete(indexKey); + + assert.deepStrictEqual( + await repo.removeFollower(firstFollow, follower.id!), + undefined, + ); + assert.ok(await repo.hasFollower(follower.id!)); + assert.deepStrictEqual(await kv.get(indexKey), [secondFollow.href]); +}); + +test("KvRepository trusts empty follower indexes", async () => { + const kv = new RecordingListMemoryKvStore(); + const repo = new KvRepository(kv); + const follower = new Person({ + id: new URL("https://example.com/ap/actor/empty-index"), + preferredUsername: "empty-index", + }); + const follow = new URL("https://example.com/ap/follow/empty-index"); + + await repo.addFollower(follow, follower); + kv.listCalls = 0; + + assert.deepStrictEqual( + await repo.removeFollower(follow, follower.id!), + follower, + ); + assert.deepStrictEqual(kv.listCalls, 0); + assert.ok(!await repo.hasFollower(follower.id!)); +}); + +test("KvRepository removes requests with missing followers", async () => { + const kv = new MemoryKvStore(); + const repo = new KvRepository(kv); + const follower = new Person({ + id: new URL("https://example.com/ap/actor/missing-follower"), + preferredUsername: "missing-follower", + }); + const follow = new URL("https://example.com/ap/follow/missing-follower"); + const followRequestKey: KvKey = [ + "_botkit", + "followRequests", + follow.href, + ]; + const indexKey: KvKey = [ + "_botkit", + "followRequests", + "followers", + follower.id!.href, + ]; + + await repo.addFollower(follow, follower); + await kv.delete(["_botkit", "followers", follower.id!.href]); + + assert.deepStrictEqual( + await repo.removeFollower(follow, follower.id!), + undefined, + ); + assert.deepStrictEqual(await kv.get(followRequestKey), undefined); + assert.deepStrictEqual(await kv.get(indexKey), undefined); + assert.ok(!await repo.hasFollower(follower.id!)); +}); + +test("KvRepository uses follower indexes when adding requests", async () => { + const kv = new RecordingListMemoryKvStore(); + const repo = new KvRepository(kv); + const follower = new Person({ + id: new URL("https://example.com/ap/actor/index-add"), + preferredUsername: "index-add", + }); + const firstFollow = new URL("https://example.com/ap/follow/index-add/1"); + const secondFollow = new URL("https://example.com/ap/follow/index-add/2"); + const indexKey: KvKey = [ + "_botkit", + "followRequests", + "followers", + follower.id!.href, + ]; + + await repo.addFollower(firstFollow, follower); + kv.listCalls = 0; + await repo.addFollower(secondFollow, follower); + + assert.deepStrictEqual(await kv.get(indexKey), [ + firstFollow.href, + secondFollow.href, + ]); + assert.deepStrictEqual(kv.listCalls, 0); + + kv.listCalls = 0; + assert.deepStrictEqual( + await repo.removeFollower(firstFollow, follower.id!), + undefined, + ); + assert.deepStrictEqual(kv.listCalls, 0); + assert.deepStrictEqual(await kv.get(indexKey), [secondFollow.href]); +}); + +test("KvRepository rebuilds legacy requests for new follower rows", async () => { + const kv = new MemoryKvStore(); + const repo = new KvRepository(kv); + const follower = new Person({ + id: new URL("https://example.com/ap/actor/legacy-new-row"), + preferredUsername: "legacy-new-row", + }); + const legacyFollow = new URL( + "https://example.com/ap/follow/legacy-new-row/1", + ); + const newFollow = new URL( + "https://example.com/ap/follow/legacy-new-row/2", + ); + const indexKey: KvKey = [ + "_botkit", + "followRequests", + "followers", + follower.id!.href, + ]; + + await kv.set( + ["_botkit", "followRequests", legacyFollow.href], + follower.id!.href, + ); + await repo.addFollower(newFollow, follower); + + assert.deepStrictEqual( + await repo.removeFollower(newFollow, follower.id!), + undefined, + ); + assert.ok(await repo.hasFollower(follower.id!)); + assert.deepStrictEqual(await kv.get(indexKey), [legacyFollow.href]); +}); + +test("KvRepository recovers legacy follower locks", async () => { + const kv = new MemoryKvStore(); + const repo = new KvRepository(kv); + const follower = new Person({ + id: new URL("https://example.com/ap/actor/legacy-lock"), + preferredUsername: "legacy-lock", + }); + const lockKey: KvKey = ["_botkit", "followers", "lock"]; + + await kv.set(lockKey, follower.id!.href); + await repo.addFollower( + new URL("https://example.com/ap/follow/legacy-lock"), + follower, + ); + + assert.ok(await repo.hasFollower(follower.id!)); + assert.notDeepStrictEqual(await kv.get(lockKey), follower.id!.href); +}); + +test("KvRepository supports non-CAS follower mutations", async () => { + const repo = new KvRepository(new NonCasMemoryKvStore()); + const follower = new Person({ + id: new URL("https://example.com/ap/actor/no-cas"), + preferredUsername: "no-cas", + }); + const follow = new URL("https://example.com/ap/follow/no-cas"); + + await repo.addFollower(follow, follower); + assert.ok(await repo.hasFollower(follower.id!)); + assert.deepStrictEqual( + await repo.removeFollower(follow, follower.id!), + follower, + ); + assert.ok(!await repo.hasFollower(follower.id!)); +}); + +test("KvRepository preserves errors when lock release fails", async () => { + const repo = new KvRepository(new FailingReleaseMemoryKvStore()); + const follower = new Person({ + id: new URL("https://example.com/ap/actor/release-failure"), + preferredUsername: "release-failure", + }); + + await assert.rejects( + () => + repo.addFollower( + new URL("https://example.com/ap/follow/release-failure"), + follower, + ), + { name: "TypeError", message: "Write failed." }, + ); +}); + for (const name in factories) { const factory = factories[name]; @@ -634,6 +1011,85 @@ for (const name in factories) { assert.deepStrictEqual(await repo.hasFollower(followerB.id!), false); }); + test("followers with multiple follow requests", async () => { + 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 repo.addFollower(followA, follower); + await repo.addFollower(followB, follower); + assert.deepStrictEqual(await repo.countFollowers(), 1); + assert.ok(await repo.hasFollower(follower.id!)); + + assert.deepStrictEqual( + await repo.removeFollower(followA, follower.id!), + undefined, + ); + assert.deepStrictEqual(await repo.countFollowers(), 1); + assert.ok(await repo.hasFollower(follower.id!)); + + assert.deepStrictEqual( + await (await repo.removeFollower(followB, follower.id!))?.toJsonLd(), + await follower.toJsonLd(), + ); + assert.deepStrictEqual(await repo.countFollowers(), 0); + assert.deepStrictEqual(await repo.hasFollower(follower.id!), false); + }); + + test("followers with reassigned follow requests", async () => { + const oldFollower = new Person({ + id: new URL("https://example.com/ap/actor/alice"), + preferredUsername: "alice", + }); + const newFollower = new Person({ + id: new URL("https://example.com/ap/actor/bob"), + preferredUsername: "bob", + }); + 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 repo.addFollower(followA, oldFollower); + await repo.addFollower(followB, oldFollower); + await repo.addFollower(followA, newFollower); + + assert.deepStrictEqual(await repo.countFollowers(), 2); + assert.ok(await repo.hasFollower(oldFollower.id!)); + assert.ok(await repo.hasFollower(newFollower.id!)); + + assert.deepStrictEqual( + await (await repo.removeFollower(followB, oldFollower.id!)) + ?.toJsonLd(), + await oldFollower.toJsonLd(), + ); + assert.deepStrictEqual(await repo.countFollowers(), 1); + assert.deepStrictEqual(await repo.hasFollower(oldFollower.id!), false); + assert.ok(await repo.hasFollower(newFollower.id!)); + + await repo.addFollower(followA, oldFollower); + assert.deepStrictEqual(await repo.countFollowers(), 1); + assert.ok(await repo.hasFollower(oldFollower.id!)); + assert.deepStrictEqual(await repo.hasFollower(newFollower.id!), false); + assert.deepStrictEqual( + await Promise.all( + (await Array.fromAsync(repo.getFollowers())).map((follower) => + follower.toJsonLd() + ), + ), + [await oldFollower.toJsonLd()], + ); + }); + test("sent follows", async () => { const follow = new Follow({ id: new URL( diff --git a/packages/botkit/src/repository.ts b/packages/botkit/src/repository.ts index 2b9a050..befab6d 100644 --- a/packages/botkit/src/repository.ts +++ b/packages/botkit/src/repository.ts @@ -29,6 +29,27 @@ export type { KvKey, KvStore } from "@fedify/fedify/federation"; export { Announce, Create } from "@fedify/vocab"; const logger = getLogger(["botkit", "repository"]); +const kvLockPollIntervalMs = 100; +const uuidPattern = + /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; + +interface KvLock { + readonly id: string; + readonly released?: boolean; +} + +function isKvLock(value: unknown): value is KvLock { + return typeof value === "object" && value != null && "id" in value && + typeof value.id === "string"; +} + +function isReleasedKvLock(value: unknown): value is KvLock { + return isKvLock(value) && value.released === true; +} + +function isLegacyKvLock(value: unknown): value is string { + return typeof value === "string" && !uuidPattern.test(value); +} /** * A UUID (universally unique identifier). @@ -123,8 +144,9 @@ export interface Repository { * Removes a follower from the repository. * @param followId The URL of the follow request. * @param followerId The ID of the actor to remove. - * @returns The removed actor. If the follower does not exist or the follow - * request is not about the follower, `undefined` will be returned. + * @returns The removed actor. If the follower does not exist, the follow + * request is not about the follower, or the actor still has another + * active follow request, `undefined` will be returned. */ removeFollower(followId: URL, followerId: URL): Promise; @@ -334,6 +356,7 @@ export interface KvStoreRepositoryPrefixes { export class KvRepository implements Repository { readonly kv: KvStore; readonly prefixes: KvStoreRepositoryPrefixes; + private readonly nonCasLocks = new Map>(); /** * Creates a new key-value store repository. @@ -511,58 +534,287 @@ export class KvRepository implements Repository { if (follower.id == null) { throw new TypeError("The follower ID is missing."); } - const followerKey: KvKey = [...this.prefixes.followers, follower.id.href]; - await this.kv.set( - followerKey, - await follower.toJsonLd({ format: "compact" }), + const followRequestIdString = followRequestId.href; + const followerId = follower.id.href; + await this.withKvLock( + this.getFollowRequestLockKey(followRequestIdString), + async () => { + const followRequestKey = this.getFollowRequestKey( + followRequestIdString, + ); + const previousFollowerId = await this.kv.get(followRequestKey); + const followerJson = await follower.toJsonLd({ format: "compact" }); + await this.withKvLock(this.getFollowersLockKey(), async () => { + const followerKey: KvKey = [...this.prefixes.followers, followerId]; + await this.kv.set(followerKey, followerJson); + await this.addFollowerIdLocked(followerId); + await this.addFollowRequestForFollowerLocked( + followerId, + followRequestIdString, + ); + await this.kv.set(followRequestKey, followerId); + if ( + previousFollowerId != null && previousFollowerId !== followerId + ) { + await this.removeFollowRequestForFollowerLocked( + previousFollowerId, + followRequestIdString, + ); + await this.cleanupFollowerLocked(previousFollowerId); + } + }); + }, ); - const lockKey: KvKey = [...this.prefixes.followers, "lock"]; - const listKey: KvKey = this.prefixes.followers; - do { - await this.kv.set(lockKey, follower.id.href); - const list = await this.kv.get(listKey) ?? []; - if (!list.includes(follower.id.href)) list.push(follower.id.href); - await this.kv.set(listKey, list); - } while (await this.kv.get(lockKey) !== follower.id.href); - const followRequestKey: KvKey = [ - ...this.prefixes.followRequests, - followRequestId.href, - ]; - await this.kv.set(followRequestKey, follower.id.href); } async removeFollower( followRequestId: URL, actorId: URL, ): Promise { - const followRequestKey: KvKey = [ - ...this.prefixes.followRequests, - followRequestId.href, - ]; - const followerId = await this.kv.get(followRequestKey); - if (followerId == null) return undefined; - const followerKey: KvKey = [...this.prefixes.followers, followerId]; - if (followerId !== actorId.href) return undefined; - const followerJson = await this.kv.get(followerKey); - if (followerJson == null) return undefined; - let follower: Object; - try { - follower = await Object.fromJsonLd(followerJson); - } catch { - return undefined; + const followRequestIdString = followRequestId.href; + return await this.withKvLock( + this.getFollowRequestLockKey(followRequestIdString), + async () => { + const followRequestKey = this.getFollowRequestKey( + followRequestIdString, + ); + const followerId = await this.kv.get(followRequestKey); + if (followerId == null || followerId !== actorId.href) { + return undefined; + } + return await this.withKvLock(this.getFollowersLockKey(), async () => { + const currentFollowerId = await this.kv.get(followRequestKey); + if (currentFollowerId !== followerId) return undefined; + const followerKey: KvKey = [...this.prefixes.followers, followerId]; + const followerJson = await this.kv.get(followerKey); + await this.kv.delete(followRequestKey); + await this.removeFollowRequestForFollowerLocked( + followerId, + followRequestIdString, + ); + const removed = await this.cleanupFollowerLocked(followerId); + if (followerJson == null) return undefined; + let follower: Object; + try { + follower = await Object.fromJsonLd(followerJson); + } catch { + return undefined; + } + if (!isActor(follower)) return undefined; + return removed ? follower : undefined; + }); + }, + ); + } + + private async addFollowerIdLocked(followerId: string): Promise { + const listKey: KvKey = this.prefixes.followers; + const list = await this.kv.get(listKey) ?? []; + if (!list.includes(followerId)) { + await this.kv.set(listKey, [...list, followerId]); + } + } + + private async cleanupFollowerLocked(followerId: string): Promise { + if (await this.hasFollowRequestForFollowerLocked(followerId)) { + return false; } - if (!isActor(follower)) return undefined; - const lockKey: KvKey = [...this.prefixes.followers, "lock"]; const listKey: KvKey = this.prefixes.followers; - do { - await this.kv.set(lockKey, followerId); - let list = await this.kv.get(listKey) ?? []; - list = list.filter((id) => id !== followerId); - await this.kv.set(listKey, list); - } while (await this.kv.get(lockKey) !== followerId); - await this.kv.delete(followerKey); - await this.kv.delete(followRequestKey); - return follower; + const list = await this.kv.get(listKey) ?? []; + await this.kv.set( + listKey, + list.filter((id) => id !== followerId), + ); + await this.kv.delete([...this.prefixes.followers, followerId]); + await this.kv.delete(this.getFollowerFollowRequestsKey(followerId)); + return true; + } + + private getFollowRequestKey(followRequestId: string): KvKey { + return [...this.prefixes.followRequests, followRequestId]; + } + + private getFollowRequestLockKey(followRequestId: string): KvKey { + return [...this.prefixes.followRequests, "lock", followRequestId]; + } + + private getFollowerFollowRequestsKey(followerId: string): KvKey { + return [...this.prefixes.followRequests, "followers", followerId]; + } + + private getFollowersLockKey(): KvKey { + return [...this.prefixes.followers, "lock"]; + } + + private async addFollowRequestForFollowerLocked( + followerId: string, + followRequestId: string, + ): Promise { + const followRequestIds = await this + .getIndexedFollowRequestsForFollowerLocked( + followerId, + ) ?? await this.rebuildFollowRequestsForFollowerLocked(followerId); + if (!followRequestIds.includes(followRequestId)) { + await this.kv.set(this.getFollowerFollowRequestsKey(followerId), [ + ...followRequestIds, + followRequestId, + ]); + } + } + + private async removeFollowRequestForFollowerLocked( + followerId: string, + followRequestId: string, + ): Promise { + const followRequestIds = await this + .getIndexedFollowRequestsForFollowerLocked( + followerId, + ) ?? await this.rebuildFollowRequestsForFollowerLocked(followerId); + await this.kv.set( + this.getFollowerFollowRequestsKey(followerId), + followRequestIds.filter((id) => id !== followRequestId), + ); + } + + private async hasFollowRequestForFollowerLocked( + followerId: string, + ): Promise { + const indexKey = this.getFollowerFollowRequestsKey(followerId); + const followRequestIds = await this + .getIndexedFollowRequestsForFollowerLocked( + followerId, + ); + if (followRequestIds != null) { + return await this.hasCurrentFollowRequestForFollowerLocked( + followerId, + followRequestIds, + indexKey, + ); + } + const rebuiltFollowRequestIds = await this + .rebuildFollowRequestsForFollowerLocked( + followerId, + ); + return await this.hasCurrentFollowRequestForFollowerLocked( + followerId, + rebuiltFollowRequestIds, + indexKey, + ); + } + + private async hasCurrentFollowRequestForFollowerLocked( + followerId: string, + followRequestIds: readonly string[], + indexKey: KvKey, + ): Promise { + const currentFollowRequestIds: string[] = []; + for (const followRequestId of followRequestIds) { + if ( + await this.kv.get(this.getFollowRequestKey(followRequestId)) === + followerId + ) { + currentFollowRequestIds.push(followRequestId); + } + } + if (currentFollowRequestIds.length !== followRequestIds.length) { + await this.kv.set( + indexKey, + currentFollowRequestIds, + ); + } + return currentFollowRequestIds.length > 0; + } + + private async getIndexedFollowRequestsForFollowerLocked( + followerId: string, + ): Promise { + return await this.kv.get( + this.getFollowerFollowRequestsKey(followerId), + ); + } + + private async rebuildFollowRequestsForFollowerLocked( + followerId: string, + ): Promise { + const indexKey = this.getFollowerFollowRequestsKey(followerId); + const indexedFollowRequestIds: string[] = []; + const followRequestKeyLength = this.prefixes.followRequests.length + 1; + for await (const entry of this.kv.list(this.prefixes.followRequests)) { + if (entry.key.length !== followRequestKeyLength) continue; + const followRequestId = entry.key[this.prefixes.followRequests.length]; + if (typeof followRequestId !== "string") continue; + if (entry.value === followerId) { + indexedFollowRequestIds.push(followRequestId); + } + } + await this.kv.set(indexKey, indexedFollowRequestIds); + return indexedFollowRequestIds; + } + + private async withKvLock( + lockKey: KvKey, + operation: () => Promise, + ): Promise { + const cas = this.kv.cas?.bind(this.kv); + if (cas == null) { + return await this.withNonCasKvLock(lockKey, operation); + } + const lock: KvLock = { id: crypto.randomUUID() }; + const lockTtl = Temporal.Duration.from({ minutes: 5 }); + const lockReleaseTtl = Temporal.Duration.from({ + milliseconds: kvLockPollIntervalMs, + }); + while (true) { + if (await cas(lockKey, undefined, lock, { ttl: lockTtl })) { + break; + } + const currentLock = await this.kv.get(lockKey); + if ( + (isLegacyKvLock(currentLock) || isReleasedKvLock(currentLock)) && + await cas(lockKey, currentLock, lock, { ttl: lockTtl }) + ) { + break; + } + await new Promise((resolve) => setTimeout(resolve, kvLockPollIntervalMs)); + } + try { + return await operation(); + } finally { + try { + const currentLock = await this.kv.get(lockKey); + if (isKvLock(currentLock) && currentLock.id === lock.id) { + await cas(lockKey, currentLock, { ...currentLock, released: true }, { + ttl: lockReleaseTtl, + }); + } + } catch (error) { + logger.warn("Failed to release KV lock: {error}", { error }); + } + } + } + + private async withNonCasKvLock( + lockKey: KvKey, + operation: () => Promise, + ): Promise { + const encodedLockKey = JSON.stringify(lockKey); + const previousLock = this.nonCasLocks.get(encodedLockKey) ?? + Promise.resolve(); + let releaseLock: () => void; + const lock = new Promise((resolve) => { + releaseLock = resolve; + }); + const tail = previousLock.then(() => lock, () => lock); + this.nonCasLocks.set(encodedLockKey, tail); + await previousLock.catch(() => {}); + try { + return await operation(); + } finally { + releaseLock!(); + if (this.nonCasLocks.get(encodedLockKey) === tail) { + this.nonCasLocks.delete(encodedLockKey); + } + } } async hasFollower(followerId: URL): Promise { @@ -851,8 +1103,14 @@ export class MemoryRepository implements Repository { if (follower.id == null) { throw new TypeError("The follower ID is missing."); } + const previousFollowerId = this.followRequests[followId.href]; this.followers.set(follower.id.href, follower); this.followRequests[followId.href] = follower.id.href; + if ( + previousFollowerId != null && previousFollowerId !== follower.id.href + ) { + this.cleanupFollower(previousFollowerId); + } return Promise.resolve(); } @@ -863,8 +1121,16 @@ export class MemoryRepository implements Repository { } delete this.followRequests[followId.href]; const follower = this.followers.get(followerId.href); - this.followers.delete(followerId.href); - return Promise.resolve(follower); + const removed = this.cleanupFollower(followerId.href); + return Promise.resolve(removed ? follower : undefined); + } + + private cleanupFollower(followerId: string): boolean { + if (globalThis.Object.values(this.followRequests).includes(followerId)) { + return false; + } + this.followers.delete(followerId); + return true; } hasFollower(followerId: URL): Promise { @@ -1060,9 +1326,7 @@ export class MemoryCachedRepository implements Repository { followId, followerId, ); - if (removedFollower !== undefined) { - await this.cache.removeFollower(followId, followerId); - } + await this.cache.removeFollower(followId, followerId); return removedFollower; }