From d9b08592da41447242772016f27ace3ec95e5208 Mon Sep 17 00:00:00 2001 From: Dennis Landmann Date: Sun, 19 Apr 2026 20:35:00 +0200 Subject: [PATCH] fix(messages): refetch row via REST on realtime insert + optimistic sender update --- apps/desktop/src-tauri/tauri.conf.json | 2 +- .../src/lib/useConversationMessages.ts | 96 +++++++++++++++++-- 2 files changed, 87 insertions(+), 11 deletions(-) diff --git a/apps/desktop/src-tauri/tauri.conf.json b/apps/desktop/src-tauri/tauri.conf.json index 8085c38..ef4d402 100644 --- a/apps/desktop/src-tauri/tauri.conf.json +++ b/apps/desktop/src-tauri/tauri.conf.json @@ -1,7 +1,7 @@ { "$schema": "https://schema.tauri.app/config/2", "productName": "ChatApp", - "version": "0.3.5", + "version": "0.3.6", "identifier": "com.meinname.chatapp", "build": { "beforeDevCommand": "pnpm vite:dev", diff --git a/apps/desktop/src/lib/useConversationMessages.ts b/apps/desktop/src/lib/useConversationMessages.ts index 601c575..1179d54 100644 --- a/apps/desktop/src/lib/useConversationMessages.ts +++ b/apps/desktop/src/lib/useConversationMessages.ts @@ -98,26 +98,81 @@ export function useConversationMessages({ conversationId, userId, deviceId }: Ar } }, [conversationId, decryptBatch]); - // Realtime INSERT handler — decrypt + append (with retry for envelope race). + // 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 (!deviceId) return; - const msg = rowToMessage(row); - let decrypted: DecryptedMessage = { ...msg, plaintext: null }; + 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, 120 * (attempt + 1))); + 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] }; + if (prev.messages.some((m) => m.id === decrypted!.id)) return prev; + return { ...prev, messages: [...prev.messages, decrypted!] }; }); }, - [deviceId, decryptBatch], + [conversationId, deviceId, decryptBatch], ); const handleUpdate = useCallback( @@ -219,7 +274,7 @@ export function useConversationMessages({ conversationId, userId, deviceId }: Ar blobNonceHexByHandleId.set(res.handle.id, bytesToPgHex(res.nonce)); } - // 2. Send message (inserts messages + envelopes in one helper). + // 2. Send message (inserts messages + per-conversation key bundles). const msg = await sendEncryptedMessage({ client: supabase, conversationId, @@ -230,7 +285,28 @@ export function useConversationMessages({ conversationId, userId, deviceId }: Ar ...(handles.length > 0 ? { attachmentHandles: handles } : {}), }); - // 3. Insert public attachment metadata rows pointing at the new message. + // 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);