optimistic seperation (#260)

Author: tim-smart

.changeset/smart-parts-bake.md

@@ -0,0 +1,5 @@
+---
+"@effect-rx/rx": patch
+---
+
+seperate optimistic state from mutations

docs/rx/Rx.ts.md

@@ -27,6 +27,9 @@ Added in v1.0.0
   - [windowFocusSignal](#windowfocussignal)
 - [KeyValueStore](#keyvaluestore)
   - [kvs](#kvs)
+- [Optimistic](#optimistic)
+  - [optimistic](#optimistic-1)
+  - [optimisticFn](#optimisticfn)
 - [Serializable](#serializable)
   - [Serializable (interface)](#serializable-interface)
   - [SerializableTypeId](#serializabletypeid)
@@ -48,7 +51,6 @@ Added in v1.0.0
   - [transform](#transform)
   - [withFallback](#withfallback)
   - [withLabel](#withlabel)
-  - [withOptimisticSet](#withoptimisticset)
 - [constructors](#constructors)
   - [context](#context)
   - [family](#family)
@@ -234,6 +236,42 @@ export declare const kvs: <A>(options: {
 
 Added in v1.0.0
 
+# Optimistic
+
+## optimistic
+
+**Signature**
+
+```ts
+export declare const optimistic: <A>(
+  self: Rx<A>
+) => Writable<A, Rx<Result.Result<A extends Result.Result<infer _A, infer _E> ? _A : A, unknown>>>
+```
+
+Added in v1.0.0
+
+## optimisticFn
+
+**Signature**
+
+```ts
+export declare const optimisticFn: {
+  <A, W, XA, XE, OW = A extends Result.Result<infer _A, infer _E> ? _A : A>(options: {
+    readonly updateToValue: (value: OW, current: NoInfer<A>) => NoInfer<W>
+    readonly fn: RxResultFn<NoInfer<OW>, XA, XE>
+  }): (self: Writable<A, Rx<Result.Result<W, unknown>>>) => RxResultFn<OW, XA, XE>
+  <A, W, XA, XE, OW = A extends Result.Result<infer _A, infer _E> ? _A : A>(
+    self: Writable<A, Rx<Result.Result<W, unknown>>>,
+    options: {
+      readonly updateToValue: (value: OW, current: NoInfer<A>) => NoInfer<W>
+      readonly fn: RxResultFn<NoInfer<OW>, XA, XE>
+    }
+  ): RxResultFn<OW, XA, XE>
+}
+```
+
+Added in v1.0.0
+
 # Serializable
 
 ## Serializable (interface)
@@ -493,33 +531,6 @@ export declare const withLabel: {
 
 Added in v1.0.0
 
-## withOptimisticSet
-
-**Signature**
-
-```ts
-export declare const withOptimisticSet: {
-  <A, XA, XE, W = A extends Result.Result<infer _A, infer _E> ? _A : A>(options: {
-    readonly updateToValue: (
-      value: W,
-      current: NoInfer<A>
-    ) => A extends Result.Result<infer _A, infer _E> ? _A : NoInfer<A>
-    readonly fn: RxResultFn<NoInfer<W>, XA, XE>
-    readonly disableRefresh?: boolean | undefined
-  }): (self: Rx<A>) => Writable<A, W>
-  <A, XA, XE, W = A extends Result.Result<infer _A, infer _E> ? _A : A>(
-    self: Rx<A>,
-    options: {
-      readonly updateToValue: (value: W, current: NoInfer<A>) => A extends Result.Result<infer _A, infer _E> ? _A : A
-      readonly fn: RxResultFn<NoInfer<W>, XA, XE>
-      readonly disableRefresh?: boolean | undefined
-    }
-  ): Writable<A, W>
-}
-```
-
-Added in v1.0.0
-
 # constructors
 
 ## context

packages/rx/src/Rx.ts

@@ -1355,88 +1355,110 @@ export const debounce: {
 
 /**
  * @since 1.0.0
- * @category combinators
+ * @category Optimistic
  */
-export const withOptimisticSet: {
-  <A, XA, XE, W = A extends Result.Result<infer _A, infer _E> ? _A : A>(
+export const optimistic = <A>(
+  self: Rx<A>
+): Writable<A, Rx<Result.Result<A extends Result.Result<infer _A, infer _E> ? _A : A, unknown>>> => {
+  let counter = 0
+  const writeRx = state(
+    [
+      counter,
+      undefined as any as Rx<Result.Result<A extends Result.Result<infer _A, infer _E> ? _A : A, unknown>>
+    ] as const
+  )
+  return writable(
+    (get) => {
+      let lastValue = get.once(self)
+      const isResult = Result.isResult(lastValue)
+      get.subscribe(self, (value) => {
+        lastValue = value
+        if (!Result.isResult(value)) {
+          return get.setSelf(value)
+        }
+        const current = Option.getOrUndefined(get.self<Result.Result<any, any>>())!
+        if (Result.isSuccess(current) && Result.isSuccess(value)) {
+          if (value.timestamp >= current.timestamp) {
+            get.setSelf(value)
+          }
+        } else {
+          get.setSelf(value)
+        }
+      })
+      const transitions = new Set<Rx<Result.Result<A extends Result.Result<infer _A, infer _E> ? _A : A, unknown>>>()
+      const cancels = new Set<() => void>()
+      get.subscribe(writeRx, ([, rx]) => {
+        if (transitions.has(rx)) {
+          return
+        }
+        transitions.add(rx)
+        const cancel = get.registry.subscribe(rx, (result) => {
+          if (Result.isSuccess(result) && result.waiting) {
+            return get.setSelf(isResult ? Result.success(result.value, { waiting: true }) : result.value)
+          }
+          transitions.delete(rx)
+          cancels.delete(cancel)
+          cancel()
+          if (transitions.size === 0) {
+            if (Result.isFailure(result)) {
+              get.setSelf(lastValue)
+            }
+            get.refresh(self)
+          }
+        }, { immediate: true })
+        cancels.add(cancel)
+      })
+      get.addFinalizer(() => {
+        for (const cancel of cancels) {
+          cancel()
+        }
+        transitions.clear()
+        cancels.clear()
+      })
+      return lastValue
+    },
+    (ctx, rx) => ctx.set(writeRx, [++counter, rx]),
+    (refresh) => refresh(self)
+  )
+}
+
+/**
+ * @since 1.0.0
+ * @category Optimistic
+ */
+export const optimisticFn: {
+  <A, W, XA, XE, OW = A extends Result.Result<infer _A, infer _E> ? _A : A>(
     options: {
-      readonly updateToValue: (
-        value: W,
-        current: NoInfer<A>
-      ) => A extends Result.Result<infer _A, infer _E> ? _A : NoInfer<A>
-      readonly fn: RxResultFn<NoInfer<W>, XA, XE>
-      readonly disableRefresh?: boolean | undefined
+      readonly updateToValue: (value: OW, current: NoInfer<A>) => NoInfer<W>
+      readonly fn: RxResultFn<NoInfer<OW>, XA, XE>
     }
   ): (
-    self: Rx<A>
-  ) => Writable<A, W>
-  <A, XA, XE, W = A extends Result.Result<infer _A, infer _E> ? _A : A>(
-    self: Rx<A>,
+    self: Writable<A, Rx<Result.Result<W, unknown>>>
+  ) => RxResultFn<OW, XA, XE>
+  <A, W, XA, XE, OW = A extends Result.Result<infer _A, infer _E> ? _A : A>(
+    self: Writable<A, Rx<Result.Result<W, unknown>>>,
     options: {
-      readonly updateToValue: (
-        value: W,
-        current: NoInfer<A>
-      ) => A extends Result.Result<infer _A, infer _E> ? _A : A
-      readonly fn: RxResultFn<NoInfer<W>, XA, XE>
-      readonly disableRefresh?: boolean | undefined
+      readonly updateToValue: (value: OW, current: NoInfer<A>) => NoInfer<W>
+      readonly fn: RxResultFn<NoInfer<OW>, XA, XE>
     }
-  ): Writable<A, W>
-} = dual(2, <A, W, XA, XE>(
-  self: Rx<A>,
+  ): RxResultFn<OW, XA, XE>
+} = dual(2, <A, W, OW, XA, XE>(
+  self: Writable<A, Rx<Result.Result<W, unknown>>>,
   options: {
-    readonly updateToValue: (value: W, current: A) => A extends Result.Result<infer _A, infer _E> ? _A : A
-    readonly fn: RxResultFn<W, XA, XE>
-    readonly disableRefresh?: boolean | undefined
+    readonly updateToValue: (value: OW, current: A) => W
+    readonly fn: RxResultFn<OW, XA, XE>
   }
-): Writable<A, W> => {
-  let counter = 0
-  const argRx = state([counter, undefined as A | undefined] as const)
-  return writable((get) => {
-    let lastValue = get.once(self)
-    get.subscribe(self, (value) => {
-      lastValue = value
-      if (!Result.isResult(value)) {
-        return get.setSelf(value)
-      }
-      const current = Option.getOrUndefined(get.self<Result.Result<any, any>>())!
-      if (Result.isSuccess(current) && Result.isSuccess(value)) {
-        if (value.timestamp >= current.timestamp) {
-          get.setSelf(value)
-        }
-      } else {
-        get.setSelf(lastValue)
-      }
-    })
-    let lastSetSuccess: A | undefined
-    get.subscribe(argRx, ([, arg]) => {
-      if (arg === undefined) {
-        return
-      }
-      lastSetSuccess = arg
-      get.setSelf(arg)
-    })
-    get.subscribe(options.fn, (value) => {
-      if (value.waiting || Result.isInitial(value)) {
-        return
-      } else if (Result.isFailure(value)) {
-        return get.setSelf(lastValue)
-      }
-      if (options.disableRefresh !== true) {
-        return get.refresh(self)
-      }
-      const current = Option.getOrUndefined(get.self<Result.Success<any, any>>())!
-      if (current === lastSetSuccess) {
-        get.setSelf(Result.success(current.value))
-      }
+): RxResultFn<OW, XA, XE> => {
+  const transition = state<Result.Result<W, unknown>>(Result.initial())
+  return fn((arg: OW, get) => {
+    const value = options.updateToValue(arg, get(self))
+    get.set(transition, Result.success(value, { waiting: true }))
+    get.set(self, transition)
+    get.set(options.fn, arg)
+    return Effect.onExit(get.result(options.fn, { suspendOnWaiting: true }), (exit) => {
+      get.set(transition, Result.fromExit(Exit.as(exit, value)))
+      return Effect.void
     })
-    return lastValue
-  }, (ctx, value) => {
-    const current = ctx.get(self)
-    const arg = options.updateToValue(value, current)
-    ctx.set(argRx, [++counter, Result.isResult(current) ? Result.success(arg, { waiting: true }) as A : arg as A])
-    ctx.set(options.fn, value)
-  }, (refresh) => {
-    refresh(self)
   })
 })
 

packages/rx/test/Rx.test.ts

@@ -1045,14 +1045,15 @@ describe("Rx", () => {
     })
   })
 
-  describe("withOptimisticSet", () => {
+  describe("optimistic", () => {
     it("non-Result", async () => {
       const latch = Effect.unsafeMakeLatch()
       const r = Registry.make()
       let i = 0
       const rx = Rx.make(() => i)
-      const optimisticRx = rx.pipe(
-        Rx.withOptimisticSet({
+      const optimisticRx = rx.pipe(Rx.optimistic)
+      const fn = optimisticRx.pipe(
+        Rx.optimisticFn({
           updateToValue: (value) => value,
           fn: Rx.fn(Effect.fnUntraced(function*() {
             yield* latch.await
@@ -1063,7 +1064,7 @@ describe("Rx", () => {
 
       expect(r.get(rx)).toEqual(0)
       expect(r.get(optimisticRx)).toEqual(0)
-      r.set(optimisticRx, 1)
+      r.set(fn, 1)
       i = 2
 
       // optimistic phase: the optimistic value is set, but the true value is not
@@ -1084,7 +1085,10 @@ describe("Rx", () => {
       let i = 0
       const rx = Rx.make(Effect.sync(() => i))
       const optimisticRx = rx.pipe(
-        Rx.withOptimisticSet({
+        Rx.optimistic
+      )
+      const fn = optimisticRx.pipe(
+        Rx.optimisticFn({
           updateToValue: (value) => value,
           fn: Rx.fn(Effect.fnUntraced(function*() {
             yield* latch.await
@@ -1095,7 +1099,7 @@ describe("Rx", () => {
 
       expect(r.get(rx)).toEqual(Result.success(0))
       expect(r.get(optimisticRx)).toEqual(Result.success(0))
-      r.set(optimisticRx, 1)
+      r.set(fn, 1)
       i = 2
 
       // optimistic phase: the optimistic value is set, but the true value is not
@@ -1116,7 +1120,10 @@ describe("Rx", () => {
       const i = 0
       const rx = Rx.make(() => i)
       const optimisticRx = rx.pipe(
-        Rx.withOptimisticSet({
+        Rx.optimistic
+      )
+      const fn = optimisticRx.pipe(
+        Rx.optimisticFn({
           updateToValue: (value) => value,
           fn: Rx.fn(Effect.fnUntraced(function*() {
             yield* latch.await
@@ -1128,7 +1135,7 @@ describe("Rx", () => {
 
       expect(r.get(rx)).toEqual(0)
       expect(r.get(optimisticRx)).toEqual(0)
-      r.set(optimisticRx, 1)
+      r.set(fn, 1)
 
       // optimistic phase: the optimistic value is set, but the true value is not
       expect(r.get(rx)).toEqual(0)