feat: cursor-aware effect replay with startAfter gating

.changeset/cursor-aware-effect-replay.md

@@ -0,0 +1,6 @@
+---
+'@tanstack/db': patch
+'@tanstack/react-db': patch
+---
+
+Add cursor-aware effect replay via `startAfter` option on `createEffect` and `useLiveQueryEffect`. Sync writes can now carry an opaque sortable cursor that propagates through `ChangeMessage` and `DeltaEvent`, enabling effects to suppress callbacks during historical replay while still hydrating internal query state. `startAfter` accepts either a scalar cursor (single-source) or a `Record<string, CollectionCursor>` for per-source gating in join queries. `DeltaEvent` now includes `triggeringSource` and `cursors` fields for gated effects.

packages/db/src/collection/state.ts

@@ -12,6 +12,7 @@ import type { StandardSchemaV1 } from '@standard-schema/spec'
 import type {
   ChangeMessage,
   CollectionConfig,
+  CollectionCursor,
   OptimisticChangeMessage,
 } from '../types'
 import type { CollectionImpl } from './index.js'
@@ -28,6 +29,7 @@ interface PendingSyncedTransaction<
   operations: Array<OptimisticChangeMessage<T>>
   truncate?: boolean
   deletedKeys: Set<string | number>
+  rowCursorWrites: Map<TKey, PendingCursorWrite>
   rowMetadataWrites: Map<TKey, PendingMetadataWrite>
   collectionMetadataWrites: Map<string, PendingMetadataWrite>
   optimisticSnapshot?: {
@@ -42,6 +44,10 @@ interface PendingSyncedTransaction<
   immediate?: boolean
 }
 
+type PendingCursorWrite =
+  | { type: `set`; value: CollectionCursor }
+  | { type: `delete` }
+
 type PendingMetadataWrite = { type: `set`; value: unknown } | { type: `delete` }
 
 type InternalChangeMessage<
@@ -313,6 +319,7 @@ export class CollectionStateManager<
       type: change.type,
       value: enrichedValue,
       previousValue: enrichedPreviousValue,
+      cursor: change.cursor,
       metadata: change.metadata,
     } as ChangeMessage<WithVirtualProps<TOutput, TKey>, TKey>
   }
@@ -895,6 +902,7 @@ export class CollectionStateManager<
       }
 
       const events: Array<ChangeMessage<TOutput, TKey>> = []
+      const eventCursors = new Map<TKey, CollectionCursor>()
       const rowUpdateMode = this.config.sync.rowUpdateMode || `partial`
       const completedOptimisticOps = new Map<
         TKey,
@@ -1034,6 +1042,14 @@ export class CollectionStateManager<
           this.syncedMetadata.set(key, metadataWrite.value)
         }
 
+        for (const [key, cursorWrite] of transaction.rowCursorWrites) {
+          if (cursorWrite.type === `delete`) {
+            eventCursors.delete(key)
+            continue
+          }
+          eventCursors.set(key, cursorWrite.value)
+        }
+
         for (const [
           key,
           metadataWrite,
@@ -1162,6 +1178,7 @@ export class CollectionStateManager<
 
       // Now check what actually changed in the final visible state
       for (const key of changedKeys) {
+        const cursor = eventCursors.get(key)
         const previousVisibleValue = currentVisibleState.get(key)
         const newVisibleValue = this.get(key) // This returns the new derived state
         const previousVirtualProps = this.getVirtualPropsSnapshotForState(key, {
@@ -1235,12 +1252,14 @@ export class CollectionStateManager<
               key,
               value: newVisibleValue,
               previousValue: previousValueWithVirtualFromCompleted,
+              cursor,
             })
           } else {
             events.push({
               type: `insert`,
               key,
               value: newVisibleValue,
+              cursor,
             })
           }
         } else if (
@@ -1251,6 +1270,7 @@ export class CollectionStateManager<
             type: `delete`,
             key,
             value: previousValueWithVirtual ?? previousVisibleValue,
+            cursor,
           })
         } else if (
           previousVisibleValue !== undefined &&
@@ -1263,6 +1283,7 @@ export class CollectionStateManager<
             key,
             value: newVisibleValue,
             previousValue: previousValueWithVirtual ?? previousVisibleValue,
+            cursor,
           })
         }
       }

packages/db/src/collection/sync.ts

@@ -94,6 +94,7 @@ export class CollectionSyncManager<
               committed: false,
               operations: [],
               deletedKeys: new Set(),
+              rowCursorWrites: new Map(),
               rowMetadataWrites: new Map(),
               collectionMetadataWrites: new Map(),
               immediate: options?.immediate,
@@ -190,6 +191,13 @@ export class CollectionSyncManager<
                 value: message.metadata,
               })
             }
+
+            if (message.cursor !== undefined) {
+              pendingTransaction.rowCursorWrites.set(key, {
+                type: `set`,
+                value: message.cursor,
+              })
+            }
           },
           commit: () => {
             const pendingTransaction =
@@ -225,6 +233,7 @@ export class CollectionSyncManager<
             // Clear all operations from the current transaction
             pendingTransaction.operations = []
             pendingTransaction.deletedKeys.clear()
+            pendingTransaction.rowCursorWrites.clear()
             pendingTransaction.rowMetadataWrites.clear()
             // Intentionally preserve collectionMetadataWrites across truncate.
             // Collection-scoped metadata (for example persisted resume/reset