Add error handling primitives to Subscribable

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

@@ -1,10 +1,42 @@
 import { describe, expect, test } from "bun:test"
-import { Chunk, Effect, SubscriptionRef } from "effect"
+import { Chunk, Effect, Stream, SubscriptionRef } from "effect"
 import * as Lens from "./Lens.js"
 import * as Subscribable from "./Subscribable.js"
 
 
 describe("Subscribable", () => {
+    test("mapError transforms errors from get and changes", async () => {
+        const source = Subscribable.make({
+            get: Effect.fail("get"),
+            changes: Stream.fail("changes"),
+        })
+        const mapped = source.pipe(Subscribable.mapError((error: string) => `mapped:${error}`))
+
+        const result = await Effect.runPromise(Effect.gen(function*() {
+            const getError = yield* Effect.flip(mapped.get)
+            const changesError = yield* Effect.flip(Stream.runDrain(mapped.changes))
+            return [getError, changesError]
+        }))
+
+        expect(result).toEqual(["mapped:get", "mapped:changes"])
+    })
+
+    test("tapError observes errors from get and changes", async () => {
+        const observed: Array<string> = []
+        const source = Subscribable.make({
+            get: Effect.fail("get"),
+            changes: Stream.fail("changes"),
+        })
+        const tapped = Subscribable.tapError(source, error => Effect.sync(() => observed.push(error)))
+
+        await Effect.runPromise(Effect.gen(function*() {
+            yield* Effect.flip(tapped.get)
+            yield* Effect.flip(Stream.runDrain(tapped.changes))
+        }))
+
+        expect(observed).toEqual(["get", "changes"])
+    })
+
     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

@@ -100,6 +100,31 @@ export const mapOptionEffect: {
 })))
 
 
+/** Maps errors from both the current value and the stream of changes. */
+export const mapError: {
+    <E, E2>(f: (error: NoInfer<E>) => E2): <A, R>(self: Subscribable<A, E, R>) => Subscribable<A, E2, R>
+    <A, E, R, E2>(self: Subscribable<A, E, R>, f: (error: NoInfer<E>) => E2): Subscribable<A, E2, R>
+} = Function.dual(2, <A, E, R, E2>(
+    self: Subscribable<A, E, R>,
+    f: (error: NoInfer<E>) => E2,
+) => make({
+    get get() { return Effect.mapError(self.get, f) },
+    get changes() { return Stream.mapError(self.changes, f) },
+}))
+
+/** Runs an effect when either the current value or the stream of changes fails. */
+export const tapError: {
+    <E, B, E2, R2>(f: (error: 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: (error: 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: (error: NoInfer<E>) => Effect.Effect<B, E2, R2>,
+) => make({
+    get get() { return Effect.tapError(self.get, f) },
+    get changes() { return Stream.tapError(self.changes, f) },
+}))
+
+
 /** Narrows the focus to a field of an object. */
 export const focusObjectOn: {
     <A extends object, K extends keyof A>(key: K): <E, R>(self: Subscribable<A, E, R>) => Subscribable<A[K], E, R>