fix(query-db-collection): prevent Direct Write items from being purged

Fixes issue where items inserted via Direct Writes (writeInsert/writeUpsert)
were incorrectly purged when query results changed and didn't include those items.

**Root Cause:**
The reference counting system (queryToRows/rowToQueries) tracks which items
are in each query result. When a query result doesn't include an item and the
item's reference count is 0, it gets deleted. Direct Write items were never
added to this tracking system, so they always had a reference count of 0 and
were purged when not in query results.

**Solution:**
- Added `directWriteKeys` Set to track items inserted via Direct Writes
- Modified purging logic in `handleQueryResult()` and `cleanupQuery()` to skip
  items in the `directWriteKeys` Set
- Updated `performWriteOperations()` to add/remove keys from `directWriteKeys`:
  - Insert operations add the key
  - Delete operations remove the key
  - Upsert operations add the key if inserting new item

**Changes:**
- query.ts: Added directWriteKeys Set and modified purging checks
- manual-sync.ts: Track direct write keys in performWriteOperations
- query.test.ts: Added test case to verify Direct Writes are protected

This ensures directly-written items persist until explicitly deleted via
writeDelete(), regardless of query result changes.

Author: claude

packages/query-db-collection/src/manual-sync.ts

@@ -38,6 +38,7 @@ export interface SyncContext<
   begin: () => void
   write: (message: Omit<ChangeMessage<TRow>, `key`>) => void
   commit: () => void
+  directWriteKeys?: Set<TKey>
 }
 
 interface NormalizedOperation<
@@ -148,6 +149,8 @@ export function performWriteOperations<
           type: `insert`,
           value: resolved,
         })
+        // Track this key as a direct write
+        ctx.directWriteKeys?.add(op.key)
         break
       }
       case `update`: {
@@ -175,6 +178,8 @@ export function performWriteOperations<
           type: `delete`,
           value: currentItem,
         })
+        // Remove from direct write tracking when explicitly deleted
+        ctx.directWriteKeys?.delete(op.key)
         break
       }
       case `upsert`: {
@@ -195,6 +200,8 @@ export function performWriteOperations<
             type: `insert`,
             value: resolved,
           })
+          // Track this key as a direct write (only for new inserts)
+          ctx.directWriteKeys?.add(op.key)
         }
         break
       }

packages/query-db-collection/src/query.ts

@@ -592,6 +592,10 @@ export function queryCollectionOptions(
   // queryKey → QueryObserver's unsubscribe function
   const unsubscribes = new Map<string, () => void>()
 
+  // Track row keys that were inserted via direct writes to prevent them from being purged
+  // when they don't appear in query results
+  const directWriteKeys = new Set<string | number>()
+
   // Helper function to add a row to the internal state
   const addRow = (rowKey: string | number, hashedQueryKey: string) => {
     const rowToQueriesSet = rowToQueries.get(rowKey) || new Set()
@@ -780,7 +784,9 @@ export function queryCollectionOptions(
             const newItem = newItemsMap.get(key)
             if (!newItem) {
               const needToRemove = removeRow(key, hashedQueryKey) // returns true if the row is no longer referenced by any queries
-              if (needToRemove) {
+              // Don't remove items that were inserted via direct writes
+              // They should only be removed via explicit writeDelete calls
+              if (needToRemove && !directWriteKeys.has(key)) {
                 write({ type: `delete`, value: oldItem })
               }
             } else if (
@@ -932,7 +938,8 @@ export function queryCollectionOptions(
             // Reference count dropped to 0, we can GC the row
             rowToQueries.delete(rowKey)
 
-            if (collection.has(rowKey)) {
+            // Don't remove items that were inserted via direct writes
+            if (collection.has(rowKey) && !directWriteKeys.has(rowKey)) {
               begin()
               write({ type: `delete`, value: collection.get(rowKey) })
               commit()
@@ -957,6 +964,7 @@ export function queryCollectionOptions(
       hashToQueryKey.clear()
       queryToRows.clear()
       rowToQueries.clear()
+      directWriteKeys.clear()
       state.observers.clear()
       unsubscribeQueryCache()
 
@@ -1024,6 +1032,7 @@ export function queryCollectionOptions(
     begin: () => void
     write: (message: Omit<ChangeMessage<any>, `key`>) => void
     commit: () => void
+    directWriteKeys?: Set<string | number>
   } | null = null
 
   // Enhanced internalSync that captures write functions for manual use
@@ -1039,6 +1048,7 @@ export function queryCollectionOptions(
       begin,
       write,
       commit,
+      directWriteKeys,
     }
 
     // Call the original internalSync logic

packages/query-db-collection/tests/query.test.ts

@@ -3538,4 +3538,90 @@ describe(`QueryCollection`, () => {
       expect(collection.size).toBe(0)
     })
   })
+
+  describe(`Direct Writes with dynamic query keys`, () => {
+    it(`should not purge Direct Write items when query results change`, async () => {
+      // Reproduces the issue reported on Discord:
+      // When using Direct Writes, items inserted via writeInsert get purged
+      // when a query result comes in that doesn't include those items
+
+      interface MachineEvent {
+        id: string
+        timestamp: string
+        message: string
+      }
+
+      const getEventKey = (item: MachineEvent) => item.id
+
+      // Simulate initial data - first response returns 2 items
+      let queryCallCount = 0
+      const initialEvents: Array<MachineEvent> = [
+        { id: `1`, timestamp: `2025-11-12T22:08:00Z`, message: `Event 1` },
+        { id: `2`, timestamp: `2025-11-12T22:08:30Z`, message: `Event 2` },
+      ]
+
+      const queryFn = vi.fn(async (): Promise<Array<MachineEvent>> => {
+        queryCallCount++
+        // First call returns 2 items
+        // Second call returns only 1 item (simulating filtered data)
+        if (queryCallCount === 1) {
+          return initialEvents
+        } else {
+          return [initialEvents[0]!] // Only return first item
+        }
+      })
+
+      const config: QueryCollectionConfig<MachineEvent> = {
+        id: `machine-events-test`,
+        queryClient,
+        queryKey: [`machineEvents`, `test`],
+        queryFn,
+        getKey: getEventKey,
+      }
+
+      const options = queryCollectionOptions(config)
+      const collection = createCollection(options)
+
+      await collection.preload()
+
+      await vi.waitFor(() => {
+        expect(collection.size).toBe(2)
+      })
+
+      expect(queryFn).toHaveBeenCalledTimes(1)
+      expect(collection.has(`1`)).toBe(true)
+      expect(collection.has(`2`)).toBe(true)
+
+      // Now perform a Direct Write to add a new event
+      // This simulates receiving an event from EventSource
+      const newEvent: MachineEvent = {
+        id: `4`,
+        timestamp: `2025-11-12T22:09:30Z`,
+        message: `Event 4`,
+      }
+
+      collection.utils.writeInsert(newEvent)
+
+      // Verify the new event was added
+      expect(collection.size).toBe(3)
+      expect(collection.has(`4`)).toBe(true)
+
+      // Now refetch - the query returns different data (only item 1)
+      // This simulates a situation where the query result changes
+      // (e.g., server-side filtering, pagination, time-based filters)
+      await collection.utils.refetch()
+
+      await vi.waitFor(() => {
+        expect(queryFn).toHaveBeenCalledTimes(2)
+      })
+
+      // After refetch, the query returns only item 1
+      // Item 2 should be removed (it's not in the result and not a direct write)
+      // BUT item 4 (the directly-written item) should still exist
+      expect(collection.has(`1`)).toBe(true) // In query result
+      expect(collection.has(`2`)).toBe(false) // Not in query result, should be removed
+      expect(collection.has(`4`)).toBe(true) // Direct write, should be protected
+      expect(collection.size).toBe(2) // Item 1 + Item 4
+    })
+  })
 })