Fix race condition where streamed events could be lost

Events stream in via model_write listener while also being fetched
from the database. If the DB fetch completed before all events were
persisted, replaceModelsInStore would wipe out events that came in
via model_write.

Added mergeModelsInStore that adds fetched events without removing
existing ones. Applied to HTTP, gRPC, and WebSocket event hooks.

Author: gschier

crates/yaak-http/src/transaction.rs

@@ -342,7 +342,8 @@ mod tests {
 
     #[tokio::test]
     async fn test_transaction_single_redirect() {
-        let redirect_headers = vec![("Location".to_string(), "https://example.com/new".to_string())];
+        let redirect_headers =
+            vec![("Location".to_string(), "https://example.com/new".to_string())];
 
         let responses = vec![
             MockResponse { status: 302, headers: redirect_headers, body: vec![] },
@@ -373,7 +374,8 @@ mod tests {
 
     #[tokio::test]
     async fn test_transaction_max_redirects_exceeded() {
-        let redirect_headers = vec![("Location".to_string(), "https://example.com/loop".to_string())];
+        let redirect_headers =
+            vec![("Location".to_string(), "https://example.com/loop".to_string())];
 
         // Create more redirects than allowed
         let responses: Vec<MockResponse> = (0..12)
@@ -525,7 +527,8 @@ mod tests {
                 _request: SendableHttpRequest,
                 _event_tx: mpsc::Sender<HttpResponseEvent>,
             ) -> Result<HttpResponse> {
-                let headers = vec![("set-cookie".to_string(), "session=xyz789; Path=/".to_string())];
+                let headers =
+                    vec![("set-cookie".to_string(), "session=xyz789; Path=/".to_string())];
 
                 let body_stream: Pin<Box<dyn AsyncRead + Send>> =
                     Box::pin(std::io::Cursor::new(vec![]));
@@ -584,7 +587,10 @@ mod tests {
                 let headers = vec![
                     ("set-cookie".to_string(), "session=abc123; Path=/".to_string()),
                     ("set-cookie".to_string(), "user_id=42; Path=/".to_string()),
-                    ("set-cookie".to_string(), "preferences=dark; Path=/; Max-Age=86400".to_string()),
+                    (
+                        "set-cookie".to_string(),
+                        "preferences=dark; Path=/; Max-Age=86400".to_string(),
+                    ),
                 ];
 
                 let body_stream: Pin<Box<dyn AsyncRead + Send>> =

crates/yaak-models/guest-js/store.ts

@@ -206,6 +206,22 @@ export function replaceModelsInStore<
   });
 }
 
+export function mergeModelsInStore<
+  M extends AnyModel['model'],
+  T extends Extract<AnyModel, { model: M }>,
+>(model: M, models: T[]) {
+  mustStore().set(modelStoreDataAtom, (prev: ModelStoreData) => {
+    const existingModels = { ...prev[model] } as Record<string, T>;
+    for (const m of models) {
+      existingModels[m.id] = m;
+    }
+    return {
+      ...prev,
+      [model]: existingModels,
+    };
+  });
+}
+
 function shouldIgnoreModel({ model, updateSource }: ModelPayload) {
   // Never ignore updates from non-user sources
   if (updateSource.type !== 'window') {

src-web/hooks/useHttpResponseEvents.ts

@@ -1,6 +1,10 @@
 import { invoke } from '@tauri-apps/api/core';
 import type { HttpResponse, HttpResponseEvent } from '@yaakapp-internal/models';
-import { httpResponseEventsAtom, replaceModelsInStore } from '@yaakapp-internal/models';
+import {
+  httpResponseEventsAtom,
+  mergeModelsInStore,
+  replaceModelsInStore,
+} from '@yaakapp-internal/models';
 import { useAtomValue } from 'jotai';
 import { useEffect, useMemo } from 'react';
 
@@ -13,8 +17,10 @@ export function useHttpResponseEvents(response: HttpResponse | null) {
       return;
     }
 
+    // Use merge instead of replace to preserve events that came in via model_write
+    // while we were fetching from the database
     invoke<HttpResponseEvent[]>('cmd_get_http_response_events', { responseId: response.id }).then(
-      (events) => replaceModelsInStore('http_response_event', events),
+      (events) => mergeModelsInStore('http_response_event', events),
     );
   }, [response?.id]);
 

src-web/hooks/usePinnedGrpcConnection.ts

@@ -3,6 +3,7 @@ import type { GrpcConnection, GrpcEvent } from '@yaakapp-internal/models';
 import {
   grpcConnectionsAtom,
   grpcEventsAtom,
+  mergeModelsInStore,
   replaceModelsInStore,
 } from '@yaakapp-internal/models';
 import { atom, useAtomValue } from 'jotai';
@@ -67,8 +68,10 @@ export function useGrpcEvents(connectionId: string | null) {
       return;
     }
 
+    // Use merge instead of replace to preserve events that came in via model_write
+    // while we were fetching from the database
     invoke<GrpcEvent[]>('models_grpc_events', { connectionId }).then((events) => {
-      replaceModelsInStore('grpc_event', events);
+      mergeModelsInStore('grpc_event', events);
     });
   }, [connectionId]);
 

src-web/hooks/usePinnedWebsocketConnection.ts

@@ -1,6 +1,7 @@
 import { invoke } from '@tauri-apps/api/core';
 import type { WebsocketConnection, WebsocketEvent } from '@yaakapp-internal/models';
 import {
+  mergeModelsInStore,
   replaceModelsInStore,
   websocketConnectionsAtom,
   websocketEventsAtom,
@@ -54,8 +55,10 @@ export function useWebsocketEvents(connectionId: string | null) {
       return;
     }
 
+    // Use merge instead of replace to preserve events that came in via model_write
+    // while we were fetching from the database
     invoke<WebsocketEvent[]>('models_websocket_events', { connectionId }).then(
-      (events) => replaceModelsInStore('websocket_event', events),
+      (events) => mergeModelsInStore('websocket_event', events),
     );
   }, [connectionId]);
 

src-web/lib/model_util.test.ts

@@ -11,7 +11,9 @@ function makeEvent(
     id: 'test',
     model: 'http_response_event',
     responseId: 'resp',
-    createdAt: Date.now(),
+    workspaceId: 'ws',
+    createdAt: new Date().toISOString(),
+    updatedAt: new Date().toISOString(),
     event: { type, name, value } as HttpResponseEvent['event'],
   };
 }