Scope desktop presence subscriptions to active demand (#5830)

## Summary - replace the desktop's global kind-20001 presence subscription with one author-filtered subscription derived from active TanStack presence queries - reconcile changing demand without a delivery gap: promote only after relay EOSE, keep the last confirmed subscription on failure, discard stale opens, and close entirely when demand is empty - preserve REST presence as the initial seed and TTL/crash-recovery backstop - add transport-seam and lifecycle tests for readiness, normalization, churn, retries, close failures, reconnect ownership assumptions, and disposal ## Why The desktop currently receives presence heartbeats from every identity on the relay. A live tap measured roughly 2,700 events/minute (45/sec), about 1 MB/minute and 71.5% of readable traffic, from approximately 1,300 distinct fleet identities. Most are discarded only after WebSocket, Tauri IPC, and JS parsing. This change applies normal Nostr author filtering at relay fan-out, before those costs. It deliberately does not introduce a relay digest protocol or client-side event batching; relevant-author traffic should be small after scoping, and the existing signed-delta/REST-TTL model remains intact. ## Correctness model - active query observers are the demand source; inactive cached queries retain no authors - replacement opens before old closes and is promoted only after EOSE - timeout/CLOSED rejects and closes the candidate while preserving the last good subscription - rapid A→B→C and A→B→A churn cannot unseat current A with stale B - empty demand never sends an unfiltered subscription - RelayClient continues to own reconnect replay; the reconciler does not duplicate subscriptions on reconnect ## Validation Exact pushed head: `8845093aec0330be16efe52d3459ff67f1000ff4` Pre-push hooks passed: - desktop check and file-size ratchet - desktop TypeScript - desktop unit suite: 4,791/4,791 - branch-skew check Focused lifecycle/transport suite: 34/34 passed before commit. Independent Royal Court review found and blocked two prototype flaws (timeout-as-success and starvation-prone trailing debounce); both were fixed and the final worktree was cleared with no remaining correctness or lifecycle blockers. Signed-off-by: Wes <wesbillman@users.noreply.github.com> Co-authored-by: Carl <c7ebe626f000404285d3686e1dc74cc07cc60a9754a150041ba132e14bd3e2ec@buzz.block.builderlab.xyz>
2026-08-18 06:50:31 +02:00 · 2026-08-13 20:32:34 -07:00
parent b30f1f6129
commit df9e773a13
9 changed files with 571 additions and 35 deletions
@@ -9,6 +9,7 @@ import { getPresence } from "@/shared/api/tauri";
 import { normalizePubkey } from "@/shared/lib/pubkey";
 import { useFocusedRefetchInterval } from "@/shared/lib/useDocumentVisible";
 import {
+  activePresencePubkeys,
  mergePresenceUpdate,
  parseLivePresenceEvent,
  presenceQueryWantsPubkey,
@@ -16,6 +17,8 @@ import {
  PRESENCE_TTL_SECONDS,
  resolveAutomaticPresenceStatus,
 } from "@/features/presence/lib/presence";
+import { PresenceSubscriptionReconciler } from "@/features/presence/lib/presenceSubscriptionReconciler";
+import { openPresenceSubscription } from "@/shared/api/presenceRelaySubscription";
 import type { PresenceLookup, PresenceStatus } from "@/shared/api/types";

 const PRESENCE_STATUS_TICK_INTERVAL_MS = 30_000;
@@ -113,18 +116,16 @@ export function usePresenceQuery(
 }

 /**
- * Subscribe to kind:20001 presence events over WebSocket and update the
- * TanStack Query presence cache in-place when updates arrive. Call once
- * in AppShell. Uses setQueriesData for targeted per-pubkey updates without
- * triggering refetches. Retries with exponential backoff on failure.
+ * Keep one live presence subscription scoped to pubkeys requested by active
+ * TanStack queries. Replacement subscriptions open before the old one closes,
+ * avoiding a live-update gap while query observers change.
 */
 export function usePresenceSubscription() {
  const queryClient = useQueryClient();

  React.useEffect(() => {
-    let unsub: (() => Promise<void>) | null = null;
    let isCancelled = false;
-    let retryTimer: ReturnType<typeof setTimeout> | null = null;
+    let reconcileTimer: ReturnType<typeof setTimeout> | null = null;

    function handlePresenceEvent(event: { pubkey: string; content: string }) {
      if (isCancelled) return;
@@ -141,28 +142,37 @@ export function usePresenceSubscription() {
      );
    }

-    function subscribeWithRetry(attempt = 0) {
-      if (isCancelled) return;
-      void relayClient
-        .subscribeToPresenceUpdates(handlePresenceEvent)
-        .then((unsubFn) => {
-          if (isCancelled) {
-            void unsubFn();
-            return;
-          }
-          unsub = unsubFn;
-        })
-        .catch(() => {
-          if (!isCancelled) {
-            const delay = Math.min(1000 * 2 ** attempt, 30_000);
-            retryTimer = setTimeout(
-              () => subscribeWithRetry(attempt + 1),
-              delay,
-            );
-          }
-        });
+    const reconciler = new PresenceSubscriptionReconciler({
+      open: (authors) =>
+        openPresenceSubscription(authors, handlePresenceEvent, (...args) =>
+          relayClient.subscribeLive(...args),
+        ),
+    });
+
+    function reconcileActiveQueries() {
+      reconciler.setAuthors(
+        activePresencePubkeys(queryClient.getQueryCache().getAll()),
+      );
    }
-    subscribeWithRetry();
+
+    function scheduleReconcile() {
+      if (reconcileTimer) return;
+      reconcileTimer = setTimeout(() => {
+        reconcileTimer = null;
+        reconcileActiveQueries();
+      }, 100);
+    }
+
+    const unsubQueryCache = queryClient.getQueryCache().subscribe((event) => {
+      if (
+        event.type === "observerAdded" ||
+        event.type === "observerRemoved" ||
+        event.type === "observerOptionsUpdated"
+      ) {
+        scheduleReconcile();
+      }
+    });
+    reconcileActiveQueries();

    const unsubReconnect = relayClient.subscribeToReconnects(() => {
      if (!isCancelled)
@@ -171,9 +181,10 @@ export function usePresenceSubscription() {

    return () => {
      isCancelled = true;
+      unsubQueryCache();
      unsubReconnect();
-      if (retryTimer) clearTimeout(retryTimer);
-      if (unsub) void unsub();
+      if (reconcileTimer) clearTimeout(reconcileTimer);
+      reconciler.dispose();
    };
  }, [queryClient]);
 }
@@ -2,6 +2,7 @@ import assert from "node:assert/strict";
 import test from "node:test";

 import {
+  activePresencePubkeys,
  mergePresenceUpdate,
  parseLivePresenceEvent,
  presenceQueryWantsPubkey,
@@ -15,6 +16,22 @@ const WILL = "8e39cba681211b3782d0e4483e9343719b9b7be66515252da5491f26421896b1";
 const OTHER =
  "44b8e82baaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";

+test("active presence authors are normalized, deduplicated, and sorted", () => {
+  const queries = [
+    { queryKey: ["presence", WILL.toUpperCase(), OTHER], isActive: () => true },
+    { queryKey: ["presence", WILL], isActive: () => true },
+  ];
+  assert.deepEqual(activePresencePubkeys(queries), [OTHER, WILL].sort());
+});
+
+test("inactive and non-presence queries do not retain presence authors", () => {
+  const queries = [
+    { queryKey: ["presence", WILL], isActive: () => false },
+    { queryKey: ["profiles", OTHER], isActive: () => true },
+  ];
+  assert.deepEqual(activePresencePubkeys(queries), []);
+});
+
 test("presence heartbeat is one minute with a three-window TTL", () => {
  assert.equal(PRESENCE_HEARTBEAT_INTERVAL_MS, 60_000);
  assert.equal(PRESENCE_TTL_SECONDS, 180);
@@ -15,6 +15,19 @@ export function parseLivePresenceEvent(event: {
  return { pubkey: event.pubkey.toLowerCase(), status };
 }

+export function activePresencePubkeys(
+  queries: Array<{ queryKey: readonly unknown[]; isActive: () => boolean }>,
+): string[] {
+  const pubkeys = new Set<string>();
+  for (const query of queries) {
+    if (!query.isActive() || query.queryKey[0] !== "presence") continue;
+    for (const value of query.queryKey.slice(1)) {
+      if (typeof value === "string" && value) pubkeys.add(value.toLowerCase());
+    }
+  }
+  return [...pubkeys].sort();
+}
+
 // Presence query keys are ["presence", ...normalizedSortedPubkeys]; a query
 // "wants" an update only for a pubkey it actually requested.
 export function presenceQueryWantsPubkey(
@@ -0,0 +1,251 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import { PresenceSubscriptionReconciler } from "./presenceSubscriptionReconciler.ts";
+
+const A = "a".repeat(64);
+const B = "b".repeat(64);
+const flush = () => new Promise((resolve) => setTimeout(resolve, 0));
+
+function deferred() {
+  let resolve;
+  let reject;
+  const promise = new Promise((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  return { promise, resolve, reject };
+}
+
+test("normalizes demand and refuses to open an empty subscription", async () => {
+  const opened = [];
+  const reconciler = new PresenceSubscriptionReconciler({
+    open: async (authors) => {
+      opened.push(authors);
+      return async () => {};
+    },
+  });
+
+  reconciler.setAuthors([]);
+  reconciler.setAuthors([B, A.toUpperCase(), A]);
+  await flush();
+  assert.deepEqual(opened, [[A, B]]);
+  reconciler.dispose();
+});
+
+test("opens replacement before closing the previous subscription", async () => {
+  const actions = [];
+  const reconciler = new PresenceSubscriptionReconciler({
+    open: async (authors) => {
+      const key = authors.join("");
+      actions.push(`open:${key}`);
+      return async () => actions.push(`close:${key}`);
+    },
+  });
+
+  reconciler.setAuthors([A]);
+  await flush();
+  reconciler.setAuthors([B]);
+  await flush();
+  assert.deepEqual(actions, [`open:${A}`, `open:${B}`, `close:${A}`]);
+  reconciler.dispose();
+});
+
+test("a stale async open is closed and never installed", async () => {
+  const first = deferred();
+  const actions = [];
+  let opens = 0;
+  const reconciler = new PresenceSubscriptionReconciler({
+    open: async (authors) => {
+      opens += 1;
+      const key = authors.join("");
+      actions.push(`open:${key}`);
+      if (opens === 1) await first.promise;
+      return async () => actions.push(`close:${key}`);
+    },
+  });
+
+  reconciler.setAuthors([A]);
+  await flush();
+  reconciler.setAuthors([B]);
+  first.resolve();
+  await flush();
+  await flush();
+  assert.deepEqual(actions, [`open:${A}`, `close:${A}`, `open:${B}`]);
+  reconciler.dispose();
+});
+
+test("failed replacement preserves the previous subscription and retries", async () => {
+  const timers = [];
+  const actions = [];
+  let failB = true;
+  const reconciler = new PresenceSubscriptionReconciler({
+    open: async (authors) => {
+      const key = authors.join("");
+      actions.push(`open:${key}`);
+      if (key === B && failB) throw new Error("relay unavailable");
+      return async () => actions.push(`close:${key}`);
+    },
+    retryDelay: () => 1,
+    setTimer: (callback) => {
+      timers.push(callback);
+      return timers.length;
+    },
+    clearTimer: () => {},
+  });
+
+  reconciler.setAuthors([A]);
+  await flush();
+  reconciler.setAuthors([B]);
+  await flush();
+  assert.deepEqual(actions, [`open:${A}`, `open:${B}`]);
+  assert.equal(timers.length, 1);
+
+  failB = false;
+  timers.shift()();
+  await flush();
+  assert.deepEqual(actions, [
+    `open:${A}`,
+    `open:${B}`,
+    `open:${B}`,
+    `close:${A}`,
+  ]);
+  reconciler.dispose();
+});
+
+test("rapid A to B to C keeps A until C is confirmed", async () => {
+  const openingB = deferred();
+  const actions = [];
+  const reconciler = new PresenceSubscriptionReconciler({
+    open: async (authors) => {
+      const key = authors.join("");
+      actions.push(`open:${key}`);
+      if (key === B) await openingB.promise;
+      return async () => actions.push(`close:${key}`);
+    },
+  });
+
+  reconciler.setAuthors([A]);
+  await flush();
+  reconciler.setAuthors([B]);
+  await flush();
+  reconciler.setAuthors(["c".repeat(64)]);
+  openingB.resolve();
+  await flush();
+  await flush();
+  assert.deepEqual(actions, [
+    `open:${A}`,
+    `open:${B}`,
+    `close:${B}`,
+    `open:${"c".repeat(64)}`,
+    `close:${A}`,
+  ]);
+  reconciler.dispose();
+});
+
+test("rapid A to B to A closes stale B but retains current A", async () => {
+  const openingB = deferred();
+  const actions = [];
+  const reconciler = new PresenceSubscriptionReconciler({
+    open: async (authors) => {
+      const key = authors.join("");
+      actions.push(`open:${key}`);
+      if (key === B) await openingB.promise;
+      return async () => actions.push(`close:${key}`);
+    },
+  });
+
+  reconciler.setAuthors([A]);
+  await flush();
+  reconciler.setAuthors([B]);
+  await flush();
+  reconciler.setAuthors([A]);
+  openingB.resolve();
+  await flush();
+  assert.deepEqual(actions, [`open:${A}`, `open:${B}`, `close:${B}`]);
+  reconciler.dispose();
+});
+
+test("prior close failure does not unseat a confirmed replacement", async () => {
+  const actions = [];
+  const reconciler = new PresenceSubscriptionReconciler({
+    open: async (authors) => {
+      const key = authors.join("");
+      actions.push(`open:${key}`);
+      return async () => {
+        actions.push(`close:${key}`);
+        if (key === A) throw new Error("close failed");
+      };
+    },
+  });
+
+  reconciler.setAuthors([A]);
+  await flush();
+  reconciler.setAuthors([B]);
+  await flush();
+  reconciler.setAuthors([B]);
+  await flush();
+  assert.deepEqual(actions, [`open:${A}`, `open:${B}`, `close:${A}`]);
+  reconciler.dispose();
+});
+
+test("dispose closes current and a late replacement without double-closing current", async () => {
+  const openingB = deferred();
+  const closes = { [A]: 0, [B]: 0 };
+  const reconciler = new PresenceSubscriptionReconciler({
+    open: async (authors) => {
+      const key = authors.join("");
+      if (key === B) await openingB.promise;
+      return async () => {
+        closes[key] += 1;
+      };
+    },
+  });
+
+  reconciler.setAuthors([A]);
+  await flush();
+  reconciler.setAuthors([B]);
+  await flush();
+  reconciler.dispose();
+  openingB.resolve();
+  await flush();
+  assert.deepEqual(closes, { [A]: 1, [B]: 1 });
+});
+
+test("dispose during an in-flight open closes the late subscription", async () => {
+  const opening = deferred();
+  let closeCount = 0;
+  const reconciler = new PresenceSubscriptionReconciler({
+    open: async () => {
+      await opening.promise;
+      return async () => {
+        closeCount += 1;
+      };
+    },
+  });
+
+  reconciler.setAuthors([A]);
+  await flush();
+  reconciler.dispose();
+  opening.resolve();
+  await flush();
+  assert.equal(closeCount, 1);
+});
+
+test("clearing demand closes current without opening a replacement", async () => {
+  const actions = [];
+  const reconciler = new PresenceSubscriptionReconciler({
+    open: async (authors) => {
+      const key = authors.join("");
+      actions.push(`open:${key}`);
+      return async () => actions.push(`close:${key}`);
+    },
+  });
+
+  reconciler.setAuthors([A]);
+  await flush();
+  reconciler.setAuthors([]);
+  await flush();
+  assert.deepEqual(actions, [`open:${A}`, `close:${A}`]);
+  reconciler.dispose();
+});
@@ -0,0 +1,138 @@
+export type PresenceSubscriptionClose = () => Promise<void>;
+
+export interface PresenceSubscriptionReconcilerOptions {
+  open: (authors: string[]) => Promise<PresenceSubscriptionClose>;
+  retryDelay?: (attempt: number) => number;
+  setTimer?: (
+    callback: () => void,
+    delayMs: number,
+  ) => ReturnType<typeof setTimeout>;
+  clearTimer?: (timer: ReturnType<typeof setTimeout>) => void;
+}
+
+/**
+ * Reconciles a changing author set onto one live relay subscription.
+ *
+ * A replacement opens before the prior subscription closes, so demand changes
+ * never create a live-update gap. Opens that finish after demand changes are
+ * closed without becoming current. A failed replacement leaves the last good
+ * subscription serving its old author set while retrying with bounded backoff.
+ */
+export class PresenceSubscriptionReconciler {
+  private readonly open: PresenceSubscriptionReconcilerOptions["open"];
+  private readonly retryDelay: (attempt: number) => number;
+  private readonly setTimer: NonNullable<
+    PresenceSubscriptionReconcilerOptions["setTimer"]
+  >;
+  private readonly clearTimer: NonNullable<
+    PresenceSubscriptionReconcilerOptions["clearTimer"]
+  >;
+  private desiredAuthors: string[] = [];
+  private desiredKey = "";
+  private current: { key: string; close: PresenceSubscriptionClose } | null =
+    null;
+  private running = false;
+  private disposed = false;
+  private retryAttempt = 0;
+  private retryTimer: ReturnType<typeof setTimeout> | null = null;
+
+  constructor(options: PresenceSubscriptionReconcilerOptions) {
+    this.open = options.open;
+    this.retryDelay =
+      options.retryDelay ??
+      ((attempt) => Math.min(1000 * 2 ** attempt, 30_000));
+    this.setTimer = options.setTimer ?? setTimeout;
+    this.clearTimer = options.clearTimer ?? clearTimeout;
+  }
+
+  setAuthors(authors: string[]) {
+    if (this.disposed) return;
+    const normalized = [
+      ...new Set(authors.map((author) => author.toLowerCase())),
+    ]
+      .filter(Boolean)
+      .sort();
+    const key = normalized.join(",");
+    if (key === this.desiredKey) return;
+
+    this.desiredAuthors = normalized;
+    this.desiredKey = key;
+    this.retryAttempt = 0;
+    this.cancelRetry();
+    void this.reconcile();
+  }
+
+  dispose() {
+    if (this.disposed) return;
+    this.disposed = true;
+    this.cancelRetry();
+    const current = this.current;
+    this.current = null;
+    if (current) void current.close().catch(() => {});
+  }
+
+  private currentKey() {
+    return this.current?.key ?? "";
+  }
+
+  private cancelRetry() {
+    if (this.retryTimer === null) return;
+    this.clearTimer(this.retryTimer);
+    this.retryTimer = null;
+  }
+
+  private scheduleRetry() {
+    if (this.retryTimer !== null || this.disposed) return;
+    const delay = this.retryDelay(this.retryAttempt);
+    this.retryAttempt += 1;
+    this.retryTimer = this.setTimer(() => {
+      this.retryTimer = null;
+      void this.reconcile();
+    }, delay);
+  }
+
+  private async reconcile() {
+    if (this.running || this.disposed) return;
+    this.running = true;
+    try {
+      while (!this.disposed && this.currentKey() !== this.desiredKey) {
+        const nextAuthors = this.desiredAuthors;
+        const nextKey = this.desiredKey;
+
+        if (nextAuthors.length === 0) {
+          const previous = this.current;
+          this.current = null;
+          if (previous) await previous.close().catch(() => {});
+          continue;
+        }
+
+        let nextClose: PresenceSubscriptionClose;
+        try {
+          nextClose = await this.open(nextAuthors);
+        } catch {
+          if (nextKey === this.desiredKey) this.scheduleRetry();
+          return;
+        }
+
+        if (this.disposed || nextKey !== this.desiredKey) {
+          await nextClose().catch(() => {});
+          continue;
+        }
+
+        const previous = this.current;
+        this.current = { key: nextKey, close: nextClose };
+        this.retryAttempt = 0;
+        if (previous) await previous.close().catch(() => {});
+      }
+    } finally {
+      this.running = false;
+      if (
+        !this.disposed &&
+        this.retryTimer === null &&
+        this.currentKey() !== this.desiredKey
+      ) {
+        void this.reconcile();
+      }
+    }
+  }
+}
@@ -0,0 +1,62 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import { openPresenceSubscription } from "./presenceRelaySubscription.ts";
+import { KIND_PRESENCE_UPDATE } from "../constants/kinds.ts";
+
+const A = "a".repeat(64);
+const B = "b".repeat(64);
+
+function liveWithReadiness(readiness) {
+  const calls = [];
+  let closeCount = 0;
+  const openLive = async (filter, onEvent, onReady, readinessTimeoutMs) => {
+    calls.push({ filter, onEvent, readinessTimeoutMs });
+    onReady(readiness);
+    return async () => {
+      closeCount += 1;
+    };
+  };
+  return { calls, closeCount: () => closeCount, openLive };
+}
+
+test("presence waits for EOSE and sends only normalized authors", async () => {
+  const { calls, closeCount, openLive } = liveWithReadiness("eose");
+  const onEvent = () => {};
+  const close = await openPresenceSubscription(
+    [B, A.toUpperCase(), A],
+    onEvent,
+    openLive,
+  );
+
+  assert.deepEqual(calls, [
+    {
+      filter: { kinds: [KIND_PRESENCE_UPDATE], authors: [A, B], limit: 0 },
+      onEvent,
+      readinessTimeoutMs: 5_000,
+    },
+  ]);
+  assert.equal(closeCount(), 0);
+  await close();
+  assert.equal(closeCount(), 1);
+});
+
+for (const readiness of ["timeout", "closed"]) {
+  test(`presence ${readiness} closes the candidate and rejects`, async () => {
+    const { closeCount, openLive } = liveWithReadiness(readiness);
+    await assert.rejects(
+      openPresenceSubscription([A], () => {}, openLive),
+      readiness === "closed" ? /rejected/ : /timed out/i,
+    );
+    assert.equal(closeCount(), 1);
+  });
+}
+
+test("empty demand never reaches the relay subscribe primitive", async () => {
+  const { calls, openLive } = liveWithReadiness("eose");
+  await assert.rejects(
+    openPresenceSubscription([], () => {}, openLive),
+    /at least one author/,
+  );
+  assert.deepEqual(calls, []);
+});
@@ -0,0 +1,45 @@
+import type {
+  LiveSubscriptionReadiness,
+  RelaySubscriptionFilter,
+} from "@/shared/api/relayClientShared";
+import type { RelayEvent } from "@/shared/api/types";
+import { KIND_PRESENCE_UPDATE } from "@/shared/constants/kinds";
+
+export type OpenLiveSubscription = (
+  filter: RelaySubscriptionFilter,
+  onEvent: (event: RelayEvent) => void,
+  onReady: (readiness: LiveSubscriptionReadiness) => void,
+  readinessTimeoutMs: number,
+) => Promise<() => Promise<void>>;
+
+/** Open an author-scoped presence subscription and require relay EOSE. */
+export async function openPresenceSubscription(
+  pubkeys: string[],
+  onEvent: (event: RelayEvent) => void,
+  openLive: OpenLiveSubscription,
+) {
+  const authors = [...new Set(pubkeys.map((pubkey) => pubkey.toLowerCase()))]
+    .filter(Boolean)
+    .sort();
+  if (authors.length === 0) {
+    throw new Error("Presence subscriptions require at least one author.");
+  }
+
+  const readiness: { value: LiveSubscriptionReadiness } = { value: "timeout" };
+  const unsubscribe = await openLive(
+    { kinds: [KIND_PRESENCE_UPDATE], authors, limit: 0 },
+    onEvent,
+    (nextReadiness) => {
+      readiness.value = nextReadiness;
+    },
+    5_000,
+  );
+  if (readiness.value === "eose") return unsubscribe;
+
+  await unsubscribe();
+  throw new Error(
+    readiness.value === "closed"
+      ? "Relay rejected the presence subscription."
+      : "Timed out confirming the presence subscription.",
+  );
+}
@@ -378,10 +378,6 @@ export class RelayClient {
    );
  }

-  async subscribeToPresenceUpdates(onEvent: (event: RelayEvent) => void) {
-    return this.subscribe({ kinds: [20001], limit: 0 }, onEvent);
-  }
-
  async publishUserStatus(text: string, emoji: string): Promise<void> {
    await this.ensureConnected();
    const tags: string[][] = [["d", "general"]];
@@ -414,8 +410,9 @@ export class RelayClient {
    filter: RelaySubscriptionFilter,
    onEvent: (event: RelayEvent) => void,
    onReady?: (readiness: LiveSubscriptionReadiness) => void,
+    readinessTimeoutMs?: number,
  ) {
-    return this.subscribe(filter, onEvent, onReady);
+    return this.subscribe(filter, onEvent, onReady, readinessTimeoutMs);
  }
  async subscribeToChannelMentionEvents(
    channelId: string,
@@ -600,6 +597,7 @@ export class RelayClient {
    filter: RelaySubscriptionFilter,
    onEvent: (event: RelayEvent) => void,
    onReady?: (readiness: LiveSubscriptionReadiness) => void,
+    readinessTimeoutMs = 250,
  ) {
    await this.ensureConnected();

@@ -614,7 +612,7 @@ export class RelayClient {
    });
    const fallbackTimeout = window.setTimeout(
      () => resolveReady("timeout"),
-      250,
+      readinessTimeoutMs,
    );

    this.subscriptions.set(subId, {
@@ -34,6 +34,7 @@ export const KIND_APPROVAL_REQUEST = 46010;
 export const KIND_MEMBER_ADDED_NOTIFICATION = 44100;
 export const KIND_MEMBER_REMOVED_NOTIFICATION = 44101;
 export const KIND_TYPING_INDICATOR = 20002;
+export const KIND_PRESENCE_UPDATE = 20001;
 export const KIND_HUDDLE_REACTION = 24810;
 export const KIND_HUDDLE_STARTED = 48100;
 export const KIND_HUDDLE_PARTICIPANT_JOINED = 48101;