Combine persist/awaitSync to just one top-level mutationFn

Author: KyleAMathews

README.md

@@ -204,41 +204,33 @@ const todosConfig = {
       primaryKey: ['id'],
     }
   ),
-  mutationFn: {
-    // Persist mutations to ElectricSQL
-    persist: async (mutations, transaction) => {
-      const response = await fetch(`http://localhost:3001/api/mutations`, {
-        method: `POST`,
-        headers: {
-          "Content-Type": `application/json`,
-        },
-        body: JSON.stringify(transaction.mutations),
-      })
-      if (!response.ok) {
-        // Throwing an error will rollback the optimistic state.
-        throw new Error(`HTTP error! Status: ${response.status}`)
-      }
-
-      const result = await response.json()
-
-      return {
-        txid: result.txid,
-      }
-    },
-    // Wait for a transaction to be synced
-    awaitSync: async ({ config, persistResult: { txid: number } }) => {
-      try {
-        // Use the awaitTxid function from the ElectricSync configuration
-        // This waits for the specific transaction to be synced to the server
-        // The second parameter is an optional timeout in milliseconds
-        await config.sync.awaitTxid(persistResult.txid, 10000)
-        return true;
-      } catch (error) {
-        console.error('Error waiting for transaction to sync:', error);
-        // Throwing an error will rollback the optimistic state.
-        throw error;
-      }
-    },
+  // Persist mutations to ElectricSQL
+  mutationFn: async (mutations, transaction, config) => {
+    const response = await fetch(`http://localhost:3001/api/mutations`, {
+      method: `POST`,
+      headers: {
+        "Content-Type": `application/json`,
+      },
+      body: JSON.stringify(transaction.mutations),
+    })
+    if (!response.ok) {
+      // Throwing an error will rollback the optimistic state.
+      throw new Error(`HTTP error! Status: ${response.status}`)
+    }
+
+    const result = await response.json()
+
+    try {
+      // Use the awaitTxid function from the ElectricSync configuration
+      // This waits for the specific transaction to be synced to the server
+      // The second parameter is an optional timeout in milliseconds
+      await config.sync.awaitTxid(persistResult.txid, 10000)
+      return true;
+    } catch (error) {
+      console.error('Error waiting for transaction to sync:', error);
+      // Throwing an error will rollback the optimistic state.
+      throw error;
+    }
   },
 };
 

examples/react/todo/src/App.tsx

@@ -30,34 +30,22 @@ export default function App() {
       { primaryKey: [`id`] }
     ),
     schema: updateTodoSchema,
-    mutationFn: {
-      persist: async ({ transaction }) => {
-        const response = await fetch(`http://localhost:3001/api/mutations`, {
-          method: `POST`,
-          headers: {
-            "Content-Type": `application/json`,
-          },
-          body: JSON.stringify(transaction.mutations),
-        })
-        if (!response.ok) {
-          throw new Error(`HTTP error! Status: ${response.status}`)
-        }
-
-        const result = await response.json()
-        return {
-          txid: result.txid,
-        }
-      },
-      awaitSync: async ({
-        persistResult,
-        collection,
-      }: {
-        persistResult: { txid: number }
-        collection: Collection
-      }) => {
-        // Start waiting for the txid
-        await collection.config.sync.awaitTxid(persistResult.txid)
-      },
+    mutationFn: async ({ transaction, collection }) => {
+      const response = await fetch(`http://localhost:3001/api/mutations`, {
+        method: `POST`,
+        headers: {
+          "Content-Type": `application/json`,
+        },
+        body: JSON.stringify(transaction.mutations),
+      })
+      if (!response.ok) {
+        throw new Error(`HTTP error! Status: ${response.status}`)
+      }
+
+      const result = await response.json()
+
+      // Start waiting for the txid
+      await collection.config.sync.awaitTxid(result.txid)
     },
   })
 
@@ -83,34 +71,22 @@ export default function App() {
       { primaryKey: [`id`] }
     ),
     schema: updateConfigSchema,
-    mutationFn: {
-      persist: async ({ transaction }) => {
-        const response = await fetch(`http://localhost:3001/api/mutations`, {
-          method: `POST`,
-          headers: {
-            "Content-Type": `application/json`,
-          },
-          body: JSON.stringify(transaction.mutations),
-        })
-        if (!response.ok) {
-          throw new Error(`HTTP error! Status: ${response.status}`)
-        }
-
-        const result = await response.json()
-        return {
-          txid: result.txid,
-        }
-      },
-      awaitSync: async ({
-        persistResult,
-        collection,
-      }: {
-        persistResult: { txid: number }
-        collection: Collection
-      }) => {
-        // Start waiting for the txid
-        await collection.config.sync.awaitTxid(persistResult.txid)
-      },
+    mutationFn: async ({ transaction, collection }) => {
+      const response = await fetch(`http://localhost:3001/api/mutations`, {
+        method: `POST`,
+        headers: {
+          "Content-Type": `application/json`,
+        },
+        body: JSON.stringify(transaction.mutations),
+      })
+      if (!response.ok) {
+        throw new Error(`HTTP error! Status: ${response.status}`)
+      }
+
+      const result = await response.json()
+
+      // Start waiting for the txid
+      await collection.config.sync.awaitTxid(result.txid)
     },
   })
 

package.json

@@ -24,7 +24,6 @@
     "eslint-config-prettier": "^10.1.1",
     "eslint-plugin-prettier": "^5.2.3",
     "eslint-plugin-react": "^7.37.4",
-    "fake-indexeddb": "^6.0.0",
     "husky": "^9.1.7",
     "jsdom": "^26.0.0",
     "lint-staged": "^15.5.0",

packages/optimistic/package.json

@@ -4,8 +4,7 @@
   "version": "0.0.2",
   "dependencies": {
     "@standard-schema/spec": "^1.0.0",
-    "@tanstack/store": "^0.7.0",
-    "idb": "^8.0.2"
+    "@tanstack/store": "^0.7.0"
   },
   "devDependencies": {
     "@vitest/coverage-istanbul": "^3.0.9"

packages/optimistic/src/TransactionManager.ts

@@ -163,7 +163,6 @@ export class TransactionManager<T extends object = Record<string, unknown>> {
         updatedAt: new Date(),
         mutations,
         metadata: {},
-        isSynced: createDeferred(),
         isPersisted: createDeferred(),
       } as Transaction
     }
@@ -180,7 +179,7 @@ export class TransactionManager<T extends object = Record<string, unknown>> {
   }
 
   /**
-   * Process a transaction through persist and awaitSync
+   * Process a transaction through the mutation function
    *
    * @param transactionId - The ID of the transaction to process
    * @private
@@ -189,73 +188,32 @@ export class TransactionManager<T extends object = Record<string, unknown>> {
     const transaction = this.getTransaction(transactionId)
     if (!transaction) return
 
-    // If no mutationFn is provided, throw an error
+    // Throw an error if no mutation function is provided
     if (!this.collection.config.mutationFn) {
       throw new Error(
-        `Cannot process transaction without a mutationFn in the collection config`
+        `No mutation function provided for transaction ${transactionId}`
       )
     }
 
+    // Set transaction state to persisting
     this.setTransactionState(transactionId, `persisting`)
 
-    this.collection.config.mutationFn
-      .persist({
-        transaction: this.createLiveTransactionReference(transactionId),
+    // Create a live reference to the transaction that always returns the latest state
+    const transactionRef = this.createLiveTransactionReference(transactionId)
+
+    // Call the mutation function
+    this.collection.config
+      .mutationFn({
+        transaction: transactionRef,
         collection: this.collection,
       })
-      .then((persistResult) => {
+      .then(() => {
         const tx = this.getTransaction(transactionId)
         if (!tx) return
 
+        // Mark as persisted
         tx.isPersisted?.resolve(true)
-        if (this.collection.config.mutationFn?.awaitSync) {
-          this.setTransactionState(transactionId, `persisted_awaiting_sync`)
-
-          // Create a promise that rejects after 2 seconds
-          const timeoutPromise = new Promise<never>((_, reject) => {
-            setTimeout(() => {
-              reject(new Error(`Sync operation timed out after 2 seconds`))
-            }, this.collection.config.mutationFn?.awaitSyncTimeoutMs ?? 2000)
-          })
-
-          // Race the awaitSync promise against the timeout
-          Promise.race([
-            this.collection.config.mutationFn.awaitSync({
-              transaction: this.createLiveTransactionReference(transactionId),
-              collection: this.collection,
-              persistResult,
-            }),
-            timeoutPromise,
-          ])
-            .then(() => {
-              const updatedTx = this.getTransaction(transactionId)
-              if (!updatedTx) return
-
-              updatedTx.isSynced?.resolve(true)
-              this.setTransactionState(transactionId, `completed`)
-            })
-            // Catch awaitSync errors or timeout
-            .catch((error) => {
-              const updatedTx = this.getTransaction(transactionId)
-              if (!updatedTx) return
-
-              // Update transaction with error information
-              updatedTx.error = {
-                message: error.message || `Error during sync`,
-                error:
-                  error instanceof Error ? error : new Error(String(error)),
-              }
-
-              // Reject the isSynced promise
-              updatedTx.isSynced?.reject(error)
-
-              // Set transaction state to failed
-              this.setTransaction(updatedTx)
-              this.setTransactionState(transactionId, `failed`)
-            })
-        } else {
-          this.setTransactionState(transactionId, `completed`)
-        }
+        this.setTransactionState(transactionId, `completed`)
       })
       .catch((error) => {
         const tx = this.getTransaction(transactionId)
@@ -267,9 +225,8 @@ export class TransactionManager<T extends object = Record<string, unknown>> {
           error: error instanceof Error ? error : new Error(String(error)),
         }
 
-        // Reject both promises
+        // Reject the promise
         tx.isPersisted?.reject(tx.error.error)
-        tx.isSynced?.reject(tx.error.error)
 
         // Set transaction state to failed
         this.setTransactionState(transactionId, `failed`)

packages/optimistic/src/collection.ts

@@ -44,7 +44,14 @@ interface PendingSyncedTransaction<T extends object = Record<string, unknown>> {
  *     id: `users-${params.userId}`,
  *     sync: { ... },
  *     // mutationFn is optional - provide it if you need mutation capabilities
- *     mutationFn: { ... }
+ *     mutationFn: async (params: {
+ *       transaction: Transaction
+ *       collection: Collection<Record<string, unknown>>
+ *     }) => {
+ *       // Implement your mutation (and syncing) logic here
+ *       // Return a promise that resolves when the mutation and syncing is complete.
+ *       // When this function returns, the optimistic mutations are dropped.
+ *     }
  *   });
  *
  *   return null;

packages/optimistic/src/types.ts

@@ -29,7 +29,6 @@ export interface Transaction {
   updatedAt: Date
   mutations: Array<PendingMutation>
   metadata: Record<string, unknown>
-  isSynced?: Deferred<boolean>
   isPersisted?: Deferred<boolean>
   error?: {
     transactionId?: string // For dependency failures
@@ -88,21 +87,10 @@ export interface OptimisticChangeMessage<
   isActive?: boolean
 }
 
-export interface MutationFn<T extends object = Record<string, unknown>> {
-  persist: (params: {
-    transaction: Transaction
-    collection: Collection<T>
-  }) => Promise<any>
-
-  // Set timeout for awaiting sync (default is 2 seconds)
-  awaitSyncTimeoutMs?: number
-  awaitSync?: (params: {
-    transaction: Transaction
-    collection: Collection<T>
-
-    persistResult: any
-  }) => Promise<void>
-}
+export type MutationFn<T extends object = Record<string, unknown>> = (params: {
+  transaction: Transaction
+  collection: Collection<T>
+}) => Promise<any>
 
 /**
  * The Standard Schema interface.