import { fetchPeerPublicKeys } from '@chat-app/shared/auth'; import { type AttachmentHandle, type DecryptedMessage, decryptMessages, encryptAndUploadAttachment, fetchConversationMessages, insertAttachmentRow, MAX_ATTACHMENT_BYTES, type MessageWithCipher, sendEncryptedMessage, shareConvKeyToUser, tryGetConvKey, } from '@chat-app/shared/chat'; import { bytesToPgHex, pgBytesToBytes } from '@chat-app/shared/supabase'; import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { decryptBatch as decryptBatchWorker } from './decryptWorker'; import { loadCachedMessages, persistMessages, pruneCache, deleteCachedMessage, } from './messageCache'; import { enqueueOutbox, getOutbox, isDue, markAttempt, type OutboxItem, removeOutbox, shouldGiveUp, subscribeOutbox, } from './messageOutbox'; import { supabase } from './supabase'; import { cachedUserKey } from './userIdentity'; interface State { messages: DecryptedMessage[]; loading: boolean; error: string | null; } interface Args { conversationId: string | undefined; userId: string | undefined; deviceId: string | undefined; } type MessageChangePayload = { eventType: 'INSERT' | 'UPDATE' | 'DELETE'; new: Record; old: Record; }; function rowToMessage(row: Record): MessageWithCipher { return { id: String(row.id), conversationId: String(row.conversation_id), senderId: String(row.sender_id), senderDeviceId: row.sender_device_id ? String(row.sender_device_id) : null, replyToId: row.reply_to_id ? String(row.reply_to_id) : null, editedAt: row.edited_at ? String(row.edited_at) : null, deletedAt: row.deleted_at ? String(row.deleted_at) : null, createdAt: String(row.created_at), ciphertext: pgBytesToBytes(String(row.ciphertext ?? '\\x')), nonce: pgBytesToBytes(String(row.nonce ?? '\\x')), keyVersion: typeof row.key_version === 'number' ? row.key_version : 1, }; } export function useConversationMessages({ conversationId, userId, deviceId }: Args): State & { send: (text: string, images?: File[], replyToId?: string | null) => Promise; refresh: () => Promise; pending: OutboxItem[]; retryPending: (id: string) => void; cancelPending: (id: string) => void; } { const [state, setState] = useState({ messages: [], loading: true, error: null }); const [pending, setPending] = useState(() => conversationId ? getOutbox(conversationId) : [], ); const privateKeyRef = useRef(null); useEffect(() => { if (!conversationId) { setPending([]); return; } return subscribeOutbox((byConv) => { setPending(byConv[conversationId] ?? []); }); }, [conversationId]); // Load own user private key once per user. useEffect(() => { privateKeyRef.current = null; if (!userId) return; void cachedUserKey(userId).then((pk) => { privateKeyRef.current = pk; }); }, [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. useEffect(() => { if (!conversationId || !userId) return; let cancelled = false; const run = async (): Promise => { // Wait for the private key ref to be populated. Loop with backoff // because it's set asynchronously by another effect; bail if the // conversation switches. for (let i = 0; i < 20 && !privateKeyRef.current && !cancelled; i++) { await new Promise((r) => setTimeout(r, 100)); } const priv = privateKeyRef.current; if (!priv || cancelled) return; try { const { data: convRow, error: convErr } = await supabase .from('conversations') .select('active_key_version') .eq('id', conversationId) .single(); if (convErr || !convRow) return; // 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; const handle = await tryGetConvKey(supabase, conversationId, userId, priv, version); if (!handle || cancelled) return; 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 }>) .filter((m) => m.accepted && m.user_id !== userId) .map((m) => m.user_id); if (memberIds.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 }>; }; }; }; }; } ) .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); } } } } catch (err) { console.warn('proactive rewrap sweep failed', err); } }; void run(); return () => { cancelled = true; }; }, [conversationId, userId]); const decryptBatch = useCallback( async (messages: MessageWithCipher[]): Promise => { const priv = privateKeyRef.current; if (!priv || !userId || messages.length === 0) { return messages.map((m) => ({ ...m, plaintext: null })); } return decryptMessages({ client: supabase, messages, ownUserId: userId, ownPrivateKey: priv, // Offload the symmetric decrypt + utf-8 decode to a Web Worker so // the main thread stays responsive during bulk operations (initial // fetch, backfill after sleep). aeadBatchDelegate: decryptBatchWorker, }); }, [userId], ); const refresh = useCallback(async () => { if (!conversationId) return; try { // Only show the loading spinner on the *initial* fetch — subsequent // refreshes (focus/visibility) replace messages in-place to avoid // flickering an empty state on every wake. setState((prev) => prev.messages.length === 0 ? { ...prev, loading: true } : prev, ); const rows = await fetchConversationMessages(supabase, conversationId); const decrypted = await decryptBatch(rows); setState({ messages: decrypted, loading: false, error: null }); // Persist the fresh batch to the local cache so next conversation // switch / app start can hydrate instantly. Fire-and-forget — cache // write failure is never user-visible. void persistMessages(conversationId, decrypted); } catch (err: unknown) { setState((prev) => ({ ...prev, loading: false, error: err instanceof Error ? err.message : 'failed to load messages', })); } }, [conversationId, decryptBatch]); // Hydrate from the local SQLite cache the moment the conversation id // changes. Runs in parallel with the network fetch — whichever resolves // first populates the UI, and `refresh` will replace stale cache data // when the server response lands. On cache-miss this is a ~5ms no-op. useEffect(() => { if (!conversationId) return; let cancelled = false; void loadCachedMessages(conversationId).then((cached) => { if (cancelled || cached.length === 0) return; setState((prev) => { // Don't clobber a fresh server response that already landed. if (prev.messages.length > 0) return prev; return { messages: cached, loading: false, error: null }; }); }); return () => { cancelled = true; }; }, [conversationId]); // Prune cache once per app session. useEffect(() => { void pruneCache(); }, []); // Realtime INSERT handler — refetches the row via REST so we get the // canonical bytea encoding (postgres_changes payloads serialize bytea // differently and decoding them inline is brittle). Then decrypt + append. // Skips if the message is already in state (e.g. optimistic insert from our // own send), so the sender's cached copy isn't overwritten with a flicker. const handleInsert = useCallback( async (row: Record) => { if (!conversationId || !deviceId) return; const id = String(row.id); let alreadyHave = false; setState((prev) => { if (prev.messages.some((m) => m.id === id)) alreadyHave = true; return prev; }); if (alreadyHave) return; let decrypted: DecryptedMessage | null = null; for (let attempt = 0; attempt < 6; attempt++) { const { data, error } = await supabase .from('messages') .select( 'id, conversation_id, sender_id, sender_device_id, reply_to_id, edited_at, deleted_at, created_at, ciphertext, nonce, key_version', ) .eq('id', id) .maybeSingle(); if (error) { console.warn('handleInsert refetch failed', error); return; } if (!data) { await new Promise((r) => window.setTimeout(r, 120 * (attempt + 1))); continue; } // db-types snapshot predates the sender-key columns; cast to bypass. const r = data as unknown as { id: string; conversation_id: string; sender_id: string; sender_device_id: string | null; reply_to_id: string | null; edited_at: string | null; deleted_at: string | null; created_at: string; ciphertext: string; nonce: string; key_version: number; }; const msg: MessageWithCipher = { id: r.id, conversationId: r.conversation_id, senderId: r.sender_id, senderDeviceId: r.sender_device_id, replyToId: r.reply_to_id, editedAt: r.edited_at, deletedAt: r.deleted_at, createdAt: r.created_at, ciphertext: pgBytesToBytes(String(r.ciphertext)), nonce: pgBytesToBytes(String(r.nonce)), keyVersion: r.key_version, }; const [d] = await decryptBatch([msg]); if (d) { decrypted = d; if (d.plaintext !== null) break; } await new Promise((r) => window.setTimeout(r, 200 * (attempt + 1))); } if (!decrypted) return; setState((prev) => { if (prev.messages.some((m) => m.id === decrypted!.id)) return prev; return { ...prev, messages: [...prev.messages, decrypted!] }; }); }, [conversationId, deviceId, decryptBatch], ); const handleUpdate = useCallback( async (row: Record) => { const partial = rowToMessage(row); setState((prev) => { const idx = prev.messages.findIndex((m) => m.id === partial.id); if (idx === -1) return prev; const existing = prev.messages[idx]; if (!existing) return prev; const next = [...prev.messages]; next[idx] = { ...existing, editedAt: partial.editedAt, deletedAt: partial.deletedAt, }; return { ...prev, messages: next }; }); if (partial.editedAt && !partial.deletedAt) { // Realtime bytea encoding varies (base64 vs hex, even null for // unchanged columns on some configs). Refetch via REST to get the // canonical `\x…` hex then decrypt — same pattern as handleInsert. if (!conversationId || !deviceId) return; let decrypted: DecryptedMessage | null = null; for (let attempt = 0; attempt < 6; attempt++) { const { data, error } = await supabase .from('messages') .select( 'id, conversation_id, sender_id, sender_device_id, reply_to_id, edited_at, deleted_at, created_at, ciphertext, nonce, key_version', ) .eq('id', partial.id) .maybeSingle(); if (error) { console.warn('handleUpdate refetch failed', error); return; } if (!data) { await new Promise((r) => window.setTimeout(r, 120 * (attempt + 1))); continue; } const r = data as unknown as { id: string; conversation_id: string; sender_id: string; sender_device_id: string | null; reply_to_id: string | null; edited_at: string | null; deleted_at: string | null; created_at: string; ciphertext: string; nonce: string; key_version: number; }; const msg: MessageWithCipher = { id: r.id, conversationId: r.conversation_id, senderId: r.sender_id, senderDeviceId: r.sender_device_id, replyToId: r.reply_to_id, editedAt: r.edited_at, deletedAt: r.deleted_at, createdAt: r.created_at, ciphertext: pgBytesToBytes(String(r.ciphertext)), nonce: pgBytesToBytes(String(r.nonce)), keyVersion: r.key_version, }; const [d] = await decryptBatch([msg]); if (d) { decrypted = d; if (d.plaintext !== null) break; } await new Promise((r) => window.setTimeout(r, 200 * (attempt + 1))); } if (!decrypted) return; setState((prev) => { const idx = prev.messages.findIndex((m) => m.id === decrypted!.id); if (idx === -1) return prev; const next = [...prev.messages]; next[idx] = decrypted!; return { ...prev, messages: next }; }); } }, [conversationId, deviceId, decryptBatch], ); const handleDelete = useCallback((row: Record) => { const id = String(row.id); setState((prev) => ({ ...prev, messages: prev.messages.filter((m) => m.id !== id), })); void deleteCachedMessage(id); }, []); useEffect(() => { if (!conversationId || !userId || !deviceId) return; 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; const BATCH_BURST_THRESHOLD = 3; let burstBuffer: Array> = []; let burstTimer: number | null = null; const flushBurst = () => { const buf = burstBuffer; burstBuffer = []; if (burstTimer !== null) { window.clearTimeout(burstTimer); burstTimer = null; } if (buf.length === 0) return; if (buf.length > BATCH_BURST_THRESHOLD) { void refresh(); } else { for (const row of buf) void handleInsert(row); } }; const queueInsert = (row: Record) => { burstBuffer.push(row); if (burstTimer === null) { burstTimer = window.setTimeout(flushBurst, BATCH_WINDOW_MS); } }; const channel = supabase .channel('conv:' + conversationId) .on( 'postgres_changes', { event: '*', schema: 'public', table: 'messages', filter: 'conversation_id=eq.' + conversationId, }, (payload: MessageChangePayload) => { if (payload.eventType === 'INSERT') { queueInsert(payload.new); } else if (payload.eventType === 'UPDATE') { void handleUpdate(payload.new); } else if (payload.eventType === 'DELETE') { handleDelete(payload.old); } }, ) // When a peer device wraps the conversation-key for us (e.g. we just // registered a fresh device), re-decrypt the visible messages. .on( 'postgres_changes', { event: 'INSERT', schema: 'public', table: 'conversation_keys', filter: 'conversation_id=eq.' + conversationId, }, (payload: { new: { recipient_device_id?: string } }) => { if (payload.new?.recipient_device_id === deviceId) { void refresh(); } }, ) .subscribe(); // Refresh + reconnect on wake from background throttle (mostly Windows // WebView2). Without this, messages inserted while the window is // minimised never arrive until the user explicitly reloads. Throttled to // at most once per AWAKE_THROTTLE_MS so the inevitable cluster of // visibility/online events on focus does not cause a flicker storm. let lastAwakeRefresh = 0; const AWAKE_THROTTLE_MS = 30_000; const onAwake = () => { if (document.visibilityState !== 'visible') return; const now = Date.now(); if (now - lastAwakeRefresh < AWAKE_THROTTLE_MS) return; lastAwakeRefresh = now; void refresh(); try { channel.subscribe(); } catch { /* already live */ } }; document.addEventListener('visibilitychange', onAwake); window.addEventListener('online', onAwake); return () => { if (burstTimer !== null) window.clearTimeout(burstTimer); document.removeEventListener('visibilitychange', onAwake); window.removeEventListener('online', onAwake); void supabase.removeChannel(channel); }; }, [conversationId, userId, deviceId, refresh, handleInsert, handleUpdate, handleDelete]); const sendText = useCallback( async (convId: string, uid: string, did: string, priv: Uint8Array, text: string, replyToId: string | null): Promise => { const msg = await sendEncryptedMessage({ client: supabase, conversationId: convId, plaintext: text, senderUserId: uid, senderDeviceId: did, senderPrivateKey: priv, ...(replyToId ? { replyToId } : {}), }); setState((prev) => { if (prev.messages.some((m) => m.id === msg.id)) return prev; return { ...prev, messages: [ ...prev.messages, { ...msg, plaintext: text } as DecryptedMessage, ], }; }); }, [], ); const send = useCallback( async (text: string, images: File[] = [], replyToId: string | null = null) => { const trimmed = text.trim(); if ((!trimmed && images.length === 0) || !conversationId || !userId || !deviceId) return; const priv = privateKeyRef.current; if (!priv) throw new Error('private key not loaded'); // Slash-command: /tempmsg sends an ephemeral message // that the sender auto-deletes after the window elapses. Both peers // see the countdown via the expireMs field embedded in the plaintext // payload — no server support required. const tempMatch = /^\/tempmsg\s+(\d+)\s+([\s\S]+)$/i.exec(trimmed); if (tempMatch && images.length === 0) { const seconds = Math.min(3600, Math.max(5, parseInt(tempMatch[1]!, 10))); const body = tempMatch[2]!.trim(); const payload = JSON.stringify({ v: 1, type: 'text', text: body, attachments: [], expireMs: seconds * 1000, }); try { await sendText(conversationId, userId, deviceId, priv, payload, replyToId); } catch (err: unknown) { const msg = err instanceof Error ? err.message : 'send failed'; enqueueOutbox({ conversationId, text: payload, replyToId, error: msg, }); } return; } // Text-only path is retryable — if the network is down or the server // rejects transiently, stash in the outbox and keep the UI optimistic. // Attachments can't be deferred (large payloads, uploaded separately), // so those still surface the error immediately. if (images.length === 0) { try { await sendText(conversationId, userId, deviceId, priv, trimmed, replyToId); } catch (err: unknown) { const msg = err instanceof Error ? err.message : 'send failed'; enqueueOutbox({ conversationId, text: trimmed, replyToId, error: msg, }); } return; } // 1. Upload + encrypt each image. Collect handles + raw blob nonces // (so the public attachment row can reference the blob-level nonce). const handles: AttachmentHandle[] = []; const blobNonceHexByHandleId = new Map(); for (const file of images) { if (file.size > MAX_ATTACHMENT_BYTES) { throw new Error('attachment exceeds max size (10 MB)'); } const dims = await readImageDimensions(file); const res = await encryptAndUploadAttachment({ client: supabase, conversationId, file, mimeType: file.type || 'application/octet-stream', sizeBytes: file.size, ...(dims.width !== undefined ? { width: dims.width } : {}), ...(dims.height !== undefined ? { height: dims.height } : {}), }); handles.push(res.handle); blobNonceHexByHandleId.set(res.handle.id, bytesToPgHex(res.nonce)); } // 2. Send message (inserts messages + per-conversation key bundles). const msg = await sendEncryptedMessage({ client: supabase, conversationId, plaintext: trimmed, senderUserId: userId, senderDeviceId: deviceId, senderPrivateKey: priv, ...(handles.length > 0 ? { attachmentHandles: handles } : {}), ...(replyToId ? { replyToId } : {}), }); // 3. Optimistic insert — we already have the plaintext in hand and the // server returned the row id, so add the message to local state // immediately. Realtime will then no-op (handleInsert dedupes by id). const attachmentsPayload = handles.length === 0 ? trimmed : JSON.stringify({ v: 1, text: trimmed, attachments: handles }); setState((prev) => { if (prev.messages.some((m) => m.id === msg.id)) return prev; return { ...prev, messages: [ ...prev.messages, { ...msg, plaintext: attachmentsPayload, } as DecryptedMessage, ], }; }); // 4. Insert public attachment metadata rows pointing at the new message. for (const h of handles) { const blobNonce = blobNonceHexByHandleId.get(h.id) ?? '\\x'; await insertAttachmentRow(supabase, msg.id, h, blobNonce); } }, [conversationId, userId, deviceId, sendText], ); // Drain outbox: retry due items, remove on success, record attempt on fail. // Runs on online event, window focus, and a 10s interval. useEffect(() => { if (!conversationId || !userId || !deviceId) return; let draining = false; const drain = async (): Promise => { if (draining) return; const priv = privateKeyRef.current; if (!priv) return; if (typeof navigator !== 'undefined' && navigator.onLine === false) return; draining = true; try { const items = getOutbox(conversationId).filter(isDue); for (const item of items) { if (shouldGiveUp(item)) continue; try { await sendText( conversationId, userId, deviceId, priv, item.text, item.replyToId, ); removeOutbox(conversationId, item.id); } catch (err: unknown) { markAttempt( conversationId, item.id, err instanceof Error ? err.message : 'send failed', ); } } } finally { draining = false; } }; const interval = window.setInterval(() => { void drain(); }, 10_000); const onOnline = () => void drain(); window.addEventListener('online', onOnline); window.addEventListener('focus', onOnline); // Kick once immediately on mount for stale queued items. void drain(); return () => { window.clearInterval(interval); window.removeEventListener('online', onOnline); window.removeEventListener('focus', onOnline); }; }, [conversationId, userId, deviceId, sendText]); const retryPending = useCallback( (id: string) => { if (!conversationId) return; // Force due now; the drain loop will pick it up on next tick. markAttempt(conversationId, id, null); // Manual kick: mutate nextAttemptAt by re-enqueuing? Simpler — just // trigger a drain-ish by queueing a microtask. The interval picks up // due items within 10s, but for UX we also eagerly try here. const priv = privateKeyRef.current; if (!priv || !userId || !deviceId) return; const item = getOutbox(conversationId).find((x) => x.id === id); if (!item) return; void (async () => { try { await sendText(conversationId, userId, deviceId, priv, item.text, item.replyToId); removeOutbox(conversationId, id); } catch (err: unknown) { markAttempt( conversationId, id, err instanceof Error ? err.message : 'send failed', ); } })(); }, [conversationId, userId, deviceId, sendText], ); const cancelPending = useCallback( (id: string) => { if (!conversationId) return; removeOutbox(conversationId, id); }, [conversationId], ); return useMemo( () => ({ ...state, send, refresh, pending, retryPending, cancelPending }), [state, send, refresh, pending, retryPending, cancelPending], ); } // Best-effort image dimension probe. Falls back silently on non-images. async function readImageDimensions(file: File): Promise<{ width?: number; height?: number }> { if (!file.type.startsWith('image/')) return {}; const url = URL.createObjectURL(file); try { return await new Promise<{ width?: number; height?: number }>((resolve) => { const img = new Image(); img.onload = () => resolve({ width: img.naturalWidth, height: img.naturalHeight }); img.onerror = () => resolve({}); img.src = url; }); } finally { URL.revokeObjectURL(url); } }