feat: fixed invalid pending state after mutation cancel

Author: mbret

.eslintrc.cjs

@@ -21,8 +21,8 @@ const config = {
   plugins: ["react"],
   ignorePatterns: ["vitest.config.ts", "vite.config.ts", "test-setup.ts"],
   rules: {
-    'react/jsx-key': ['error', { checkFragmentShorthand: true }],
-    'react-hooks/exhaustive-deps': 'error',
+    "react/jsx-key": ["error", { checkFragmentShorthand: true }],
+    "react-hooks/exhaustive-deps": "error",
     "@typescript-eslint/explicit-function-return-type": 0,
     "react/display-name": 0,
     "@typescript-eslint/no-invalid-void-type": 0,
@@ -33,4 +33,4 @@ const config = {
   }
 }
 
-module.exports = config
+module.exports = config

README.md

@@ -1,8 +1,8 @@
 # reactjrx
 
-`reactjrx` is an innovative library designed to bridge RxJS and React, offering developers a powerful set of tools to integrate observables into their React applications seamlessly. 
+`reactjrx` is an innovative library designed to bridge RxJS and React, offering developers a powerful set of tools to integrate observables into their React applications seamlessly.
 
-It simplifies the reactive programming approach by providing easy-to-use bindings for observing and reacting to data changes, coupled with advanced state management and queries helpers akin to `react-query` for handling effects, mutations, and asynchronous data fetching. 
+It simplifies the reactive programming approach by providing easy-to-use bindings for observing and reacting to data changes, coupled with advanced state management and queries helpers akin to `react-query` for handling effects, mutations, and asynchronous data fetching.
 
 Tailored for projects where RxJS is a core dependency, `reactjrx` enhances the development experience by enabling more dynamic, efficient, and responsive UIs, making it an essential addition for developers looking to leverage the full potential of RxJS within their React applications.
 

src/lib/binding/types.ts

@@ -1,4 +1,4 @@
-import { type Observable } from "rxjs";
+import { type Observable } from "rxjs"
 
 export type SubscribeSource<Data> =
   | (() => Observable<Data>)

src/lib/queries/client/focusManager.ts

@@ -1,12 +1,4 @@
-import {
-  filter,
-  fromEvent,
-  map,
-  merge,
-  skip,
-  tap,
-  BehaviorSubject,
-} from "rxjs"
+import { filter, fromEvent, map, merge, skip, tap, BehaviorSubject } from "rxjs"
 
 export class FocusManager {
   readonly #visibility$ = merge(

src/lib/queries/client/keys/hashKey.test.ts

@@ -10,7 +10,7 @@ it("should return correct serialized key", () => {
   expect(hashKey(["todos", { page: 2, status: "foo" }])).toBe(
     `["todos",{"page":2,"status":"foo"}]`
   )
-  expect(
-    hashKey(["todos", { page: 2, status: "foo", bar: undefined }])
-  ).toBe(`["todos",{"page":2,"status":"foo"}]`)
+  expect(hashKey(["todos", { page: 2, status: "foo", bar: undefined }])).toBe(
+    `["todos",{"page":2,"status":"foo"}]`
+  )
 })

src/lib/queries/client/keys/matchKey.test.ts

@@ -6,9 +6,7 @@ it("should compare keys", () => {
   expect(matchKey([], [], { exact: true })).toBe(true)
   expect(matchKey([], ["foo"], { exact: false })).toBe(false)
   expect(matchKey(["foo"], [], { exact: false })).toBe(true)
-  expect(
-    matchKey(["foo", "bar"], ["foo"], { exact: false })
-  ).toBe(true)
+  expect(matchKey(["foo", "bar"], ["foo"], { exact: false })).toBe(true)
   expect(matchKey(["foo"], ["foo", "bar"], { exact: true })).toBe(false)
   expect(matchKey(["foo", "bar"], ["foo"], { exact: false })).toBe(true)
   expect(matchKey(["foo"], ["foo", "bar"], { exact: false })).toBe(false)

src/lib/queries/client/mutations/cache/MutationCache.ts

@@ -12,7 +12,7 @@ import {
   combineLatest,
   EMPTY
 } from "rxjs"
-import { createPredicateForFilters } from "../filters"
+import { createPredicateForFilters } from "../utils/filters"
 import {
   type MutationCacheConfig,
   type MutationCacheNotifyEvent

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

@@ -10,10 +10,12 @@ import {
   concat,
   toArray,
   mergeMap,
-  NEVER,
-  startWith
+  startWith,
+  map,
+  filter,
+  scan
 } from "rxjs"
-import { getDefaultMutationState } from "../defaultMutationState"
+import { getDefaultMutationState } from "../utils/defaultMutationState"
 import { type DefaultError } from "../../types"
 import { type MutationCache } from "../cache/MutationCache"
 import { makeObservable } from "../../utils/makeObservable"
@@ -25,6 +27,7 @@ import {
 import { executeMutation } from "./executeMutation"
 import { trackSubscriptions } from "../../../../utils/operators/trackSubscriptions"
 import { observeUntilFinished } from "./observeUntilFinished"
+import { distinctUntilStateChanged } from "../utils/distinctUntilStateChanged"
 
 interface MutationConfig<TData, TError, TVariables, TContext> {
   mutationCache: MutationCache
@@ -59,18 +62,24 @@ export class Mutation<
     this.options = options
     this.state = state ?? this.state
 
-    const execution$ = this.#executeSubject.pipe(
+    const resetPendingState$ = this.#cancelSubject.pipe(
+      filter(() => this.state.status === "pending"),
+      map(() => ({
+        status: "idle" as const
+      }))
+    )
+
+    const executionState$ = this.#executeSubject.pipe(
       switchMap((variables) =>
         executeMutation<TData, TError, TVariables, TContext>({
           options: {
             ...this.options,
             onMutate: (variables) => {
-              const onCacheMutate$ = makeObservable(
-                () =>
-                  mutationCache.config.onMutate?.(
-                    variables,
-                    this as Mutation<any, any, any, any>
-                  )
+              const onCacheMutate$ = makeObservable(() =>
+                mutationCache.config.onMutate?.(
+                  variables,
+                  this as Mutation<any, any, any, any>
+                )
               ) as Observable<TContext>
 
               // eslint-disable-next-line @typescript-eslint/promise-function-async
@@ -87,53 +96,50 @@ export class Mutation<
               return context$
             },
             onError: (error, variables, context) => {
-              const onCacheError$ = makeObservable(
-                () =>
-                  mutationCache.config.onError?.(
-                    error as any,
-                    variables,
-                    context,
-                    this as Mutation<any, any, any, any>
-                  )
+              const onCacheError$ = makeObservable(() =>
+                mutationCache.config.onError?.(
+                  error as any,
+                  variables,
+                  context,
+                  this as Mutation<any, any, any, any>
+                )
               )
 
-              const onOptionError$ = makeObservable(
-                () => this.options.onError?.(error, variables, context)
+              const onOptionError$ = makeObservable(() =>
+                this.options.onError?.(error, variables, context)
               )
 
               return concat(onCacheError$, onOptionError$).pipe(toArray())
             },
             onSettled: (data, error, variables, context) => {
-              const onCacheSuccess$ = makeObservable(
-                () =>
-                  mutationCache.config.onSettled?.(
-                    data,
-                    error as Error,
-                    variables,
-                    context,
-                    this as Mutation<any, any, any, any>
-                  )
+              const onCacheSuccess$ = makeObservable(() =>
+                mutationCache.config.onSettled?.(
+                  data,
+                  error as Error,
+                  variables,
+                  context,
+                  this as Mutation<any, any, any, any>
+                )
               )
 
-              const onOptionSettled$ = makeObservable(
-                () => this.options.onSettled?.(data, error, variables, context)
+              const onOptionSettled$ = makeObservable(() =>
+                this.options.onSettled?.(data, error, variables, context)
               )
 
               return concat(onCacheSuccess$, onOptionSettled$).pipe(toArray())
             },
             onSuccess: (data, variables, context) => {
-              const onCacheSuccess$ = makeObservable(
-                () =>
-                  mutationCache.config.onSuccess?.(
-                    data,
-                    variables,
-                    context,
-                    this as Mutation<any, any, any, any>
-                  )
+              const onCacheSuccess$ = makeObservable(() =>
+                mutationCache.config.onSuccess?.(
+                  data,
+                  variables,
+                  context,
+                  this as Mutation<any, any, any, any>
+                )
               )
 
-              const onOptionSuccess$ = makeObservable(
-                () => this.options.onSuccess?.(data, variables, context)
+              const onOptionSuccess$ = makeObservable(() =>
+                this.options.onSuccess?.(data, variables, context)
               )
 
               return concat(onCacheSuccess$, onOptionSuccess$).pipe(toArray())
@@ -145,25 +151,27 @@ export class Mutation<
       )
     )
 
-    this.state$ = merge(
-      execution$,
-      /**
-       * We keep state forever since only a explicit destroy
-       * may terminate the mutation
-       */
-      NEVER
-    ).pipe(
+    /**
+     * @important
+     * This observable needs to complete on cancelSubject
+     * otherwise the state will never be freed
+     */
+    const stateChange$ = merge(resetPendingState$, executionState$)
+
+    this.state$ = stateChange$.pipe(
       startWith(this.state),
+      scan(
+        (acc, partialState) => ({
+          ...acc,
+          ...partialState
+        }),
+        this.state
+      ),
+      distinctUntilStateChanged,
       tap((value) => {
-        this.state = { ...this.state, ...value }
+        this.state = value
       }),
-      takeUntil(this.#cancelSubject),
-      /**
-       * refCount as true somewhat make NEVER complete when there are
-       * no more observers. I thought I should have to complete manually (which is
-       * why we still cancel the observable when we remove it from cache)
-       */
-      shareReplay({ bufferSize: 1, refCount: false }),
+      shareReplay(1),
       trackSubscriptions((count) => {
         this.#observerCount.next(count)
       })
@@ -196,7 +204,11 @@ export class Mutation<
     return this.execute(this.state.variables as TVariables)
   }
 
-  // @todo merge with query
+  /**
+   * Cancel if needed and finalize the mutation.
+   * The mutation will be garbage collected automatically
+   * when no more observers are listening
+   */
   cancel() {
     this.#cancelSubject.next()
     this.#cancelSubject.complete()

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

@@ -9,14 +9,12 @@ import {
   iif,
   catchError,
   scan,
-  distinctUntilChanged,
   shareReplay
 } from "rxjs"
 import { type MutationOptions, type MutationState } from "./types"
 import { makeObservable } from "../../utils/makeObservable"
 import { type DefaultError } from "../../types"
-import { getDefaultMutationState } from "../defaultMutationState"
-import { shallowEqual } from "../../../../utils/shallowEqual"
+import { getDefaultMutationState } from "../utils/defaultMutationState"
 import { onlineManager } from "../../onlineManager"
 import { waitForNetworkOnError } from "./waitForNetworkOnError"
 import { delayWhenNetworkOnline } from "./delayWhenNetworkOnline"
@@ -227,11 +225,7 @@ export const executeMutation = <
         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)
-    )
+    }, getDefaultMutationState<TData, TError, TVariables, TContext>())
   )
 
   return mutation$

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

@@ -1,7 +1,9 @@
 import { type Observable, takeWhile } from "rxjs"
 import { type MutationState } from "./types"
 
-export const observeUntilFinished = <T extends MutationState<any, any, any, any>>(
+export const observeUntilFinished = <
+  T extends MutationState<any, any, any, any>
+>(
   source: Observable<T>
 ) =>
   source.pipe(

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

@@ -37,7 +37,9 @@ export const waitForNetworkOnError = <
            * timer needed to be iso RQ, so the state returned by mutation include both previous and next one
            */
           timer(1).pipe(
-            mergeMap(() => throwError(() => error).pipe(delayWhenNetworkOnline<T>()))
+            mergeMap(() =>
+              throwError(() => error).pipe(delayWhenNetworkOnline<T>())
+            )
           )
         )
       } else {

src/lib/queries/client/mutations/observers/MutationObserver.ts

@@ -10,10 +10,12 @@ import {
   last,
   filter,
   tap,
-  ignoreElements
+  ignoreElements,
+  distinctUntilChanged,
+  scan
 } from "rxjs"
 import { type MutateOptions } from "../types"
-import { getDefaultMutationState } from "../defaultMutationState"
+import { getDefaultMutationState } from "../utils/defaultMutationState"
 import { type QueryClient } from "../../QueryClient"
 import { type DefaultError } from "../../types"
 import { type Mutation } from "../mutation/Mutation"
@@ -27,6 +29,8 @@ import { type MutationState } from "../mutation/types"
 import { isDefined } from "../../../../utils/isDefined"
 import { matchKey } from "../../keys/matchKey"
 import { observeUntilFinished } from "../mutation/observeUntilFinished"
+import { shallowEqual } from "../../../../utils/shallowEqual"
+import { distinctUntilStateChanged } from "../utils/distinctUntilStateChanged"
 
 /**
  * Provide API to observe mutations results globally.
@@ -193,10 +197,35 @@ export class MutationObserver<
     )
 
     const mutationResult$ = this.#mutationRunner.state$.pipe(
+      distinctUntilChanged(
+        (
+          { mutation: prevMutation, state: { data: prevData, ...prev } },
+          { mutation: currentMutation, state: { data: currData, ...curr } }
+        ) =>
+          prevMutation === currentMutation &&
+          shallowEqual(prev, curr) &&
+          shallowEqual(prevData, currData)
+      ),
+      map(({ state }) => state),
+      scan<
+        MutationState<TData, TError, TVariables, TContext>,
+        MutationState<TData, TError, TVariables, TContext>
+      >((acc, current) => {
+        return {
+          ...acc,
+          ...current,
+          ...(current.status === "pending" && {
+            data: current.data ?? acc.data
+          }),
+          ...(current.status === "pending" && {
+            error: current.error ?? acc.error
+          })
+        }
+      }),
       map((state) => this.getObserverResultFromState(state))
     )
 
-    const currentMutationCancelled$ = this.#currentMutationSubject.pipe(
+    const defaultStateOnMutationCancel$ = this.#currentMutationSubject.pipe(
       filter(isDefined),
       switchMap((mutation) => mutation.mutation.cancelled$),
       map(() => this.getObserverResultFromState(getDefaultMutationState()))
@@ -205,8 +234,8 @@ export class MutationObserver<
     const result$ = merge(
       this.observed$,
       mutationResult$,
-      currentMutationCancelled$
-    )
+      defaultStateOnMutationCancel$
+    ).pipe(distinctUntilStateChanged)
 
     return { result$, lastValue }
   }