diff --git a/.changeset/sweet-kings-wear.md b/.changeset/sweet-kings-wear.md
new file mode 100644
index 000000000..cfddc8891
--- /dev/null
+++ b/.changeset/sweet-kings-wear.md
@@ -0,0 +1,6 @@
+---
+"@tanstack/react-optimistic": patch
+"@tanstack/optimistic": patch
+---
+
+Make transactions first class & move ownership of mutationFn from collections to transactions
diff --git a/README.md b/README.md
index ec857d083..232e26b73 100644
--- a/README.md
+++ b/README.md
@@ -58,12 +58,25 @@ The library uses proxies to create immutable snapshots and track changes:
 The primary hook for interacting with collections in React components.
 
 ```typescript
+// Create a collection
 const { data, insert, update, delete: deleteFn } = useCollection({
   id: 'todos',
   sync: { /* sync configuration */ },
-  mutationFn: { /* mutation functions */ },
   schema: /* optional schema */
 });
+
+// Create a mutation
+const mutation = useOptimisticMutation({
+  mutationFn: async ({ mutations }) => {
+    // Implement your mutation logic here
+    // This function is called when mutations are committed
+  }
+});
+
+// Use the mutation with collection operations
+mutation.mutate(() => {
+  insert({ text: 'New todo' });
+});
 ```
 
 Returns:
@@ -167,10 +180,22 @@ const todoCollection = useCollection({
 
 ## Transaction Management
 
-The library includes a robust transaction management system:
+The library includes a simple yet powerful transaction management system. Transactions are created using the `createTransaction` function:
+
+```typescript
+const tx = createTransaction({
+  mutationFn: async ({ transaction }) => {
+    // Implement your mutation logic here
+    // This function is called when the transaction is committed
+  },
+})
 
-- `TransactionManager`: Handles transaction lifecycle, persistence, and retry logic
-- `TransactionStore`: Provides persistent storage for transactions using IndexedDB
+// Apply mutations within the transaction
+tx.mutate(() => {
+  // All collection operations (insert/update/delete) within this callback
+  // will be part of this transaction
+})
+```
 
 Transactions progress through several states:
 
@@ -204,35 +229,56 @@ const todosConfig = {
       primaryKey: ['id'],
     }
   ),
-  // 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),
+};
+
+// In your component
+function TodoList() {
+  const { data, insert, update, delete: deleteFn } = useCollection(todosConfig)
+
+  // Create a mutation for handling all todo operations
+  const todoMutation = useOptimisticMutation({
+    mutationFn: async ({ transaction }) => {
+      // Filter out collection from mutations before sending to server
+      const payload = transaction.mutations.map(m => {
+        const { collection, ...payload } = m
+        return payload
+      })
+
+      const response = await fetch(`http://localhost:3001/api/mutations`, {
+        method: `POST`,
+        headers: {
+          "Content-Type": `application/json`,
+        },
+        body: JSON.stringify(payload),
+      })
+      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
+        await transaction.mutations[0].collection.config.sync.awaitTxid(result.txid)
+      } catch (error) {
+        console.error('Error waiting for transaction to sync:', error);
+        // Throwing an error will rollback the optimistic state.
+        throw error;
+      }
+    },
+  })
+
+  // Use the mutation for any todo operations
+  const addTodo = () => {
+    todoMutation.mutate(() => {
+      insert({ title: 'New todo', completed: false })
     })
-    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;
-    }
-  },
-};
+  // ... rest of your component
+}
 
 // In a route loader
 export async function loader() {
diff --git a/examples/react/todo/package.json b/examples/react/todo/package.json
index cfec16b6a..b1fa984a9 100644
--- a/examples/react/todo/package.json
+++ b/examples/react/todo/package.json
@@ -44,8 +44,8 @@
     "db:generate": "drizzle-kit generate",
     "db:push": "tsx scripts/migrate.ts",
     "db:studio": "drizzle-kit studio",
-    "dev": "docker-compose up -d && concurrently \"pnpm api:dev\" \"vite\"",
-    "lint": "eslint .",
+    "dev": "docker compose up -d && concurrently \"pnpm api:dev\" \"vite\"",
+    "lint": "eslint . --fix",
     "preview": "vite preview"
   },
   "type": "module"
diff --git a/examples/react/todo/src/App.tsx b/examples/react/todo/src/App.tsx
index c9634a78c..973756e37 100644
--- a/examples/react/todo/src/App.tsx
+++ b/examples/react/todo/src/App.tsx
@@ -1,94 +1,112 @@
 import React, { useState } from "react"
-import { createElectricSync, useCollection } from "@tanstack/react-optimistic"
+import {
+  Collection,
+  createElectricSync,
+  createTransaction,
+  useLiveQuery,
+} from "@tanstack/react-optimistic"
 import { DevTools } from "./DevTools"
 import { updateConfigSchema, updateTodoSchema } from "./db/validation"
+import type { MutationFn, PendingMutation } from "@tanstack/react-optimistic"
 import type { UpdateConfig, UpdateTodo } from "./db/validation"
-import type { Collection } from "@tanstack/react-optimistic"
 import type { FormEvent } from "react"
 
-export default function App() {
-  const [newTodo, setNewTodo] = useState(``)
+const todoMutationFn: MutationFn = async ({ transaction }) => {
+  const payload = transaction.mutations.map(
+    (m: PendingMutation<UpdateTodo>) => {
+      const { collection, ...rest } = m
+      return rest
+    }
+  )
+  const response = await fetch(`http://localhost:3001/api/mutations`, {
+    method: `POST`,
+    headers: {
+      "Content-Type": `application/json`,
+    },
+    body: JSON.stringify(payload),
+  })
+  if (!response.ok) {
+    throw new Error(`HTTP error! Status: ${response.status}`)
+  }
 
-  const {
-    data: todos,
-    insert,
-    update,
-    delete: deleteTodo,
-  } = useCollection<UpdateTodo>({
-    id: `todos`,
-    sync: createElectricSync(
-      {
-        url: `http://localhost:3003/v1/shape`,
-        params: {
-          table: `todos`,
-        },
-        parser: {
-          // Parse timestamp columns into JavaScript Date objects
-          timestamptz: (date: string) => new Date(date),
-        },
-      },
-      { primaryKey: [`id`] }
-    ),
-    schema: updateTodoSchema,
-    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()
 
-      const result = await response.json()
+  // Start waiting for the txid
+  await transaction.mutations[0]!.collection.config.sync.awaitTxid(result.txid)
+}
 
-      // Start waiting for the txid
-      await collection.config.sync.awaitTxid(result.txid)
+const configMutationFn: MutationFn = async ({ transaction }) => {
+  const payload = transaction.mutations.map(
+    (m: PendingMutation<UpdateConfig>) => {
+      const { collection, ...rest } = m
+      return rest
+    }
+  )
+  const response = await fetch(`http://localhost:3001/api/mutations`, {
+    method: `POST`,
+    headers: {
+      "Content-Type": `application/json`,
     },
+    body: JSON.stringify(payload),
   })
+  if (!response.ok) {
+    throw new Error(`HTTP error! Status: ${response.status}`)
+  }
 
-  const {
-    data: configData,
-    update: updateConfig,
-    insert: insertConfig,
-  } = useCollection<UpdateConfig>({
-    id: `config`,
-    sync: createElectricSync(
-      {
-        url: `http://localhost:3003/v1/shape`,
-        params: {
-          table: `config`,
-        },
-        parser: {
-          // Parse timestamp columns into JavaScript Date objects
-          timestamptz: (date: string) => {
-            return new Date(date)
-          },
-        },
+  const result = await response.json()
+
+  // Start waiting for the txid
+  await transaction.mutations[0]!.collection.config.sync.awaitTxid(result.txid)
+}
+
+const todoCollection = new Collection<UpdateTodo>({
+  id: `todos`,
+  sync: createElectricSync(
+    {
+      url: `http://localhost:3003/v1/shape`,
+      params: {
+        table: `todos`,
+      },
+      parser: {
+        // Parse timestamp columns into JavaScript Date objects
+        timestamptz: (date: string) => new Date(date),
+      },
+    },
+    { primaryKey: [`id`] }
+  ),
+  schema: updateTodoSchema,
+})
+
+const configCollection = new Collection<UpdateConfig>({
+  id: `config`,
+  sync: createElectricSync(
+    {
+      url: `http://localhost:3003/v1/shape`,
+      params: {
+        table: `config`,
       },
-      { primaryKey: [`id`] }
-    ),
-    schema: updateConfigSchema,
-    mutationFn: async ({ transaction, collection }) => {
-      const response = await fetch(`http://localhost:3001/api/mutations`, {
-        method: `POST`,
-        headers: {
-          "Content-Type": `application/json`,
+      parser: {
+        // Parse timestamp columns into JavaScript Date objects
+        timestamptz: (date: string) => {
+          return new Date(date)
         },
-        body: JSON.stringify(transaction.mutations),
-      })
-      if (!response.ok) {
-        throw new Error(`HTTP error! Status: ${response.status}`)
-      }
+      },
+    },
+    { primaryKey: [`id`] }
+  ),
+  schema: updateConfigSchema,
+})
+
+export default function App() {
+  const [newTodo, setNewTodo] = useState(``)
 
-      const result = await response.json()
+  const { data: todos } = useLiveQuery((q) =>
+    q.from({ todoCollection }).keyBy(`@id`).select(`@id`, `@text`, `@completed`)
+  )
 
-      // Start waiting for the txid
-      await collection.config.sync.awaitTxid(result.txid)
-    },
-  })
+  const { data: configData } = useLiveQuery((q) =>
+    q.from({ configCollection }).keyBy(`@id`).select(`@id`, `@key`, `@value`)
+  )
 
   // Define a more robust type-safe helper function to get config values
   const getConfigValue = (key: string): string => {
@@ -104,19 +122,26 @@ export default function App() {
   const setConfigValue = (key: string, value: string): void => {
     for (const config of configData) {
       if (config.key === key) {
-        updateConfig(config, (draft) => {
-          draft.value = value
-        })
+        createTransaction({ mutationFn: configMutationFn }).mutate(() =>
+          configCollection.update(
+            Array.from(configCollection.state.values())[0]!,
+            (draft) => {
+              draft.value = value
+            }
+          )
+        )
 
         return
       }
     }
 
     // If the config doesn't exist yet, create it
-    insertConfig({
-      key,
-      value,
-    })
+    createTransaction({ mutationFn: configMutationFn }).mutate(() =>
+      configCollection.insert({
+        key,
+        value,
+      })
+    )
   }
 
   const backgroundColor = getConfigValue(`backgroundColor`)
@@ -176,17 +201,29 @@ export default function App() {
     e.preventDefault()
     if (!newTodo.trim()) return
 
-    insert({
-      text: newTodo,
-      completed: false,
-    })
+    const tx = createTransaction({ mutationFn: todoMutationFn })
+    tx.mutate(() =>
+      todoCollection.insert({
+        text: newTodo,
+        completed: false,
+        id: Math.round(Math.random() * 1000000),
+      })
+    )
     setNewTodo(``)
   }
 
   const toggleTodo = (todo: UpdateTodo) => {
-    update(todo, (draft) => {
-      draft.completed = !draft.completed
-    })
+    const tx = createTransaction({ mutationFn: todoMutationFn })
+    tx.mutate(() =>
+      todoCollection.update(
+        Array.from(todoCollection.state.values()).find(
+          (t) => t.id === todo.id
+        )!,
+        (draft) => {
+          draft.completed = !draft.completed
+        }
+      )
+    )
   }
 
   const activeTodos = todos.filter((todo) => !todo.completed)
@@ -233,13 +270,23 @@ export default function App() {
                   className="absolute left-0 w-12 h-full text-[30px] text-[#e6e6e6] hover:text-[#4d4d4d]"
                   onClick={() => {
                     const allCompleted = completedTodos.length === todos.length
-                    update(
-                      allCompleted ? completedTodos : activeTodos,
-                      (drafts) => {
-                        drafts.forEach(
-                          (draft) => (draft.completed = !allCompleted)
-                        )
-                      }
+                    const tx = createTransaction({ mutationFn: todoMutationFn })
+                    const todosToToggle = allCompleted
+                      ? completedTodos
+                      : activeTodos
+                    const togglingIds = new Set()
+                    todosToToggle.forEach((t) => togglingIds.add(t.id))
+                    tx.mutate(() =>
+                      todoCollection.update(
+                        Array.from(todoCollection.state.values()).filter((t) =>
+                          togglingIds.has(t.id)
+                        ),
+                        (drafts) => {
+                          drafts.forEach(
+                            (draft) => (draft.completed = !allCompleted)
+                          )
+                        }
+                      )
                     )
                   }}
                 >
@@ -281,7 +328,18 @@ export default function App() {
                           {todo.text}
                         </label>
                         <button
-                          onClick={() => deleteTodo(todo)}
+                          onClick={() => {
+                            const tx = createTransaction({
+                              mutationFn: todoMutationFn,
+                            })
+                            tx.mutate(() =>
+                              todoCollection.delete(
+                                Array.from(todoCollection.state.values()).find(
+                                  (t) => t.id === todo.id
+                                )!
+                              )
+                            )
+                          }}
                           className="hidden group-hover:block absolute right-[10px] w-[40px] h-[40px] my-auto top-0 bottom-0 text-[30px] text-[#cc9a9a] hover:text-[#af5b5e] transition-colors"
                         >
                           ×
@@ -300,7 +358,16 @@ export default function App() {
                   {completedTodos.length > 0 && (
                     <button
                       onClick={() => {
-                        deleteTodo(completedTodos)
+                        const tx = createTransaction({
+                          mutationFn: todoMutationFn,
+                        })
+                        tx.mutate(() =>
+                          todoCollection.delete(
+                            Array.from(todoCollection.state.values()).filter(
+                              (t) => completedTodos.some((ct) => ct.id === t.id)
+                            )
+                          )
+                        )
                       }}
                       className="text-inherit hover:underline"
                     >
diff --git a/examples/react/todo/src/db/validation.ts b/examples/react/todo/src/db/validation.ts
index 15ac08d51..1fc244b5c 100644
--- a/examples/react/todo/src/db/validation.ts
+++ b/examples/react/todo/src/db/validation.ts
@@ -25,6 +25,9 @@ export type UpdateConfig = z.infer<typeof updateConfigSchema>
 
 // Validation functions
 export const validateInsertTodo = (data: unknown): InsertTodo => {
+  if (data.text === `really hard todo`) {
+    throw new Error(`we don't want to do really hard todos`)
+  }
   return insertTodoSchema.parse(data)
 }
 
diff --git a/package.json b/package.json
index 0a58c3648..7f3698f11 100644
--- a/package.json
+++ b/package.json
@@ -63,7 +63,8 @@
     "changeset": "changeset",
     "changeset:publish": "changeset publish",
     "changeset:version": "changeset version && pnpm install --no-frozen-lockfile",
-    "lint": "eslint . --fix",
+    "lint": "pnpm --filter \"./packages/**\" lint",
+    "lint-all": "eslint . --fix",
     "prepack": "vite build",
     "prepare": "husky",
     "test": "pnpm --filter \"./packages/**\" test"
diff --git a/packages/optimistic/README.md b/packages/optimistic/README.md
index 6fc680019..f505bff4a 100644
--- a/packages/optimistic/README.md
+++ b/packages/optimistic/README.md
@@ -128,10 +128,22 @@ const todoCollection = createCollection({
 
 ## Transaction Management
 
-The library includes a robust transaction management system:
+The library includes a simple yet powerful transaction management system. Transactions are created using the `createTransaction` function:
 
-- `TransactionManager`: Handles transaction lifecycle, persistence, and retry logic
-- `TransactionStore`: Provides persistent storage for transactions using IndexedDB
+```typescript
+const tx = createTransaction({
+  mutationFn: async ({ transaction }) => {
+    // Implement your mutation logic here
+    // This function is called when the transaction is committed
+  },
+})
+
+// Apply mutations within the transaction
+tx.mutate(() => {
+  // All collection operations (insert/update/delete) within this callback
+  // will be part of this transaction
+})
+```
 
 Transactions progress through several states:
 
diff --git a/packages/optimistic/package.json b/packages/optimistic/package.json
index 8975b5d4f..71a763639 100644
--- a/packages/optimistic/package.json
+++ b/packages/optimistic/package.json
@@ -48,6 +48,7 @@
   "scripts": {
     "build": "vite build",
     "dev": "vite build --watch",
+    "lint": "eslint . --fix",
     "test": "npx vitest --run"
   },
   "sideEffects": false,
diff --git a/packages/optimistic/src/TransactionManager.ts b/packages/optimistic/src/TransactionManager.ts
deleted file mode 100644
index 00bea38e9..000000000
--- a/packages/optimistic/src/TransactionManager.ts
+++ /dev/null
@@ -1,285 +0,0 @@
-import { Store } from "@tanstack/store"
-import { SortedMap } from "./SortedMap"
-import { createDeferred } from "./deferred"
-import type { Collection } from "./collection"
-import type { PendingMutation, Transaction, TransactionState } from "./types"
-
-// Singleton instance of TransactionManager with type map
-
-const transactionManagerInstances = new Map<string, TransactionManager<any>>()
-
-/**
- * Get the global TransactionManager instance for a specific type
- * Creates a new instance if one doesn't exist for that type
- *
- * @param collection - The collection this manager is associated with
- * @returns The TransactionManager instance
- */
-export function getTransactionManager<
-  T extends object = Record<string, unknown>,
->(collection?: Collection<T>): TransactionManager<T> {
-  if (!collection) {
-    throw new Error(
-      `TransactionManager not initialized. You must provide its collection on the first call.`
-    )
-  }
-
-  if (!transactionManagerInstances.has(collection.id)) {
-    transactionManagerInstances.set(
-      collection.id,
-      new TransactionManager(collection)
-    )
-  }
-  return transactionManagerInstances.get(collection.id) as TransactionManager<T>
-}
-
-export class TransactionManager<T extends object = Record<string, unknown>> {
-  private collection: Collection<T>
-  public transactions: Store<SortedMap<string, Transaction>>
-
-  /**
-   * Creates a new TransactionManager instance
-   *
-   * @param collection - The collection this manager is associated with
-   */
-  constructor(collection: Collection<T>) {
-    this.collection = collection
-    // Initialize store with SortedMap that sorts by createdAt
-    this.transactions = new Store(
-      new SortedMap<string, Transaction>(
-        (a, b) => a.createdAt.getTime() - b.createdAt.getTime()
-      )
-    )
-  }
-
-  /**
-   * Retrieves a transaction by its ID
-   *
-   * @param id - The unique identifier of the transaction
-   * @returns The transaction if found, undefined otherwise
-   */
-  getTransaction(id: string): Transaction | undefined {
-    const transaction = this.transactions.state.get(id)
-
-    return transaction
-  }
-
-  private setTransaction(transaction: Transaction): void {
-    this.transactions.setState((sortedMap) => {
-      sortedMap.set(transaction.id, transaction)
-      return sortedMap
-    })
-
-    this.collection.commitPendingTransactions()
-  }
-
-  /**
-   * Create a live transaction reference that always returns the latest values
-   * @param id Transaction ID
-   * @returns A proxy that always gets the latest transaction values
-   */
-  createLiveTransactionReference(id: string): Transaction {
-    const self: TransactionManager<T> = this
-    return new Proxy(
-      {
-        // Implement the toObject method directly on the proxy target
-        toObject() {
-          const transaction = self.getTransaction(id)
-          if (!transaction) {
-            throw new Error(`Transaction with id ${id} not found`)
-          }
-
-          // Create a shallow copy of the transaction without the toObject method
-
-          const { toObject, ...transactionData } = transaction
-          return { ...transactionData }
-        },
-      } as Transaction,
-      {
-        get(target, prop) {
-          // If the property is toObject, return the method from the target
-          if (prop === `toObject`) {
-            return target.toObject
-          }
-
-          // Otherwise, get the latest transaction data
-          const latest = self.getTransaction(id)
-          if (!latest) {
-            throw new Error(`Transaction with id ${id} not found`)
-          }
-          return latest[prop as keyof Transaction]
-        },
-        set() {
-          // We don't allow direct setting of properties on the transaction
-          console.warn(
-            `Direct modification of transaction properties is not allowed. Use setTransactionState if updating the state`
-          )
-          return true
-        },
-      }
-    )
-  }
-
-  /**
-   * Applies mutations to the current transaction. A given transaction accumulates mutations
-   * within a single event loop.
-   *
-   * @param mutations - Array of pending mutations to apply
-   * @returns A live reference to the created or updated transaction
-   */
-  applyTransaction(mutations: Array<PendingMutation>): Transaction {
-    // See if there's an existing transaction with overlapping queued mutation.
-    const mutationKeys = mutations.map((m) => m.key)
-    let transaction: Transaction | undefined = Array.from(
-      this.transactions.state.values()
-    ).filter(
-      (t) =>
-        t.state === `pending` &&
-        t.mutations.some((m) => mutationKeys.includes(m.key))
-    )[0]
-
-    // If there's a transaction, overwrite matching mutations.
-    if (transaction) {
-      for (const newMutation of mutations) {
-        const existingIndex = transaction.mutations.findIndex(
-          (m) => m.key === newMutation.key
-        )
-
-        if (existingIndex >= 0) {
-          // Replace existing mutation
-          // TODO this won't work for cases where different mutations modify different keys
-          transaction.mutations[existingIndex] = newMutation
-        } else {
-          // Insert new mutation
-          transaction.mutations.push(newMutation)
-        }
-      }
-    } else {
-      // Create a new transaction if none exists
-      transaction = {
-        id: crypto.randomUUID(),
-        state: `pending`,
-        createdAt: new Date(),
-        updatedAt: new Date(),
-        mutations,
-        metadata: {},
-        isPersisted: createDeferred(),
-      } as Transaction
-    }
-
-    this.setTransaction(transaction)
-
-    // Start processing in the next event loop tick.
-    setTimeout(() => {
-      this.processTransaction(transaction.id)
-    }, 0)
-
-    // Return a live reference to the transaction
-    return this.createLiveTransactionReference(transaction.id)
-  }
-
-  /**
-   * Process a transaction through the mutation function
-   *
-   * @param transactionId - The ID of the transaction to process
-   * @private
-   */
-  private processTransaction(transactionId: string): void {
-    const transaction = this.getTransaction(transactionId)
-    if (!transaction) return
-
-    // Throw an error if no mutation function is provided
-    if (!this.collection.config.mutationFn) {
-      throw new Error(
-        `No mutation function provided for transaction ${transactionId}`
-      )
-    }
-
-    // Set transaction state to persisting
-    this.setTransactionState(transactionId, `persisting`)
-
-    // 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(() => {
-        const tx = this.getTransaction(transactionId)
-        if (!tx) return
-
-        // Mark as persisted
-        tx.isPersisted?.resolve(true)
-        this.setTransactionState(transactionId, `completed`)
-      })
-      .catch((error) => {
-        const tx = this.getTransaction(transactionId)
-        if (!tx) return
-
-        // Update transaction with error information
-        tx.error = {
-          message: error instanceof Error ? error.message : String(error),
-          error: error instanceof Error ? error : new Error(String(error)),
-        }
-
-        // Reject the promise
-        tx.isPersisted?.reject(tx.error.error)
-
-        // Set transaction state to failed
-        this.setTransactionState(transactionId, `failed`)
-      })
-  }
-
-  /**
-   * Updates the state of a transaction
-   *
-   * @param id - The unique identifier of the transaction
-   * @param newState - The new state to set
-   * @throws Error if the transaction is not found
-   */
-  setTransactionState(id: string, newState: TransactionState): void {
-    const transaction = this.getTransaction(id)
-    if (!transaction) {
-      throw new Error(`Transaction ${id} not found`)
-    }
-
-    // Force a small delay to ensure updatedAt is different
-    const updatedTransaction: Transaction = {
-      ...transaction,
-      state: newState,
-      updatedAt: new Date(Date.now() + 1),
-    }
-
-    this.setTransaction(updatedTransaction)
-  }
-
-  /**
-   * Gets all active transactions (those not in completed or failed state)
-   *
-   * @returns Array of active transactions
-   */
-  getActiveTransactions(): Array<Transaction> {
-    return Array.from(this.transactions.state.values()).filter(
-      (tx) => tx.state !== `completed` && tx.state !== `failed`
-    )
-  }
-
-  /**
-   * Checks if two sets of mutations have overlapping keys
-   *
-   * @param mutations1 - First set of mutations
-   * @param mutations2 - Second set of mutations
-   * @returns True if there are overlapping mutations, false otherwise
-   */
-  hasOverlappingMutations(
-    mutations1: Array<PendingMutation>,
-    mutations2: Array<PendingMutation>
-  ): boolean {
-    const ids1 = new Set(mutations1.map((m) => m.original.id))
-    const ids2 = new Set(mutations2.map((m) => m.original.id))
-    return Array.from(ids1).some((id) => ids2.has(id))
-  }
-}
diff --git a/packages/optimistic/src/collection.ts b/packages/optimistic/src/collection.ts
index 0c38007d4..614d6788c 100644
--- a/packages/optimistic/src/collection.ts
+++ b/packages/optimistic/src/collection.ts
@@ -1,6 +1,7 @@
 import { Derived, Store, batch } from "@tanstack/store"
 import { withArrayChangeTracking, withChangeTracking } from "./proxy"
-import { getTransactionManager } from "./TransactionManager"
+import { getActiveTransaction } from "./transactions"
+import { SortedMap } from "./SortedMap"
 import type {
   ChangeMessage,
   CollectionConfig,
@@ -43,15 +44,6 @@ interface PendingSyncedTransaction<T extends object = Record<string, unknown>> {
  *   await preloadCollection({
  *     id: `users-${params.userId}`,
  *     sync: { ... },
- *     // mutationFn is optional - provide it if you need mutation capabilities
- *     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;
@@ -59,7 +51,7 @@ interface PendingSyncedTransaction<T extends object = Record<string, unknown>> {
  * ```
  *
  * @template T - The type of items in the collection
- * @param config - Configuration for the collection, including id, sync, and optional mutationFn
+ * @param config - Configuration for the collection, including id and sync
  * @returns Promise that resolves when the initial sync is finished
  */
 export function preloadCollection<T extends object = Record<string, unknown>>(
@@ -89,7 +81,6 @@ export function preloadCollection<T extends object = Record<string, unknown>>(
         new Collection<T>({
           id: config.id,
           sync: config.sync,
-          mutationFn: config.mutationFn,
           schema: config.schema,
         })
       )
@@ -154,8 +145,7 @@ export class SchemaValidationError extends Error {
 }
 
 export class Collection<T extends object = Record<string, unknown>> {
-  public transactionManager!: ReturnType<typeof getTransactionManager<T>>
-
+  public transactions: Store<SortedMap<string, Transaction>>
   public optimisticOperations: Derived<Array<OptimisticChangeMessage<T>>>
   public derivedState: Derived<Map<string, T>>
   public derivedArray: Derived<Array<T>>
@@ -195,7 +185,11 @@ export class Collection<T extends object = Record<string, unknown>> {
       throw new Error(`Collection requires a sync config`)
     }
 
-    this.transactionManager = getTransactionManager<T>(this)
+    this.transactions = new Store(
+      new SortedMap<string, Transaction>(
+        (a, b) => a.createdAt.getTime() - b.createdAt.getTime()
+      )
+    )
 
     // Copies of live mutations are stored here and removed once the transaction completes.
     this.optimisticOperations = new Derived({
@@ -225,7 +219,7 @@ export class Collection<T extends object = Record<string, unknown>> {
 
         return result
       },
-      deps: [this.transactionManager.transactions],
+      deps: [this.transactions],
     })
     this.optimisticOperations.mount()
 
@@ -373,7 +367,7 @@ export class Collection<T extends object = Record<string, unknown>> {
    */
   commitPendingTransactions = () => {
     if (
-      !Array.from(this.transactions.values()).some(
+      !Array.from(this.transactions.state.values()).some(
         ({ state }) => state === `persisting`
       )
     ) {
@@ -525,7 +519,7 @@ export class Collection<T extends object = Record<string, unknown>> {
       h = (Math.imul(31, h) + str.charCodeAt(i)) | 0
     }
 
-    return Math.abs(h).toString(36)
+    return `${this.id}/${Math.abs(h).toString(36)}`
   }
 
   /**
@@ -534,7 +528,6 @@ export class Collection<T extends object = Record<string, unknown>> {
    * @param config - Optional configuration including metadata and custom keys
    * @returns A Transaction object representing the insert operation(s)
    * @throws {SchemaValidationError} If the data fails schema validation
-   * @throws {Error} If mutationFn is not provided
    * @example
    * // Insert a single item
    * insert({ text: "Buy groceries", completed: false })
@@ -549,15 +542,13 @@ export class Collection<T extends object = Record<string, unknown>> {
    * insert({ text: "Buy groceries" }, { key: "grocery-task" })
    */
   insert = (data: T | Array<T>, config?: InsertConfig) => {
-    // Throw error if mutationFn is not provided
-    if (!this.config.mutationFn) {
-      throw new Error(
-        `Cannot use mutation operators without providing a mutationFn in the collection config`
-      )
+    const transaction = getActiveTransaction()
+    if (typeof transaction === `undefined`) {
+      throw `no transaction found when calling collection.insert`
     }
 
     const items = Array.isArray(data) ? data : [data]
-    const mutations: Array<PendingMutation> = []
+    const mutations: Array<PendingMutation<T>> = []
 
     // Handle keys - convert to array if string, or generate if not provided
     let keys: Array<string>
@@ -579,7 +570,7 @@ export class Collection<T extends object = Record<string, unknown>> {
       const validatedData = this.validateData(item, `insert`)
       const key = keys[index]!
 
-      const mutation: PendingMutation = {
+      const mutation: PendingMutation<T> = {
         mutationId: crypto.randomUUID(),
         original: {},
         modified: validatedData as Record<string, unknown>,
@@ -590,12 +581,20 @@ export class Collection<T extends object = Record<string, unknown>> {
         type: `insert`,
         createdAt: new Date(),
         updatedAt: new Date(),
+        collection: this,
       }
 
       mutations.push(mutation)
     })
 
-    return this.transactionManager.applyTransaction(mutations)
+    transaction.applyMutations(mutations)
+
+    this.transactions.setState((sortedMap) => {
+      sortedMap.set(transaction.id, transaction)
+      return sortedMap
+    })
+
+    return transaction
   }
 
   /**
@@ -605,7 +604,6 @@ export class Collection<T extends object = Record<string, unknown>> {
    * @param maybeCallback - Update callback if config was provided
    * @returns A Transaction object representing the update operation(s)
    * @throws {SchemaValidationError} If the updated data fails schema validation
-   * @throws {Error} If mutationFn is not provided
    * @example
    * // Update a single item
    * update(todo, (draft) => { draft.completed = true })
@@ -636,16 +634,15 @@ export class Collection<T extends object = Record<string, unknown>> {
     configOrCallback: ((draft: TItem | Array<TItem>) => void) | OperationConfig,
     maybeCallback?: (draft: TItem | Array<TItem>) => void
   ) {
-    // Throw error if mutationFn is not provided
-    if (!this.config.mutationFn) {
-      throw new Error(
-        `Cannot use mutation operators without providing a mutationFn in the collection config`
-      )
-    }
-
     if (typeof items === `undefined`) {
       throw new Error(`The first argument to update is missing`)
     }
+
+    const transaction = getActiveTransaction()
+    if (typeof transaction === `undefined`) {
+      throw `no transaction found when calling collection.update`
+    }
+
     const isArray = Array.isArray(items)
     const itemsArray = Array.isArray(items) ? items : [items]
     const callback =
@@ -685,7 +682,7 @@ export class Collection<T extends object = Record<string, unknown>> {
     }
 
     // Create mutations for each object that has changes
-    const mutations: Array<PendingMutation> = keys
+    const mutations: Array<PendingMutation<T>> = keys
       .map((key, index) => {
         const changes = changesArray[index]
 
@@ -714,16 +711,24 @@ export class Collection<T extends object = Record<string, unknown>> {
           type: `update`,
           createdAt: new Date(),
           updatedAt: new Date(),
+          collection: this,
         }
       })
-      .filter(Boolean) as Array<PendingMutation>
+      .filter(Boolean) as Array<PendingMutation<T>>
 
     // If no changes were made, return early
     if (mutations.length === 0) {
       throw new Error(`No changes were made to any of the objects`)
     }
 
-    return this.transactionManager.applyTransaction(mutations)
+    transaction.applyMutations(mutations)
+
+    this.transactions.setState((sortedMap) => {
+      sortedMap.set(transaction.id, transaction)
+      return sortedMap
+    })
+
+    return transaction
   }
 
   /**
@@ -731,7 +736,6 @@ export class Collection<T extends object = Record<string, unknown>> {
    * @param items - Single item/key or array of items/keys to delete
    * @param config - Optional configuration including metadata
    * @returns A Transaction object representing the delete operation(s)
-   * @throws {Error} If mutationFn is not provided
    * @example
    * // Delete a single item
    * delete(todo)
@@ -746,15 +750,13 @@ export class Collection<T extends object = Record<string, unknown>> {
     items: Array<T | string> | T | string,
     config?: OperationConfig
   ) => {
-    // Throw error if mutationFn is not provided
-    if (!this.config.mutationFn) {
-      throw new Error(
-        `Cannot use mutation operators without providing a mutationFn in the collection config`
-      )
+    const transaction = getActiveTransaction()
+    if (typeof transaction === `undefined`) {
+      throw `no transaction found when calling collection.delete`
     }
 
     const itemsArray = Array.isArray(items) ? items : [items]
-    const mutations: Array<PendingMutation> = []
+    const mutations: Array<PendingMutation<T>> = []
 
     for (const item of itemsArray) {
       let key: string
@@ -774,7 +776,7 @@ export class Collection<T extends object = Record<string, unknown>> {
         )
       }
 
-      const mutation: PendingMutation = {
+      const mutation: PendingMutation<T> = {
         mutationId: crypto.randomUUID(),
         original: (this.state.get(key) || {}) as Record<string, unknown>,
         modified: { _deleted: true },
@@ -788,6 +790,7 @@ export class Collection<T extends object = Record<string, unknown>> {
         type: `delete`,
         createdAt: new Date(),
         updatedAt: new Date(),
+        collection: this,
       }
 
       mutations.push(mutation)
@@ -801,7 +804,14 @@ export class Collection<T extends object = Record<string, unknown>> {
       }
     })
 
-    return this.transactionManager.applyTransaction(mutations)
+    transaction.applyMutations(mutations)
+
+    this.transactions.setState((sortedMap) => {
+      sortedMap.set(transaction.id, transaction)
+      return sortedMap
+    })
+
+    return transaction
   }
 
   /**
@@ -862,15 +872,6 @@ export class Collection<T extends object = Record<string, unknown>> {
     })
   }
 
-  /**
-   * Gets the current transactions in the collection
-   *
-   * @returns A SortedMap of all transactions in the collection
-   */
-  get transactions() {
-    return this.transactionManager.transactions.state
-  }
-
   /**
    * Returns the current state of the collection as an array of changes
    * @returns An array of changes
diff --git a/packages/optimistic/src/index.ts b/packages/optimistic/src/index.ts
index 6977607a1..259cb02d5 100644
--- a/packages/optimistic/src/index.ts
+++ b/packages/optimistic/src/index.ts
@@ -1,9 +1,9 @@
 // Re-export all public APIs
 export * from "./collection"
 export * from "./SortedMap"
-export * from "./TransactionManager"
+export * from "./transactions"
 export * from "./types"
 export * from "./errors"
 export * from "./utils"
 export * from "./proxy"
-export * from "./query"
+export * from "./query/index.js"
diff --git a/packages/optimistic/src/query/compiled-query.ts b/packages/optimistic/src/query/compiled-query.ts
index 6e27d6f05..219978062 100644
--- a/packages/optimistic/src/query/compiled-query.ts
+++ b/packages/optimistic/src/query/compiled-query.ts
@@ -22,7 +22,7 @@ export class CompiledQuery<TResults extends object = Record<string, unknown>> {
   private inputs: Record<string, RootStreamBuilder<any>>
   private inputCollections: Record<string, Collection<any>>
   private resultCollection: Collection<TResults>
-  private state: `compiled` | `running` | `stopped` = `compiled`
+  public state: `compiled` | `running` | `stopped` = `compiled`
   private version = 0
   private unsubscribeEffect?: () => void
 
diff --git a/packages/optimistic/src/query/index.ts b/packages/optimistic/src/query/index.ts
index f4aaec166..f9a228905 100644
--- a/packages/optimistic/src/query/index.ts
+++ b/packages/optimistic/src/query/index.ts
@@ -1,12 +1,5 @@
-export {
-  queryBuilder,
-  BaseQueryBuilder,
-  type QueryBuilder,
-  type InitialQueryBuilder,
-  type ResultFromQueryBuilder,
-  type ResultsFromContext,
-} from "./query-builder.js"
+export * from "./query-builder.js"
 export * from "./compiled-query.js"
 export * from "./pipeline-compiler.js"
-export type { Query } from "./schema.js"
-export type { Context, Schema } from "./types.js"
+export * from "./schema.js"
+export * from "./types.js"
diff --git a/packages/optimistic/src/transactions.ts b/packages/optimistic/src/transactions.ts
new file mode 100644
index 000000000..d6e83215f
--- /dev/null
+++ b/packages/optimistic/src/transactions.ts
@@ -0,0 +1,198 @@
+import { createDeferred } from "./deferred"
+import type { Deferred } from "./deferred"
+import type {
+  PendingMutation,
+  TransactionConfig,
+  TransactionState,
+} from "./types"
+
+function generateUUID() {
+  // Check if crypto.randomUUID is available (modern browsers and Node.js 15+)
+  if (
+    typeof crypto !== `undefined` &&
+    typeof crypto.randomUUID === `function`
+  ) {
+    return crypto.randomUUID()
+  }
+
+  // Fallback implementation for older environments
+  return `xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx`.replace(/[xy]/g, function (c) {
+    const r = (Math.random() * 16) | 0
+    const v = c === `x` ? r : (r & 0x3) | 0x8
+    return v.toString(16)
+  })
+}
+
+const transactions: Array<Transaction> = []
+
+export function createTransaction(config: TransactionConfig): Transaction {
+  if (typeof config.mutationFn === `undefined`) {
+    throw `mutationFn is required when creating a transaction`
+  }
+
+  let transactionId = config.id
+  if (!transactionId) {
+    transactionId = generateUUID()
+  }
+  const newTransaction = new Transaction({ ...config, id: transactionId })
+  transactions.push(newTransaction)
+
+  return newTransaction
+}
+
+let transactionStack: Array<Transaction> = []
+
+export function getActiveTransaction(): Transaction | undefined {
+  if (transactionStack.length > 0) {
+    return transactionStack.slice(-1)[0]
+  } else {
+    return undefined
+  }
+}
+
+function registerTransaction(tx: Transaction) {
+  transactionStack.push(tx)
+}
+
+function unregisterTransaction(tx: Transaction) {
+  transactionStack = transactionStack.filter((t) => t.id !== tx.id)
+}
+
+export class Transaction {
+  public id: string
+  public state: TransactionState
+  public mutationFn
+  public mutations: Array<PendingMutation<any>>
+  public isPersisted: Deferred<Transaction>
+  public autoCommit: boolean
+  public createdAt: Date
+  public metadata: Record<string, unknown>
+  public error?: {
+    message: string
+    error: Error
+  }
+
+  constructor(config: TransactionConfig) {
+    this.id = config.id!
+    this.mutationFn = config.mutationFn
+    this.state = `pending`
+    this.mutations = []
+    this.isPersisted = createDeferred()
+    this.autoCommit = config.autoCommit ?? true
+    this.createdAt = new Date()
+    this.metadata = config.metadata ?? {}
+  }
+
+  setState(newState: TransactionState) {
+    this.state = newState
+  }
+
+  mutate(callback: () => void): Transaction {
+    if (this.state !== `pending`) {
+      throw `You can no longer call .mutate() as the transaction is no longer pending`
+    }
+
+    registerTransaction(this)
+    try {
+      callback()
+    } finally {
+      unregisterTransaction(this)
+    }
+
+    if (this.autoCommit) {
+      this.commit()
+    }
+
+    return this
+  }
+
+  applyMutations(mutations: Array<PendingMutation<any>>): void {
+    for (const newMutation of mutations) {
+      const existingIndex = this.mutations.findIndex(
+        (m) => m.key === newMutation.key
+      )
+
+      if (existingIndex >= 0) {
+        // Replace existing mutation
+        this.mutations[existingIndex] = newMutation
+      } else {
+        // Insert new mutation
+        this.mutations.push(newMutation)
+      }
+    }
+  }
+
+  rollback(config?: { isSecondaryRollback?: boolean }): Transaction {
+    const isSecondaryRollback = config?.isSecondaryRollback ?? false
+    if (this.state === `completed`) {
+      throw `You can no longer call .rollback() as the transaction is already completed`
+    }
+
+    this.setState(`failed`)
+
+    // See if there's any other transactions w/ mutations on the same keys
+    // and roll them back as well.
+    if (!isSecondaryRollback) {
+      const mutationKeys = new Set()
+      this.mutations.forEach((m) => mutationKeys.add(m.key))
+      transactions.forEach(
+        (t) =>
+          t.state === `pending` &&
+          t.mutations.some((m) => mutationKeys.has(m.key)) &&
+          t.rollback({ isSecondaryRollback: true })
+      )
+    }
+
+    // Reject the promise
+    this.isPersisted.reject(this.error?.error)
+
+    this.touchCollection()
+
+    return this
+  }
+
+  // Tell collection that something has changed with the transaction
+  touchCollection(): void {
+    const hasCalled = new Set()
+    this.mutations.forEach((mutation) => {
+      if (!hasCalled.has(mutation.collection.id)) {
+        mutation.collection.transactions.setState((state) => state)
+        mutation.collection.commitPendingTransactions()
+        hasCalled.add(mutation.collection.id)
+      }
+    })
+  }
+
+  async commit(): Promise<Transaction> {
+    if (this.state !== `pending`) {
+      throw `You can no longer call .commit() as the transaction is no longer pending`
+    }
+
+    this.setState(`persisting`)
+
+    if (this.mutations.length === 0) {
+      this.setState(`completed`)
+    }
+
+    // Run mutationFn
+    try {
+      await this.mutationFn({ transaction: this })
+
+      this.setState(`completed`)
+      this.touchCollection()
+
+      this.isPersisted.resolve(this)
+    } catch (error) {
+      // Update transaction with error information
+      this.error = {
+        message: error instanceof Error ? error.message : String(error),
+        error: error instanceof Error ? error : new Error(String(error)),
+      }
+
+      // rollback the transaction
+      return this.rollback()
+    }
+
+    return this
+  }
+}
diff --git a/packages/optimistic/src/types.ts b/packages/optimistic/src/types.ts
index 4ccdfe087..f381b322b 100644
--- a/packages/optimistic/src/types.ts
+++ b/packages/optimistic/src/types.ts
@@ -1,10 +1,14 @@
-import type { Collection } from "../src/collection"
-import type { Deferred } from "../src/deferred"
+import type { Collection } from "./collection"
 import type { StandardSchemaV1 } from "@standard-schema/spec"
+import type { Transaction } from "./transactions"
 
 export type TransactionState = `pending` | `persisting` | `completed` | `failed`
 
-export interface PendingMutation {
+/**
+ * Represents a pending mutation within a transaction
+ * Contains information about the original and modified data, as well as metadata
+ */
+export interface PendingMutation<T extends object = Record<string, unknown>> {
   mutationId: string
   original: Record<string, unknown>
   modified: Record<string, unknown>
@@ -15,29 +19,29 @@ export interface PendingMutation {
   syncMetadata: Record<string, unknown>
   createdAt: Date
   updatedAt: Date
+  collection: Collection<T>
 }
 
-export interface Transaction {
-  id: string
-  state: TransactionState
-  createdAt: Date
-  updatedAt: Date
-  mutations: Array<PendingMutation>
-  metadata: Record<string, unknown>
-  isPersisted?: Deferred<boolean>
-  error?: {
-    transactionId?: string // For dependency failures
-    message: string
-    error: Error
-  }
-  /**
-   * Get a plain object representation of the transaction
-   * This is useful for creating clones or serializing the transaction
-   */
-  toObject: () => Omit<Transaction, `toObject`>
+/**
+ * Configuration options for creating a new transaction
+ */
+export type MutationFnParams = {
+  transaction: Transaction
 }
 
-export type TransactionWithoutToObject = Omit<Transaction, `toObject`>
+export type MutationFn = (params: MutationFnParams) => Promise<any>
+
+export interface TransactionConfig {
+  /** Unique identifier for the transaction */
+  id?: string
+  /* If the transaction should autocommit after a mutate call or should commit be called explicitly */
+  autoCommit?: boolean
+  mutationFn: MutationFn
+  /** Custom metadata to associate with the transaction */
+  metadata?: Record<string, unknown>
+}
+
+export type { Transaction }
 
 type Value<TExtensions = never> =
   | string
@@ -83,11 +87,6 @@ export interface OptimisticChangeMessage<
   isActive?: boolean
 }
 
-export type MutationFn<T extends object = Record<string, unknown>> = (params: {
-  transaction: Transaction
-  collection: Collection<T>
-}) => Promise<any>
-
 /**
  * The Standard Schema interface.
  * This follows the standard-schema specification: https://github.com/standard-schema/standard-schema
@@ -118,7 +117,6 @@ export interface InsertConfig {
 export interface CollectionConfig<T extends object = Record<string, unknown>> {
   id: string
   sync: SyncConfig<T>
-  mutationFn?: MutationFn<T>
   schema?: StandardSchema<T>
 }
 
diff --git a/packages/optimistic/tests/TransactionManager.test.ts b/packages/optimistic/tests/TransactionManager.test.ts
deleted file mode 100644
index 623c1d9fa..000000000
--- a/packages/optimistic/tests/TransactionManager.test.ts
+++ /dev/null
@@ -1,381 +0,0 @@
-import { beforeEach, describe, expect, it } from "vitest"
-import { TransactionManager } from "../src/TransactionManager"
-import { Collection } from "../src/collection"
-import type { PendingMutation } from "../src/types"
-
-describe(`TransactionManager`, () => {
-  let collection: Collection
-  let manager: TransactionManager
-
-  beforeEach(() => {
-    collection = new Collection({
-      id: `foo`,
-      sync: {
-        sync: () => {},
-      },
-      mutationFn: async () => {
-        await new Promise((resolve) => setTimeout(resolve, 1))
-      },
-    })
-    manager = new TransactionManager(collection)
-  })
-
-  const createMockMutation = (id: string): PendingMutation => ({
-    mutationId: id,
-    original: { id, value: `original` },
-    modified: { id, value: `modified` },
-    changes: { value: `modified` },
-    type: `insert`,
-    key: id,
-    metadata: null,
-    createdAt: new Date(),
-    updatedAt: new Date(),
-    syncMetadata: {},
-  })
-
-  describe(`Basic Transaction Management`, () => {
-    it(`should create a transaction in pending state`, () => {
-      const mutations = [createMockMutation(`test-1`)]
-      const transaction = manager.applyTransaction(mutations)
-
-      expect(transaction.id).toBeDefined()
-      expect(transaction.state).toBe(`pending`)
-      expect(transaction.mutations).toEqual(mutations)
-    })
-
-    it(`should update transaction state`, () => {
-      const mutations = [createMockMutation(`test-2`)]
-      const transaction = manager.applyTransaction(mutations)
-
-      // Add a small delay to ensure timestamps are different
-      const beforeUpdate = transaction.updatedAt
-      manager.setTransactionState(transaction.id, `pending`)
-
-      const updated = manager.getTransaction(transaction.id)
-      expect(updated?.state).toBe(`pending`)
-      expect(updated?.updatedAt.getTime()).toBeGreaterThan(
-        beforeUpdate.getTime()
-      )
-    })
-
-    it(`should throw when updating non-existent transaction`, () => {
-      expect(() =>
-        manager.setTransactionState(`non-existent`, `completed`)
-      ).toThrow(`Transaction non-existent not found`)
-    })
-  })
-
-  describe(`Ordered vs Parallel Transactions`, () => {
-    it(`should not queue transactions`, () => {
-      // Create multiple parallel transactions modifying same object
-      const tx1 = manager.applyTransaction([createMockMutation(`object-1`)])
-      const tx2 = manager.applyTransaction([createMockMutation(`object-1`)])
-      const tx3 = manager.applyTransaction([createMockMutation(`object-1`)])
-
-      // All should be in pending state and not queued
-      expect(tx1.state).toBe(`pending`)
-      expect(tx2.state).toBe(`pending`)
-      expect(tx3.state).toBe(`pending`)
-    })
-  })
-
-  describe(`Transaction Ordering`, () => {
-    it(`should maintain transactions sorted by createdAt`, async () => {
-      // Create transactions with different timestamps
-      const now = Date.now()
-      const timestamps = [now, now - 1000, now - 2000]
-
-      // Create transactions in reverse chronological order
-      await Promise.all(
-        timestamps.map((timestamp, i) => {
-          const tx = manager.applyTransaction([
-            createMockMutation(`test-${i + 1}`),
-          ])
-          // Force the createdAt time
-          const updatedTx = {
-            ...tx.toObject(),
-            createdAt: new Date(timestamp),
-          }
-
-          manager.transactions.setState((sortedMap) => {
-            // @ts-expect-error this is fine for a test
-            sortedMap.set(updatedTx.id, updatedTx)
-            return sortedMap
-          })
-          return updatedTx
-        })
-      )
-
-      // Verify transactions are returned in chronological order (oldest first)
-      const sortedTransactions = Array.from(manager.transactions.state.values())
-      expect(sortedTransactions[0]?.createdAt.getTime()).toBe(timestamps[2]) // Oldest
-      expect(sortedTransactions[1]?.createdAt.getTime()).toBe(timestamps[1])
-      expect(sortedTransactions[2]?.createdAt.getTime()).toBe(timestamps[0]) // Newest
-    })
-
-    it(`should create a new transaction when no existing transactions with overlapping keys exist`, () => {
-      // Create a new transaction
-      const mutations = [createMockMutation(`test-apply-1`)]
-      const transaction = manager.applyTransaction(mutations)
-
-      // Verify transaction was created with the expected properties
-      expect(transaction.id).toBeDefined()
-      expect(transaction.state).toBe(`pending`)
-      expect(transaction.mutations).toEqual(mutations)
-    })
-
-    it(`should overwrite mutations for the same key in existing pending transactions`, () => {
-      // Create first transaction with a mutation
-      const originalMutation = {
-        ...createMockMutation(`test-apply-2`),
-        modified: { id: `test-apply-2`, value: `original-value` },
-        changes: { value: `original-value` },
-      }
-
-      manager.applyTransaction([originalMutation])
-
-      // Create second transaction with a mutation - this should be queued behind the first.
-      const tx1 = manager.applyTransaction([originalMutation])
-      expect(tx1.mutations[0]?.modified.value).toBe(`original-value`)
-
-      // Apply a new transaction with a mutation for the same key but different value.
-      const newMutation = {
-        ...createMockMutation(`test-apply-2`),
-        modified: { id: `test-apply-2`, value: `updated-value` },
-        changes: { value: `updated-value` },
-      }
-
-      const tx2 = manager.applyTransaction([newMutation])
-
-      // Should reuse the same transaction ID
-      expect(tx2.id).toBe(tx1.id)
-
-      // Should have updated the mutation
-      expect(tx2.mutations.length).toBe(1)
-      expect(tx2.mutations[0]?.modified.value).toBe(`updated-value`)
-      expect(tx2.mutations[0]?.changes.value).toBe(`updated-value`)
-    })
-
-    it(`should add new mutations while preserving existing ones for different keys`, () => {
-      // Create first transaction with a mutation
-      const mutation1 = createMockMutation(`test-apply-3a`)
-      const tx1 = manager.applyTransaction([mutation1])
-
-      // Apply a new transaction with a mutation for a different key
-      const mutation2 = createMockMutation(`test-apply-3b`)
-      const tx2 = manager.applyTransaction([mutation2])
-
-      // Should create a new transaction since keys don't overlap
-      expect(tx2.id).not.toBe(tx1.id)
-      expect(tx2.mutations.length).toBe(1)
-      expect(tx2.mutations[0]?.key).toBe(`test-apply-3b`)
-    })
-
-    it(`should handle multiple transactions with overlapping keys`, () => {
-      // Create first transaction with mutations for keys A and B
-      const mutationA1 = {
-        ...createMockMutation(`key-A`),
-        modified: { id: `key-A`, value: `A-original` },
-        changes: { value: `A-original` },
-      }
-
-      const mutationB1 = {
-        ...createMockMutation(`key-B`),
-        modified: { id: `key-B`, value: `B-original` },
-        changes: { value: `B-original` },
-      }
-
-      // Apply an initial one so the second is queued behind it.
-      manager.applyTransaction([mutationA1, mutationB1])
-      const tx1 = manager.applyTransaction([mutationA1, mutationB1])
-
-      // Create second transaction with mutations for keys B and C
-      const mutationB2 = {
-        ...createMockMutation(`key-B`),
-        modified: { id: `key-B`, value: `B-updated` },
-        changes: { value: `B-updated` },
-      }
-
-      const mutationC1 = {
-        ...createMockMutation(`key-C`),
-        modified: { id: `key-C`, value: `C-original` },
-        changes: { value: `C-original` },
-      }
-
-      // Apply the new transaction
-      const tx2 = manager.applyTransaction([mutationB2, mutationC1])
-
-      // Should update tx1 since it has an overlapping key (B)
-      expect(tx2.id).toBe(tx1.id)
-
-      // Should have 3 mutations now (A, B-updated, C)
-      expect(tx2.mutations.length).toBe(3)
-
-      // Find each mutation by key
-      const mutationA = tx2.mutations.find((m) => m.key === `key-A`)
-      const mutationB = tx2.mutations.find((m) => m.key === `key-B`)
-      const mutationC = tx2.mutations.find((m) => m.key === `key-C`)
-
-      // Verify A is unchanged
-      expect(mutationA).toBeDefined()
-      expect(mutationA?.modified.value).toBe(`A-original`)
-
-      // Verify B is updated
-      expect(mutationB).toBeDefined()
-      expect(mutationB?.modified.value).toBe(`B-updated`)
-
-      // Verify C is added
-      expect(mutationC).toBeDefined()
-      expect(mutationC?.modified.value).toBe(`C-original`)
-    })
-
-    it(`should handle the case where mutations don't overlap at all`, () => {
-      // Create three transactions with non-overlapping mutations
-      const tx1 = manager.applyTransaction([createMockMutation(`key-1`)])
-      const tx2 = manager.applyTransaction([createMockMutation(`key-2`)])
-
-      // Apply a transaction with a new non-overlapping key
-      const tx3 = manager.applyTransaction([createMockMutation(`key-3`)])
-
-      // Should be a new transaction
-      expect(tx3.id).not.toBe(tx1.id)
-      expect(tx3.id).not.toBe(tx2.id)
-      expect(tx3.mutations.length).toBe(1)
-      expect(tx3.mutations[0]?.key).toBe(`key-3`)
-
-      // Original transactions should be unchanged
-      const updatedTx1 = manager.getTransaction(tx1.id)
-      const updatedTx2 = manager.getTransaction(tx2.id)
-
-      expect(updatedTx1?.mutations.length).toBe(1)
-      expect(updatedTx1?.mutations[0]?.key).toBe(`key-1`)
-
-      expect(updatedTx2?.mutations.length).toBe(1)
-      expect(updatedTx2?.mutations[0]?.key).toBe(`key-2`)
-    })
-
-    it(`should only consider active transactions for applying updates`, () => {
-      // Create a transaction and mark it as completed
-      const mutation1 = createMockMutation(`completed-key`)
-      const tx1 = manager.applyTransaction([mutation1])
-      manager.setTransactionState(tx1.id, `completed`)
-
-      // Apply a new transaction with the same key
-      const mutation2 = {
-        ...createMockMutation(`completed-key`),
-        modified: { id: `completed-key`, value: `new-value` },
-        changes: { value: `new-value` },
-      }
-
-      const tx2 = manager.applyTransaction([mutation2])
-
-      // Should create a new transaction since the existing one is completed
-      expect(tx2.id).not.toBe(tx1.id)
-      expect(tx2.mutations.length).toBe(1)
-      expect(tx2.mutations[0]?.modified.value).toBe(`new-value`)
-    })
-  })
-
-  describe(`Error Handling`, () => {
-    it(`should reject isPersisted persist fails`, async () => {
-      // Create a collection with a persist function that throws an error
-      const errorCollection = new Collection({
-        id: `foo`,
-        sync: {
-          sync: () => {},
-        },
-        // eslint-disable-next-line @typescript-eslint/require-await
-        mutationFn: async () => {
-          throw new Error(`Persist error`)
-        },
-      })
-      const errorManager = new TransactionManager(errorCollection)
-
-      // Apply a transaction
-      const mutations = [createMockMutation(`error-test-5`)]
-      const transaction = errorManager.applyTransaction(mutations)
-
-      await expect(transaction.isPersisted?.promise).rejects.toThrow(
-        `Persist error`
-      )
-
-      // Verify the transaction state
-      expect(transaction.state).toBe(`failed`)
-      expect(transaction.error?.message).toBe(`Persist error`)
-    })
-
-    it(`should handle non-Error objects thrown during persist`, async () => {
-      // Create a collection with a persist function that throws a non-Error object
-      const nonErrorCollection = new Collection({
-        id: `non-error-object`,
-        sync: {
-          sync: () => {},
-        },
-        // eslint-disable-next-line @typescript-eslint/require-await
-        mutationFn: async () => {
-          // Throw a string instead of an Error object
-          throw `String error message`
-        },
-      })
-      const nonErrorManager = new TransactionManager(nonErrorCollection)
-
-      // Apply a transaction
-      const mutations = [createMockMutation(`non-error-test`)]
-      const transaction = nonErrorManager.applyTransaction(mutations)
-
-      // The promise should reject with a converted Error
-      await expect(transaction.isPersisted?.promise).rejects.toThrow(
-        `String error message`
-      )
-      transaction.isPersisted?.promise.catch(() => {})
-
-      // Verify the transaction state and error handling
-      expect(transaction.state).toBe(`failed`)
-      expect(transaction.error?.message).toBe(`String error message`)
-      expect(transaction.error?.error).toBeInstanceOf(Error)
-    })
-
-    // TODO figure out why this isn't working.
-    // it(`should handle non-Error objects thrown during awaitSync`, async () => {
-    //   // Create a collection with an awaitSync function that throws a non-Error object
-    //   const nonErrorSyncCollection = new Collection({
-    //     id: `non-error-sync-object`,
-    //     sync: {
-    //       sync: () => {},
-    //     },
-    //     mutationFn: {
-    //       persist: () => {
-    //         return Promise.resolve({ success: true })
-    //       },
-    //       awaitSync: () => {
-    //         // Throw a number instead of an Error object
-    //         throw 123
-    //       },
-    //     },
-    //   })
-    //   const nonErrorSyncManager = new TransactionManager(
-    //     nonErrorSyncCollection
-    //   )
-    //
-    //   // Apply a transaction
-    //   const mutations = [createMockMutation(`non-error-sync-test`)]
-    //   const transaction = nonErrorSyncManager.applyTransaction(
-    //     mutations,
-    //
-    //   )
-    //
-    //   // The promise should reject with a converted Error
-    //   // await expect(transaction.isPersisted?.promise).rejects.toThrow(`123`)
-    //   try {
-    //     await transaction.isPersisted?.promise
-    //   } catch (e) {
-    //     console.log(e)
-    //   }
-    //
-    //   // Verify the transaction state and error handling
-    //   expect(transaction.state).toBe(`failed`)
-    //   expect(transaction.error?.message).toBe(`123`)
-    //   expect(transaction.error?.error).toBeInstanceOf(Error)
-    // })
-  })
-})
diff --git a/packages/optimistic/tests/collection-subscribe-changes.test.ts b/packages/optimistic/tests/collection-subscribe-changes.test.ts
index 7f3b2cd3b..59245e893 100644
--- a/packages/optimistic/tests/collection-subscribe-changes.test.ts
+++ b/packages/optimistic/tests/collection-subscribe-changes.test.ts
@@ -1,10 +1,13 @@
 import { describe, expect, it, vi } from "vitest"
 import mitt from "mitt"
 import { Collection } from "../src/collection"
+import { createTransaction } from "../src/transactions"
 import type {
   ChangeMessage,
   ChangesPayload,
   PendingMutation,
+  Transaction,
+  TransactionConfig,
 } from "../src/types"
 
 // Helper function to wait for changes to be processed
@@ -34,7 +37,6 @@ describe(`Collection.subscribeChanges`, () => {
           commit()
         },
       },
-      mutationFn: async () => {},
     })
 
     // Wait for initial sync to complete
@@ -90,7 +92,6 @@ describe(`Collection.subscribeChanges`, () => {
           commit()
         },
       },
-      mutationFn: async () => {},
     })
 
     // Subscribe to changes
@@ -207,12 +208,13 @@ describe(`Collection.subscribeChanges`, () => {
           })
         },
       },
-      mutationFn: async ({ transaction }) => {
-        emitter.emit(`sync`, transaction.mutations)
-        return Promise.resolve()
-      },
     })
 
+    const mutationFn = async ({ transaction }) => {
+      emitter.emit(`sync`, transaction.mutations)
+      return Promise.resolve()
+    }
+
     // Subscribe to changes
     const unsubscribe = collection.subscribeChanges(callback)
 
@@ -220,7 +222,13 @@ describe(`Collection.subscribeChanges`, () => {
     callback.mockReset()
 
     // Perform optimistic insert
-    collection.insert({ value: `optimistic value` }, { key: `optimisticItem` })
+    const tx = createTransaction({ mutationFn })
+    tx.mutate(() =>
+      collection.insert(
+        { value: `optimistic value` },
+        { key: `optimisticItem` }
+      )
+    )
 
     // Verify that insert was emitted immediately (optimistically)
     expect(callback).toHaveBeenCalledTimes(1)
@@ -249,10 +257,13 @@ describe(`Collection.subscribeChanges`, () => {
     if (!item) {
       throw new Error(`Item not found`)
     }
-    collection.update(item, (draft) => {
-      draft.value = `updated optimistic value`
-      draft.updated = true
-    })
+    const updateTx = createTransaction({ mutationFn })
+    updateTx.mutate(() =>
+      collection.update(item, (draft) => {
+        draft.value = `updated optimistic value`
+        draft.updated = true
+      })
+    )
 
     // Verify that update was emitted
     expect(callback).toHaveBeenCalledTimes(1)
@@ -278,7 +289,8 @@ describe(`Collection.subscribeChanges`, () => {
     callback.mockReset()
 
     // Perform optimistic delete
-    collection.delete(`optimisticItem`)
+    const deleteTx = createTransaction({ mutationFn })
+    deleteTx.mutate(() => collection.delete(`optimisticItem`))
 
     // Verify that delete was emitted
     expect(callback).toHaveBeenCalledTimes(1)
@@ -327,12 +339,13 @@ describe(`Collection.subscribeChanges`, () => {
           commit()
         },
       },
-      mutationFn: async ({ transaction }) => {
-        emitter.emit(`sync`, transaction.mutations)
-        return Promise.resolve()
-      },
     })
 
+    const mutationFn = async ({ transaction }) => {
+      emitter.emit(`sync`, transaction.mutations)
+      return Promise.resolve()
+    }
+
     // Subscribe to changes
     const unsubscribe = collection.subscribeChanges(callback)
 
@@ -363,9 +376,12 @@ describe(`Collection.subscribeChanges`, () => {
     callback.mockReset()
 
     // Now add an optimistic item
-    const tx = collection.insert(
-      { value: `optimistic value` },
-      { key: `optimisticItem` }
+    const tx = createTransaction({ mutationFn })
+    tx.mutate(() =>
+      collection.insert(
+        { value: `optimistic value` },
+        { key: `optimisticItem` }
+      )
     )
 
     // Verify optimistic insert was emitted - this is the synchronous optimistic update
@@ -380,13 +396,12 @@ describe(`Collection.subscribeChanges`, () => {
     ])
     callback.mockReset()
 
-    await tx.isPersisted?.promise
+    await tx.isPersisted.promise
 
     // Verify synced update was emitted
-    expect(callback).toHaveBeenCalledTimes(2)
-    // This is called 3 times:
-    // 1. Set transaction state to persisting
-    // 2. The sync operation arrives and is applied to the state
+    expect(callback).toHaveBeenCalledTimes(1)
+    // This is called 1 time when the mutationFn call returns
+    // and the optimistic state is dropped and the synced state applied.
     callback.mockReset()
 
     // Update both items in optimistic and synced ways
@@ -394,9 +409,12 @@ describe(`Collection.subscribeChanges`, () => {
     const optItem = collection.state.get(`optimisticItem`)
     let updateTx
     if (optItem) {
-      updateTx = collection.update(optItem, (draft) => {
-        draft.value = `updated optimistic value`
-      })
+      updateTx = createTransaction({ mutationFn })
+      updateTx.mutate(() =>
+        collection.update(optItem, (draft) => {
+          draft.value = `updated optimistic value`
+        })
+      )
     }
 
     // We don't await here as the optimistic update is sync
@@ -417,14 +435,12 @@ describe(`Collection.subscribeChanges`, () => {
     ])
     callback.mockReset()
 
-    await updateTx?.isPersisted?.promise
+    await updateTx?.isPersisted.promise
 
     // Verify synced update was emitted
-    expect(callback).toHaveBeenCalledTimes(2)
-
-    // This is called 3 times:
-    // 1. Set transaction state to persisting
-    // 2. The sync operation arrives and is applied to the state
+    expect(callback).toHaveBeenCalledTimes(1)
+    // This is called 1 time when the mutationFn call returns
+    // and the optimistic state is dropped and the synced state applied.
     callback.mockReset()
 
     // Then update the synced item with a synced update
@@ -496,11 +512,11 @@ describe(`Collection.subscribeChanges`, () => {
           })
         },
       },
-      mutationFn: async ({ transaction }) => {
-        emitter.emit(`sync`, transaction.mutations)
-        return Promise.resolve()
-      },
     })
+    const mutationFn = async ({ transaction }) => {
+      emitter.emit(`sync`, transaction.mutations)
+      return Promise.resolve()
+    }
 
     // Subscribe to changes
     const unsubscribe = collection.subscribeChanges(callback)
@@ -516,11 +532,14 @@ describe(`Collection.subscribeChanges`, () => {
     callback.mockReset()
 
     // Insert multiple items at once
-    collection.insert([
-      { value: `batch1` },
-      { value: `batch2` },
-      { value: `batch3` },
-    ])
+    const tx1 = createTransaction({ mutationFn })
+    tx1.mutate(() =>
+      collection.insert([
+        { value: `batch1` },
+        { value: `batch2` },
+        { value: `batch3` },
+      ])
+    )
 
     // Verify only the 3 new items were emitted, not the existing ones
     expect(callback).toHaveBeenCalledTimes(1)
@@ -537,10 +556,10 @@ describe(`Collection.subscribeChanges`, () => {
     await waitForChanges()
 
     // Verify synced update was emitted
-    expect(callback).toHaveBeenCalledTimes(2)
-    // This is called 3 times:
-    // 1. Set transaction state to persisting
-    // 2. The sync operation arrives and is applied to the state
+    expect(callback).toHaveBeenCalledTimes(1)
+    // This is called when the mutationFn returns and
+    // the optimistic state is dropped and synced state is
+    // applied.
     callback.mockReset()
 
     // Update one item only
@@ -548,9 +567,12 @@ describe(`Collection.subscribeChanges`, () => {
     if (!itemToUpdate) {
       throw new Error(`Item not found`)
     }
-    collection.update(itemToUpdate, (draft) => {
-      draft.value = `updated value`
-    })
+    const tx2 = createTransaction({ mutationFn })
+    tx2.mutate(() =>
+      collection.update(itemToUpdate, (draft) => {
+        draft.value = `updated value`
+      })
+    )
 
     // Verify only the updated item was emitted
     expect(callback).toHaveBeenCalledTimes(1)
@@ -582,8 +604,8 @@ describe(`Collection.subscribeChanges`, () => {
           commit()
         },
       },
-      mutationFn: async () => {},
     })
+    const mutationFn = async () => {}
 
     // Subscribe to changes
     const unsubscribe = collection.subscribeChanges(callback)
@@ -598,7 +620,8 @@ describe(`Collection.subscribeChanges`, () => {
     unsubscribe()
 
     // Insert an item
-    collection.insert({ value: `test value` })
+    const tx = createTransaction({ mutationFn })
+    tx.mutate(() => collection.insert({ value: `test value` }))
 
     // Callback shouldn't be called after unsubscribe
     expect(callback).not.toHaveBeenCalled()
diff --git a/packages/optimistic/tests/collection.test.ts b/packages/optimistic/tests/collection.test.ts
index f797dca9b..2d3f69171 100644
--- a/packages/optimistic/tests/collection.test.ts
+++ b/packages/optimistic/tests/collection.test.ts
@@ -2,8 +2,10 @@ import { describe, expect, it, vi } from "vitest"
 import mitt from "mitt"
 import { z } from "zod"
 import { Collection, SchemaValidationError } from "../src/collection"
+import { createTransaction } from "../src/transactions"
 import type {
   ChangeMessage,
+  MutationFn,
   OptimisticChangeMessage,
   PendingMutation,
 } from "../src/types"
@@ -13,21 +15,10 @@ describe(`Collection`, () => {
     expect(() => new Collection()).toThrow(`Collection requires a sync config`)
   })
 
-  it(`should allow creating a collection without a mutationFn`, () => {
-    // This should not throw an error
-    const collection = new Collection({
-      id: `foo`,
-      sync: { sync: async () => {} },
-    })
-
-    // Verify that the collection was created successfully
-    expect(collection).toBeInstanceOf(Collection)
-  })
-
-  it(`should throw an error when trying to use mutation operations without a mutationFn`, async () => {
+  it(`should throw an error when trying to use mutation operations outside of a transaction`, async () => {
     // Create a collection with sync but no mutationFn
     const collection = new Collection<{ value: string }>({
-      id: `no-mutation-fn`,
+      id: `foo`,
       sync: {
         sync: ({ begin, write, commit }) => {
           // Immediately execute the sync cycle
@@ -51,25 +42,19 @@ describe(`Collection`, () => {
     // Verify that insert throws an error
     expect(() => {
       collection.insert({ value: `new value` }, { key: `new-key` })
-    }).toThrow(
-      `Cannot use mutation operators without providing a mutationFn in the collection config`
-    )
+    }).toThrow(`no transaction found when calling collection.insert`)
 
     // Verify that update throws an error
     expect(() => {
       collection.update(collection.state.get(`initial`)!, (draft) => {
         draft.value = `updated value`
       })
-    }).toThrow(
-      `Cannot use mutation operators without providing a mutationFn in the collection config`
-    )
+    }).toThrow(`no transaction found when calling collection.update`)
 
     // Verify that delete throws an error
     expect(() => {
       collection.delete(`initial`)
-    }).toThrow(
-      `Cannot use mutation operators without providing a mutationFn in the collection config`
-    )
+    }).toThrow(`no transaction found when calling collection.delete`)
   })
 
   it(`It shouldn't expose any state until the initial sync is finished`, () => {
@@ -107,7 +92,6 @@ describe(`Collection`, () => {
           expect(collection.state).toEqual(expectedData)
         },
       },
-      mutationFn: async () => {},
     })
   })
 
@@ -137,38 +121,42 @@ describe(`Collection`, () => {
           })
         },
       },
-      mutationFn: ({ transaction }) => {
-        // Redact time-based and random fields
-        const redactedTransaction = {
-          ...transaction.toObject(),
-          mutations: {
-            ...transaction.mutations.map((mutation) => {
-              return {
-                ...mutation,
-                createdAt: `[REDACTED]`,
-                updatedAt: `[REDACTED]`,
-                mutationId: `[REDACTED]`,
-              }
-            }),
-          },
-        }
+    })
+
+    const mutationFn: MutationFn = ({ transaction }) => {
+      // Redact time-based and random fields
+      const redactedTransaction = {
+        ...transaction,
+        mutations: {
+          ...transaction.mutations.map((mutation) => {
+            return {
+              ...mutation,
+              createdAt: `[REDACTED]`,
+              updatedAt: `[REDACTED]`,
+              mutationId: `[REDACTED]`,
+            }
+          }),
+        },
+      }
 
-        // Call the mock function with the redacted transaction
-        persistMock({ transaction: redactedTransaction })
+      // Call the mock function with the redacted transaction
+      persistMock({ transaction: redactedTransaction })
 
-        // Call the mock function with the transaction
-        syncMock({ transaction })
+      // Call the mock function with the transaction
+      syncMock({ transaction })
 
-        emitter.emit(`sync`, transaction.mutations)
-        return Promise.resolve()
-      },
-    })
+      emitter.emit(`sync`, transaction.mutations)
+      return Promise.resolve()
+    }
 
     // Test insert with auto-generated key
     const data = { value: `bar` }
-    const transaction = collection.insert(data)
+    // TODO create transaction manually with the above mutationFn & get assertions passing.
+    const tx = createTransaction({ mutationFn })
+    tx.mutate(() => collection.insert(data))
+
     // @ts-expect-error possibly undefined is ok in test
-    const insertedKey = transaction.mutations[0].key
+    const insertedKey = tx.mutations[0].key
 
     // The merged value should immediately contain the new insert
     expect(collection.state).toEqual(new Map([[insertedKey, { value: `bar` }]]))
@@ -176,7 +164,7 @@ describe(`Collection`, () => {
     // check there's a transaction in peristing state
     expect(
       // @ts-expect-error possibly undefined is ok in test
-      Array.from(collection.transactions.values())[0].mutations[0].changes
+      tx.mutations[0].changes
     ).toEqual({
       value: `bar`,
     })
@@ -190,8 +178,6 @@ describe(`Collection`, () => {
     }
     expect(collection.optimisticOperations.state[0]).toEqual(insertOperation)
 
-    await transaction.isPersisted?.promise
-
     // Check persist data (moved outside the persist callback)
     // @ts-expect-error possibly undefined is ok in test
     const persistData = persistMock.mock.calls[0][0]
@@ -204,7 +190,7 @@ describe(`Collection`, () => {
       value: `bar`,
     })
 
-    await transaction.isPersisted?.promise
+    await tx.isPersisted.promise
 
     // @ts-expect-error possibly undefined is ok in test
     const syncData = syncMock.mock.calls[0][0]
@@ -219,43 +205,54 @@ describe(`Collection`, () => {
     // optimistic update is gone & synced data & comibned state are all updated.
     expect(
       // @ts-expect-error possibly undefined is ok in test
-      Array.from(collection.transactions.values())[0].state
+      Array.from(collection.transactions.state.values())[0].state
     ).toMatchInlineSnapshot(`"completed"`)
+    expect(collection.state).toEqual(new Map([[insertedKey, { value: `bar` }]]))
     expect(
       collection.optimisticOperations.state.filter((o) => o.isActive)
     ).toEqual([])
-    expect(collection.state).toEqual(new Map([[insertedKey, { value: `bar` }]]))
 
     // Test insert with provided key
-    collection.insert({ value: `baz` }, { key: `custom-key` })
+    const tx2 = createTransaction({ mutationFn })
+    tx2.mutate(() => collection.insert({ value: `baz` }, { key: `custom-key` }))
     expect(collection.state.get(`custom-key`)).toEqual({ value: `baz` })
+    await tx2.isPersisted.promise
 
     // Test bulk insert
+    const tx3 = createTransaction({ mutationFn })
     const bulkData = [{ value: `item1` }, { value: `item2` }]
-    collection.insert(bulkData)
+    tx3.mutate(() => collection.insert(bulkData))
     const keys = Array.from(collection.state.keys())
     // @ts-expect-error possibly undefined is ok in test
     expect(collection.state.get(keys[2])).toEqual(bulkData[0])
     // @ts-expect-error possibly undefined is ok in test
     expect(collection.state.get(keys[3])).toEqual(bulkData[1])
+    await tx3.isPersisted.promise
 
+    const tx4 = createTransaction({ mutationFn })
     // Test update with callback
-    collection.update([collection.state.get(insertedKey)!], (item) => {
-      // @ts-expect-error possibly undefined is ok in test
-      item[0].value = `bar2`
-    })
+    tx4.mutate(() =>
+      collection.update([collection.state.get(insertedKey)!], (item) => {
+        // @ts-expect-error possibly undefined is ok in test
+        item[0].value = `bar2`
+      })
+    )
 
     // The merged value should contain the update.
     expect(collection.state.get(insertedKey)).toEqual({ value: `bar2` })
+    await tx4.isPersisted.promise
 
+    const tx5 = createTransaction({ mutationFn })
     // Test update with config and callback
-    collection.update(
-      collection.state.get(insertedKey),
-      { metadata: { updated: true } },
-      (item) => {
-        item.value = `bar3`
-        item.newProp = `new value`
-      }
+    tx5.mutate(() =>
+      collection.update(
+        collection.state.get(insertedKey)!,
+        { metadata: { updated: true } },
+        (item) => {
+          item.value = `bar3`
+          item.newProp = `new value`
+        }
+      )
     )
 
     // The merged value should contain the update
@@ -264,6 +261,40 @@ describe(`Collection`, () => {
       newProp: `new value`,
     })
 
+    await tx5.isPersisted.promise
+
+    // If there are two updates, the second should overwrite the first.
+    const tx55 = createTransaction({ mutationFn })
+    // Test update with config and callback
+    tx55.mutate(() => {
+      collection.update(
+        collection.state.get(insertedKey)!,
+        { metadata: { updated: true } },
+        (item) => {
+          item.value = `bar3.1`
+          item.newProp = `new value.1`
+        }
+      )
+      collection.update(
+        collection.state.get(insertedKey)!,
+        { metadata: { updated: true } },
+        (item) => {
+          item.value = `bar3`
+          item.newProp = `new value`
+        }
+      )
+    })
+
+    // The merged value should contain the update
+    expect(collection.state.get(insertedKey)).toEqual({
+      value: `bar3`,
+      newProp: `new value`,
+    })
+    expect(tx55.mutations).toHaveLength(1)
+
+    await tx55.isPersisted.promise
+
+    const tx6 = createTransaction({ mutationFn })
     // Test bulk update
     const items = [
       // @ts-expect-error possibly undefined is ok in test
@@ -271,41 +302,54 @@ describe(`Collection`, () => {
       // @ts-expect-error possibly undefined is ok in test
       collection.state.get(keys[3])!,
     ]
-    collection.update(items, { metadata: { bulkUpdate: true } }, (drafts) => {
-      drafts.forEach((draft) => {
-        draft.value += `-updated`
+    tx6.mutate(() =>
+      collection.update(items, { metadata: { bulkUpdate: true } }, (drafts) => {
+        drafts.forEach((draft) => {
+          draft.value += `-updated`
+        })
       })
-    })
+    )
 
     // Check bulk updates
     // @ts-expect-error possibly undefined is ok in test
     expect(collection.state.get(keys[2])).toEqual({ value: `item1-updated` })
     // @ts-expect-error possibly undefined is ok in test
     expect(collection.state.get(keys[3])).toEqual({ value: `item2-updated` })
+    await tx6.isPersisted.promise
 
+    const tx7 = createTransaction({ mutationFn })
     const toBeDeleted = collection.state.get(insertedKey)!
     // Test delete single item
-    collection.delete(toBeDeleted)
+    tx7.mutate(() => collection.delete(toBeDeleted))
     expect(collection.state.has(insertedKey)).toBe(false)
     expect(collection.objectKeyMap.has(toBeDeleted)).toBe(false)
+    await tx7.isPersisted.promise
 
     // Test delete with metadata
-    collection.delete(collection.state.get(`custom-key`), {
-      metadata: { reason: `test` },
-    })
+    const tx8 = createTransaction({ mutationFn })
+    tx8.mutate(() =>
+      collection.delete(collection.state.get(`custom-key`)!, {
+        metadata: { reason: `test` },
+      })
+    )
     expect(collection.state.has(`custom-key`)).toBe(false)
+    await tx8.isPersisted.promise
 
     // Test bulk delete
-    collection.delete([
-      // @ts-expect-error possibly undefined is ok in test
-      collection.state.get(keys[2])!,
-      // @ts-expect-error possibly undefined is ok in test
-      collection.state.get(keys[3])!,
-    ])
+    const tx9 = createTransaction({ mutationFn })
+    tx9.mutate(() =>
+      collection.delete([
+        // @ts-expect-error possibly undefined is ok in test
+        collection.state.get(keys[2])!,
+        // @ts-expect-error possibly undefined is ok in test
+        collection.state.get(keys[3])!,
+      ])
+    )
     // @ts-expect-error possibly undefined is ok in test
     expect(collection.state.has(keys[2])).toBe(false)
     // @ts-expect-error possibly undefined is ok in test
     expect(collection.state.has(keys[3])).toBe(false)
+    await tx9.isPersisted.promise
   })
 
   it(`synced updates should be applied while there's an ongoing transaction`, async () => {
@@ -331,29 +375,34 @@ describe(`Collection`, () => {
           })
         },
       },
-      mutationFn: ({ transaction }) => {
-        // Sync something and check that that it isn't applied because
-        // we're still in the middle of persisting a transaction.
-        emitter.emit(`update`, [
-          { key: `the-key`, type: `insert`, changes: { bar: `value` } },
-          // This update is ignored because the optimistic update overrides it.
-          { key: `foo`, type: `update`, changes: { bar: `value2` } },
-        ])
-        expect(collection.state).toEqual(new Map([[`foo`, { value: `bar` }]]))
-        // Remove it so we don't have to assert against it below
-        emitter.emit(`update`, [{ key: `the-key`, type: `delete` }])
-
-        emitter.emit(`update`, transaction.mutations)
-        return Promise.resolve()
-      },
     })
 
+    const mutationFn: MutationFn = ({ transaction }) => {
+      // Sync something and check that that it isn't applied because
+      // we're still in the middle of persisting a transaction.
+      emitter.emit(`update`, [
+        { key: `the-key`, type: `insert`, changes: { bar: `value` } },
+        // This update is ignored because the optimistic update overrides it.
+        { key: `foo`, type: `update`, changes: { bar: `value2` } },
+      ])
+      expect(collection.state).toEqual(new Map([[`foo`, { value: `bar` }]]))
+      // Remove it so we don't have to assert against it below
+      emitter.emit(`update`, [{ key: `the-key`, type: `delete` }])
+
+      emitter.emit(`update`, transaction.mutations)
+      return Promise.resolve()
+    }
+
+    const tx1 = createTransaction({ mutationFn })
+
     // insert
-    const transaction = collection.insert(
-      {
-        value: `bar`,
-      },
-      { key: `foo` }
+    tx1.mutate(() =>
+      collection.insert(
+        {
+          value: `bar`,
+        },
+        { key: `foo` }
+      )
     )
 
     // The merged value should immediately contain the new insert
@@ -362,7 +411,7 @@ describe(`Collection`, () => {
     // check there's a transaction in peristing state
     expect(
       // @ts-expect-error possibly undefined is ok in test
-      Array.from(collection.transactions.values())[0].mutations[0].changes
+      Array.from(collection.transactions.state.values())[0].mutations[0].changes
     ).toEqual({
       value: `bar`,
     })
@@ -376,7 +425,7 @@ describe(`Collection`, () => {
     }
     expect(collection.optimisticOperations.state[0]).toEqual(insertOperation)
 
-    await transaction.isPersisted?.promise
+    await tx1.isPersisted.promise
 
     expect(collection.state).toEqual(new Map([[`foo`, { value: `bar` }]]))
   })
@@ -390,8 +439,8 @@ describe(`Collection`, () => {
           commit()
         },
       },
-      mutationFn: async () => {},
     })
+    const mutationFn = async () => {}
 
     // Insert multiple items with a sparse key array
     const items = [
@@ -401,21 +450,24 @@ describe(`Collection`, () => {
       { value: `item4` },
     ]
 
+    const tx1 = createTransaction({ mutationFn })
     // Only provide keys for first and third items
-    const transaction = collection.insert(items, {
-      key: [`key1`, undefined, `key3`],
-    })
+    tx1.mutate(() =>
+      collection.insert(items, {
+        key: [`key1`, undefined, `key3`],
+      })
+    )
 
     // Get all keys from the transaction
-    const keys = transaction.mutations.map((m) => m.key)
+    const keys = tx1.mutations.map((m) => m.key)
 
     // Verify explicit keys were used
     expect(keys[0]).toBe(`key1`)
     expect(keys[2]).toBe(`key3`)
 
     // Verify auto-generated keys for undefined positions
-    expect(keys[1]).toHaveLength(6)
-    expect(keys[3]).toHaveLength(6)
+    expect(keys[1]).toHaveLength(43)
+    expect(keys[3]).toHaveLength(43)
 
     // Verify all items were inserted with correct values
     // @ts-expect-error possibly undefined is ok in test
@@ -427,11 +479,14 @@ describe(`Collection`, () => {
     // @ts-expect-error possibly undefined is ok in test
     expect(collection.state.get(keys[3])).toEqual({ value: `item4` })
 
+    const tx2 = createTransaction({ mutationFn })
     // Test error case: more keys than items
     expect(() => {
-      collection.insert([{ value: `test` }], {
-        key: [`key1`, `key2`],
-      })
+      tx2.mutate(() =>
+        collection.insert([{ value: `test` }], {
+          key: [`key1`, `key2`],
+        })
+      )
     }).toThrow(`More keys provided than items to insert`)
   })
 
@@ -444,30 +499,38 @@ describe(`Collection`, () => {
           commit()
         },
       },
-      mutationFn: () => Promise.resolve(),
     })
 
+    const mutationFn = () => Promise.resolve()
+
     // Add an item to the collection
     const item = { name: `Test Item` }
-    collection.insert(item)
+    const tx1 = createTransaction({ mutationFn })
+    tx1.mutate(() => collection.insert(item))
 
     // Should throw when trying to delete an object not in the collection
     const notInCollection = { name: `Not In Collection` }
-    expect(() => collection.delete(notInCollection)).toThrow(
+    const tx2 = createTransaction({ mutationFn })
+    expect(() => tx2.mutate(() => collection.delete(notInCollection))).toThrow(
       `Object not found in collection`
     )
 
     // Should throw when trying to delete an invalid type
+    const tx3 = createTransaction({ mutationFn })
     // @ts-expect-error testing error handling with invalid type
-    expect(() => collection.delete(123)).toThrow(
+    expect(() => tx3.mutate(() => collection.delete(123))).toThrow(
       `Invalid item type for delete - must be an object or string key`
     )
 
     // Should not throw when deleting by string key (even if key doesn't exist)
-    expect(() => collection.delete(`non-existent-key`)).not.toThrow()
+    const tx4 = createTransaction({ mutationFn })
+    expect(() =>
+      tx4.mutate(() => collection.delete(`non-existent-key`))
+    ).not.toThrow()
 
     // Should not throw when deleting an object that exists in the collection
-    expect(() => collection.delete(item)).not.toThrow()
+    const tx5 = createTransaction({ mutationFn })
+    expect(() => tx5.mutate(() => collection.delete(item))).not.toThrow()
   })
 })
 
@@ -489,9 +552,9 @@ describe(`Collection with schema validation`, () => {
           commit()
         },
       },
-      mutationFn: async () => {},
       schema: userSchema,
     })
+    const mutationFn = async () => {}
 
     // Valid data should work
     const validUser = {
@@ -500,7 +563,8 @@ describe(`Collection with schema validation`, () => {
       email: `alice@example.com`,
     }
 
-    collection.insert(validUser, { key: `user1` })
+    const tx1 = createTransaction({ mutationFn })
+    tx1.mutate(() => collection.insert(validUser, { key: `user1` }))
 
     // Invalid data should throw SchemaValidationError
     const invalidUser = {
@@ -510,7 +574,8 @@ describe(`Collection with schema validation`, () => {
     }
 
     try {
-      collection.insert(invalidUser, { key: `user2` })
+      const tx2 = createTransaction({ mutationFn })
+      tx2.mutate(() => collection.insert(invalidUser, { key: `user2` }))
       // Should not reach here
       expect(true).toBe(false)
     } catch (error) {
@@ -532,15 +597,21 @@ describe(`Collection with schema validation`, () => {
     }
 
     // Partial updates should work with valid data
-    collection.update(collection.state.get(`user1`), (draft) => {
-      draft.age = 31
-    })
+    const tx3 = createTransaction({ mutationFn })
+    tx3.mutate(() =>
+      collection.update(collection.state.get(`user1`)!, (draft) => {
+        draft.age = 31
+      })
+    )
 
     // Partial updates should fail with invalid data
     try {
-      collection.update(collection.state.get(`user1`), (draft) => {
-        draft.age = -1
-      })
+      const tx4 = createTransaction({ mutationFn })
+      tx4.mutate(() =>
+        collection.update(collection.state.get(`user1`)!, (draft) => {
+          draft.age = -1
+        })
+      )
       // Should not reach here
       expect(true).toBe(false)
     } catch (error) {
diff --git a/packages/optimistic/tests/object-key-map.test.ts b/packages/optimistic/tests/object-key-map.test.ts
index 2fff637fa..016ddb933 100644
--- a/packages/optimistic/tests/object-key-map.test.ts
+++ b/packages/optimistic/tests/object-key-map.test.ts
@@ -1,9 +1,11 @@
 import { beforeEach, describe, expect, it } from "vitest"
 import { z } from "zod"
 import { Collection } from "../src/collection"
+import { createTransaction } from "../src/transactions"
 
 describe(`Object-Key Association`, () => {
   let collection: Collection<{ name: string; age: number }>
+  let mutationFn: () => {}
 
   beforeEach(() => {
     const schema = z.object({
@@ -17,14 +19,15 @@ describe(`Object-Key Association`, () => {
       sync: {
         sync: async () => {},
       },
-      mutationFn: async () => {},
     })
+    mutationFn = async () => {}
   })
 
   it(`should associate an object with its key after insert`, () => {
     // Insert an object
     const data = { name: `John`, age: 30 }
-    collection.insert(data, { key: `user1` })
+    const tx1 = createTransaction({ mutationFn })
+    tx1.mutate(() => collection.insert(data, { key: `user1` }))
 
     const item = collection.state.get(`user1`)
 
@@ -32,9 +35,12 @@ describe(`Object-Key Association`, () => {
     expect(item).toBeDefined()
 
     // Update using the object reference
-    collection.update(item, (draft) => {
-      draft.age = 31
-    })
+    const tx2 = createTransaction({ mutationFn })
+    tx2.mutate(() =>
+      collection.update(item, (draft) => {
+        draft.age = 31
+      })
+    )
 
     // Verify the update worked
     const updated = collection.state.get(`user1`)
@@ -46,20 +52,26 @@ describe(`Object-Key Association`, () => {
     const johnData = { name: `John`, age: 30 }
     const janeData = { name: `Jane`, age: 28 }
 
-    collection.insert([johnData, janeData], {
-      key: [`user1`, `user2`],
-    })
+    const tx1 = createTransaction({ mutationFn })
+    tx1.mutate(() =>
+      collection.insert([johnData, janeData], {
+        key: [`user1`, `user2`],
+      })
+    )
 
     const john = collection.state.get(`user1`)
     const jane = collection.state.get(`user2`)
 
     // Update multiple objects using their references
-    collection.update([john!, jane!], (items) => {
-      if (items[0] && items[1]) {
-        items[0].age = 31
-        items[1].name = `Jane Doe`
-      }
-    })
+    const tx2 = createTransaction({ mutationFn })
+    tx2.mutate(() =>
+      collection.update([john!, jane!], (items) => {
+        if (items[0] && items[1]) {
+          items[0].age = 31
+          items[1].name = `Jane Doe`
+        }
+      })
+    )
 
     // Verify updates
     expect(collection.state.get(`user1`)).toEqual({ name: `John`, age: 31 })
@@ -69,12 +81,14 @@ describe(`Object-Key Association`, () => {
   it(`should handle delete with object reference`, () => {
     // Insert an object
     const data = { name: `John`, age: 30 }
-    collection.insert(data, { key: `user1` })
+    const tx1 = createTransaction({ mutationFn })
+    tx1.mutate(() => collection.insert(data, { key: `user1` }))
 
     const john = collection.state.get(`user1`)
 
     // Delete using the object reference
-    collection.delete(john)
+    const tx2 = createTransaction({ mutationFn })
+    tx2.mutate(() => collection.delete(john))
 
     // Verify deletion
     expect(collection.state.get(`user1`)).toBeUndefined()
@@ -83,19 +97,26 @@ describe(`Object-Key Association`, () => {
   it(`should maintain object-key association after updates`, () => {
     // Insert an object
     const data = { name: `John`, age: 30 }
-    collection.insert(data, { key: `user1` })
+    const tx1 = createTransaction({ mutationFn })
+    tx1.mutate(() => collection.insert(data, { key: `user1` }))
 
     const john = collection.state.get(`user1`)
 
     // First update
-    collection.update(john, (item) => {
-      item.age = 31
-    })
+    const tx2 = createTransaction({ mutationFn })
+    tx2.mutate(() =>
+      collection.update(john, (item) => {
+        item.age = 31
+      })
+    )
 
     // Second update using the same object reference
-    collection.update(john, (item) => {
-      item.name = `John Doe`
-    })
+    const tx3 = createTransaction({ mutationFn })
+    tx3.mutate(() =>
+      collection.update(john, (item) => {
+        item.name = `John Doe`
+      })
+    )
 
     // Verify both updates worked
     const updated = collection.state.get(`user1`)
@@ -106,10 +127,13 @@ describe(`Object-Key Association`, () => {
     const unknownObject = { name: `Unknown`, age: 25 }
 
     // Try to update using an object that wasn't inserted
+    const tx = createTransaction({ mutationFn })
     expect(() => {
-      collection.update(unknownObject, (item) => {
-        item.age = 26
-      })
+      tx.mutate(() =>
+        collection.update(unknownObject, (item) => {
+          item.age = 26
+        })
+      )
     }).toThrow()
   })
 
@@ -119,9 +143,12 @@ describe(`Object-Key Association`, () => {
     const janeData = { name: `Jane`, age: 28 }
     const bobData = { name: `Bob`, age: 35 }
 
-    collection.insert([johnData, janeData, bobData], {
-      key: [`user1`, `user2`, `user3`],
-    })
+    const tx = createTransaction({ mutationFn })
+    tx.mutate(() =>
+      collection.insert([johnData, janeData, bobData], {
+        key: [`user1`, `user2`, `user3`],
+      })
+    )
 
     const john = collection.state.get(`user1`)
     const jane = collection.state.get(`user2`)
diff --git a/packages/optimistic/tests/query/function-integration.test.ts b/packages/optimistic/tests/query/function-integration.test.ts
index e35cfa503..1769c3cac 100644
--- a/packages/optimistic/tests/query/function-integration.test.ts
+++ b/packages/optimistic/tests/query/function-integration.test.ts
@@ -230,7 +230,7 @@ describe(`Query Function Integration`, () => {
       expect(results[0].id).toBe(1) // Alice
       expect(results[0].joined.getFullYear()).toBe(2023)
       expect(results[0].joined.getMonth()).toBe(0) // January (0-indexed)
-      expect(results[0].joined.getDate()).toBe(15)
+      expect(results[0].joined.getUTCDate()).toBe(15)
     })
   })
 
diff --git a/packages/optimistic/tests/query/functions.test.ts b/packages/optimistic/tests/query/functions.test.ts
index 0b265dcc8..6c32e6765 100644
--- a/packages/optimistic/tests/query/functions.test.ts
+++ b/packages/optimistic/tests/query/functions.test.ts
@@ -200,7 +200,7 @@ describe(`Query > Functions`, () => {
         expect(result).toBeInstanceOf(Date)
         expect((result as Date).getFullYear()).toBe(2023)
         expect((result as Date).getMonth()).toBe(0) // January = 0
-        expect((result as Date).getDate()).toBe(15)
+        expect((result as Date).getUTCDate()).toBe(15)
 
         // Test other date formats
         const isoResult = evaluateFunction(`DATE`, `2023-02-20T12:30:45Z`)
diff --git a/packages/optimistic/tests/transaction-types.test.ts b/packages/optimistic/tests/transaction-types.test.ts
new file mode 100644
index 000000000..8d9dc6578
--- /dev/null
+++ b/packages/optimistic/tests/transaction-types.test.ts
@@ -0,0 +1,94 @@
+import { describe, expect, it } from "vitest"
+import type {
+  MutationFactoryConfig,
+  PendingMutation,
+  Transaction,
+  TransactionConfig,
+  TransactionState,
+} from "../src/types"
+
+describe(`Transaction Types`, () => {
+  it(`should validate PendingMutation structure with collectionId`, () => {
+    // Type assertion test - this will fail at compile time if the type is incorrect
+    const pendingMutation: PendingMutation = {
+      mutationId: `test-mutation-1`,
+      original: { id: 1, name: `Original` },
+      modified: { id: 1, name: `Modified` },
+      changes: { name: `Modified` },
+      key: `1`,
+      type: `update`,
+      metadata: null,
+      syncMetadata: {},
+      createdAt: new Date(),
+      updatedAt: new Date(),
+      collectionId: `users`, // New required field
+    }
+
+    expect(pendingMutation.collectionId).toBe(`users`)
+    expect(pendingMutation.type).toBe(`update`)
+  })
+
+  it(`should validate TransactionConfig structure`, () => {
+    // Empty config is valid
+    const minimalConfig: TransactionConfig = {}
+    expect(minimalConfig).toBeDefined()
+
+    // Full config
+    const fullConfig: TransactionConfig = {
+      id: `custom-transaction-id`,
+      metadata: { source: `user-form` },
+    }
+
+    expect(fullConfig.id).toBe(`custom-transaction-id`)
+    expect(fullConfig.metadata).toEqual({ source: `user-form` })
+  })
+
+  it(`should validate Transaction structure`, () => {
+    const mockToObject = () => ({
+      id: `test-transaction`,
+      state: `pending` as TransactionState,
+      createdAt: new Date(),
+      updatedAt: new Date(),
+      mutations: [],
+      metadata: {},
+    })
+
+    const transaction: Transaction = {
+      id: `test-transaction`,
+      state: `pending`,
+      createdAt: new Date(),
+      updatedAt: new Date(),
+      mutations: [],
+      metadata: {},
+      toObject: mockToObject,
+    }
+
+    expect(transaction.id).toBe(`test-transaction`)
+
+    // Test toObject method
+    const plainObject = transaction.toObject()
+    expect(plainObject.id).toBe(`test-transaction`)
+    expect(plainObject).not.toHaveProperty(`toObject`)
+  })
+
+  it(`should validate MutationFactoryConfig structure`, () => {
+    // Minimal config with just the required mutationFn
+    const minimalConfig: MutationFactoryConfig = {
+      mutationFn: async () => {
+        return Promise.resolve({ success: true })
+      },
+    }
+
+    expect(typeof minimalConfig.mutationFn).toBe(`function`)
+
+    // Full config with metadata
+    const fullConfig: MutationFactoryConfig = {
+      mutationFn: async () => {
+        return Promise.resolve({ success: true })
+      },
+      metadata: { source: `signup-form` },
+    }
+
+    expect(fullConfig.metadata).toEqual({ source: `signup-form` })
+  })
+})
diff --git a/packages/optimistic/tests/transactions.test.ts b/packages/optimistic/tests/transactions.test.ts
new file mode 100644
index 000000000..1d3a86327
--- /dev/null
+++ b/packages/optimistic/tests/transactions.test.ts
@@ -0,0 +1,222 @@
+import { describe, expect, it } from "vitest"
+import { createTransaction } from "../src/transactions"
+import { Collection } from "../src/collection"
+
+describe(`Transactions`, () => {
+  it(`calling createTransaction creates a transaction`, () => {
+    const transaction = createTransaction({
+      mutationFn: async () => Promise.resolve(),
+      metadata: { foo: true },
+    })
+
+    expect(transaction.commit).toBeTruthy()
+    expect(transaction.metadata.foo).toBeTruthy()
+  })
+  it(`goes straight to completed if you call commit w/o any mutations`, () => {
+    const transaction = createTransaction({
+      mutationFn: async () => Promise.resolve(),
+    })
+
+    transaction.commit()
+    expect(transaction.state).toBe(`completed`)
+  })
+  it(`thows an error if you don't pass in mutationFn`, () => {
+    // @ts-expect-error missing argument on purpose
+    expect(() => createTransaction({})).toThrowError(
+      `mutationFn is required when creating a transaction`
+    )
+  })
+  it(`thows an error if call mutate or commit or rollback when it's completed`, async () => {
+    const transaction = createTransaction({
+      mutationFn: async () => Promise.resolve(),
+    })
+
+    await transaction.commit()
+
+    await expect(transaction.commit()).rejects.toThrowError(
+      `You can no longer call .commit() as the transaction is no longer pending`
+    )
+    expect(() => transaction.rollback()).toThrowError(
+      `You can no longer call .rollback() as the transaction is already completed`
+    )
+    expect(() => transaction.mutate(() => {})).toThrowError(
+      `You can no longer call .mutate() as the transaction is no longer pending`
+    )
+  })
+  it(`should allow manually controlling the transaction lifecycle`, () => {
+    const transaction = createTransaction({
+      mutationFn: async () => Promise.resolve(),
+      autoCommit: false,
+    })
+    const collection = new Collection<{ value: string; newProp?: string }>({
+      id: `foo`,
+      sync: {
+        sync: () => {},
+      },
+    })
+
+    transaction.mutate(() => {
+      collection.insert({ value: `foo-me`, newProp: `something something` })
+    })
+    transaction.mutate(() => {
+      collection.insert({ value: `foo-me2`, newProp: `something something2` })
+    })
+
+    expect(transaction.mutations).toHaveLength(2)
+
+    transaction.commit()
+  })
+  it(`should allow mutating multiple collections`, () => {
+    const transaction = createTransaction({
+      mutationFn: async () => Promise.resolve(),
+      autoCommit: false,
+    })
+    const collection1 = new Collection<{ value: string; newProp?: string }>({
+      id: `foo`,
+      sync: {
+        sync: () => {},
+      },
+    })
+    const collection2 = new Collection<{ value: string; newProp?: string }>({
+      id: `foo2`,
+      sync: {
+        sync: () => {},
+      },
+    })
+
+    transaction.mutate(() => {
+      collection1.insert({ value: `foo-me`, newProp: `something something` })
+      collection2.insert({ value: `foo-me`, newProp: `something something` })
+    })
+
+    expect(transaction.mutations).toHaveLength(2)
+
+    transaction.commit()
+  })
+  it(`should allow devs to roll back manual transactions`, () => {
+    const transaction = createTransaction({
+      mutationFn: async () => Promise.resolve(),
+      autoCommit: false,
+    })
+    const collection = new Collection<{ value: string; newProp?: string }>({
+      id: `foo`,
+      sync: {
+        sync: () => {},
+      },
+    })
+
+    transaction.mutate(() => {
+      collection.insert({ value: `foo-me`, newProp: `something something` })
+    })
+
+    transaction.rollback()
+
+    transaction.isPersisted.promise.catch(() => {})
+    expect(transaction.state).toBe(`failed`)
+  })
+  it(`should rollback if the mutationFn throws an error`, async () => {
+    const transaction = createTransaction({
+      mutationFn: async () => {
+        await Promise.resolve()
+        throw new Error(`bad`)
+      },
+      autoCommit: false,
+    })
+    const collection = new Collection<{ value: string; newProp?: string }>({
+      id: `foo`,
+      sync: {
+        sync: () => {},
+      },
+    })
+
+    transaction.mutate(() => {
+      collection.insert({ value: `foo-me`, newProp: `something something` })
+    })
+
+    transaction.commit()
+
+    await expect(transaction.isPersisted.promise).rejects.toThrow(`bad`)
+    transaction.isPersisted.promise.catch(() => {})
+    expect(transaction.state).toBe(`failed`)
+    expect(transaction.error?.message).toBe(`bad`)
+    expect(transaction.error?.error).toBeInstanceOf(Error)
+  })
+  it(`should handle string errors as well`, async () => {
+    const transaction = createTransaction({
+      mutationFn: async () => {
+        await Promise.resolve()
+        throw `bad`
+      },
+      autoCommit: false,
+    })
+    const collection = new Collection<{ value: string; newProp?: string }>({
+      id: `foo`,
+      sync: {
+        sync: () => {},
+      },
+    })
+
+    transaction.mutate(() => {
+      collection.insert({ value: `foo-me`, newProp: `something something` })
+    })
+
+    transaction.commit()
+
+    await expect(transaction.isPersisted.promise).rejects.toThrow(`bad`)
+    transaction.isPersisted.promise.catch(() => {})
+    expect(transaction.state).toBe(`failed`)
+    expect(transaction.error?.message).toBe(`bad`)
+    expect(transaction.error?.error).toBeInstanceOf(Error)
+  })
+  it(`should, when rolling back, find any other pending transactions w/ overlapping mutations and roll them back as well`, async () => {
+    const transaction1 = createTransaction({
+      mutationFn: async () => Promise.resolve(),
+      autoCommit: false,
+    })
+    const transaction2 = createTransaction({
+      mutationFn: async () => Promise.resolve(),
+      autoCommit: false,
+    })
+    const transaction3 = createTransaction({
+      mutationFn: async () => Promise.resolve(),
+      autoCommit: false,
+    })
+    const collection = new Collection<{ value: string; newProp?: string }>({
+      id: `foo`,
+      sync: {
+        sync: () => {},
+      },
+    })
+
+    transaction1.mutate(() => {
+      collection.insert({ value: `foo-me`, newProp: `something something` })
+    })
+
+    transaction2.mutate(() => {
+      collection.state.forEach((object) => {
+        collection.update(object, (draft) => {
+          draft.value = `foo-me-2`
+        })
+      })
+    })
+
+    transaction2.commit()
+    await transaction2.isPersisted.promise
+
+    transaction3.mutate(() => {
+      collection.state.forEach((object) => {
+        collection.update(object, (draft) => {
+          draft.value = `foo-me-3`
+        })
+      })
+    })
+
+    transaction1.rollback()
+    transaction1.isPersisted.promise.catch(() => {})
+    transaction3.isPersisted.promise.catch(() => {})
+
+    expect(transaction1.state).toBe(`failed`)
+    expect(transaction2.state).toBe(`completed`)
+    expect(transaction3.state).toBe(`failed`)
+  })
+})
diff --git a/packages/react-optimistic/README.md b/packages/react-optimistic/README.md
index 43e75ccf9..53be3743a 100644
--- a/packages/react-optimistic/README.md
+++ b/packages/react-optimistic/README.md
@@ -135,48 +135,89 @@ const todosConfig = {
       },
     },
     {
-      // Electric client instance
-      client: electric,
+      // Primary key for the todos table
+      primaryKey: ['id'],
     }
   ),
-  // Define mutation functions for backend persistence
-  mutationFn: {
-    insert: async (items) => {
-      // Insert items into Electric database
-      const db = electric.db
-      await db.todos.createMany({
-        data: items,
-      })
-    },
-    update: async (items) => {
-      // Update items in Electric database
-      const db = electric.db
-      for (const item of items) {
-        await db.todos.update({
-          where: { id: item.id },
-          data: item,
-        })
-      }
-    },
-    delete: async (items) => {
-      // Delete items from Electric database
-      const db = electric.db
-      await db.todos.deleteMany({
-        where: {
-          id: {
-            in: items.map((item) => item.id),
-          },
-        },
-      })
-    },
-  },
 }
 
 // Use the collection in a component
 function TodoList() {
   const { data, insert, update, delete: deleteFn } = useCollection(todosConfig)
 
-  // Now you can use these functions to interact with your data
-  // with automatic optimistic updates and backend persistence
+  // Create a mutation for handling all todo operations
+  const todoMutation = useOptimisticMutation({
+    mutationFn: async ({ transaction }) => {
+      const payload = transaction.mutations.map(m => {
+        const { collection, ...payload } = m
+        return payload
+      })
+
+      const response = await fetch(`http://localhost:3001/api/mutations`, {
+        method: `POST`,
+        headers: {
+          "Content-Type": `application/json`,
+        },
+        body: JSON.stringify(payload),
+      })
+
+      if (!response.ok) {
+        throw new Error(`HTTP error! Status: ${response.status}`)
+      }
+
+      const result = await response.json()
+      await transaction.mutations[0].collection.config.sync.awaitTxid(result.txid)
+    }
+  })
+
+  const addTodo = () => {
+    todoMutation.mutate(() => {
+      insert({ title: 'New todo', completed: false })
+    })
+  }
+
+  const toggleTodo = (todo) => {
+    todoMutation.mutate(() => {
+      update(todo, (draft) => {
+        draft.completed = !draft.completed
+      })
+    })
+  }
+
+  const removeTodo = (todo) => {
+    todoMutation.mutate(() => {
+      deleteFn(todo)
+    })
+  }
+
+  return (
+    <div>
+      <button
+        onClick={addTodo}
+        disabled={todoMutation.isPending}
+      >
+        {todoMutation.isPending ? 'Saving...' : 'Add Todo'}
+      </button>
+      <ul>
+        {data.map(todo => (
+          <li key={todo.id}>
+            <input
+              type="checkbox"
+              checked={todo.completed}
+              onChange={() => toggleTodo(todo)}
+              disabled={todoMutation.isPending}
+            />
+            {todo.title}
+            <button
+              onClick={() => removeTodo(todo)}
+              disabled={todoMutation.isPending}
+            >
+              Delete
+            </button>
+          </li>
+        ))}
+      </ul>
+    </div>
+  )
 }
 ```
diff --git a/packages/react-optimistic/package.json b/packages/react-optimistic/package.json
index ecd8b75a8..557251ab2 100644
--- a/packages/react-optimistic/package.json
+++ b/packages/react-optimistic/package.json
@@ -58,7 +58,8 @@
   "scripts": {
     "build": "vite build",
     "dev": "vite build --watch",
-    "test": "npx vitest --run"
+    "test": "npx vitest --run",
+    "lint": "eslint . --fix"
   },
   "sideEffects": false,
   "type": "module",
diff --git a/packages/react-optimistic/src/index.ts b/packages/react-optimistic/src/index.ts
index f037b5d53..bac2ca6f2 100644
--- a/packages/react-optimistic/src/index.ts
+++ b/packages/react-optimistic/src/index.ts
@@ -1,7 +1,12 @@
 // Re-export all public APIs
 export * from "./useCollection"
+export * from "./useOptimisticMutation"
 export * from "./useLiveQuery"
 export * from "./electric"
 
 // Re-export everything from @tanstack/optimistic
 export * from "@tanstack/optimistic"
+
+// Re-export some stuff explicitly to ensure the type & value is exported
+export { Collection } from "@tanstack/optimistic"
+export { createTransaction } from "@tanstack/optimistic"
diff --git a/packages/react-optimistic/src/useCollection.ts b/packages/react-optimistic/src/useCollection.ts
index ee308713f..27071f78b 100644
--- a/packages/react-optimistic/src/useCollection.ts
+++ b/packages/react-optimistic/src/useCollection.ts
@@ -40,15 +40,8 @@ export function useCollections() {
           snapshotCache = null // Invalidate cache when state changes
           callback()
         })
-        const transactionsUnsub =
-          collection.transactionManager.transactions.subscribe(() => {
-            snapshotCache = null // Invalidate cache when transactions change
-            callback()
-          })
-        return () => {
-          derivedStateUnsub()
-          transactionsUnsub()
-        }
+
+        return derivedStateUnsub
       })
 
       return () => {
@@ -72,7 +65,7 @@ export function useCollections() {
       for (const [id, collection] of collectionsStore.state) {
         snapshot.set(id, {
           state: collection.derivedState.state,
-          transactions: collection.transactionManager.transactions.state,
+          transactions: collection.transactions.state,
         })
       }
       snapshotCache = snapshot
@@ -94,7 +87,7 @@ export function useCollections() {
       for (const [id, collection] of collectionsStore.state) {
         snapshot.set(id, {
           state: collection.derivedState.state,
-          transactions: collection.transactionManager.transactions.state,
+          transactions: collection.transactions.state,
         })
       }
       snapshotCache = snapshot
@@ -308,7 +301,6 @@ export function useCollection<T extends object, R = any>(
         new Collection<T>({
           id: config.id,
           sync: config.sync,
-          mutationFn: config.mutationFn,
           schema: config.schema,
         })
       )
diff --git a/packages/react-optimistic/src/useLiveQuery.ts b/packages/react-optimistic/src/useLiveQuery.ts
index 3fe740975..cc0551096 100644
--- a/packages/react-optimistic/src/useLiveQuery.ts
+++ b/packages/react-optimistic/src/useLiveQuery.ts
@@ -1,4 +1,4 @@
-import { useEffect, useMemo } from "react"
+import { useEffect, useMemo, useState } from "react"
 import { useStore } from "@tanstack/react-store"
 import { compileQuery, queryBuilder } from "@tanstack/optimistic"
 import type {
@@ -22,22 +22,31 @@ export function useLiveQuery<
   ) => QueryBuilder<TResultContext>,
   deps: Array<unknown> = []
 ): UseLiveQueryReturn<ResultsFromContext<TResultContext>> {
+  const [restart, forceRestart] = useState(0)
+
   const compiledQuery = useMemo(() => {
     const query = queryFn(queryBuilder())
     const compiled = compileQuery(query)
     compiled.start()
     return compiled
-  }, deps)
+  }, [...deps, restart])
 
+  const state = useStore(compiledQuery.results.derivedState)
+  let data: Array<ResultsFromContext<TResultContext>> | undefined
+
+  // Clean up on unmount
   useEffect(() => {
+    if (compiledQuery.state === `stopped`) {
+      forceRestart((count) => {
+        return (count += 1)
+      })
+    }
+
     return () => {
       compiledQuery.stop()
     }
   }, [compiledQuery])
 
-  const state = useStore(compiledQuery.results.derivedState)
-  let data: Array<ResultsFromContext<TResultContext>> | undefined
-
   return {
     state,
     get data() {
diff --git a/packages/react-optimistic/src/useOptimisticMutation.ts b/packages/react-optimistic/src/useOptimisticMutation.ts
new file mode 100644
index 000000000..dcd1f7d22
--- /dev/null
+++ b/packages/react-optimistic/src/useOptimisticMutation.ts
@@ -0,0 +1,15 @@
+import { createTransaction } from "@tanstack/optimistic"
+import type { Transaction, TransactionConfig } from "@tanstack/optimistic"
+
+export function useOptimisticMutation(config: TransactionConfig) {
+  return {
+    mutate: (callback: () => void): Transaction => {
+      const transaction = createTransaction(config)
+      transaction.mutate(callback)
+      return transaction
+    },
+    createTransaction: (): Transaction => {
+      return createTransaction({ ...config, autoCommit: false })
+    },
+  }
+}
diff --git a/packages/react-optimistic/tests/electric.test.ts b/packages/react-optimistic/tests/electric.test.ts
index f71829460..8a3bb1225 100644
--- a/packages/react-optimistic/tests/electric.test.ts
+++ b/packages/react-optimistic/tests/electric.test.ts
@@ -1,5 +1,5 @@
 import { beforeEach, describe, expect, it, vi } from "vitest"
-import { Collection } from "@tanstack/optimistic"
+import { Collection, createTransaction } from "@tanstack/optimistic"
 import { createElectricSync } from "../src/electric"
 import type { PendingMutation, Transaction } from "@tanstack/optimistic"
 import type { Message, Row } from "@electric-sql/client"
@@ -48,7 +48,6 @@ describe(`Electric Integration`, () => {
     collection = new Collection({
       id: `test`,
       sync: electricSync,
-      mutationFn: vi.fn().mockResolvedValue(undefined),
     })
   })
 
@@ -362,17 +361,17 @@ describe(`Electric Integration`, () => {
       const testCollection = new Collection({
         id: `ofo`,
         sync: electricSync,
-        mutationFn: testMutationFn,
       })
 
-      let transaction = testCollection.insert(
-        { id: 1, name: `Test item 1` },
-        { key: `item1` }
+      const tx1 = createTransaction({ mutationFn: testMutationFn })
+
+      let transaction = tx1.mutate(() =>
+        testCollection.insert({ id: 1, name: `Test item 1` }, { key: `item1` })
       )
 
-      await transaction.isPersisted?.promise
+      await transaction.isPersisted.promise
 
-      transaction = testCollection.transactions.get(transaction.id)!
+      transaction = testCollection.transactions.state.get(transaction.id)!
 
       // Verify the mutation function was called correctly
       expect(testMutationFn).toHaveBeenCalledTimes(1)
@@ -414,32 +413,27 @@ describe(`Electric Integration`, () => {
 
   it(`Transaction proxy with toObject method works correctly`, () => {
     // Create a collection with a simple mutation function
-    const testCollection = new Collection({
+    new Collection({
       id: `foo`,
       sync: createElectricSync({ url: `foo` }, { primaryKey: [`id`] }), // Add primary key
-      mutationFn: async () => {},
     })
 
     // Create a transaction
-    const transaction = testCollection.transactionManager.applyTransaction(
-      [] // No mutations
-    )
+    const transaction = createTransaction({
+      mutationFn: () => {
+        return Promise.resolve()
+      },
+    })
 
     // Test that we can access properties
     expect(transaction.id).toBeDefined()
     expect(transaction.state).toBe(`pending`)
 
-    // Test the toObject method
-    const transactionObj = transaction.toObject()
-    expect(transactionObj).toBeDefined()
-    expect(transactionObj.id).toBe(transaction.id)
-    expect(transactionObj.state).toBe(transaction.state)
-
     // Test that we can create a modified clone
     const modifiedTransaction = {
-      ...transaction.toObject(),
+      ...transaction,
       metadata: {
-        ...transaction.toObject().metadata,
+        ...transaction.metadata,
         testKey: `testValue`,
       },
     }
diff --git a/packages/react-optimistic/tests/useCollection.test.tsx b/packages/react-optimistic/tests/useCollection.test.tsx
index db3451d38..7aaf42fb9 100644
--- a/packages/react-optimistic/tests/useCollection.test.tsx
+++ b/packages/react-optimistic/tests/useCollection.test.tsx
@@ -1,8 +1,10 @@
 import { describe, expect, it, vi } from "vitest"
 import { act, renderHook } from "@testing-library/react"
 import mitt from "mitt"
+import { createTransaction } from "@tanstack/optimistic"
 import { useCollection } from "../src/useCollection"
-import type { PendingMutation } from "@tanstack/optimistic"
+import { useOptimisticMutation } from "../src/useOptimisticMutation"
+import type { MutationFn, PendingMutation } from "@tanstack/optimistic"
 
 describe(`useCollection`, () => {
   it(`should handle insert, update, and delete operations`, async () => {
@@ -28,6 +30,10 @@ describe(`useCollection`, () => {
             })
           },
         },
+      })
+    )
+    const mutationHook = renderHook(() =>
+      useOptimisticMutation({
         mutationFn: ({ transaction }) => {
           persistMock()
           act(() => {
@@ -37,6 +43,7 @@ describe(`useCollection`, () => {
         },
       })
     )
+    const { mutate } = mutationHook.result.current
 
     // Initial state should be empty
     expect(result.current.state.size).toBe(0)
@@ -44,7 +51,9 @@ describe(`useCollection`, () => {
 
     // Test single insert with explicit key
     await act(async () => {
-      await result.current.insert({ name: `Alice` }, { key: `user1` })
+      await Promise.resolve(
+        mutate(() => result.current.insert({ name: `Alice` }, { key: `user1` }))
+      )
     })
 
     // Verify insert
@@ -54,9 +63,13 @@ describe(`useCollection`, () => {
 
     // Test bulk insert with sparse keys
     await act(async () => {
-      await result.current.insert([{ name: `Bob` }, { name: `Charlie` }], {
-        key: [`user2`, undefined],
-      })
+      await Promise.resolve(
+        mutate(() =>
+          result.current.insert([{ name: `Bob` }, { name: `Charlie` }], {
+            key: [`user2`, undefined],
+          })
+        )
+      )
     })
 
     // Get the auto-generated key for Charlie
@@ -72,16 +85,17 @@ describe(`useCollection`, () => {
 
     // Test update with callback
     const updateTransaction = await act(async () => {
-      return await result.current.update(
-        result.current.state.get(`user1`)!,
-        (item) => {
-          item.name = `Alice Smith`
-        }
+      return Promise.resolve(
+        mutate(() =>
+          result.current.update(result.current.state.get(`user1`)!, (item) => {
+            item.name = `Alice Smith`
+          })
+        )
       )
     })
 
     await act(async () => {
-      await updateTransaction.isPersisted?.promise
+      await updateTransaction.isPersisted.promise
     })
 
     // Verify update
@@ -94,18 +108,22 @@ describe(`useCollection`, () => {
         result.current.state.get(`user1`)!,
         result.current.state.get(`user2`)!,
       ]
-      return await result.current.update(
-        items,
-        { metadata: { bulkUpdate: true } },
-        (drafts) => {
-          drafts.forEach((draft, i) => {
-            if (i === 0) {
-              draft.name = draft.name + ` Jr.`
-            } else if (i === 1) {
-              draft.name = draft.name + ` Sr.`
+      return await Promise.resolve(
+        mutate(() =>
+          result.current.update(
+            items,
+            { metadata: { bulkUpdate: true } },
+            (drafts) => {
+              drafts.forEach((draft, i) => {
+                if (i === 0) {
+                  draft.name = draft.name + ` Jr.`
+                } else if (i === 1) {
+                  draft.name = draft.name + ` Sr.`
+                }
+              })
             }
-          })
-        }
+          )
+        )
       )
     })
 
@@ -119,7 +137,9 @@ describe(`useCollection`, () => {
 
     // Test single delete
     await act(async () => {
-      await result.current.delete(result.current.state.get(`user1`)!)
+      await Promise.resolve(
+        mutate(() => result.current.delete(result.current.state.get(`user1`)!))
+      )
     })
 
     // Verify single delete
@@ -132,9 +152,13 @@ describe(`useCollection`, () => {
         result.current.state.get(`user2`)!,
         result.current.state.get(charlieKey!)!,
       ]
-      await result.current.delete(items, {
-        metadata: { reason: `bulk cleanup` },
-      })
+      await Promise.resolve(
+        mutate(() =>
+          result.current.delete(items, {
+            metadata: { reason: `bulk cleanup` },
+          })
+        )
+      )
     })
 
     // Verify all items are deleted
@@ -146,14 +170,33 @@ describe(`useCollection`, () => {
     expect(persistMock).toHaveBeenCalledTimes(6) // 2 inserts + 2 updates + 2 deletes
   })
 
-  it(`should expose state, items, and data properties correctly`, async () => {
+  it(`should allow you to do manually committed transactions`, async () => {
     const emitter = mitt()
     const persistMock = vi.fn().mockResolvedValue(undefined)
 
     // Setup initial hook render
     const { result } = renderHook(() =>
-      useCollection({
-        id: `test-properties`,
+      useCollection<{ name: string }>({
+        id: `test-collection`,
+        sync: {
+          sync: ({ begin, write, commit }) => {
+            emitter.on(`*`, (_, mutations) => {
+              begin()
+              ;(mutations as Array<PendingMutation>).forEach((mutation) => {
+                write({
+                  key: mutation.key,
+                  type: mutation.type,
+                  value: mutation.changes as { name: string },
+                })
+              })
+              commit()
+            })
+          },
+        },
+      })
+    )
+    const mutationHook = renderHook(() =>
+      useOptimisticMutation({
         mutationFn: ({ transaction }) => {
           persistMock()
           act(() => {
@@ -161,6 +204,35 @@ describe(`useCollection`, () => {
           })
           return Promise.resolve()
         },
+      })
+    )
+    const transaction = mutationHook.result.current.createTransaction()
+
+    // Initial state should be empty
+    expect(result.current.state.size).toBe(0)
+    expect(result.current.data).toEqual([])
+
+    // Test single insert with explicit key
+    await act(async () => {
+      await Promise.resolve()
+      transaction.mutate(() =>
+        result.current.insert({ name: `Alice` }, { key: `user1` })
+      )
+
+      transaction.commit()
+    })
+
+    expect(transaction.state).toBe(`completed`)
+  })
+
+  it(`should expose state, items, and data properties correctly`, async () => {
+    const emitter = mitt()
+    const persistMock = vi.fn().mockResolvedValue(undefined)
+
+    // Setup initial hook render
+    const { result } = renderHook(() =>
+      useCollection({
+        id: `test-properties`,
         sync: {
           sync: ({ begin, write, commit }) => {
             emitter.on(`*`, (_, mutations) => {
@@ -178,6 +250,13 @@ describe(`useCollection`, () => {
         },
       })
     )
+    const mutationFn: MutationFn = ({ transaction }) => {
+      persistMock()
+      act(() => {
+        emitter.emit(`update`, transaction.mutations)
+      })
+      return Promise.resolve()
+    }
 
     // Initial state should be empty
     expect(result.current.state).toBeInstanceOf(Map)
@@ -187,13 +266,17 @@ describe(`useCollection`, () => {
 
     // Insert some test data
     await act(async () => {
-      await result.current.insert(
-        [
-          { id: 1, name: `Item 1` },
-          { id: 2, name: `Item 2` },
-          { id: 3, name: `Item 3` },
-        ],
-        { key: [`key1`, `key2`, `key3`] }
+      const tx = createTransaction({ mutationFn })
+      await Promise.resolve()
+      tx.mutate(() =>
+        result.current.insert(
+          [
+            { id: 1, name: `Item 1` },
+            { id: 2, name: `Item 2` },
+            { id: 3, name: `Item 3` },
+          ],
+          { key: [`key1`, `key2`, `key3`] }
+        )
       )
       emitter.emit(`update`, [
         { key: `key1`, type: `insert`, changes: { id: 1, name: `Item 1` } },
@@ -223,13 +306,6 @@ describe(`useCollection`, () => {
     const { result } = renderHook(() =>
       useCollection<{ id: number; name: string }>({
         id: `test-selector`,
-        mutationFn: ({ transaction }) => {
-          persistMock()
-          act(() => {
-            emitter.emit(`update`, transaction.mutations)
-          })
-          return Promise.resolve()
-        },
         sync: {
           sync: ({ begin, write, commit }) => {
             emitter.on(`*`, (_, mutations) => {
@@ -248,6 +324,13 @@ describe(`useCollection`, () => {
         },
       })
     )
+    const mutationFn: MutationFn = ({ transaction }) => {
+      persistMock()
+      act(() => {
+        emitter.emit(`update`, transaction.mutations)
+      })
+      return Promise.resolve()
+    }
 
     // Initial state
     expect(result.current.state).toBeInstanceOf(Map)
@@ -257,13 +340,17 @@ describe(`useCollection`, () => {
 
     // Insert some test data
     await act(async () => {
-      await result.current.insert(
-        [
-          { id: 1, name: `Alice` },
-          { id: 2, name: `Bob` },
-          { id: 3, name: `Charlie` },
-        ],
-        { key: [`key1`, `key2`, `key3`] }
+      await Promise.resolve()
+      const tx = createTransaction({ mutationFn })
+      tx.mutate(() =>
+        result.current.insert(
+          [
+            { id: 1, name: `Alice` },
+            { id: 2, name: `Bob` },
+            { id: 3, name: `Charlie` },
+          ],
+          { key: [`key1`, `key2`, `key3`] }
+        )
       )
       emitter.emit(`update`, [
         { key: `key1`, type: `insert`, changes: { id: 1, name: `Alice` } },
diff --git a/packages/react-optimistic/tests/useLiveQuery.test.tsx b/packages/react-optimistic/tests/useLiveQuery.test.tsx
index 1e914ff43..2c2a3053b 100644
--- a/packages/react-optimistic/tests/useLiveQuery.test.tsx
+++ b/packages/react-optimistic/tests/useLiveQuery.test.tsx
@@ -95,10 +95,6 @@ describe(`Query Collections`, () => {
           })
         },
       },
-      mutationFn: async ({ transaction }) => {
-        emitter.emit(`sync`, transaction.mutations)
-        return Promise.resolve()
-      },
     })
 
     // Sync from initial state
@@ -249,10 +245,6 @@ describe(`Query Collections`, () => {
           )
         },
       },
-      mutationFn: async ({ transaction }) => {
-        emitter.emit(`sync-person`, transaction.mutations)
-        return Promise.resolve()
-      },
     })
 
     // Create issue collection
@@ -427,10 +419,6 @@ describe(`Query Collections`, () => {
           })
         },
       },
-      mutationFn: async ({ transaction }) => {
-        emitter.emit(`sync`, transaction.mutations)
-        return Promise.resolve()
-      },
     })
 
     // Sync from initial state
@@ -526,10 +514,6 @@ describe(`Query Collections`, () => {
           })
         },
       },
-      mutationFn: async ({ transaction }) => {
-        emitter.emit(`sync`, transaction.mutations)
-        return Promise.resolve()
-      },
     })
 
     // Mock console.log to track when compiledQuery.stop() is called