From fa51a97b02fdd0b4a42a6f166b85dd484c5c05e1 Mon Sep 17 00:00:00 2001 From: Shakker Date: Sun, 26 Jul 2026 05:09:40 +0100 Subject: [PATCH] fix: enforce Buzz room membership --- .../buzz/src/buzz-bus.lifecycle.test.ts | 95 +++- extensions/buzz/src/buzz-bus.ts | 479 ++++++++++++++++-- extensions/buzz/src/room-membership.ts | 110 ++++ 3 files changed, 624 insertions(+), 60 deletions(-) create mode 100644 extensions/buzz/src/room-membership.ts diff --git a/extensions/buzz/src/buzz-bus.lifecycle.test.ts b/extensions/buzz/src/buzz-bus.lifecycle.test.ts index f620afb54957..8c12b538a679 100644 --- a/extensions/buzz/src/buzz-bus.lifecycle.test.ts +++ b/extensions/buzz/src/buzz-bus.lifecycle.test.ts @@ -1,4 +1,4 @@ -import { finalizeEvent, type Event } from "nostr-tools"; +import { finalizeEvent, getPublicKey, type Event, type Filter } from "nostr-tools"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; @@ -8,8 +8,15 @@ const relayMocks = vi.hoisted(() => ({ publish: vi.fn<(event: Event) => Promise>(), subscriptionClose: vi.fn(), close: vi.fn(), - onevent: undefined as ((event: Event) => void) | undefined, - closeHandler: undefined as ((reason: string) => void) | undefined, + membershipEvents: [] as Event[], + subscriptions: [] as Array<{ + filter: Filter; + handlers: { + onevent: (event: Event) => void; + oneose?: () => void; + onclose: (reason: string) => void; + }; + }>, })); vi.mock("nostr-tools", async (importOriginal) => { @@ -24,11 +31,23 @@ vi.mock("nostr-tools", async (importOriginal) => { close = relayMocks.close; subscribe( - _filters: unknown, - handlers: { onevent: (event: Event) => void; onclose: (reason: string) => void }, + filters: Filter[], + handlers: { + onevent: (event: Event) => void; + oneose?: () => void; + onclose: (reason: string) => void; + }, ) { - relayMocks.onevent = handlers.onevent; - relayMocks.closeHandler = handlers.onclose; + const filter = filters[0] ?? {}; + relayMocks.subscriptions.push({ filter, handlers }); + if (filter.kinds?.includes(39002)) { + for (const event of relayMocks.membershipEvents) { + handlers.onevent(event); + } + handlers.oneose?.(); + } else if (filter.kinds?.includes(40099)) { + handlers.oneose?.(); + } return { close: relayMocks.subscriptionClose }; } }, @@ -40,6 +59,9 @@ import { sendBuzzTextOneShot, startBuzzBus } from "./buzz-bus.js"; const PRIVATE_KEY = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f"; const SENDER_PRIVATE_KEY = "0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20"; const ACCOUNT_ID = "default"; +const CHANNEL_ID = "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"; +const BOT_PUBLIC_KEY = getPublicKey(Uint8Array.from(Buffer.from(PRIVATE_KEY, "hex"))); +const SENDER_PUBLIC_KEY = getPublicKey(Uint8Array.from(Buffer.from(SENDER_PRIVATE_KEY, "hex"))); const tempDirs = useAutoCleanupTempDirTracker(afterEach); let previousStateDir: string | undefined; let stateDir: string; @@ -50,8 +72,22 @@ describe("Buzz bus lifecycle", () => { stateDir = tempDirs.make("openclaw-buzz-dedupe-"); process.env.OPENCLAW_STATE_DIR = stateDir; vi.clearAllMocks(); - relayMocks.onevent = undefined; - relayMocks.closeHandler = undefined; + relayMocks.subscriptions.length = 0; + relayMocks.membershipEvents = [ + { + id: "membership-1", + kind: 39002, + pubkey: "f".repeat(64), + created_at: 1_700_000_000, + content: "", + sig: "e".repeat(128), + tags: [ + ["d", CHANNEL_ID], + ["p", BOT_PUBLIC_KEY, "", "bot"], + ["p", SENDER_PUBLIC_KEY, "", "member"], + ], + }, + ]; relayMocks.connect.mockResolvedValue(); relayMocks.auth.mockRejectedValue(new Error("auth rejected")); relayMocks.publish.mockResolvedValue(""); @@ -71,7 +107,7 @@ describe("Buzz bus lifecycle", () => { accountId: ACCOUNT_ID, relayUrl: "wss://buzz.example.com", privateKey: PRIVATE_KEY, - channelIds: ["7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"], + channelIds: [CHANNEL_ID], onMessage: async () => {}, }), ).rejects.toThrow("auth rejected"); @@ -86,7 +122,7 @@ describe("Buzz bus lifecycle", () => { const messageId = await sendBuzzTextOneShot({ relayUrl: "wss://buzz.example.com", privateKey: PRIVATE_KEY, - channelId: "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c", + channelId: CHANNEL_ID, text: "hello", threadId: "root-id", replyToId: "parent-id", @@ -98,7 +134,7 @@ describe("Buzz bus lifecycle", () => { kind: 9, content: "hello", tags: [ - ["h", "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"], + ["h", CHANNEL_ID], ["e", "root-id", "", "root"], ["e", "parent-id", "", "reply"], ], @@ -114,7 +150,7 @@ describe("Buzz bus lifecycle", () => { sendBuzzTextOneShot({ relayUrl: "wss://buzz.example.com", privateKey: PRIVATE_KEY, - channelId: "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c", + channelId: CHANNEL_ID, text: "hello", }), ).rejects.toThrow("rejected"); @@ -129,7 +165,7 @@ describe("Buzz bus lifecycle", () => { accountId: ACCOUNT_ID, relayUrl: "wss://buzz.example.com", privateKey: PRIVATE_KEY, - channelIds: ["7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"], + channelIds: [CHANNEL_ID], onMessage, }); const event = finalizeEvent( @@ -137,13 +173,16 @@ describe("Buzz bus lifecycle", () => { kind: 9, created_at: 1_700_000_000, content: "hello", - tags: [["h", "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"]], + tags: [["h", CHANNEL_ID]], }, Uint8Array.from(Buffer.from(SENDER_PRIVATE_KEY, "hex")), ); - relayMocks.onevent?.(event); - relayMocks.onevent?.(event); + const messageSubscription = relayMocks.subscriptions.find((entry) => + entry.filter.kinds?.includes(9), + ); + messageSubscription?.handlers.onevent(event); + messageSubscription?.handlers.onevent(event); await vi.waitFor(() => expect(onMessage).toHaveBeenCalledOnce()); await bus.close(); @@ -157,7 +196,7 @@ describe("Buzz bus lifecycle", () => { accountId: ACCOUNT_ID, relayUrl: "wss://buzz.example.com", privateKey: PRIVATE_KEY, - channelIds: ["7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"], + channelIds: [CHANNEL_ID], onMessage: async () => { throw new Error("dispatch failed"); }, @@ -169,12 +208,14 @@ describe("Buzz bus lifecycle", () => { kind: 9, created_at: 1_700_000_000, content: "hello", - tags: [["h", "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"]], + tags: [["h", CHANNEL_ID]], }, Uint8Array.from(Buffer.from(SENDER_PRIVATE_KEY, "hex")), ); - relayMocks.onevent?.(event); + relayMocks.subscriptions + .find((entry) => entry.filter.kinds?.includes(9)) + ?.handlers.onevent(event); await vi.waitFor(() => expect(onMessageError).toHaveBeenCalledWith(expect.any(Error))); expect(onFatalError).not.toHaveBeenCalled(); @@ -188,7 +229,7 @@ describe("Buzz bus lifecycle", () => { kind: 9, created_at: Math.floor(Date.now() / 1000), content: "hello", - tags: [["h", "7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"]], + tags: [["h", CHANNEL_ID]], }, Uint8Array.from(Buffer.from(SENDER_PRIVATE_KEY, "hex")), ); @@ -197,10 +238,12 @@ describe("Buzz bus lifecycle", () => { accountId: ACCOUNT_ID, relayUrl: "wss://buzz.example.com", privateKey: PRIVATE_KEY, - channelIds: ["7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"], + channelIds: [CHANNEL_ID], onMessage: firstOnMessage, }); - relayMocks.onevent?.(event); + relayMocks.subscriptions + .find((entry) => entry.filter.kinds?.includes(9)) + ?.handlers.onevent(event); await vi.waitFor(() => expect(firstOnMessage).toHaveBeenCalledOnce()); await firstBus.close(); @@ -209,10 +252,12 @@ describe("Buzz bus lifecycle", () => { accountId: ACCOUNT_ID, relayUrl: "wss://buzz.example.com", privateKey: PRIVATE_KEY, - channelIds: ["7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"], + channelIds: [CHANNEL_ID], onMessage: secondOnMessage, }); - relayMocks.onevent?.(event); + relayMocks.subscriptions + .findLast((entry) => entry.filter.kinds?.includes(9)) + ?.handlers.onevent(event); await new Promise((resolve) => { setTimeout(resolve, 100); }); diff --git a/extensions/buzz/src/buzz-bus.ts b/extensions/buzz/src/buzz-bus.ts index c4315dee3291..1b301ec29213 100644 --- a/extensions/buzz/src/buzz-bus.ts +++ b/extensions/buzz/src/buzz-bus.ts @@ -7,6 +7,14 @@ import { } from "./message-event.js"; import { syncBuzzProfile } from "./profile.js"; import { authenticateBuzzRelay, createBuzzAuthSigner, parseBuzzAuthTag } from "./relay-auth.js"; +import { + BUZZ_ROOM_MEMBERSHIP_KIND, + BUZZ_ROOM_SYSTEM_KIND, + isNewerBuzzRoomMembership, + parseBuzzRoomMembershipChangeEvent, + parseBuzzRoomMembershipEvent, + type BuzzRoomMembership, +} from "./room-membership.js"; import { decodeBuzzPrivateKey, resolveBuzzPublicKey } from "./types.js"; const MESSAGE_KIND = 9; @@ -16,6 +24,9 @@ const REPLAY_TTL_MS = 30 * 24 * 60 * 60 * 1000; const REPLAY_MAX_ENTRIES = 10_000; const REPLAY_STATE_MAX_ENTRIES = 50_000; const REPLAY_NAMESPACE_PREFIX = "buzz.inbound-dedupe"; +const MEMBERSHIP_READY_TIMEOUT_MS = 10_000; +const MEMBERSHIP_REFRESH_DELAYS_MS = [100, 500, 1_500, 3_000] as const; +const MEMBERSHIP_EVENT_CACHE_MAX_ENTRIES = 10_000; export interface BuzzBus { publicKey: string; @@ -123,6 +134,381 @@ async function connectAuthenticatedBuzzRelay(params: { } } +async function sleepWithSignal(delayMs: number, signal?: AbortSignal): Promise { + signal?.throwIfAborted(); + await new Promise((resolve, reject) => { + let settled = false; + let timer: ReturnType | undefined; + const finish = (error?: unknown) => { + if (settled) { + return; + } + settled = true; + if (timer) { + clearTimeout(timer); + } + signal?.removeEventListener("abort", onAbort); + if (error === undefined) { + resolve(); + } else { + reject(error); + } + }; + const onAbort = () => + finish(signal?.reason ?? new Error("Buzz room membership refresh aborted")); + timer = setTimeout(() => finish(), delayMs); + signal?.addEventListener("abort", onAbort, { once: true }); + if (signal?.aborted) { + onAbort(); + } + }); +} + +async function queryBuzzRoomMemberships(params: { + relay: Relay; + channelIds: string[]; + timeoutMs?: number; + signal?: AbortSignal; +}): Promise> { + const configuredRooms = new Set(params.channelIds); + const memberships = new Map(); + return await new Promise>((resolve, reject) => { + let settled = false; + let subscription: ReturnType | undefined; + let timeout: ReturnType | undefined; + const finish = (error?: unknown) => { + if (settled) { + return; + } + settled = true; + if (timeout) { + clearTimeout(timeout); + } + params.signal?.removeEventListener("abort", onAbort); + subscription?.close("membership snapshot loaded"); + if (error === undefined) { + resolve(memberships); + } else { + reject( + error instanceof Error + ? error + : new Error("Buzz room membership query failed", { cause: error }), + ); + } + }; + const onAbort = () => + finish(params.signal?.reason ?? new Error("Buzz room membership query aborted")); + timeout = setTimeout( + () => finish(new Error("Timed out loading Buzz room membership")), + params.timeoutMs ?? MEMBERSHIP_READY_TIMEOUT_MS, + ); + params.signal?.addEventListener("abort", onAbort, { once: true }); + subscription = params.relay.subscribe( + [ + { + kinds: [BUZZ_ROOM_MEMBERSHIP_KIND], + "#d": params.channelIds, + limit: params.channelIds.length, + }, + ], + { + onevent: (event) => { + const membership = parseBuzzRoomMembershipEvent(event); + if ( + !membership || + !configuredRooms.has(membership.roomId) || + !isNewerBuzzRoomMembership(membership, memberships.get(membership.roomId)) + ) { + return; + } + memberships.set(membership.roomId, membership); + }, + oneose: () => finish(), + onclose: (reason) => { + if (reason !== "membership snapshot loaded") { + finish(new Error(`Buzz room membership query closed: ${reason}`)); + } + }, + }, + ); + if (settled) { + subscription.close("membership snapshot loaded"); + } + if (params.signal?.aborted) { + onAbort(); + } + }); +} + +async function createBuzzRoomMembershipTracker(params: { + relay: Relay; + channelIds: string[]; + botPublicKey: string; + since: number; + onFatalError?: (error: Error) => void; + signal?: AbortSignal; +}): Promise<{ + isMember: (channelId: string, publicKey: string) => boolean; + subscriptions: Array>; +}> { + type BufferedSystemEvent = { event: Event; historical: boolean }; + type ExpectedMembership = "present" | "absent"; + type RefreshState = { + generation: number; + lastAttemptedGeneration: number; + promise: Promise; + }; + + let initialized = false; + const historicalRooms = new Set(); + const bufferedEvents: BufferedSystemEvent[] = []; + const seenEventIds = new Map(); + const blockedRooms = new Set(); + const deniedMembers = new Map>(); + const pendingMemberships = new Map>(); + const refreshes = new Map(); + let memberships = new Map(); + + const markSystemEventSeen = (eventId: string): boolean => { + if (seenEventIds.has(eventId)) { + return false; + } + seenEventIds.set(eventId, true); + if (seenEventIds.size > MEMBERSHIP_EVENT_CACHE_MAX_ENTRIES) { + const oldestEventId = seenEventIds.keys().next().value; + if (oldestEventId) { + seenEventIds.delete(oldestEventId); + } + } + return true; + }; + const reportSystemEventError = (error: unknown) => { + if (params.signal?.aborted) { + return; + } + params.onFatalError?.(error instanceof Error ? error : new Error(String(error))); + params.relay.close(); + }; + + const refreshMembership = async (channelId: string, state: RefreshState): Promise => { + const baseline = memberships.get(channelId); + if (!baseline) { + throw new Error(`Missing Buzz room membership for ${channelId}`); + } + for (const delayMs of MEMBERSHIP_REFRESH_DELAYS_MS) { + const generation = state.generation; + state.lastAttemptedGeneration = generation; + await sleepWithSignal(delayMs, params.signal); + if (state.generation !== generation) { + continue; + } + let refreshed: BuzzRoomMembership | undefined; + try { + refreshed = ( + await queryBuzzRoomMemberships({ + relay: params.relay, + channelIds: [channelId], + timeoutMs: 3_000, + signal: params.signal, + }) + ).get(channelId); + } catch (error) { + if (params.signal?.aborted) { + throw error; + } + continue; + } + if (state.generation !== generation || !refreshed) { + continue; + } + const pending = pendingMemberships.get(channelId); + const pendingMatches = + !pending || + [...pending].every( + ([publicKey, expected]) => refreshed.members.has(publicKey) === (expected === "present"), + ); + const botMembershipChanged = pending?.has(params.botPublicKey) === true; + if ( + !pendingMatches || + (botMembershipChanged && !isNewerBuzzRoomMembership(refreshed, baseline)) + ) { + continue; + } + if ( + refreshed.roles.get(params.botPublicKey) !== "bot" || + !refreshed.members.has(params.botPublicKey) + ) { + blockedRooms.add(channelId); + throw new Error(`Buzz bot no longer has the Bot role in room ${channelId}`); + } + memberships.set(channelId, refreshed); + pendingMemberships.delete(channelId); + deniedMembers.delete(channelId); + blockedRooms.delete(channelId); + return; + } + if (state.generation !== state.lastAttemptedGeneration) { + return; + } + blockedRooms.add(channelId); + throw new Error(`Could not refresh Buzz room membership for ${channelId}`); + }; + + const refreshMembershipOnce = (channelId: string): Promise => { + const current = refreshes.get(channelId); + if (current) { + current.generation += 1; + return current.promise; + } + const state = { + generation: 1, + lastAttemptedGeneration: 0, + promise: Promise.resolve(), + } satisfies RefreshState; + state.promise = refreshMembership(channelId, state).finally(() => { + if (refreshes.get(channelId) === state) { + refreshes.delete(channelId); + } + if ( + state.generation !== state.lastAttemptedGeneration && + pendingMemberships.has(channelId) && + !params.signal?.aborted + ) { + void refreshMembershipOnce(channelId).catch(reportSystemEventError); + } + }); + refreshes.set(channelId, state); + return state.promise; + }; + + const handleSystemEvent = (event: Event): Promise | undefined => { + if (!markSystemEventSeen(event.id)) { + return; + } + const channelId = event.tags + .find((tag) => tag[0] === "h")?.[1] + ?.trim() + .toLowerCase(); + if (!channelId) { + return; + } + const membership = memberships.get(channelId); + if (!membership) { + return; + } + const change = parseBuzzRoomMembershipChangeEvent(event, membership); + if (!change) { + return; + } + // System events invalidate membership; the relay-signed roster decides the + // final state. Removals deny immediately, while joins wait for confirmation. + const expected = change.type === "member_joined" ? "present" : "absent"; + const pending = pendingMemberships.get(channelId) ?? new Map(); + pending.set(change.targetPublicKey, expected); + pendingMemberships.set(channelId, pending); + if (expected === "absent") { + const denied = deniedMembers.get(channelId) ?? new Set(); + denied.add(change.targetPublicKey); + deniedMembers.set(channelId, denied); + } + if (change.targetPublicKey === params.botPublicKey) { + blockedRooms.add(channelId); + } + return refreshMembershipOnce(channelId); + }; + + let resolveHistorical: (() => void) | undefined; + let rejectHistorical: ((error: Error) => void) | undefined; + const historicalReady = new Promise((resolve, reject) => { + resolveHistorical = resolve; + rejectHistorical = reject; + }); + const historicalTimeout = setTimeout(() => { + rejectHistorical?.(new Error("Timed out loading Buzz room membership changes")); + }, MEMBERSHIP_READY_TIMEOUT_MS); + const subscriptions = params.channelIds.map((channelId) => + params.relay.subscribe( + [ + { + kinds: [BUZZ_ROOM_SYSTEM_KIND], + "#h": [channelId], + since: params.since, + }, + ], + { + onevent: (event) => { + if (!initialized) { + bufferedEvents.push({ event, historical: !historicalRooms.has(channelId) }); + return; + } + void handleSystemEvent(event)?.catch(reportSystemEventError); + }, + oneose: () => { + historicalRooms.add(channelId); + if (historicalRooms.size === params.channelIds.length) { + resolveHistorical?.(); + } + }, + onclose: (reason) => { + if (!historicalRooms.has(channelId)) { + rejectHistorical?.( + new Error(`Buzz membership subscription closed for ${channelId}: ${reason}`), + ); + } else if ( + reason !== "shutdown" && + reason !== "relay connection closed by us" && + !params.signal?.aborted + ) { + params.onFatalError?.( + new Error(`Buzz membership subscription closed for ${channelId}: ${reason}`), + ); + } + }, + }, + ), + ); + + try { + await historicalReady; + memberships = await queryBuzzRoomMemberships(params); + } catch (error) { + for (const subscription of subscriptions) { + subscription.close("membership tracker setup failed"); + } + throw error; + } finally { + clearTimeout(historicalTimeout); + } + + for (const channelId of params.channelIds) { + if (memberships.get(channelId)?.roles.get(params.botPublicKey) !== "bot") { + for (const subscription of subscriptions) { + subscription.close("membership tracker setup failed"); + } + throw new Error(`Buzz bot does not have the Bot role in configured room ${channelId}`); + } + } + + // Each room subscription reaches EOSE before the snapshot query starts, so + // the snapshot owns historical state. Only events received after that room's + // EOSE can be newer than the loaded snapshot and need an in-memory overlay. + const liveEvents = bufferedEvents + .filter((entry) => !entry.historical) + .map((entry) => entry.event); + for (const event of liveEvents) { + void handleSystemEvent(event)?.catch(reportSystemEventError); + } + initialized = true; + + return { + isMember: (channelId, publicKey) => + !blockedRooms.has(channelId) && + !deniedMembers.get(channelId)?.has(publicKey.trim().toLowerCase()) && + memberships.get(channelId)?.members.has(publicKey.trim().toLowerCase()) === true, + subscriptions, + }; +} + export async function sendBuzzTextOneShot(params: { relayUrl: string; privateKey: string; @@ -168,6 +554,10 @@ export async function startBuzzBus(options: { const publicKey = resolveBuzzPublicKey(options.privateKey); const authTag = parseBuzzAuthTag(options.authTag ?? ""); const sessionStartedAt = Math.floor(Date.now() / 1000); + const lifecycleAbort = new AbortController(); + const signal = options.signal + ? AbortSignal.any([options.signal, lifecycleAbort.signal]) + : lifecycleAbort.signal; const replayGuard = createChannelReplayGuard({ dedupe: { pluginId: "buzz", @@ -186,7 +576,7 @@ export async function startBuzzBus(options: { relayUrl: options.relayUrl, secretKey, authTag, - signal: options.signal, + signal, }); let subscriptions: Array> = []; let stopPresenceHeartbeat = () => {}; @@ -198,6 +588,7 @@ export async function startBuzzBus(options: { return event.id; }, close: async () => { + lifecycleAbort.abort(new Error("Buzz bus closed")); stopPresenceHeartbeat(); for (const subscription of subscriptions) { subscription.close("shutdown"); @@ -208,40 +599,57 @@ export async function startBuzzBus(options: { }; try { - subscriptions = options.channelIds.map((channelId) => - relay.subscribe( - [ + const membershipTracker = await createBuzzRoomMembershipTracker({ + relay, + channelIds: options.channelIds, + botPublicKey: publicKey, + since: sessionStartedAt, + onFatalError: options.onFatalError, + signal, + }); + subscriptions.push(...membershipTracker.subscriptions); + + subscriptions.push( + ...options.channelIds.map((channelId) => + relay.subscribe( + [ + { + kinds: [MESSAGE_KIND], + "#h": [channelId], + since: options.since ?? sessionStartedAt, + }, + ], { - kinds: [MESSAGE_KIND], - "#h": [channelId], - since: options.since ?? sessionStartedAt, + onevent: (event) => { + if (event.pubkey === publicKey) { + return; + } + if (!membershipTracker.isMember(channelId, event.pubkey)) { + return; + } + const message = parseBuzzMessageEvent(event); + if (!message || message.channelId !== channelId) { + return; + } + // Relay reconnects can replay signed events. Only admitted room + // members reach the persistent dedupe store or agent pipeline. + void replayGuard + .processGuarded(event, async () => { + await options.onMessage(message, bus); + }) + .catch((error: unknown) => { + options.onMessageError?.( + error instanceof Error ? error : new Error(String(error)), + ); + }); + }, + onclose: (reason) => { + if (reason !== "shutdown" && reason !== "relay connection closed by us") { + options.onFatalError?.(new Error(`Buzz subscription closed: ${reason}`)); + } + }, }, - ], - { - onevent: (event) => { - // Relay reconnects can replay signed events. Guard by immutable event id - // before any authorization, command, or agent work can run twice. - void replayGuard - .processGuarded(event, async () => { - if (event.pubkey === publicKey) { - return; - } - const message = parseBuzzMessageEvent(event); - if (!message) { - return; - } - await options.onMessage(message, bus); - }) - .catch((error: unknown) => { - options.onMessageError?.(error instanceof Error ? error : new Error(String(error))); - }); - }, - onclose: (reason) => { - if (reason !== "shutdown" && reason !== "relay connection closed by us") { - options.onFatalError?.(new Error(`Buzz subscription closed: ${reason}`)); - } - }, - }, + ), ), ); // Buzz presence is a separate ephemeral protocol, not a property of the @@ -260,7 +668,7 @@ export async function startBuzzBus(options: { publicKey, displayName: options.profileName, authTag, - signal: options.signal, + signal, }) .then((result) => { if (result.status === "published") { @@ -268,7 +676,7 @@ export async function startBuzzBus(options: { } }) .catch((error: unknown) => { - if (options.signal?.aborted) { + if (signal.aborted) { return; } options.onProfileError?.( @@ -283,6 +691,7 @@ export async function startBuzzBus(options: { } catch (error) { // Every failed startup must release the socket before ownership returns to // the gateway-level reconnect loop. + lifecycleAbort.abort(error); relay.close(); throw error; } diff --git a/extensions/buzz/src/room-membership.ts b/extensions/buzz/src/room-membership.ts new file mode 100644 index 000000000000..c38df66d93d0 --- /dev/null +++ b/extensions/buzz/src/room-membership.ts @@ -0,0 +1,110 @@ +import type { Event } from "nostr-tools"; +import { BUZZ_CHANNEL_ID_PATTERN } from "./target.js"; + +export const BUZZ_ROOM_MEMBERSHIP_KIND = 39002; +export const BUZZ_ROOM_SYSTEM_KIND = 40099; + +const HEX_PUBLIC_KEY_PATTERN = /^[0-9a-f]{64}$/u; +const MEMBERSHIP_CHANGE_TYPES = new Set(["member_joined", "member_left", "member_removed"]); + +export type BuzzRoomMembershipChange = { + type: "member_joined" | "member_left" | "member_removed"; + targetPublicKey: string; +}; + +export type BuzzRoomMembership = { + roomId: string; + createdAt: number; + eventId: string; + publisherPublicKey: string; + members: ReadonlySet; + roles: ReadonlyMap; +}; + +export function parseBuzzRoomMembershipEvent(event: Event): BuzzRoomMembership | undefined { + if (event.kind !== BUZZ_ROOM_MEMBERSHIP_KIND) { + return undefined; + } + const roomId = event.tags + .find((tag) => tag[0] === "d")?.[1] + ?.trim() + .toLowerCase(); + if (!roomId || !BUZZ_CHANNEL_ID_PATTERN.test(roomId)) { + return undefined; + } + + const members = new Set(); + const roles = new Map(); + for (const tag of event.tags) { + if (tag[0] !== "p") { + continue; + } + const publicKey = tag[1]?.trim().toLowerCase(); + if (!publicKey || !HEX_PUBLIC_KEY_PATTERN.test(publicKey)) { + continue; + } + members.add(publicKey); + const role = tag[3]?.trim().toLowerCase(); + if (role) { + roles.set(publicKey, role); + } + } + + return { + roomId, + createdAt: event.created_at, + eventId: event.id, + publisherPublicKey: event.pubkey.toLowerCase(), + members, + roles, + }; +} + +export function isNewerBuzzRoomMembership( + candidate: BuzzRoomMembership, + current: BuzzRoomMembership | undefined, +): boolean { + return ( + !current || + candidate.createdAt > current.createdAt || + (candidate.createdAt === current.createdAt && candidate.eventId < current.eventId) + ); +} + +export function parseBuzzRoomMembershipChangeEvent( + event: Event, + membership: BuzzRoomMembership, +): BuzzRoomMembershipChange | undefined { + if ( + event.kind !== BUZZ_ROOM_SYSTEM_KIND || + event.pubkey.toLowerCase() !== membership.publisherPublicKey || + !event.tags.some((tag) => tag[0] === "h" && tag[1]?.toLowerCase() === membership.roomId) + ) { + return undefined; + } + try { + const content = JSON.parse(event.content) as { + type?: unknown; + target?: unknown; + actor?: unknown; + }; + if (typeof content.type !== "string" || !MEMBERSHIP_CHANGE_TYPES.has(content.type)) { + return undefined; + } + const target = + typeof content.target === "string" + ? content.target.trim().toLowerCase() + : content.type === "member_left" && typeof content.actor === "string" + ? content.actor.trim().toLowerCase() + : ""; + if (!HEX_PUBLIC_KEY_PATTERN.test(target)) { + return undefined; + } + return { + type: content.type as BuzzRoomMembershipChange["type"], + targetPublicKey: target, + }; + } catch { + return undefined; + } +}