feat: add persisted sync metadata support

packages/db/src/collection/state.ts

@@ -28,6 +28,8 @@ interface PendingSyncedTransaction<
   operations: Array<OptimisticChangeMessage<T>>
   truncate?: boolean
   deletedKeys: Set<string | number>
+  rowMetadataWrites: Map<TKey, PendingMetadataWrite>
+  collectionMetadataWrites: Map<string, PendingMetadataWrite>
   optimisticSnapshot?: {
     upserts: Map<TKey, T>
     deletes: Set<TKey>
@@ -40,6 +42,8 @@ interface PendingSyncedTransaction<
   immediate?: boolean
 }
 
+type PendingMetadataWrite = { type: `set`; value: unknown } | { type: `delete` }
+
 type InternalChangeMessage<
   T extends object = Record<string, unknown>,
   TKey extends string | number = string | number,
@@ -70,6 +74,7 @@ export class CollectionStateManager<
   > = []
   public syncedData: SortedMap<TKey, TOutput>
   public syncedMetadata = new Map<TKey, unknown>()
+  public syncedCollectionMetadata = new Map<string, unknown>()
 
   // Optimistic state tracking - make public for testing
   public optimisticUpserts = new Map<TKey, TOutput>()
@@ -870,6 +875,9 @@ export class CollectionStateManager<
         for (const operation of transaction.operations) {
           changedKeys.add(operation.key as TKey)
         }
+        for (const [key] of transaction.rowMetadataWrites) {
+          changedKeys.add(key)
+        }
       }
 
       // Use pre-captured state if available (from optimistic scenarios),
@@ -959,26 +967,6 @@ export class CollectionStateManager<
           const key = operation.key as TKey
           this.syncedKeys.add(key)
 
-          // Update metadata
-          switch (operation.type) {
-            case `insert`:
-              this.syncedMetadata.set(key, operation.metadata)
-              break
-            case `update`:
-              this.syncedMetadata.set(
-                key,
-                Object.assign(
-                  {},
-                  this.syncedMetadata.get(key),
-                  operation.metadata,
-                ),
-              )
-              break
-            case `delete`:
-              this.syncedMetadata.delete(key)
-              break
-          }
-
           // Determine origin: 'local' for local-only collections or pending local changes
           const origin: VirtualOrigin =
             this.isLocalOnly ||
@@ -1025,6 +1013,7 @@ export class CollectionStateManager<
             }
             case `delete`:
               this.syncedData.delete(key)
+              this.syncedMetadata.delete(key)
               // Clean up origin and pending tracking for deleted rows
               this.rowOrigins.delete(key)
               this.pendingLocalChanges.delete(key)
@@ -1036,6 +1025,25 @@ export class CollectionStateManager<
               break
           }
         }
+
+        for (const [key, metadataWrite] of transaction.rowMetadataWrites) {
+          if (metadataWrite.type === `delete`) {
+            this.syncedMetadata.delete(key)
+            continue
+          }
+          this.syncedMetadata.set(key, metadataWrite.value)
+        }
+
+        for (const [
+          key,
+          metadataWrite,
+        ] of transaction.collectionMetadataWrites) {
+          if (metadataWrite.type === `delete`) {
+            this.syncedCollectionMetadata.delete(key)
+            continue
+          }
+          this.syncedCollectionMetadata.set(key, metadataWrite.value)
+        }
       }
 
       // After applying synced operations, if this commit included a truncate,
@@ -1365,6 +1373,7 @@ export class CollectionStateManager<
   public cleanup(): void {
     this.syncedData.clear()
     this.syncedMetadata.clear()
+    this.syncedCollectionMetadata.clear()
     this.optimisticUpserts.clear()
     this.optimisticDeletes.clear()
     this.pendingOptimisticUpserts.clear()

packages/db/src/collection/sync.ts

@@ -18,6 +18,7 @@ import type {
   LoadSubsetOptions,
   OptimisticChangeMessage,
   SyncConfigRes,
+  SyncMetadataApi,
 } from '../types'
 import type { CollectionImpl } from './index.js'
 import type { CollectionStateManager } from './state'
@@ -93,6 +94,8 @@ export class CollectionSyncManager<
               committed: false,
               operations: [],
               deletedKeys: new Set(),
+              rowMetadataWrites: new Map(),
+              collectionMetadataWrites: new Map(),
               immediate: options?.immediate,
             })
           },
@@ -169,6 +172,23 @@ export class CollectionSyncManager<
 
             if (messageType === `delete`) {
               pendingTransaction.deletedKeys.add(key)
+              pendingTransaction.rowMetadataWrites.set(key, { type: `delete` })
+            } else if (messageType === `insert`) {
+              if (message.metadata !== undefined) {
+                pendingTransaction.rowMetadataWrites.set(key, {
+                  type: `set`,
+                  value: message.metadata,
+                })
+              } else {
+                pendingTransaction.rowMetadataWrites.set(key, {
+                  type: `delete`,
+                })
+              }
+            } else if (message.metadata !== undefined) {
+              pendingTransaction.rowMetadataWrites.set(key, {
+                type: `set`,
+                value: message.metadata,
+              })
             }
           },
           commit: () => {
@@ -205,6 +225,7 @@ export class CollectionSyncManager<
             // Clear all operations from the current transaction
             pendingTransaction.operations = []
             pendingTransaction.deletedKeys.clear()
+            pendingTransaction.rowMetadataWrites.clear()
 
             // Mark the transaction as a truncate operation. During commit, this triggers:
             // - Delete events for all previously synced keys (excluding optimistic-deleted keys)
@@ -220,6 +241,7 @@ export class CollectionSyncManager<
               deletes: new Set(this.state.optimisticDeletes),
             }
           },
+          metadata: this.createSyncMetadataApi(),
         }),
       )
 
@@ -245,6 +267,113 @@ export class CollectionSyncManager<
     }
   }
 
+  private getActivePendingSyncTransaction() {
+    const pendingTransaction =
+      this.state.pendingSyncedTransactions[
+        this.state.pendingSyncedTransactions.length - 1
+      ]
+
+    if (!pendingTransaction) {
+      throw new NoPendingSyncTransactionWriteError()
+    }
+    if (pendingTransaction.committed) {
+      throw new SyncTransactionAlreadyCommittedWriteError()
+    }
+
+    return pendingTransaction
+  }
+
+  private createSyncMetadataApi(): SyncMetadataApi<TKey> {
+    return {
+      row: {
+        get: (key) => {
+          const pendingTransaction =
+            this.state.pendingSyncedTransactions[
+              this.state.pendingSyncedTransactions.length - 1
+            ]
+          const pendingWrite = pendingTransaction?.rowMetadataWrites.get(key)
+          if (pendingWrite) {
+            return pendingWrite.type === `delete`
+              ? undefined
+              : pendingWrite.value
+          }
+          if (pendingTransaction?.truncate) {
+            return undefined
+          }
+          return this.state.syncedMetadata.get(key)
+        },
+        set: (key, metadata) => {
+          const pendingTransaction = this.getActivePendingSyncTransaction()
+          pendingTransaction.rowMetadataWrites.set(key, {
+            type: `set`,
+            value: metadata,
+          })
+        },
+        delete: (key) => {
+          const pendingTransaction = this.getActivePendingSyncTransaction()
+          pendingTransaction.rowMetadataWrites.set(key, {
+            type: `delete`,
+          })
+        },
+      },
+      collection: {
+        get: (key) => {
+          const pendingTransaction =
+            this.state.pendingSyncedTransactions[
+              this.state.pendingSyncedTransactions.length - 1
+            ]
+          const pendingWrite =
+            pendingTransaction?.collectionMetadataWrites.get(key)
+          if (pendingWrite) {
+            return pendingWrite.type === `delete`
+              ? undefined
+              : pendingWrite.value
+          }
+          return this.state.syncedCollectionMetadata.get(key)
+        },
+        set: (key, value) => {
+          const pendingTransaction = this.getActivePendingSyncTransaction()
+          pendingTransaction.collectionMetadataWrites.set(key, {
+            type: `set`,
+            value,
+          })
+        },
+        delete: (key) => {
+          const pendingTransaction = this.getActivePendingSyncTransaction()
+          pendingTransaction.collectionMetadataWrites.set(key, {
+            type: `delete`,
+          })
+        },
+        list: (prefix) => {
+          const merged = new Map(this.state.syncedCollectionMetadata)
+          const pendingTransaction =
+            this.state.pendingSyncedTransactions[
+              this.state.pendingSyncedTransactions.length - 1
+            ]
+          if (pendingTransaction) {
+            for (const [
+              key,
+              pendingWrite,
+            ] of pendingTransaction.collectionMetadataWrites) {
+              if (pendingWrite.type === `delete`) {
+                merged.delete(key)
+              } else {
+                merged.set(key, pendingWrite.value)
+              }
+            }
+          }
+
+          return Array.from(merged.entries())
+            .filter(([key]) => (prefix ? key.startsWith(prefix) : true))
+            .map(([key, value]) => ({
+              key,
+              value,
+            }))
+        },
+      },
+    }
+  }
+
   /**
    * Preload the collection data by starting sync if not already started
    * Multiple concurrent calls will share the same promise

packages/db/src/types.ts

@@ -339,6 +339,7 @@ export interface SyncConfig<
     commit: () => void
     markReady: () => void
     truncate: () => void
+    metadata?: SyncMetadataApi<TKey>
   }) => void | CleanupFn | SyncConfigRes
 
   /**
@@ -357,6 +358,25 @@ export interface SyncConfig<
   rowUpdateMode?: `partial` | `full`
 }
 
+export interface SyncMetadataApi<
+  TKey extends string | number = string | number,
+> {
+  row: {
+    get: (key: TKey) => unknown | undefined
+    set: (key: TKey, metadata: unknown) => void
+    delete: (key: TKey) => void
+  }
+  collection: {
+    get: (key: string) => unknown | undefined
+    set: (key: string, value: unknown) => void
+    delete: (key: string) => void
+    list: (prefix?: string) => ReadonlyArray<{
+      key: string
+      value: unknown
+    }>
+  }
+}
+
 export interface ChangeMessage<
   T extends object = Record<string, unknown>,
   TKey extends string | number = string | number,