feat: added more tests

src/lib/queries/client/mutations/mutation/Mutation.ts

@@ -10,7 +10,9 @@ import {
   shareReplay,
   takeWhile,
   BehaviorSubject,
-  concat
+  concat,
+  toArray,
+  mergeMap
 } from "rxjs"
 import { getDefaultMutationState } from "../defaultMutationState"
 import { type DefaultError } from "../../types"
@@ -67,11 +69,59 @@ export class Mutation<
     const execution$ = this.executeSubject.pipe(
       switchMap((variables) =>
         executeMutation({
-          mutation: this,
-          mutationCache: this.mutationCache,
           options: {
             ...this.options,
-            // @todo test onError, onSettled
+            onMutate: (variables) => {
+              const onCacheMutate$ = functionAsObservable(
+                () =>
+                  mutationCache.config.onMutate?.(
+                    variables,
+                    this as Mutation<any, any, any, any>
+                  )
+              ) as Observable<TContext>
+
+              const onOptionMutate$ = functionAsObservable(
+                // eslint-disable-next-line @typescript-eslint/promise-function-async
+                () => this.options.onMutate?.(variables) ?? undefined
+              )
+
+              return onCacheMutate$.pipe(mergeMap(() => onOptionMutate$))
+            },
+            onError: (error, variables, context) => {
+              const onCacheError$ = functionAsObservable(
+                () =>
+                  mutationCache.config.onError?.(
+                    error as any,
+                    variables,
+                    context,
+                    this as Mutation<any, any, any, any>
+                  )
+              )
+
+              const onOptionError$ = functionAsObservable(
+                () => this.options.onError?.(error, variables, context)
+              )
+
+              return concat(onCacheError$, onOptionError$).pipe(toArray())
+            },
+            onSettled: (data, error, variables, context) => {
+              const onCacheSuccess$ = functionAsObservable(
+                () =>
+                  mutationCache.config.onSettled?.(
+                    data,
+                    error as Error,
+                    variables,
+                    context,
+                    this as Mutation<any, any, any, any>
+                  )
+              )
+
+              const onOptionSettled$ = functionAsObservable(
+                () => this.options.onSettled?.(data, error, variables, context)
+              )
+
+              return concat(onCacheSuccess$, onOptionSettled$).pipe(toArray())
+            },
             onSuccess: (data, variables, context) => {
               const onCacheSuccess$ = functionAsObservable(
                 () =>
@@ -87,7 +137,7 @@ export class Mutation<
                 () => this.options.onSuccess?.(data, variables, context)
               )
 
-              return concat(onCacheSuccess$, onOptionSuccess$)
+              return concat(onCacheSuccess$, onOptionSuccess$).pipe(toArray())
             }
           },
           state: this.state,

src/lib/queries/client/mutations/mutation/executeMutation.ts

@@ -5,19 +5,19 @@ import {
   switchMap,
   concat,
   toArray,
-  mergeMap,
   takeWhile,
   share,
   iif,
-  catchError
+  catchError,
+  scan,
+  distinctUntilChanged,
 } from "rxjs"
 import { type MutationOptions, type MutationState } from "./types"
 import { functionAsObservable } from "../../utils/functionAsObservable"
 import { retryOnError } from "../../operators"
 import { type DefaultError } from "../../types"
-import { mergeResults } from "../operators"
-import { type Mutation } from "./Mutation"
-import { type MutationCache } from "../cache/MutationCache"
+import { getDefaultMutationState } from "../defaultMutationState"
+import { shallowEqual } from "../../../../utils/shallowEqual"
 
 export const executeMutation = <
   TData = unknown,
@@ -27,15 +27,11 @@ export const executeMutation = <
 >({
   variables,
   state,
-  options,
-  mutation,
-  mutationCache
+  options
 }: {
   variables: TVariables
   state: MutationState<TData, TError, TVariables, TContext>
   options: MutationOptions<TData, TError, TVariables, TContext>
-  mutation: Mutation<TData, TError, TVariables, TContext>
-  mutationCache: MutationCache
 }) => {
   type LocalState = MutationState<TData, TError, TVariables, TContext>
 
@@ -46,18 +42,6 @@ export const executeMutation = <
 
   const mutationFn = options.mutationFn ?? defaultFn
 
-  const onCacheMutate$ = iif(
-    () => isPaused,
-    of(null),
-    functionAsObservable(
-      () =>
-        mutationCache.config.onMutate?.(
-          variables,
-          mutation as Mutation<any, any, any>
-        )
-    )
-  )
-
   const onOptionMutate$ = iif(
     () => isPaused,
     of(state.context),
@@ -67,10 +51,7 @@ export const executeMutation = <
     )
   )
 
-  const onMutate$ = onCacheMutate$.pipe(
-    mergeMap(() => onOptionMutate$),
-    share()
-  )
+  const onMutate$ = onOptionMutate$.pipe(share())
 
   type QueryState = Omit<Partial<LocalState>, "data"> & {
     // add layer to allow undefined as mutation result
@@ -80,27 +61,15 @@ export const executeMutation = <
   const onError = (error: TError, context: TContext, attempt: number) => {
     console.error(error)
 
-    const onCacheError$ = functionAsObservable(
-      () =>
-        mutationCache.config.onError?.<TData, TError, TVariables, TContext>(
-          error as Error,
-          variables,
-          context,
-          mutation
-        )
-    )
-
     const onError$ = functionAsObservable(
       () => options.onError?.(error, variables, context)
     )
 
-    return concat(onCacheError$, onError$).pipe(
+    return onError$.pipe(
       catchError(() => of(error)),
-      toArray(),
       map(
-        (): QueryState => ({
+        (): Omit<QueryState, "result"> => ({
           failureCount: attempt,
-          result: undefined,
           error,
           failureReason: error,
           context,
@@ -136,7 +105,12 @@ export const executeMutation = <
               failureReason: error
             }),
           catchError: (attempt, error) =>
-            onError(error, context as TContext, attempt)
+            onError(error, context as TContext, attempt).pipe(
+              map((data) => ({
+                ...data,
+                result: undefined
+              }))
+            )
         }),
         takeWhile(
           ({ result, error }) =>
@@ -159,13 +133,15 @@ export const executeMutation = <
 
   const mutation$ = merge(
     initState$,
-    onMutate$.pipe(map((context) => ({ context }))),
+    // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
+    onMutate$.pipe(map((context) => ({ context }) as Partial<LocalState>)),
     queryRunner$.pipe(
       switchMap(({ result: mutationData, error, ...restState }) => {
         if (!mutationData && !error)
+          // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
           return of({
             ...restState
-          })
+          } as Partial<LocalState>)
 
         const onSuccess$ = error
           ? of(null)
@@ -178,18 +154,6 @@ export const executeMutation = <
                 )
             )
 
-        // to pass as option from cache not here
-        const onCacheSettled$ = functionAsObservable(
-          () =>
-            mutationCache.config.onSettled?.(
-              mutationData?.data,
-              error as any,
-              variables,
-              restState.context,
-              mutation as Mutation<any, any, any>
-            )
-        )
-
         const onOptionSettled$ = functionAsObservable(
           () =>
             options.onSettled?.(
@@ -200,7 +164,7 @@ export const executeMutation = <
             )
         )
 
-        const onSettled$ = concat(onCacheSettled$, onOptionSettled$).pipe(
+        const onSettled$ = onOptionSettled$.pipe(
           catchError((error) => (mutationData ? of(mutationData) : of(error)))
         )
 
@@ -223,14 +187,32 @@ export const executeMutation = <
                 } satisfies Partial<LocalState>)
           ),
           catchError((error) =>
-            onError(error, restState.context as TContext, 0)
+            onError(error, restState.context as TContext, 0).pipe(
+              map((data) => ({
+                ...data,
+                data: undefined
+              }))
+            )
           )
         )
 
         return result$
       })
     )
-  ).pipe(mergeResults)
+  ).pipe(
+    scan((acc, current) => {
+      return {
+        ...acc,
+        ...current,
+        data: current.data ?? acc.data,
+        error: current.error ?? acc.error
+      }
+    }, getDefaultMutationState<TData, TError, TVariables, TContext>()),
+    distinctUntilChanged(
+      ({ data: prevData, ...prev }, { data: currData, ...curr }) =>
+        shallowEqual(prev, curr) && shallowEqual(prevData, currData)
+    )
+  )
 
   return mutation$
 }

src/lib/queries/client/mutations/mutation/mutation.options.test.ts

@@ -0,0 +1,113 @@
+import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"
+import { type QueryClient } from "../../createClient"
+import { createQueryClient, waitForTimeout } from "../../../../../tests/utils"
+import { MutationObserver } from "../observers/MutationObserver"
+import { waitFor } from "@testing-library/react"
+import { noop } from "rxjs"
+import { MutationCache } from "../cache/MutationCache"
+
+describe("mutations", () => {
+  let queryClient: QueryClient
+
+  beforeEach(() => {
+    queryClient = createQueryClient()
+    queryClient.mount()
+  })
+
+  afterEach(() => {
+    queryClient.clear()
+  })
+
+  test("mutation onError callbacks should see updated options", async () => {
+    const onError = vi.fn()
+
+    const mutation = new MutationObserver(queryClient, {
+      mutationFn: async () => {
+        await waitForTimeout(10)
+
+        throw new Error("error")
+      },
+      onError: () => {
+        onError(1)
+      }
+    })
+
+    void mutation.mutate().catch(noop)
+
+    mutation.setOptions({
+      onError: () => {
+        onError(2)
+      }
+    })
+
+    await waitFor(() => {
+      expect(onError).toHaveBeenCalledTimes(1)
+    })
+
+    expect(onError).toHaveBeenCalledWith(2)
+  })
+
+  test("mutation onSettled callbacks should see updated options", async () => {
+    const onSettled = vi.fn()
+
+    const mutation = new MutationObserver(queryClient, {
+      mutationFn: async () => {
+        return "update"
+      },
+      onSettled: () => {
+        onSettled(1)
+      }
+    })
+
+    void mutation.mutate()
+
+    mutation.setOptions({
+      onSettled: () => {
+        onSettled(2)
+      }
+    })
+
+    await waitFor(() => {
+      expect(onSettled).toHaveBeenCalledTimes(1)
+    })
+
+    expect(onSettled).toHaveBeenCalledWith(2)
+  })
+
+  test("mutation onMutate callbacks should see updated options", async () => {
+    const onMutate = vi.fn()
+
+    const queryClient2 = createQueryClient({
+      mutationCache: new MutationCache({
+        onMutate: async () => {
+          await waitForTimeout(10)
+
+          return ""
+        }
+      })
+    })
+
+    const mutation = new MutationObserver(queryClient2, {
+      mutationFn: async () => {
+        return "update"
+      },
+      onMutate: () => {
+        onMutate(1)
+      }
+    })
+
+    void mutation.mutate()
+
+    mutation.setOptions({
+      onMutate: () => {
+        onMutate(2)
+      }
+    })
+
+    await waitFor(() => {
+      expect(onMutate).toHaveBeenCalledTimes(1)
+    })
+
+    expect(onMutate).toHaveBeenCalledWith(2)
+  })
+})