Virtual props implementation

.changeset/shiny-planes-laugh.md

@@ -0,0 +1,6 @@
+---
+'@tanstack/db': patch
+---
+
+Implement virtual properties end-to-end, including live query behavior and
+typing support for virtual metadata on rows.

docs/guides/live-queries.md

@@ -29,6 +29,18 @@ const activeUsers = createCollection(liveQueryCollectionOptions({
 
 The result types are automatically inferred from your query structure, providing full TypeScript support. When you use a `select` clause, the result type matches your projection. Without `select`, you get the full schema with proper join optionality.
 
+## Virtual properties
+
+Live query results include computed, read-only virtual properties on every row:
+
+- `$synced`: `true` when the row is confirmed by sync; `false` when it is still optimistic.
+- `$origin`: `"local"` if the last confirmed change came from this client, otherwise `"remote"`.
+- `$key`: the row key for the result.
+- `$collectionId`: the source collection ID.
+
+These props can be used in `where`, `select`, and `orderBy` clauses. They are added to
+query outputs automatically and should not be persisted back to storage.
+
 ## Table of Contents
 
 - [Creating Live Query Collections](#creating-live-query-collections)

packages/db/src/collection/change-events.ts

@@ -22,6 +22,7 @@ import type {
 import type { CollectionImpl } from './index.js'
 import type { SingleRowRefProxy } from '../query/builder/ref-proxy'
 import type { BasicExpression, OrderBy } from '../query/ir.js'
+import type { WithVirtualProps } from '../virtual-props.js'
 
 /**
  * Returns the current state of the collection as an array of changes
@@ -58,14 +59,14 @@ export function currentStateAsChanges<
   T extends object,
   TKey extends string | number,
 >(
-  collection: CollectionLike<T, TKey>,
+  collection: CollectionLike<WithVirtualProps<T, TKey>, TKey>,
   options: CurrentStateAsChangesOptions = {},
-): Array<ChangeMessage<T>> | void {
+): Array<ChangeMessage<WithVirtualProps<T, TKey>, TKey>> | void {
   // Helper function to collect filtered results
   const collectFilteredResults = (
-    filterFn?: (value: T) => boolean,
-  ): Array<ChangeMessage<T>> => {
-    const result: Array<ChangeMessage<T>> = []
+    filterFn?: (value: WithVirtualProps<T, TKey>) => boolean,
+  ): Array<ChangeMessage<WithVirtualProps<T, TKey>, TKey>> => {
+    const result: Array<ChangeMessage<WithVirtualProps<T, TKey>, TKey>> = []
     for (const [key, value] of collection.entries()) {
       // If no filter function is provided, include all items
       if (filterFn?.(value) ?? true) {
@@ -106,7 +107,7 @@ export function currentStateAsChanges<
     }
 
     // Convert keys to change messages
-    const result: Array<ChangeMessage<T>> = []
+    const result: Array<ChangeMessage<WithVirtualProps<T, TKey>, TKey>> = []
     for (const key of orderedKeys) {
       const value = collection.get(key)
       if (value !== undefined) {
@@ -138,7 +139,7 @@ export function currentStateAsChanges<
 
     if (optimizationResult.canOptimize) {
       // Use index optimization
-      const result: Array<ChangeMessage<T>> = []
+      const result: Array<ChangeMessage<WithVirtualProps<T, TKey>, TKey>> = []
       for (const key of optimizationResult.matchingKeys) {
         const value = collection.get(key)
         if (value !== undefined) {
@@ -241,9 +242,12 @@ export function createFilterFunctionFromExpression<T extends object>(
  * @param options - The subscription options containing the where clause
  * @returns A filtered callback function
  */
-export function createFilteredCallback<T extends object>(
+export function createFilteredCallback<
+  T extends object,
+  TKey extends string | number = string | number,
+>(
   originalCallback: (changes: Array<ChangeMessage<T>>) => void,
-  options: SubscribeChangesOptions,
+  options: SubscribeChangesOptions<T, TKey>,
 ): (changes: Array<ChangeMessage<T>>) => void {
   const filterFn = createFilterFunctionFromExpression(options.whereExpression!)
 

packages/db/src/collection/changes.ts

@@ -10,6 +10,8 @@ import type { CollectionLifecycleManager } from './lifecycle.js'
 import type { CollectionSyncManager } from './sync.js'
 import type { CollectionEventsManager } from './events.js'
 import type { CollectionImpl } from './index.js'
+import type { CollectionStateManager } from './state.js'
+import type { WithVirtualProps } from '../virtual-props.js'
 
 export class CollectionChangesManager<
   TOutput extends object = Record<string, unknown>,
@@ -21,6 +23,7 @@ export class CollectionChangesManager<
   private sync!: CollectionSyncManager<TOutput, TKey, TSchema, TInput>
   private events!: CollectionEventsManager
   private collection!: CollectionImpl<TOutput, TKey, any, TSchema, TInput>
+  private state!: CollectionStateManager<TOutput, TKey, TSchema, TInput>
 
   public activeSubscribersCount = 0
   public changeSubscriptions = new Set<CollectionSubscription>()
@@ -37,11 +40,13 @@ export class CollectionChangesManager<
     sync: CollectionSyncManager<TOutput, TKey, TSchema, TInput>
     events: CollectionEventsManager
     collection: CollectionImpl<TOutput, TKey, any, TSchema, TInput>
+    state: CollectionStateManager<TOutput, TKey, TSchema, TInput>
   }) {
     this.lifecycle = deps.lifecycle
     this.sync = deps.sync
     this.events = deps.events
     this.collection = deps.collection
+    this.state = deps.state
   }
 
   /**
@@ -55,6 +60,16 @@ export class CollectionChangesManager<
     }
   }
 
+  /**
+   * Enriches a change message with virtual properties ($synced, $origin, $key, $collectionId).
+   * Uses the "add-if-missing" pattern to preserve virtual properties from upstream collections.
+   */
+  private enrichChangeWithVirtualProps(
+    change: ChangeMessage<TOutput, TKey>,
+  ): ChangeMessage<WithVirtualProps<TOutput, TKey>, TKey> {
+    return this.state.enrichChangeMessage(change)
+  }
+
   /**
    * Emit events either immediately or batch them for later emission
    */
@@ -70,35 +85,43 @@ export class CollectionChangesManager<
     }
 
     // Either we're not batching, or we're forcing emission (user action or ending batch cycle)
-    let eventsToEmit = changes
+    let rawEvents = changes
 
     if (forceEmit) {
       // Force emit is used to end a batch (e.g. after a sync commit). Combine any
       // buffered optimistic events with the final changes so subscribers see the
       // whole picture, even if the sync diff is empty.
       if (this.batchedEvents.length > 0) {
-        eventsToEmit = [...this.batchedEvents, ...changes]
+        rawEvents = [...this.batchedEvents, ...changes]
       }
       this.batchedEvents = []
       this.shouldBatchEvents = false
     }
 
-    if (eventsToEmit.length === 0) {
+    if (rawEvents.length === 0) {
       return
     }
 
+    // Enrich all change messages with virtual properties
+    // This uses the "add-if-missing" pattern to preserve pass-through semantics
+    const enrichedEvents: Array<
+      ChangeMessage<WithVirtualProps<TOutput, TKey>, TKey>
+    > = rawEvents.map((change) => this.enrichChangeWithVirtualProps(change))
+
     // Emit to all listeners
     for (const subscription of this.changeSubscriptions) {
-      subscription.emitEvents(eventsToEmit)
+      subscription.emitEvents(enrichedEvents)
     }
   }
 
   /**
    * Subscribe to changes in the collection
    */
   public subscribeChanges(
-    callback: (changes: Array<ChangeMessage<TOutput>>) => void,
-    options: SubscribeChangesOptions<TOutput> = {},
+    callback: (
+      changes: Array<ChangeMessage<WithVirtualProps<TOutput, TKey>>>,
+    ) => void,
+    options: SubscribeChangesOptions<TOutput, TKey> = {},
   ): CollectionSubscription {
     // Start sync and track subscriber
     this.addSubscriber()
@@ -113,7 +136,7 @@ export class CollectionChangesManager<
     const { where, ...opts } = options
     let whereExpression = opts.whereExpression
     if (where) {
-      const proxy = createSingleRowRefProxy<TOutput>()
+      const proxy = createSingleRowRefProxy<WithVirtualProps<TOutput, TKey>>()
       const result = where(proxy)
       whereExpression = toExpression(result)
     }

packages/db/src/collection/index.ts

@@ -37,6 +37,7 @@ import type { SingleRowRefProxy } from '../query/builder/ref-proxy'
 import type { StandardSchemaV1 } from '@standard-schema/spec'
 import type { BTreeIndex } from '../indexes/btree-index.js'
 import type { IndexProxy } from '../indexes/lazy-index.js'
+import type { WithVirtualProps } from '../virtual-props.js'
 
 /**
  * Enhanced Collection interface that includes both data type T and utilities TUtils
@@ -340,6 +341,7 @@ export class CollectionImpl<
       lifecycle: this._lifecycle,
       sync: this._sync,
       events: this._events,
+      state: this._state, // Required for enriching changes with virtual properties
     })
     this._events.setDeps({
       collection: this, // Required for adding to emitted events
@@ -451,8 +453,8 @@ export class CollectionImpl<
   /**
    * Get the current value for a key (virtual derived state)
    */
-  public get(key: TKey): TOutput | undefined {
-    return this._state.get(key)
+  public get(key: TKey): WithVirtualProps<TOutput, TKey> | undefined {
+    return this._state.getWithVirtualProps(key)
   }
 
   /**
@@ -479,40 +481,68 @@ export class CollectionImpl<
   /**
    * Get all values (virtual derived state)
    */
-  public *values(): IterableIterator<TOutput> {
-    yield* this._state.values()
+  public *values(): IterableIterator<WithVirtualProps<TOutput, TKey>> {
+    for (const key of this._state.keys()) {
+      const value = this.get(key)
+      if (value !== undefined) {
+        yield value
+      }
+    }
   }
 
   /**
    * Get all entries (virtual derived state)
    */
-  public *entries(): IterableIterator<[TKey, TOutput]> {
-    yield* this._state.entries()
+  public *entries(): IterableIterator<[TKey, WithVirtualProps<TOutput, TKey>]> {
+    for (const key of this._state.keys()) {
+      const value = this.get(key)
+      if (value !== undefined) {
+        yield [key, value]
+      }
+    }
   }
 
   /**
    * Get all entries (virtual derived state)
    */
-  public *[Symbol.iterator](): IterableIterator<[TKey, TOutput]> {
-    yield* this._state[Symbol.iterator]()
+  public *[Symbol.iterator](): IterableIterator<
+    [TKey, WithVirtualProps<TOutput, TKey>]
+  > {
+    yield* this.entries()
   }
 
   /**
    * Execute a callback for each entry in the collection
    */
   public forEach(
-    callbackfn: (value: TOutput, key: TKey, index: number) => void,
+    callbackfn: (
+      value: WithVirtualProps<TOutput, TKey>,
+      key: TKey,
+      index: number,
+    ) => void,
   ): void {
-    return this._state.forEach(callbackfn)
+    let index = 0
+    for (const [key, value] of this.entries()) {
+      callbackfn(value, key, index++)
+    }
   }
 
   /**
    * Create a new array with the results of calling a function for each entry in the collection
    */
   public map<U>(
-    callbackfn: (value: TOutput, key: TKey, index: number) => U,
+    callbackfn: (
+      value: WithVirtualProps<TOutput, TKey>,
+      key: TKey,
+      index: number,
+    ) => U,
   ): Array<U> {
-    return this._state.map(callbackfn)
+    const result: Array<U> = []
+    let index = 0
+    for (const [key, value] of this.entries()) {
+      result.push(callbackfn(value, key, index++))
+    }
+    return result
   }
 
   public getKeyFromItem(item: TOutput): TKey {
@@ -755,7 +785,7 @@ export class CollectionImpl<
    * }
    */
   get state() {
-    const result = new Map<TKey, TOutput>()
+    const result = new Map<TKey, WithVirtualProps<TOutput, TKey>>()
     for (const [key, value] of this.entries()) {
       result.set(key, value)
     }
@@ -768,7 +798,7 @@ export class CollectionImpl<
    *
    * @returns Promise that resolves to a Map containing all items in the collection
    */
-  stateWhenReady(): Promise<Map<TKey, TOutput>> {
+  stateWhenReady(): Promise<Map<TKey, WithVirtualProps<TOutput, TKey>>> {
     // If we already have data or collection is ready, resolve immediately
     if (this.size > 0 || this.isReady()) {
       return Promise.resolve(this.state)
@@ -793,7 +823,7 @@ export class CollectionImpl<
    *
    * @returns Promise that resolves to an Array containing all items in the collection
    */
-  toArrayWhenReady(): Promise<Array<TOutput>> {
+  toArrayWhenReady(): Promise<Array<WithVirtualProps<TOutput, TKey>>> {
     // If we already have data or collection is ready, resolve immediately
     if (this.size > 0 || this.isReady()) {
       return Promise.resolve(this.toArray)
@@ -823,7 +853,7 @@ export class CollectionImpl<
    */
   public currentStateAsChanges(
     options: CurrentStateAsChangesOptions = {},
-  ): Array<ChangeMessage<TOutput>> | void {
+  ): Array<ChangeMessage<WithVirtualProps<TOutput, TKey>>> | void {
     return currentStateAsChanges(this, options)
   }
 
@@ -870,8 +900,10 @@ export class CollectionImpl<
    * })
    */
   public subscribeChanges(
-    callback: (changes: Array<ChangeMessage<TOutput>>) => void,
-    options: SubscribeChangesOptions<TOutput> = {},
+    callback: (
+      changes: Array<ChangeMessage<WithVirtualProps<TOutput, TKey>>>,
+    ) => void,
+    options: SubscribeChangesOptions<TOutput, TKey> = {},
   ): CollectionSubscription {
     return this._changes.subscribeChanges(callback, options)
   }