diff --git a/apps/desktop/src/lib/useConversationMessages.ts b/apps/desktop/src/lib/useConversationMessages.ts index c80779d..ea6d6b9 100644 --- a/apps/desktop/src/lib/useConversationMessages.ts +++ b/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 - // member and ensure the active conv-key has a `recipient_user_id` bundle - // 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 - // local conv-key handle. Closes the legacy migration gap so peer B can - // read on first unlock without manual intervention from A. + // Proactive rewrap sweep: when a conversation opens, ensure the active + // conv-key has a `recipient_user_id` bundle for every accepted member. + // + // If any peer is missing a bundle at the active version, the previous + // implementation called `shareConvKeyToUser` for each missing peer — + // 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 - // can't unwrap our bundle at the active version (we lost the device - // key, or the bundle was wiped by the 0.18.0 reset_user_key bug, or - // 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. + // First, make sure we have a usable handle for the active version + // (this auto-rotates if we're locked out of our own bundle — the + // recovery path added in v0.21.1/v0.21.2). const handle = await getOrCreateConvKey(supabase, conversationId, { userId, privateKey: priv, }); if (cancelled) return; - // If the helper rotated, all current members with a `user_keys` - // public key were already wrapped by `rotateConvKey`. No further - // sweep work is needed. - if (handle.keyVersion > version) return; + if (handle.keyVersion > version) return; // already rotated by helper + // 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); - for (const peer of peers) { - if (cancelled) return; - const { count, error: cntErr } = await ( - supabase as unknown as { - from: (t: string) => { - select: (s: string, o?: object) => { - eq: (...a: unknown[]) => { - eq: (...a: unknown[]) => { - eq: ( - ...a: unknown[] - ) => Promise<{ count: number | null; error: unknown }>; - }; + // Count how many of the peers have a recipient_user_id bundle at + // the active version. If any are missing, rotate to V+1 — the + // rotation will wrap a fresh key for every accepted member with a + // user_keys row. + const { data: existingRows, error: rowsErr } = await ( + supabase as unknown as { + from: (t: string) => { + select: (s: string) => { + eq: (c: string, v: string) => { + eq: (c: string, v: number) => { + in: (c: string, v: string[]) => Promise<{ + 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 - // within BATCH_WINDOW_MS, collapse to a single refresh() which pulls - // the last 100 in one query — cheaper and keeps order stable. For - // lone inserts the per-id path stays so latency is unchanged. - const BATCH_WINDOW_MS = 250; + // refetches + decrypts. The first event in a quiet period fires + // `handleInsert` immediately so single incoming messages don't sit + // behind a debounce timer (previous behaviour: 250 ms blank between + // notification-sound and message body). Subsequent events arriving + // 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> = []; let burstTimer: number | null = null; @@ -513,6 +533,15 @@ export function useConversationMessages({ conversationId, userId, deviceId }: Ar } }; const queueInsert = (row: Record) => { + 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 - // registered a fresh device), re-decrypt the visible messages. + // Any conversation_keys change for this conv invalidates the cached + // 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 } }) => { - if (payload.new?.recipient_device_id === deviceId) { + (payload: { + 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(); } }, diff --git a/packages/shared/src/chat/convKeys.ts b/packages/shared/src/chat/convKeys.ts index 8bb7fc4..12b569b 100644 --- a/packages/shared/src/chat/convKeys.ts +++ b/packages/shared/src/chat/convKeys.ts @@ -28,7 +28,28 @@ export interface ConvKeyHandle { const cache = new Map(); 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,