Merge branch 'main' into kevin/pred-pushdown-query-coll

Author: KyleAMathews

.changeset/floppy-wings-mix.md

@@ -0,0 +1,5 @@
+---
+"@tanstack/db": patch
+---
+
+Stop pushing where clauses that target renamed subquery projections so alias remapping stays intact, preventing a bug where a where clause would not be executed correctly.

.changeset/large-otters-unite.md

@@ -54,15 +54,21 @@ The `electricCollectionOptions` function accepts the following options:
 
 ### Persistence Handlers
 
+Handlers are called before mutations to persist changes to your backend:
+
 - `onInsert`: Handler called before insert operations
-- `onUpdate`: Handler called before update operations  
+- `onUpdate`: Handler called before update operations
 - `onDelete`: Handler called before delete operations
 
-## Persistence Handlers
+Each handler should return `{ txid }` to wait for synchronization. For cases where your API can not return txids, use the `awaitMatch` utility function.
+
+## Persistence Handlers & Synchronization
+
+Handlers persist mutations to the backend and wait for Electric to sync the changes back. This prevents UI glitches where optimistic updates would be removed and then re-added. TanStack DB blocks sync data until the mutation is confirmed, ensuring smooth user experience.
 
-Handlers can be defined to run on mutations. They are useful to send mutations to the backend and confirming them once Electric delivers the corresponding transactions. Until confirmation, TanStack DB blocks sync data for the collection to prevent race conditions. To avoid any delays, it’s important to use a matching strategy.
+### 1. Using Txid (Recommended)
 
-The most reliable strategy is for the backend to include the transaction ID (txid) in its response, allowing the client to match each mutation with Electric’s transaction identifiers for precise confirmation. If no strategy is provided, client mutations are automatically confirmed after three seconds.
+The recommended approach uses PostgreSQL transaction IDs (txids) for precise matching. The backend returns a txid, and the client waits for that specific txid to appear in the Electric stream.
 
 ```typescript
 const todosCollection = createCollection(
@@ -74,15 +80,83 @@ const todosCollection = createCollection(
       url: '/api/todos',
       params: { table: 'todos' },
     },
-    
+
     onInsert: async ({ transaction }) => {
       const newItem = transaction.mutations[0].modified
       const response = await api.todos.create(newItem)
-      
+
+      // Return txid to wait for sync
       return { txid: response.txid }
     },
-    
-    // you can also implement onUpdate and onDelete handlers
+
+    onUpdate: async ({ transaction }) => {
+      const { original, changes } = transaction.mutations[0]
+      const response = await api.todos.update({
+        where: { id: original.id },
+        data: changes
+      })
+
+      return { txid: response.txid }
+    }
+  })
+)
+```
+
+### 2. Using Custom Match Functions
+
+For cases where txids aren't available, use the `awaitMatch` utility function to wait for synchronization with custom matching logic:
+
+```typescript
+import { isChangeMessage } from '@tanstack/electric-db-collection'
+
+const todosCollection = createCollection(
+  electricCollectionOptions({
+    id: 'todos',
+    getKey: (item) => item.id,
+    shapeOptions: {
+      url: '/api/todos',
+      params: { table: 'todos' },
+    },
+
+    onInsert: async ({ transaction, collection }) => {
+      const newItem = transaction.mutations[0].modified
+      await api.todos.create(newItem)
+
+      // Use awaitMatch utility for custom matching
+      await collection.utils.awaitMatch(
+        (message) => {
+          return isChangeMessage(message) &&
+                 message.headers.operation === 'insert' &&
+                 message.value.text === newItem.text
+        },
+        5000 // timeout in ms (optional, defaults to 3000)
+      )
+    }
+  })
+)
+```
+
+### 3. Using Simple Timeout
+
+For quick prototyping or when you're confident about timing, you can use a simple timeout. This is crude but works as almost always the data will be synced back in under 2 seconds:
+
+```typescript
+const todosCollection = createCollection(
+  electricCollectionOptions({
+    id: 'todos',
+    getKey: (item) => item.id,
+    shapeOptions: {
+      url: '/api/todos',
+      params: { table: 'todos' },
+    },
+
+    onInsert: async ({ transaction }) => {
+      const newItem = transaction.mutations[0].modified
+      await api.todos.create(newItem)
+
+      // Simple timeout approach
+      await new Promise(resolve => setTimeout(resolve, 2000))
+    }
   })
 )
 ```
@@ -162,7 +236,9 @@ export const ServerRoute = createServerFileRoute("/api/todos").methods({
 
 ## Optimistic Updates with Explicit Transactions
 
-For more advanced use cases, you can create custom actions that can do multiple mutations across collections transactionally. In this case, you need to explicitly await for the transaction ID using `utils.awaitTxId()`.
+For more advanced use cases, you can create custom actions that can do multiple mutations across collections transactionally. You can use the utility methods to wait for synchronization with different strategies:
+
+### Using Txid Strategy
 
 ```typescript
 const addTodoAction = createOptimisticAction({
@@ -184,19 +260,187 @@ const addTodoAction = createOptimisticAction({
       data: { text, completed: false }
     })
 
+    // Wait for the specific txid
     await todosCollection.utils.awaitTxId(response.txid)
   }
 })
 ```
 
+### Using Custom Match Function
+
+```typescript
+import { isChangeMessage } from '@tanstack/electric-db-collection'
+
+const addTodoAction = createOptimisticAction({
+  onMutate: ({ text }) => {
+    const tempId = crypto.randomUUID()
+    todosCollection.insert({
+      id: tempId,
+      text,
+      completed: false,
+      created_at: new Date(),
+    })
+  },
+  
+  mutationFn: async ({ text }) => {
+    await api.todos.create({
+      data: { text, completed: false }
+    })
+    
+    // Wait for matching message
+    await todosCollection.utils.awaitMatch(
+      (message) => {
+        return isChangeMessage(message) && 
+               message.headers.operation === 'insert' &&
+               message.value.text === text
+      }
+    )
+  }
+})
+```
+
 ## Utility Methods
 
 The collection provides these utility methods via `collection.utils`:
 
-- `awaitTxId(txid, timeout?)`: Manually wait for a specific transaction ID to be synchronized
+### `awaitTxId(txid, timeout?)`
+
+Manually wait for a specific transaction ID to be synchronized:
 
 ```typescript
-todosCollection.utils.awaitTxId(12345)
+// Wait for specific txid
+await todosCollection.utils.awaitTxId(12345)
+
+// With custom timeout (default is 30 seconds)
+await todosCollection.utils.awaitTxId(12345, 10000)
 ```
 
 This is useful when you need to ensure a mutation has been synchronized before proceeding with other operations.
+
+### `awaitMatch(matchFn, timeout?)`
+
+Manually wait for a custom match function to find a matching message:
+
+```typescript
+import { isChangeMessage } from '@tanstack/electric-db-collection'
+
+// Wait for a specific message pattern
+await todosCollection.utils.awaitMatch(
+  (message) => {
+    return isChangeMessage(message) &&
+           message.headers.operation === 'insert' &&
+           message.value.text === 'New Todo'
+  },
+  5000 // timeout in ms
+)
+```
+
+### Helper Functions
+
+The package exports helper functions for use in custom match functions:
+
+- `isChangeMessage(message)`: Check if a message is a data change (insert/update/delete)
+- `isControlMessage(message)`: Check if a message is a control message (up-to-date, must-refetch)
+
+```typescript
+import { isChangeMessage, isControlMessage } from '@tanstack/electric-db-collection'
+
+// Use in custom match functions
+const matchFn = (message) => {
+  if (isChangeMessage(message)) {
+    return message.headers.operation === 'insert'
+  }
+  return false
+}
+```
+
+## Debugging
+
+### Common Issue: awaitTxId Stalls or Times Out
+
+A frequent issue developers encounter is that `awaitTxId` (or the transaction's `isPersisted.promise`) stalls indefinitely, eventually timing out with no error messages. The data persists correctly to the database, but the optimistic mutation never resolves.
+
+**Root Cause:** This happens when the transaction ID (txid) returned from your API doesn't match the actual transaction ID of the mutation in Postgres. This mismatch occurs when you query `pg_current_xact_id()` **outside** the same transaction that performs the mutation.
+
+### Enable Debug Logging
+
+To diagnose txid issues, enable debug logging in your browser console:
+
+```javascript
+localStorage.debug = 'ts/db:electric'
+```
+
+This will show you when mutations start waiting for txids and when txids arrive from Electric's sync stream.
+
+This is powered by the [debug](https://www.npmjs.com/package/debug) package.
+
+**When txids DON'T match (common bug):**
+```
+ts/db:electric awaitTxId called with txid 124
+ts/db:electric new txids synced from pg [123]
+// Stalls forever - 124 never arrives!
+```
+
+In this example, the mutation happened in transaction 123, but you queried `pg_current_xact_id()` in a separate transaction (124) that ran after the mutation. The client waits for 124 which will never arrive.
+
+**When txids DO match (correct):**
+```
+ts/db:electric awaitTxId called with txid 123
+ts/db:electric new txids synced from pg [123]
+ts/db:electric awaitTxId found match for txid 123
+// Resolves immediately!
+```
+
+### The Solution: Query txid Inside the Transaction
+
+You **must** call `pg_current_xact_id()` inside the same transaction as your mutation:
+
+**❌ Wrong - txid queried outside transaction:**
+```typescript
+// DON'T DO THIS
+async function createTodo(data) {
+  const txid = await generateTxId(sql) // Wrong: separate transaction
+
+  await sql.begin(async (tx) => {
+    await tx`INSERT INTO todos ${tx(data)}`
+  })
+
+  return { txid } // This txid won't match!
+}
+```
+
+**✅ Correct - txid queried inside transaction:**
+```typescript
+// DO THIS
+async function createTodo(data) {
+  let txid!: Txid
+
+  const result = await sql.begin(async (tx) => {
+    // Call generateTxId INSIDE the transaction
+    txid = await generateTxId(tx)
+
+    const [todo] = await tx`
+      INSERT INTO todos ${tx(data)}
+      RETURNING *
+    `
+    return todo
+  })
+
+  return { todo: result, txid } // txid matches the mutation
+}
+
+async function generateTxId(tx: any): Promise<Txid> {
+  const result = await tx`SELECT pg_current_xact_id()::xid::text as txid`
+  const txid = result[0]?.txid
+
+  if (txid === undefined) {
+    throw new Error(`Failed to get transaction ID`)
+  }
+
+  return parseInt(txid, 10)
+}
+```
+
+See working examples in:
+- `examples/react/todo/src/routes/api/todos.ts`
+- `examples/react/todo/src/api/server.ts`