Add additional error handling primitives for Subscribable

packages/effect-lens-next/src/Subscribable.test.ts

@@ -37,6 +37,41 @@ describe("Subscribable", () => {
         expect(observed).toEqual(["get", "changes"])
     })
 
+    test("catch recovers get and changes with the corresponding fallback channel", async () => {
+        const source = Subscribable.make({
+            get: Effect.fail("get"),
+            changes: Stream.fail("changes"),
+        })
+        const recovered = source.pipe(Subscribable.catch(() => Subscribable.make({
+            get: Effect.succeed("fallback-get"),
+            changes: Stream.succeed("fallback-changes"),
+        })))
+
+        const result = await Effect.runPromise(Effect.gen(function*() {
+            const current = yield* recovered.get
+            const changes = yield* Stream.runCollect(recovered.changes)
+            return [current, Array.from(changes)]
+        }))
+
+        expect(result).toEqual(["fallback-get", ["fallback-changes"]])
+    })
+
+    test("orElseSucceed recovers errors from get and changes", async () => {
+        const source = Subscribable.make({
+            get: Effect.fail("get"),
+            changes: Stream.fail("changes"),
+        })
+        const recovered = Subscribable.orElseSucceed(source, () => "fallback")
+
+        const result = await Effect.runPromise(Effect.gen(function*() {
+            const current = yield* recovered.get
+            const changes = yield* Stream.runCollect(recovered.changes)
+            return [current, Array.from(changes)]
+        }))
+
+        expect(result).toEqual(["fallback", ["fallback"]])
+    })
+
     test("focusArrayLength reads the current array length and reflects updates", async () => {
         const result = await Effect.runPromise(
             Effect.flatMap(

packages/effect-lens-next/src/Subscribable.ts

@@ -1,4 +1,4 @@
-import { Array, type Cause, Chunk, Effect, Function, Iterable, Option, Pipeable, Predicate, Stream } from "effect"
+import { Array, type Cause, Chunk, Effect, Function, Iterable, Option, Pipeable, Predicate, type Result, type Schedule, Stream } from "effect"
 
 
 export const SubscribableTypeId: unique symbol = Symbol.for("@effect-fc/Lens/v4/Subscribable")
@@ -124,6 +124,82 @@ export const tapError: {
     get changes() { return Stream.tapError(self.changes, f) },
 }))
 
+const catch_: {
+    <E, B, E2, R2>(f: (error: NoInfer<E>) => Subscribable<B, E2, R2>): <A, R>(self: Subscribable<A, E, R>) => Subscribable<A | B, E2, R | R2>
+    <A, E, R, B, E2, R2>(self: Subscribable<A, E, R>, f: (error: NoInfer<E>) => Subscribable<B, E2, R2>): Subscribable<A | B, E2, R | R2>
+} = Function.dual(2, <A, E, R, B, E2, R2>(
+    self: Subscribable<A, E, R>,
+    f: (error: NoInfer<E>) => Subscribable<B, E2, R2>,
+) => make({
+    get get() { return Effect.catch(self.get, error => f(error).get) },
+    get changes() { return Stream.catch(self.changes, error => f(error).changes) },
+}))
+
+/** Recovers from typed errors with another `Subscribable`. */
+export { catch_ as catch }
+
+/** Recovers from all failure causes with another `Subscribable`. */
+export const catchCause: {
+    <E, B, E2, R2>(f: (cause: Cause.Cause<NoInfer<E>>) => Subscribable<B, E2, R2>): <A, R>(self: Subscribable<A, E, R>) => Subscribable<A | B, E2, R | R2>
+    <A, E, R, B, E2, R2>(self: Subscribable<A, E, R>, f: (cause: Cause.Cause<NoInfer<E>>) => Subscribable<B, E2, R2>): Subscribable<A | B, E2, R | R2>
+} = Function.dual(2, <A, E, R, B, E2, R2>(
+    self: Subscribable<A, E, R>,
+    f: (cause: Cause.Cause<NoInfer<E>>) => Subscribable<B, E2, R2>,
+) => make({
+    get get() { return Effect.catchCause(self.get, cause => f(cause).get) },
+    get changes() { return Stream.catchCause(self.changes, cause => f(cause).changes) },
+}))
+
+/** Runs an effect when either channel fails, exposing the complete failure cause. */
+export const tapCause: {
+    <E, B, E2, R2>(f: (cause: Cause.Cause<NoInfer<E>>) => Effect.Effect<B, E2, R2>): <A, R>(self: Subscribable<A, E, R>) => Subscribable<A, E | E2, R | R2>
+    <A, E, R, B, E2, R2>(self: Subscribable<A, E, R>, f: (cause: Cause.Cause<NoInfer<E>>) => Effect.Effect<B, E2, R2>): Subscribable<A, E | E2, R | R2>
+} = Function.dual(2, <A, E, R, B, E2, R2>(
+    self: Subscribable<A, E, R>,
+    f: (cause: Cause.Cause<NoInfer<E>>) => Effect.Effect<B, E2, R2>,
+) => make({
+    get get() { return Effect.tapCause(self.get, f) },
+    get changes() { return Stream.tapCause(self.changes, f) },
+}))
+
+/** Falls back to another `Subscribable` when either channel fails. */
+export const orElse: {
+    <B, E2, R2>(that: () => Subscribable<B, E2, R2>): <A, E, R>(self: Subscribable<A, E, R>) => Subscribable<A | B, E2, R | R2>
+    <A, E, R, B, E2, R2>(self: Subscribable<A, E, R>, that: () => Subscribable<B, E2, R2>): Subscribable<A | B, E2, R | R2>
+} = Function.dual(2, <A, E, R, B, E2, R2>(
+    self: Subscribable<A, E, R>,
+    that: () => Subscribable<B, E2, R2>,
+) => catch_(self, that))
+
+/** Replaces typed errors from either channel with a lazily evaluated value. */
+export const orElseSucceed: {
+    <B>(value: () => B): <A, E, R>(self: Subscribable<A, E, R>) => Subscribable<A | B, never, R>
+    <A, E, R, B>(self: Subscribable<A, E, R>, value: () => B): Subscribable<A | B, never, R>
+} = Function.dual(2, <A, E, R, B>(self: Subscribable<A, E, R>, value: () => B) => make({
+    get get() { return Effect.orElseSucceed(self.get, value) },
+    get changes() { return Stream.orElseSucceed(self.changes, value) },
+}))
+
+/** Retries failures from both channels according to the supplied schedule. */
+export const retry: {
+    <E, X, E2, R2>(policy: Schedule.Schedule<X, NoInfer<E>, E2, R2>): <A, R>(self: Subscribable<A, E, R>) => Subscribable<A, E | E2, R | R2>
+    <A, E, R, X, E2, R2>(self: Subscribable<A, E, R>, policy: Schedule.Schedule<X, NoInfer<E>, E2, R2>): Subscribable<A, E | E2, R | R2>
+} = Function.dual(2, <A, E, R, X, E2, R2>(
+    self: Subscribable<A, E, R>,
+    policy: Schedule.Schedule<X, NoInfer<E>, E2, R2>,
+) => make({
+    get get() { return Effect.retry(self.get, policy) },
+    get changes() { return Stream.retry(self.changes, policy) },
+}))
+
+/** Converts typed failures from both channels into `Result` values. */
+export const result = <A, E, R>(
+    self: Subscribable<A, E, R>,
+): Subscribable<Result.Result<A, E>, never, R> => make({
+    get get() { return Effect.result(self.get) },
+    get changes() { return Stream.result(self.changes) },
+})
+
 
 /** Narrows the focus to a field of an object. */
 export const focusObjectOn: {