fix(conv-key): rotate-instead-of-share + cache invalidation; fire-first realtime inserts (no sound-vs-text gap)

Friend-DM "Nachricht nicht lesbar" recurred even after v0.21.3 because the
proactive sweep called shareConvKeyToUser, which reads the module-level
conv-key cache first. After a server-side cleanup the cache still held the
stale locally-bootstrapped key, so each side wrapped its own different key
for the peer and the bundles diverged anew.

Switch the sweep to rotate_conv_key when any peer's user-id bundle is
missing at the active version: a fresh symmetric key is generated, wrapped
for every member at their CURRENT pubkey, and the active version is bumped
under a row-level FOR UPDATE lock. Concurrent rotations are race-safe — the
loser sees "new version must be greater" and bails; the winner's bundles
propagate via realtime.

Realtime conversation_keys subscription now invalidates the cache for the
affected (conversationId, key_version) on any INSERT/UPDATE/DELETE — so
admin cleanups, peer rotations, or device wraps can no longer leave a
stale entry in this client's session cache.

queueInsert now fires the first event of a quiet period immediately and
only collapses follow-up bursts. BATCH_WINDOW_MS dropped 250 → 80 ms.
This closes the ~250 ms gap between the notification sound and the
message body appearing.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

apps/desktop/src/lib/useConversationMessages.ts

@@ -1,6 +1,6 @@
-import { fetchPeerPublicKeys } from '@chat-app/shared/auth';
 import {
   type AttachmentHandle,
+  clearConvKeyCache,
   type DecryptedMessage,
   decryptMessages,
   encryptAndUploadAttachment,
@@ -9,8 +9,8 @@ import {
   insertAttachmentRow,
   MAX_ATTACHMENT_BYTES,
   type MessageWithCipher,
+  rotateConvKey,
   sendEncryptedMessage,
-  shareConvKeyToUser,
 } from '@chat-app/shared/chat';
 import { bytesToPgHex, pgBytesToBytes } from '@chat-app/shared/supabase';
 import { useCallback, useEffect, useMemo, useRef, useState } from 'react';
@@ -125,12 +125,25 @@ export function useConversationMessages({ conversationId, userId, deviceId }: Ar
     });
   }, [userId]);
 
-  // Proactive rewrap sweep: when a conversation opens, walk every accepted
+  // Proactive rewrap sweep: when a conversation opens, ensure the active
-  // member and ensure the active conv-key has a `recipient_user_id` bundle
+  // conv-key has a `recipient_user_id` bundle for every accepted member.
-  // for them. Members who are missing one (typically peers who haven't yet
+  //
-  // migrated to the per-user key model) get a best-effort wrap from the
+  // If any peer is missing a bundle at the active version, the previous
-  // local conv-key handle. Closes the legacy migration gap so peer B can
+  // implementation called `shareConvKeyToUser` for each missing peer —
-  // read on first unlock without manual intervention from A.
+  // that helper reads from the module-level conv-key cache first, and if
+  // the cache held a STALE locally-generated key (from a buggy bootstrap
+  // race in an earlier app version), the stale key got propagated to the
+  // peer's row. Both sides then encrypt with mutually un-mergeable keys
+  // and every message is "Nachricht nicht lesbar" forever (incident:
+  // conv aae12d84).
+  //
+  // The replacement: when any peer is missing, call `rotateConvKey` once.
+  // Rotation generates a fresh symmetric key locally, fetches each member's
+  // CURRENT pubkey, wraps the fresh key for everyone, and atomically bumps
+  // `active_key_version` via the `rotate_conv_key` RPC (FOR UPDATE lock
+  // serialises concurrent rotations). This bypasses the cache entirely:
+  // the new version's cache entry is the just-rotated key, and the stale
+  // entry at the old version is irrelevant because nobody reads it any more.
   useEffect(() => {
     if (!conversationId || !userId) return;
     let cancelled = false;
@@ -154,72 +167,76 @@ export function useConversationMessages({ conversationId, userId, deviceId }: Ar
         // db-types snapshot predates the active_key_version column; cast via unknown.
         const version = (convRow as unknown as { active_key_version: number }).active_key_version;
 
-        // Use `getOrCreateConvKey` rather than `tryGetConvKey` so that if we
+        // First, make sure we have a usable handle for the active version
-        // can't unwrap our bundle at the active version (we lost the device
+        // (this auto-rotates if we're locked out of our own bundle — the
-        // key, or the bundle was wiped by the 0.18.0 reset_user_key bug, or
+        // recovery path added in v0.21.1/v0.21.2).
-        // we only ever had a legacy `recipient_device_id` row), the helper
-        // auto-rotates the conv-key to version+1 and wraps fresh bundles
-        // for every member with a `user_keys` row. This is the only path
-        // that recovers stuck-legacy conversations on the receive side —
-        // `tryGetConvKey` just returned null and left the chat permanently
-        // un-decryptable for the locked-out party.
         const handle = await getOrCreateConvKey(supabase, conversationId, {
           userId,
           privateKey: priv,
         });
         if (cancelled) return;
-        // If the helper rotated, all current members with a `user_keys`
+        if (handle.keyVersion > version) return; // already rotated by helper
-        // public key were already wrapped by `rotateConvKey`. No further
-        // sweep work is needed.
-        if (handle.keyVersion > version) return;
 
+        // Check membership state on the server.
         const { data: members, error: mErr } = await supabase
           .from('conversation_members')
           .select('user_id, accepted')
           .eq('conversation_id', conversationId);
         if (mErr || !members) return;
-        const memberIds = (members as Array<{ user_id: string; accepted: boolean }>)
+        const peerIds = (members as Array<{ user_id: string; accepted: boolean }>)
           .filter((m) => m.accepted && m.user_id !== userId)
           .map((m) => m.user_id);
-        if (memberIds.length === 0) return;
+        if (peerIds.length === 0) return;
 
-        const peers = await fetchPeerPublicKeys(supabase, memberIds);
+        // Count how many of the peers have a recipient_user_id bundle at
-        for (const peer of peers) {
+        // the active version. If any are missing, rotate to V+1 — the
-          if (cancelled) return;
+        // rotation will wrap a fresh key for every accepted member with a
-          const { count, error: cntErr } = await (
+        // user_keys row.
-            supabase as unknown as {
+        const { data: existingRows, error: rowsErr } = await (
-              from: (t: string) => {
+          supabase as unknown as {
-                select: (s: string, o?: object) => {
+            from: (t: string) => {
-                  eq: (...a: unknown[]) => {
+              select: (s: string) => {
-                    eq: (...a: unknown[]) => {
+                eq: (c: string, v: string) => {
-                      eq: (
+                  eq: (c: string, v: number) => {
-                        ...a: unknown[]
+                    in: (c: string, v: string[]) => Promise<{
-                      ) => Promise<{ count: number | null; error: unknown }>;
+                      data: Array<{ recipient_user_id: string }> | null;
-                    };
+                      error: unknown;
+                    }>;
                   };
                 };
               };
-            }
+            };
-          )
-            .from('conversation_keys')
-            .select('recipient_user_id', { count: 'exact', head: true })
-            .eq('conversation_id', conversationId)
-            .eq('recipient_user_id', peer.userId)
-            .eq('key_version', version);
-          if (cntErr) continue;
-          if ((count ?? 0) === 0) {
-            try {
-              await shareConvKeyToUser(
-                supabase,
-                conversationId,
-                peer.userId,
-                peer.publicKey,
-                { userId, privateKey: priv },
-              );
-            } catch (err) {
-              console.warn('proactive rewrap failed for', peer.userId, err);
-            }
           }
+        )
+          .from('conversation_keys')
+          .select('recipient_user_id')
+          .eq('conversation_id', conversationId)
+          .eq('key_version', version)
+          .in('recipient_user_id', peerIds);
+        if (rowsErr) return;
+        const wrappedPeerIds = new Set(
+          (existingRows ?? []).map((r) => r.recipient_user_id),
+        );
+        const missing = peerIds.filter((id) => !wrappedPeerIds.has(id));
+        if (missing.length === 0) return;
+
+        // At least one peer is missing a bundle — rotate. We deliberately do
+        // NOT use the cached conv-key here. The rotation generates a fresh
+        // key wrapped to every current member's CURRENT pubkey, so any
+        // staleness in the local cache for the OLD version is irrelevant
+        // going forward.
+        try {
+          await rotateConvKey(supabase, conversationId, {
+            userId,
+            privateKey: priv,
+          });
+        } catch (err) {
+          // Most likely cause: a concurrent peer also called rotate and
+          // won the race; their bumped active_key_version makes our
+          // `p_new_version <= cur_version` and the RPC raises. That's fine —
+          // the next chat-open / send will fetch the new active version and
+          // unwrap the bundle that peer wrapped for us.
+          console.warn('proactive rotate failed (likely concurrent rotation)', err);
         }
       } catch (err) {
         console.warn('proactive rewrap sweep failed', err);
@@ -490,11 +507,14 @@ export function useConversationMessages({ conversationId, userId, deviceId }: Ar
     void refresh();
 
     // Batch INSERT bursts so a paste / backfill doesn't fire N parallel
-    // refetches + decrypts. If more than BATCH_BURST_THRESHOLD ids arrive
+    // refetches + decrypts. The first event in a quiet period fires
-    // within BATCH_WINDOW_MS, collapse to a single refresh() which pulls
+    // `handleInsert` immediately so single incoming messages don't sit
-    // the last 100 in one query — cheaper and keeps order stable. For
+    // behind a debounce timer (previous behaviour: 250 ms blank between
-    // lone inserts the per-id path stays so latency is unchanged.
+    // notification-sound and message body). Subsequent events arriving
-    const BATCH_WINDOW_MS = 250;
+    // within BATCH_WINDOW_MS of the first are buffered; if the burst grows
+    // past BATCH_BURST_THRESHOLD the buffered tail collapses into one
+    // `refresh()` instead of N individual refetches.
+    const BATCH_WINDOW_MS = 80;
     const BATCH_BURST_THRESHOLD = 3;
     let burstBuffer: Array<Record<string, unknown>> = [];
     let burstTimer: number | null = null;
@@ -513,6 +533,15 @@ export function useConversationMessages({ conversationId, userId, deviceId }: Ar
       }
     };
     const queueInsert = (row: Record<string, unknown>) => {
+      if (burstBuffer.length === 0 && burstTimer === null) {
+        // First event in a quiet period — fire immediately so the user sees
+        // the message right when they hear the notification sound. Arm a
+        // short window in case a burst follows; follow-ups go through the
+        // buffer and may collapse into a refresh.
+        void handleInsert(row);
+        burstTimer = window.setTimeout(flushBurst, BATCH_WINDOW_MS);
+        return;
+      }
       burstBuffer.push(row);
       if (burstTimer === null) {
         burstTimer = window.setTimeout(flushBurst, BATCH_WINDOW_MS);
@@ -539,18 +568,46 @@ export function useConversationMessages({ conversationId, userId, deviceId }: Ar
           }
         },
       )
-      // When a peer device wraps the conversation-key for us (e.g. we just
+      // Any conversation_keys change for this conv invalidates the cached
-      // registered a fresh device), re-decrypt the visible messages.
+      // conv-key for the affected version. The module-level cache in
+      // shared/chat/convKeys.ts otherwise holds the previously-unwrapped key
+      // forever within a session — which is exactly what propagated the
+      // stale local bootstrap key in conv aae12d84, recreating divergent
+      // bundles after a server-side cleanup. Clearing on any INSERT/UPDATE/
+      // DELETE for the conv forces the next `getOrCreateConvKey` /
+      // `tryGetConvKey` call to re-fetch the canonical bundle from the
+      // server. Cheap (a single Map.delete), defensive, and avoids stale-
+      // cache propagation across all of {peer rotation, device wrap, admin
+      // cleanup}.
+      //
+      // We also keep the historical "device wrap → refresh" trigger so a
+      // freshly-registered device of our own re-decrypts in place.
       .on(
         'postgres_changes',
         {
-          event: 'INSERT',
+          event: '*',
           schema: 'public',
           table: 'conversation_keys',
           filter: 'conversation_id=eq.' + conversationId,
         },
-        (payload: { new: { recipient_device_id?: string } }) => {
+        (payload: {
-          if (payload.new?.recipient_device_id === deviceId) {
+          eventType: 'INSERT' | 'UPDATE' | 'DELETE';
+          new: { recipient_device_id?: string; key_version?: number };
+          old: { recipient_device_id?: string; key_version?: number };
+        }) => {
+          const v =
+            payload.eventType === 'DELETE'
+              ? payload.old?.key_version
+              : payload.new?.key_version;
+          if (typeof v === 'number') {
+            clearConvKeyCache(conversationId, v);
+          } else {
+            clearConvKeyCache(conversationId);
+          }
+          if (
+            payload.eventType === 'INSERT' &&
+            payload.new?.recipient_device_id === deviceId
+          ) {
             void refresh();
           }
         },

packages/shared/src/chat/convKeys.ts

@@ -28,7 +28,28 @@ export interface ConvKeyHandle {
 const cache = new Map<string, ConvKeyHandle>();
 const cacheKey = (convId: string, v: number) => convId + '@' + v;
 
-export function clearConvKeyCache(): void { cache.clear(); }
+// Clear the in-memory conv-key cache. Three modes:
+//   * no args     → clear everything (e.g. on logout)
+//   * convId only → clear all key-version entries for this conversation
+//   * convId + v  → clear just the specific (conv, version) entry
+//
+// Callers that observe a peer rotation or a server-side conv-keys mutation
+// MUST invalidate the affected entries so subsequent `getOrCreateConvKey` /
+// `tryGetConvKey` calls re-fetch the canonical bundle from the server
+// instead of returning a now-stale cached key.
+export function clearConvKeyCache(conversationId?: string, keyVersion?: number): void {
+  if (conversationId === undefined) {
+    cache.clear();
+    return;
+  }
+  if (keyVersion !== undefined) {
+    cache.delete(cacheKey(conversationId, keyVersion));
+    return;
+  }
+  for (const key of Array.from(cache.keys())) {
+    if (key.startsWith(conversationId + '@')) cache.delete(key);
+  }
+}
 
 async function listMemberPublicKeys(
   client: AppSupabaseClient,