Merge branch 'next' of git.valverde.cloud:Thilawyn/reffuse into next

Co-authored-by: Julien Valverdé <julien.valverde@mailo.com>
Reviewed-on: https://gitea:3000/Thilawyn/reffuse/pulls/14

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@reffuse/monorepo",
-  "packageManager": "bun@1.2.9",
+  "packageManager": "bun@1.2.13",
   "private": true,
   "workspaces": [
     "./packages/*"
@@ -15,9 +15,9 @@
     "clean:node": "rm -rf node_modules"
   },
   "devDependencies": {
-    "npm-check-updates": "^17.1.18",
+    "npm-check-updates": "^18.0.1",
     "npm-sort": "^0.0.4",
-    "turbo": "^2.5.0",
+    "turbo": "^2.5.3",
     "typescript": "^5.8.3"
   }
 }

packages/example/package.json

@@ -11,41 +11,41 @@
     "preview": "vite preview"
   },
   "devDependencies": {
-    "@eslint/js": "^9.24.0",
+    "@eslint/js": "^9.26.0",
-    "@tanstack/react-router": "^1.115.3",
+    "@tanstack/react-router": "^1.120.3",
-    "@tanstack/react-router-devtools": "^1.115.3",
+    "@tanstack/react-router-devtools": "^1.120.3",
-    "@tanstack/router-plugin": "^1.115.3",
+    "@tanstack/router-plugin": "^1.120.3",
     "@thilawyn/thilaschema": "^0.1.4",
-    "@types/react": "^19.1.1",
+    "@types/react": "^19.1.4",
-    "@types/react-dom": "^19.1.2",
+    "@types/react-dom": "^19.1.5",
-    "@vitejs/plugin-react": "^4.3.4",
+    "@vitejs/plugin-react": "^4.4.1",
-    "eslint": "^9.24.0",
+    "eslint": "^9.26.0",
     "eslint-plugin-react-hooks": "^5.2.0",
-    "eslint-plugin-react-refresh": "^0.4.19",
+    "eslint-plugin-react-refresh": "^0.4.20",
-    "globals": "^16.0.0",
+    "globals": "^16.1.0",
     "react": "^19.1.0",
     "react-dom": "^19.1.0",
-    "typescript-eslint": "^8.29.1",
+    "typescript-eslint": "^8.32.1",
-    "vite": "^6.2.6"
+    "vite": "^6.3.5"
   },
   "dependencies": {
-    "@effect/platform": "^0.80.8",
+    "@effect/platform": "^0.82.1",
-    "@effect/platform-browser": "^0.59.8",
+    "@effect/platform-browser": "^0.62.1",
     "@radix-ui/themes": "^3.2.1",
     "@reffuse/extension-lazyref": "workspace:*",
     "@reffuse/extension-query": "workspace:*",
     "@typed/async-data": "^0.13.1",
     "@typed/id": "^0.17.2",
     "@typed/lazy-ref": "^0.3.3",
-    "effect": "^3.14.8",
+    "effect": "^3.15.1",
-    "lucide-react": "^0.487.0",
+    "lucide-react": "^0.510.0",
     "mobx": "^6.13.7",
     "reffuse": "workspace:*"
   },
   "overrides": {
-    "effect": "^3.14.8",
+    "effect": "^3.15.1",
-    "@effect/platform": "^0.80.8",
+    "@effect/platform": "^0.82.1",
-    "@effect/platform-browser": "^0.59.8",
+    "@effect/platform-browser": "^0.62.1",
     "@typed/lazy-ref": "^0.3.3",
     "@typed/async-data": "^0.13.1"
   }

packages/example/src/VQueryErrorHandler.tsx

@@ -1,7 +1,7 @@
 import { AlertDialog, Button, Flex, Text } from "@radix-ui/themes"
 import { Cause, Console, Effect, Either, flow, Match, Option, Stream } from "effect"
 import { useState } from "react"
-import { AppQueryErrorHandler } from "./query"
+import { AppQueryClient } from "./query"
 import { R } from "./reffuse"
 
 
@@ -9,8 +9,8 @@ export function VQueryErrorHandler() {
     const [open, setOpen] = useState(false)
 
     const error = R.useSubscribeStream(
-        R.useMemo(() => AppQueryErrorHandler.pipe(
+        R.useMemo(() => AppQueryClient.pipe(
-            Effect.map(handler => handler.errors.pipe(
+            Effect.map(client => client.errorHandler.errors.pipe(
                 Stream.changes,
                 Stream.tap(Console.error),
                 Stream.tap(() => Effect.sync(() => setOpen(true))),

packages/example/src/domain/Todo.ts

@@ -1,6 +1,5 @@
 import { ThSchema } from "@thilawyn/thilaschema"
-import { GetRandomValues, makeUuid4 } from "@typed/id"
+import { Schema } from "effect"
-import { Effect, Schema } from "effect"
 
 
 export class Todo extends Schema.Class<Todo>("Todo")({
@@ -18,9 +17,4 @@ export const TodoFromJsonStruct = Schema.Struct({
     ThSchema.assertEncodedJsonifiable
 )
 
-export const TodoFromJson = TodoFromJsonStruct.pipe(Schema.compose(Todo))
+export const TodoFromJson = Schema.compose(TodoFromJsonStruct, Todo)
-
-
-export const generateUniqueID = makeUuid4.pipe(
-    Effect.provide(GetRandomValues.CryptoRandom)
-)

packages/example/src/query.ts

@@ -9,11 +9,7 @@ export class AppQueryErrorHandler extends QueryErrorHandler.Service<AppQueryErro
     "AppQueryErrorHandler",
 
     (self, failure, defect) => self.pipe(
-        Effect.catchTags({
+        Effect.catchTag("RequestError", "ResponseError", failure),
-            RequestError: failure,
-            ResponseError: failure,
-        }),
-
         Effect.catchAllDefect(defect),
     ),
 ) {}

packages/example/src/reffuse.ts

@@ -3,12 +3,11 @@ import { Clipboard, Geolocation, Permissions } from "@effect/platform-browser"
 import { LazyRefExtension } from "@reffuse/extension-lazyref"
 import { QueryExtension } from "@reffuse/extension-query"
 import { Reffuse, ReffuseContext } from "reffuse"
-import { AppQueryClient, AppQueryErrorHandler } from "./query"
+import { AppQueryClient } from "./query"
 
 
 export const RootContext = ReffuseContext.make<
     | AppQueryClient
-    | AppQueryErrorHandler
     | Clipboard.Clipboard
     | Geolocation.Geolocation
     | Permissions.Permissions

packages/example/src/routeTree.gen.ts

@@ -19,6 +19,7 @@ import { Route as LazyrefImport } from './routes/lazyref'
 import { Route as CountImport } from './routes/count'
 import { Route as BlankImport } from './routes/blank'
 import { Route as IndexImport } from './routes/index'
+import { Route as StreamsPullImport } from './routes/streams/pull'
 import { Route as QueryUsequeryImport } from './routes/query/usequery'
 import { Route as QueryUsemutationImport } from './routes/query/usemutation'
 import { Route as QueryServiceImport } from './routes/query/service'
@@ -73,6 +74,12 @@ const IndexRoute = IndexImport.update({
   getParentRoute: () => rootRoute,
 } as any)
 
+const StreamsPullRoute = StreamsPullImport.update({
+  id: '/streams/pull',
+  path: '/streams/pull',
+  getParentRoute: () => rootRoute,
+} as any)
+
 const QueryUsequeryRoute = QueryUsequeryImport.update({
   id: '/query/usequery',
   path: '/query/usequery',
@@ -172,6 +179,13 @@ declare module '@tanstack/react-router' {
       preLoaderRoute: typeof QueryUsequeryImport
       parentRoute: typeof rootRoute
     }
+    '/streams/pull': {
+      id: '/streams/pull'
+      path: '/streams/pull'
+      fullPath: '/streams/pull'
+      preLoaderRoute: typeof StreamsPullImport
+      parentRoute: typeof rootRoute
+    }
   }
 }
 
@@ -189,6 +203,7 @@ export interface FileRoutesByFullPath {
   '/query/service': typeof QueryServiceRoute
   '/query/usemutation': typeof QueryUsemutationRoute
   '/query/usequery': typeof QueryUsequeryRoute
+  '/streams/pull': typeof StreamsPullRoute
 }
 
 export interface FileRoutesByTo {
@@ -203,6 +218,7 @@ export interface FileRoutesByTo {
   '/query/service': typeof QueryServiceRoute
   '/query/usemutation': typeof QueryUsemutationRoute
   '/query/usequery': typeof QueryUsequeryRoute
+  '/streams/pull': typeof StreamsPullRoute
 }
 
 export interface FileRoutesById {
@@ -218,6 +234,7 @@ export interface FileRoutesById {
   '/query/service': typeof QueryServiceRoute
   '/query/usemutation': typeof QueryUsemutationRoute
   '/query/usequery': typeof QueryUsequeryRoute
+  '/streams/pull': typeof StreamsPullRoute
 }
 
 export interface FileRouteTypes {
@@ -234,6 +251,7 @@ export interface FileRouteTypes {
     | '/query/service'
     | '/query/usemutation'
     | '/query/usequery'
+    | '/streams/pull'
   fileRoutesByTo: FileRoutesByTo
   to:
     | '/'
@@ -247,6 +265,7 @@ export interface FileRouteTypes {
     | '/query/service'
     | '/query/usemutation'
     | '/query/usequery'
+    | '/streams/pull'
   id:
     | '__root__'
     | '/'
@@ -260,6 +279,7 @@ export interface FileRouteTypes {
     | '/query/service'
     | '/query/usemutation'
     | '/query/usequery'
+    | '/streams/pull'
   fileRoutesById: FileRoutesById
 }
 
@@ -275,6 +295,7 @@ export interface RootRouteChildren {
   QueryServiceRoute: typeof QueryServiceRoute
   QueryUsemutationRoute: typeof QueryUsemutationRoute
   QueryUsequeryRoute: typeof QueryUsequeryRoute
+  StreamsPullRoute: typeof StreamsPullRoute
 }
 
 const rootRouteChildren: RootRouteChildren = {
@@ -289,6 +310,7 @@ const rootRouteChildren: RootRouteChildren = {
   QueryServiceRoute: QueryServiceRoute,
   QueryUsemutationRoute: QueryUsemutationRoute,
   QueryUsequeryRoute: QueryUsequeryRoute,
+  StreamsPullRoute: StreamsPullRoute,
 }
 
 export const routeTree = rootRoute
@@ -311,7 +333,8 @@ export const routeTree = rootRoute
         "/todos",
         "/query/service",
         "/query/usemutation",
-        "/query/usequery"
+        "/query/usequery",
+        "/streams/pull"
       ]
     },
     "/": {
@@ -346,6 +369,9 @@ export const routeTree = rootRoute
     },
     "/query/usequery": {
       "filePath": "query/usequery.tsx"
+    },
+    "/streams/pull": {
+      "filePath": "streams/pull.tsx"
     }
   }
 }

packages/example/src/routes/count.tsx

@@ -1,6 +1,6 @@
 import { R } from "@/reffuse"
 import { createFileRoute } from "@tanstack/react-router"
-import { Ref } from "effect"
+import { Effect, Ref } from "effect"
 
 
 export const Route = createFileRoute("/count")({
@@ -11,14 +11,13 @@ function Count() {
 
     const runSync = R.useRunSync()
 
-    const countRef = R.useRef(0)
+    const countRef = R.useRef(() => Effect.succeed(0))
-    const [count] = R.useRefState(countRef)
+    const [count] = R.useSubscribeRefs(countRef)
 
 
     return (
         <div className="container mx-auto">
-            {/* <button onClick={() => setCount((count) => count + 1)}> */}
+            <button onClick={() => runSync(Ref.update(countRef, count => count + 1))}>
-            <button onClick={() => Ref.update(countRef, count => count + 1).pipe(runSync)}>
                 count is {count}
             </button>
         </div>

packages/example/src/routes/query/service.tsx

@@ -14,7 +14,7 @@ export const Route = createFileRoute("/query/service")({
 
 function RouteComponent() {
     const query = R.useQuery({
-        key: R.useStreamFromValues(["uuid4", 10 as number]),
+        key: R.useStreamFromReactiveValues(["uuid4", 10 as number]),
         query: ([, count]) => Console.log(`Querying ${ count } IDs...`).pipe(
             Effect.andThen(Effect.sleep("500 millis")),
             Effect.andThen(HttpClient.get(`https://www.uuidtools.com/api/generate/v4/count/${ count }`)),

packages/example/src/routes/query/usequery.tsx

@@ -20,7 +20,7 @@ function RouteComponent() {
     const [count, setCount] = useState(1)
 
     const query = R.useQuery({
-        key: R.useStreamFromValues(["uuid4", count]),
+        key: R.useStreamFromReactiveValues(["uuid4", count]),
         query: ([, count]) => Console.log(`Querying ${ count } IDs...`).pipe(
             Effect.andThen(Effect.sleep("500 millis")),
             Effect.andThen(HttpClient.get(`https://www.uuidtools.com/api/generate/v4/count/${ count }`)),

packages/example/src/routes/streams/pull.tsx

@@ -0,0 +1,34 @@
+import { R } from "@/reffuse"
+import { Button, Flex, Text } from "@radix-ui/themes"
+import { createFileRoute } from "@tanstack/react-router"
+import { Chunk, Effect, Exit, Option, Queue, Random, Scope, Stream } from "effect"
+import { useMemo, useState } from "react"
+
+
+export const Route = createFileRoute("/streams/pull")({
+    component: RouteComponent
+})
+
+function RouteComponent() {
+    const stream = useMemo(() => Stream.repeatEffect(Random.nextInt), [])
+    const streamScope = R.useScope([stream], { finalizerExecutionMode: "fork" })
+
+    const queue = R.useMemo(() => Effect.provideService(Stream.toQueueOfElements(stream), Scope.Scope, streamScope), [streamScope])
+
+    const [value, setValue] = useState(Option.none<number>())
+    const pullLatest = R.useCallbackSync(() => Queue.takeAll(queue).pipe(
+        Effect.flatMap(Chunk.last),
+        Effect.flatMap(Exit.matchEffect({
+            onSuccess: Effect.succeed,
+            onFailure: Effect.fail,
+        })),
+        Effect.tap(v => Effect.sync(() => setValue(Option.some(v)))),
+    ), [queue])
+
+    return (
+        <Flex direction="column" align="center" gap="2">
+            {Option.isSome(value) && <Text>{value.value}</Text>}
+            <Button onClick={pullLatest}>Pull latest</Button>
+        </Flex>
+    )
+}

packages/example/src/routes/tests.tsx

@@ -2,7 +2,11 @@ import { R } from "@/reffuse"
 import { Button, Flex, Text } from "@radix-ui/themes"
 import { createFileRoute } from "@tanstack/react-router"
 import { GetRandomValues, makeUuid4 } from "@typed/id"
-import { Console, Effect, Ref } from "effect"
+import { Console, Effect, Option } from "effect"
+import { useEffect, useState } from "react"
+
+
+const makeUuid = Effect.provide(makeUuid4, GetRandomValues.CryptoRandom)
 
 
 export const Route = createFileRoute("/tests")({
@@ -10,48 +14,34 @@ export const Route = createFileRoute("/tests")({
 })
 
 function RouteComponent() {
-    const deepRef = R.useRef({ value: "poulet" })
+    const runSync = R.useRunSync()
-    const deepValueRef = R.useSubRef(deepRef, ["value"])
 
-    // const value = R.useMemoScoped(Effect.addFinalizer(() => Console.log("cleanup")).pipe(
+    const [uuid, setUuid] = useState(R.useMemo(() => makeUuid, []))
-    //     Effect.andThen(makeUuid4),
+    const generateUuid = R.useCallbackSync(() => makeUuid.pipe(
-    //     Effect.provide(GetRandomValues.CryptoRandom),
+        Effect.tap(v => Effect.sync(() => setUuid(v)))
-    // ), [])
-    // console.log(value)
-
-    R.useFork(() => Effect.addFinalizer(() => Console.log("cleanup")).pipe(
-        Effect.andThen(Console.log("ouient")),
-        Effect.delay("1 second"),
     ), [])
 
+    const uuidStream = R.useStreamFromReactiveValues([uuid])
+    const uuidStreamLatestValue = R.useSubscribeStream(uuidStream)
 
-    const uuidRef = R.useRef("none")
+    const [, scopeLayer] = R.useScope([uuid])
-    const anotherRef = R.useRef(69)
-
-
-    const logValue = R.useCallbackSync(Effect.fn(function*(value: string) {
-        yield* Effect.log(value)
-    }), [])
-
-    const generateUuid = R.useCallbackSync(() => makeUuid4.pipe(
-        Effect.provide(GetRandomValues.CryptoRandom),
-        Effect.tap(v => Ref.set(uuidRef, v)),
-        Effect.tap(v => Ref.set(deepValueRef, v)),
-    ), [])
 
+    useEffect(() => Effect.addFinalizer(() => Console.log("Scope cleanup!")).pipe(
+        Effect.andThen(Console.log("Scope changed")),
+        Effect.provide(scopeLayer),
+        runSync,
+    ), [scopeLayer, runSync])
 
     return (
-        <Flex direction="row" justify="center" align="center" gap="2">
+        <Flex direction="column" justify="center" align="center" gap="2">
-            <R.SubscribeRefs refs={[uuidRef, anotherRef]}>
+            <Text>{uuid}</Text>
-                {(uuid, anotherRef) => <Text>{uuid} / {anotherRef}</Text>}
+            <Button onClick={generateUuid}>Generate UUID</Button>
-            </R.SubscribeRefs>
+            <Text>
-
+                {Option.match(uuidStreamLatestValue, {
-            <R.SubscribeRefs refs={[deepRef, deepValueRef]}>
+                    onSome: ([v]) => v,
-                {(deep, deepValue) => <Text>{JSON.stringify(deep)} / {deepValue}</Text>}
+                    onNone: () => <></>,
-            </R.SubscribeRefs>
+                })}
-
+            </Text>
-            <Button onClick={() => logValue("test")}>Log value</Button>
-            <Button onClick={() => generateUuid()}>Generate UUID</Button>
         </Flex>
     )
 }

packages/example/src/routes/todos.tsx

@@ -26,7 +26,7 @@ function Todos() {
 
     return (
         <Container>
-            <TodosContext.Provider layer={todosLayer}>
+            <TodosContext.Provider layer={todosLayer} finalizerExecutionMode="fork">
                 <VTodos />
             </TodosContext.Provider>
         </Container>

packages/example/src/todos/services/TodosState.ts

@@ -2,68 +2,43 @@ import { Todo } from "@/domain"
 import { KeyValueStore } from "@effect/platform"
 import { BrowserKeyValueStore } from "@effect/platform-browser"
 import { PlatformError } from "@effect/platform/Error"
-import { Chunk, Context, Effect, identity, Layer, ParseResult, Ref, Schema, SubscriptionRef } from "effect"
+import { Chunk, Context, Effect, identity, Layer, ParseResult, Ref, Schema, Stream, SubscriptionRef } from "effect"
 
 
 export class TodosState extends Context.Tag("TodosState")<TodosState, {
     readonly todos: SubscriptionRef.SubscriptionRef<Chunk.Chunk<Todo.Todo>>
-
+    readonly load: Effect.Effect<void, PlatformError | ParseResult.ParseError>
-    readonly readFromLocalStorage: Effect.Effect<void, PlatformError | ParseResult.ParseError>
+    readonly save: Effect.Effect<void, PlatformError | ParseResult.ParseError>
-    readonly saveToLocalStorage: Effect.Effect<void, PlatformError | ParseResult.ParseError>
-
-    readonly prepend: (todo: Todo.Todo) => Effect.Effect<void>
-    readonly replace: (index: number, todo: Todo.Todo) => Effect.Effect<void>
-    readonly remove: (index: number) => Effect.Effect<void>
-    // readonly moveUp: (index: number) => Effect.Effect<void, Cause.NoSuchElementException>
-    // readonly moveDown: (index: number) => Effect.Effect<void, Cause.NoSuchElementException>
 }>() {}
 
 
 export const make = (key: string) => Layer.effect(TodosState, Effect.gen(function*() {
-    const todos = yield* SubscriptionRef.make(Chunk.empty<Todo.Todo>())
-
     const readFromLocalStorage = KeyValueStore.KeyValueStore.pipe(
         Effect.flatMap(kv => kv.get(key)),
         Effect.flatMap(identity),
-        Effect.flatMap(Schema.parseJson().pipe(
+        Effect.flatMap(Schema.decode(
-            Schema.compose(Schema.Chunk(Todo.TodoFromJson)),
+            Schema.compose(Schema.parseJson(), Schema.Chunk(Todo.TodoFromJson))
-            Schema.decode,
         )),
-        Effect.flatMap(v => Ref.set(todos, v)),
+        Effect.catchTag("NoSuchElementException", () => Effect.succeed(Chunk.empty<Todo.Todo>())),
-
-        Effect.catchTag("NoSuchElementException", () => Ref.set(todos, Chunk.empty())),
-
         Effect.provide(BrowserKeyValueStore.layerLocalStorage),
     )
 
-    const saveToLocalStorage = Effect.all([KeyValueStore.KeyValueStore, todos]).pipe(
+    const writeToLocalStorage = (values: Chunk.Chunk<Todo.Todo>) => KeyValueStore.KeyValueStore.pipe(
-        Effect.flatMap(([kv, values]) => values.pipe(
+        Effect.flatMap(kv => values.pipe(
-            Schema.parseJson().pipe(
+            Schema.encode(
-                Schema.compose(Schema.Chunk(Todo.TodoFromJson)),
+                Schema.compose(Schema.parseJson(), Schema.Chunk(Todo.TodoFromJson))
-                Schema.encode,
             ),
             Effect.flatMap(v => kv.set(key, v)),
         )),
-
         Effect.provide(BrowserKeyValueStore.layerLocalStorage),
     )
 
-    const prepend = (todo: Todo.Todo) => Ref.update(todos, Chunk.prepend(todo))
+    const todos = yield* SubscriptionRef.make(yield* readFromLocalStorage)
-    const replace = (index: number, todo: Todo.Todo) => Ref.update(todos, Chunk.replace(index, todo))
+    const load = Effect.flatMap(readFromLocalStorage, v => Ref.set(todos, v))
-    const remove = (index: number) => Ref.update(todos, Chunk.remove(index))
+    const save = Effect.flatMap(todos, writeToLocalStorage)
 
-    // const moveUp = (index: number) => Effect.gen(function*() {
+    // Sync changes with local storage
+    yield* Effect.forkScoped(Stream.runForEach(todos.changes, writeToLocalStorage))
 
-    // })
+    return { todos, load, save }
-
-    yield* readFromLocalStorage
-
-    return {
-        todos,
-        readFromLocalStorage,
-        saveToLocalStorage,
-        prepend,
-        replace,
-        remove,
-    }
 }))

packages/example/src/todos/views/VNewTodo.tsx

@@ -1,25 +1,27 @@
 import { Todo } from "@/domain"
 import { Box, Button, Card, Flex, TextArea } from "@radix-ui/themes"
-import { Effect, Option, SubscriptionRef } from "effect"
+import { GetRandomValues, makeUuid4 } from "@typed/id"
+import { Chunk, Effect, Option, Ref } from "effect"
 import { R } from "../reffuse"
 import { TodosState } from "../services"
 
 
-const createEmptyTodo = Todo.generateUniqueID.pipe(
+const createEmptyTodo = makeUuid4.pipe(
-    Effect.map(id => Todo.Todo.make({
+    Effect.map(id => Todo.Todo.make({ id, content: "", completedAt: Option.none()}, true)),
-        id,
+    Effect.provide(GetRandomValues.CryptoRandom),
-        content: "",
-        completedAt: Option.none(),
-    }, true))
 )
 
 
 export function VNewTodo() {
 
-    const runSync = R.useRunSync()
+    const todoRef = R.useRef(() => createEmptyTodo)
+    const [content, setContent] = R.useRefState(R.useSubRefFromPath(todoRef, ["content"]))
 
-    const todoRef = R.useMemo(() => createEmptyTodo.pipe(Effect.flatMap(SubscriptionRef.make)), [])
+    const add = R.useCallbackSync(() => Effect.all([TodosState.TodosState, todoRef]).pipe(
-    const [todo, setTodo] = R.useRefState(todoRef)
+        Effect.flatMap(([state, todo]) => Ref.update(state.todos, Chunk.prepend(todo))),
+        Effect.andThen(createEmptyTodo),
+        Effect.flatMap(v => Ref.set(todoRef, v)),
+    ), [todoRef])
 
 
     return (
@@ -27,23 +29,12 @@ export function VNewTodo() {
             <Card>
                 <Flex direction="column" align="stretch" gap="2">
                     <TextArea
-                        value={todo.content}
+                        value={content}
-                        onChange={e => setTodo(prev =>
+                        onChange={e => setContent(e.target.value)}
-                            Todo.Todo.make({ ...prev, content: e.target.value }, true)
-                        )}
                     />
 
                     <Flex direction="row" justify="center" align="center">
-                        <Button
+                        <Button onClick={add}>Add</Button>
-                            onClick={() => TodosState.TodosState.pipe(
-                                Effect.flatMap(state => state.prepend(todo)),
-                                Effect.andThen(createEmptyTodo),
-                                Effect.map(setTodo),
-                                runSync,
-                            )}
-                        >
-                            Add
-                        </Button>
                     </Flex>
                 </Flex>
             </Card>

packages/example/src/todos/views/VTodo.tsx

@@ -1,20 +1,28 @@
 import { Todo } from "@/domain"
 import { Box, Card, Flex, IconButton, TextArea } from "@radix-ui/themes"
-import { Effect } from "effect"
+import { Effect, Ref, Stream, SubscriptionRef } from "effect"
 import { Delete } from "lucide-react"
 import { useState } from "react"
 import { R } from "../reffuse"
-import { TodosState } from "../services"
 
 
 export interface VTodoProps {
-    readonly index: number
+    readonly todoRef: SubscriptionRef.SubscriptionRef<Todo.Todo>
-    readonly todo: Todo.Todo
+    readonly remove: Effect.Effect<void>
 }
 
-export function VTodo({ index, todo }: VTodoProps) {
+export function VTodo({ todoRef, remove }: VTodoProps) {
 
     const runSync = R.useRunSync()
+
+    const localTodoRef = R.useRef(() => todoRef)
+    const [content, setContent] = R.useRefState(R.useSubRefFromPath(localTodoRef, ["content"]))
+
+    R.useFork(() => localTodoRef.changes.pipe(
+        Stream.debounce("250 millis"),
+        Stream.runForEach(v => Ref.set(todoRef, v)),
+    ), [localTodoRef])
+
     const editorMode = useState(false)
 
 
@@ -23,14 +31,8 @@ export function VTodo({ index, todo }: VTodoProps) {
             <Card>
                 <Flex direction="column" align="stretch" gap="1">
                     <TextArea
-                        value={todo.content}
+                        value={content}
-                        onChange={e => TodosState.TodosState.pipe(
+                        onChange={e => setContent(e.target.value)}
-                            Effect.flatMap(state => state.replace(
-                                index,
-                                Todo.Todo.make({ ...todo, content: e.target.value }, true),
-                            )),
-                            runSync,
-                        )}
                         disabled={!editorMode}
                     />
 
@@ -38,12 +40,7 @@ export function VTodo({ index, todo }: VTodoProps) {
                         <Box></Box>
 
                         <Flex direction="row" align="center" gap="1">
-                            <IconButton
+                            <IconButton onClick={() => runSync(remove)}>
-                                onClick={() => TodosState.TodosState.pipe(
-                                    Effect.flatMap(state => state.remove(index)),
-                                    runSync,
-                                )}
-                            >
                                 <Delete />
                             </IconButton>
                         </Flex>

packages/example/src/todos/views/VTodos.tsx

@@ -1,5 +1,5 @@
 import { Box, Flex } from "@radix-ui/themes"
-import { Chunk, Effect, Stream } from "effect"
+import { Chunk, Effect, Ref } from "effect"
 import { R } from "../reffuse"
 import { TodosState } from "../services"
 import { VNewTodo } from "./VNewTodo"
@@ -8,15 +8,8 @@ import { VTodo } from "./VTodo"
 
 export function VTodos() {
 
-    // Sync changes to the todos with the local storage
+    const todosRef = R.useMemo(() => Effect.map(TodosState.TodosState, state => state.todos), [])
-    R.useFork(() => TodosState.TodosState.pipe(
+    const [todos] = R.useSubscribeRefs(todosRef)
-        Effect.flatMap(state =>
-            Stream.runForEach(state.todos.changes, () => state.saveToLocalStorage)
-        )
-    ), [])
-
-    const todosRef = R.useMemo(() => TodosState.TodosState.pipe(Effect.map(state => state.todos)), [])
-    const [todos] = R.useRefState(todosRef)
 
 
     return (
@@ -27,7 +20,16 @@ export function VTodos() {
 
             {Chunk.map(todos, (todo, index) => (
                 <Box key={todo.id} width="500px">
-                    <VTodo index={index} todo={todo} />
+                    <R.SubRefFromGetSet
+                        parent={todosRef}
+                        getter={parentValue => Chunk.unsafeGet(parentValue, index)}
+                        setter={(parentValue, value) => Chunk.replace(parentValue, index, value)}
+                    >
+                        {ref => <VTodo
+                            todoRef={ref}
+                            remove={Ref.update(todosRef, Chunk.remove(index))}
+                        />}
+                    </R.SubRefFromGetSet>
                 </Box>
             ))}
         </Flex>

packages/extension-lazyref/package.json

@@ -35,7 +35,7 @@
   "peerDependencies": {
     "@typed/lazy-ref": "^0.3.0",
     "@types/react": "^19.0.0",
-    "effect": "^3.13.0",
+    "effect": "^3.15.0",
     "react": "^19.0.0",
     "reffuse": "^0.1.8"
   }

packages/extension-query/package.json

@@ -37,7 +37,7 @@
     "@effect/platform-browser": "^0.56.0",
     "@typed/async-data": "^0.13.0",
     "@types/react": "^19.0.0",
-    "effect": "^3.13.0",
+    "effect": "^3.15.0",
     "react": "^19.0.0",
     "reffuse": "^0.1.6"
   }

packages/extension-query/src/QueryErrorHandler.ts

@@ -1,4 +1,4 @@
-import { Cause, Context, Effect, identity, Layer, Queue, Stream } from "effect"
+import { Cause, Context, Effect, identity, Layer, PubSub, Stream } from "effect"
 import type { Mutable } from "effect/Types"
 
 
@@ -36,17 +36,17 @@ export const Service = <Self, HandledE = never>() => (
         const TagClass = Context.Tag(id)() as ServiceResult<Self, Id, FallbackA, HandledE>
 
         (TagClass as Mutable<typeof TagClass>).Live = Layer.effect(TagClass, Effect.gen(function*() {
-            const queue = yield* Queue.unbounded<Cause.Cause<HandledE>>()
+            const pubsub = yield* PubSub.unbounded<Cause.Cause<HandledE>>()
-            const errors = Stream.fromQueue(queue)
+            const errors = Stream.fromPubSub(pubsub)
 
             const handle = <A, E, R>(
                 self: Effect.Effect<A, E, R>
             ): Effect.Effect<A | FallbackA, Exclude<E, HandledE>, R> => f(
                 self as unknown as Effect.Effect<never, HandledE, never>,
-                (failure: HandledE) => Queue.offer(queue, Cause.fail(failure)).pipe(
+                (failure: HandledE) => PubSub.publish(pubsub, Cause.fail(failure)).pipe(
                     Effect.andThen(Effect.failCause(Cause.empty))
                 ),
-                (defect: unknown) => Queue.offer(queue, Cause.die(defect)).pipe(
+                (defect: unknown) => PubSub.publish(pubsub, Cause.die(defect)).pipe(
                     Effect.andThen(Effect.failCause(Cause.empty))
                 ),
             )

packages/reffuse/package.json

@@ -1,6 +1,6 @@
 {
   "name": "reffuse",
-  "version": "0.1.9",
+  "version": "0.1.11",
   "type": "module",
   "files": [
     "./README.md",
@@ -35,7 +35,7 @@
   },
   "peerDependencies": {
     "@types/react": "^19.0.0",
-    "effect": "^3.13.0",
+    "effect": "^3.15.0",
     "react": "^19.0.0"
   }
 }

packages/reffuse/src/ReffuseContext.ts

@@ -1,4 +1,4 @@
-import { Array, Context, Effect, ExecutionStrategy, Exit, Layer, Ref, Runtime, Scope } from "effect"
+import { Array, Context, Effect, ExecutionStrategy, Exit, Layer, Match, Ref, Runtime, Scope } from "effect"
 import * as React from "react"
 import * as ReffuseRuntime from "./ReffuseRuntime.js"
 
@@ -25,6 +25,8 @@ export type R<T> = T extends ReffuseContext<infer R> ? R : never
 export type ReactProvider<R> = React.FC<{
     readonly layer: Layer.Layer<R, unknown, Scope.Scope>
     readonly scope?: Scope.Scope
+    readonly finalizerExecutionStrategy?: ExecutionStrategy.ExecutionStrategy
+    readonly finalizerExecutionMode?: "sync" | "fork"
     readonly children?: React.ReactNode
 }>
 
@@ -32,16 +34,25 @@ const makeProvider = <R>(Context: React.Context<Context.Context<R>>): ReactProvi
     return function ReffuseContextReactProvider(props) {
         const runtime = ReffuseRuntime.useRuntime()
         const runSync = React.useMemo(() => Runtime.runSync(runtime), [runtime])
+        const runFork = React.useMemo(() => Runtime.runFork(runtime), [runtime])
 
         const makeScope = React.useMemo(() => props.scope
-            ? Scope.fork(props.scope, ExecutionStrategy.sequential)
+            ? Scope.fork(props.scope, props.finalizerExecutionStrategy ?? ExecutionStrategy.sequential)
-            : Scope.make(),
+            : Scope.make(props.finalizerExecutionStrategy ?? ExecutionStrategy.sequential),
         [props.scope])
 
-        const makeContext = React.useCallback((scope: Scope.CloseableScope) => Effect.context<R>().pipe(
+        const makeContext = (scope: Scope.CloseableScope) => Effect.context<R>().pipe(
             Effect.provide(props.layer),
             Effect.provideService(Scope.Scope, scope),
-        ), [props.layer])
+        )
+
+        const closeScope = (scope: Scope.CloseableScope) => Scope.close(scope, Exit.void).pipe(
+            effect => Match.value(props.finalizerExecutionMode ?? "sync").pipe(
+                Match.when("sync", () => { runSync(effect) }),
+                Match.when("fork", () => { runFork(effect) }),
+                Match.exhaustive,
+            )
+        )
 
         const [isInitialRun, initialScope, initialValue] = React.useMemo(() => Effect.Do.pipe(
             Effect.bind("isInitialRun", () => Ref.make(true)),
@@ -57,7 +68,7 @@ const makeProvider = <R>(Context: React.Context<Context.Context<R>>): ReactProvi
             Effect.if({
                 onTrue: () => Ref.set(isInitialRun, false).pipe(
                     Effect.map(() =>
-                        () => runSync(Scope.close(initialScope, Exit.void))
+                        () => closeScope(initialScope)
                     )
                 ),
 
@@ -68,13 +79,13 @@ const makeProvider = <R>(Context: React.Context<Context.Context<R>>): ReactProvi
                         Effect.sync(() => setValue(context))
                     ),
                     Effect.map(({ scope }) =>
-                        () => runSync(Scope.close(scope, Exit.void))
+                        () => closeScope(scope)
                     ),
                 ),
             }),
 
             runSync,
-        ), [makeScope, makeContext, runSync])
+        ), [makeScope, runSync, runFork])
 
         return React.createElement(Context, { ...props, value })
     }
@@ -84,6 +95,7 @@ export type AsyncReactProvider<R> = React.FC<{
     readonly layer: Layer.Layer<R, unknown, Scope.Scope>
     readonly scope?: Scope.Scope
     readonly finalizerExecutionStrategy?: ExecutionStrategy.ExecutionStrategy
+    readonly finalizerExecutionMode?: "sync" | "fork"
     readonly fallback?: React.ReactNode
     readonly children?: React.ReactNode
 }>
@@ -112,7 +124,7 @@ const makeAsyncProvider = <R>(Context: React.Context<Context.Context<R>>): Async
 
             const scope = runSync(props.scope
                 ? Scope.fork(props.scope, props.finalizerExecutionStrategy ?? ExecutionStrategy.sequential)
-                : Scope.make(props.finalizerExecutionStrategy)
+                : Scope.make(props.finalizerExecutionStrategy ?? ExecutionStrategy.sequential)
             )
 
             Effect.context<R>().pipe(
@@ -126,7 +138,13 @@ const makeAsyncProvider = <R>(Context: React.Context<Context.Context<R>>): Async
                 effect => runFork(effect, { ...props, scope }),
             )
 
-            return () => { runFork(Scope.close(scope, Exit.void)) }
+            return () => Scope.close(scope, Exit.void).pipe(
+                effect => Match.value(props.finalizerExecutionMode ?? "sync").pipe(
+                    Match.when("sync", () => { runSync(effect) }),
+                    Match.when("fork", () => { runFork(effect) }),
+                    Match.exhaustive,
+                )
+            )
         }, [props.layer, runSync, runFork])
 
         return React.createElement(React.Suspense, {

packages/reffuse/src/ReffuseNamespace.ts

@@ -1,4 +1,4 @@
-import { type Context, Effect, ExecutionStrategy, Exit, type Fiber, type Layer, Option, pipe, Pipeable, Queue, Ref, Runtime, Scope, Stream, SubscriptionRef } from "effect"
+import { type Context, Effect, ExecutionStrategy, Exit, type Fiber, Layer, Match, Option, pipe, Pipeable, PubSub, Ref, Runtime, Scope, Stream, SubscriptionRef } from "effect"
 import * as React from "react"
 import * as ReffuseContext from "./ReffuseContext.js"
 import * as ReffuseRuntime from "./ReffuseRuntime.js"
@@ -14,6 +14,11 @@ export interface ScopeOptions {
     readonly finalizerExecutionStrategy?: ExecutionStrategy.ExecutionStrategy
 }
 
+export interface UseScopeOptions extends RenderOptions, ScopeOptions {
+    readonly scope?: Scope.Scope
+    readonly finalizerExecutionMode?: "sync" | "fork"
+}
+
 export type RefsA<T extends readonly SubscriptionRef.SubscriptionRef<any>[]> = {
     [K in keyof T]: Effect.Effect.Success<T[K]>
 }
@@ -23,7 +28,8 @@ export abstract class ReffuseNamespace<R> {
     declare ["constructor"]: ReffuseNamespaceClass<R>
 
     constructor() {
-        this.SubRef = this.SubRef.bind(this as any) as any
+        this.SubRefFromGetSet = this.SubRefFromGetSet.bind(this as any) as any
+        this.SubRefFromPath = this.SubRefFromPath.bind(this as any) as any
         this.SubscribeRefs = this.SubscribeRefs.bind(this as any) as any
         this.RefState = this.RefState.bind(this as any) as any
         this.SubscribeStream = this.SubscribeStream.bind(this as any) as any
@@ -88,6 +94,56 @@ export abstract class ReffuseNamespace<R> {
         ), [runtime, context])
     }
 
+    useScope<R>(
+        this: ReffuseNamespace<R>,
+        deps: React.DependencyList = [],
+        options?: UseScopeOptions,
+    ): readonly [scope: Scope.Scope, layer: Layer.Layer<Scope.Scope>] {
+        const runSync = this.useRunSync()
+        const runFork = this.useRunFork()
+
+        const makeScope = React.useMemo(() => options?.scope
+            ? Scope.fork(options.scope, options.finalizerExecutionStrategy ?? ExecutionStrategy.sequential)
+            : Scope.make(options?.finalizerExecutionStrategy ?? ExecutionStrategy.sequential),
+        [options?.scope])
+
+        const closeScope = (scope: Scope.CloseableScope) => Scope.close(scope, Exit.void).pipe(
+            effect => Match.value(options?.finalizerExecutionMode ?? "sync").pipe(
+                Match.when("sync", () => { runSync(effect) }),
+                Match.when("fork", () => { runFork(effect) }),
+                Match.exhaustive,
+            )
+        )
+
+        const [isInitialRun, initialScope] = React.useMemo(() => runSync(
+            Effect.all([Ref.make(true), makeScope])
+        ), [makeScope])
+
+        const [scope, setScope] = React.useState(initialScope)
+
+        React.useEffect(() => isInitialRun.pipe(
+            Effect.if({
+                onTrue: () => Effect.as(
+                    Ref.set(isInitialRun, false),
+                    () => closeScope(initialScope),
+                ),
+
+                onFalse: () => makeScope.pipe(
+                    Effect.tap(v => Effect.sync(() => setScope(v))),
+                    Effect.map(v => () => closeScope(v)),
+                ),
+            }),
+
+            runSync,
+        ), [
+            makeScope,
+            ...options?.doNotReExecuteOnRuntimeOrContextChange ? [] : [runSync, runFork],
+            ...deps,
+        ])
+
+        return React.useMemo(() => [scope, Layer.succeed(Scope.Scope, scope)] as const, [scope])
+    }
+
     /**
      * Reffuse equivalent to `React.useMemo`.
      *
@@ -111,53 +167,6 @@ export abstract class ReffuseNamespace<R> {
         ])
     }
 
-    useMemoScoped<A, E, R>(
-        this: ReffuseNamespace<R>,
-        effect: () => Effect.Effect<A, E, R | Scope.Scope>,
-        deps: React.DependencyList,
-        options?: RenderOptions & ScopeOptions,
-    ): A {
-        const runSync = this.useRunSync()
-
-        const [isInitialRun, initialScope, initialValue] = React.useMemo(() => Effect.Do.pipe(
-            Effect.bind("isInitialRun", () => Ref.make(true)),
-            Effect.bind("scope", () => Scope.make(options?.finalizerExecutionStrategy)),
-            Effect.bind("value", ({ scope }) => Effect.provideService(effect(), Scope.Scope, scope)),
-            Effect.map(({ isInitialRun, scope, value }) => [isInitialRun, scope, value] as const),
-            runSync,
-        ), [])
-
-        const [value, setValue] = React.useState(initialValue)
-
-        React.useEffect(() => isInitialRun.pipe(
-            Effect.if({
-                onTrue: () => Ref.set(isInitialRun, false).pipe(
-                    Effect.map(() =>
-                        () => runSync(Scope.close(initialScope, Exit.void))
-                    )
-                ),
-
-                onFalse: () => Effect.Do.pipe(
-                    Effect.bind("scope", () => Scope.make(options?.finalizerExecutionStrategy)),
-                    Effect.bind("value", ({ scope }) => Effect.provideService(effect(), Scope.Scope, scope)),
-                    Effect.tap(({ value }) =>
-                        Effect.sync(() => setValue(value))
-                    ),
-                    Effect.map(({ scope }) =>
-                        () => runSync(Scope.close(scope, Exit.void))
-                    ),
-                ),
-            }),
-
-            runSync,
-        ), [
-            ...options?.doNotReExecuteOnRuntimeOrContextChange ? [] : [runSync],
-            ...deps,
-        ])
-
-        return value
-    }
-
     /**
      * Reffuse equivalent to `React.useEffect`.
      *
@@ -378,18 +387,39 @@ export abstract class ReffuseNamespace<R> {
         ])
     }
 
-    useRef<A, R>(
+    useRef<A, E, R>(
         this: ReffuseNamespace<R>,
-        value: A,
+        initialValue: () => Effect.Effect<A, E, R>,
     ): SubscriptionRef.SubscriptionRef<A> {
         return this.useMemo(
-            () => SubscriptionRef.make(value),
+            () => Effect.flatMap(initialValue(), SubscriptionRef.make),
             [],
             { doNotReExecuteOnRuntimeOrContextChange: true }, // Do not recreate the ref when the context changes
         )
     }
 
-    useSubRef<B, const P extends PropertyPath.Paths<B>, R>(
+    useRefFromReactiveValue<A, R>(
+        this: ReffuseNamespace<R>,
+        value: A,
+    ): SubscriptionRef.SubscriptionRef<A> {
+        const ref = this.useRef(() => Effect.succeed(value))
+        this.useEffect(() => Ref.set(ref, value), [value], { doNotReExecuteOnRuntimeOrContextChange: true })
+        return ref
+    }
+
+    useSubRefFromGetSet<A, B, R>(
+        this: ReffuseNamespace<R>,
+        parent: SubscriptionRef.SubscriptionRef<B>,
+        getter: (parentValue: B) => A,
+        setter: (parentValue: B, value: A) => B,
+    ): SubscriptionSubRef.SubscriptionSubRef<A, B> {
+        return React.useMemo(
+            () => SubscriptionSubRef.makeFromGetSet(parent, getter, setter),
+            [parent],
+        )
+    }
+
+    useSubRefFromPath<B, const P extends PropertyPath.Paths<B>, R>(
         this: ReffuseNamespace<R>,
         parent: SubscriptionRef.SubscriptionRef<B>,
         path: P,
@@ -455,36 +485,77 @@ export abstract class ReffuseNamespace<R> {
         return [reactStateValue, setValue]
     }
 
-    useStreamFromValues<const A extends React.DependencyList, R>(
+    useStreamFromReactiveValues<const A extends React.DependencyList, R>(
         this: ReffuseNamespace<R>,
         values: A,
     ): Stream.Stream<A> {
-        const [queue, stream] = this.useMemo(() => Queue.unbounded<A>().pipe(
+        const [, scopeLayer] = this.useScope([], { finalizerExecutionMode: "fork" })
-            Effect.map(queue => [queue, Stream.fromQueue(queue)] as const)
-        ), [])
 
-        this.useEffect(() => Queue.offer(queue, values), values)
+        const { latest, pubsub, stream } = this.useMemo(() => Effect.Do.pipe(
+            Effect.bind("latest", () => Ref.make(values)),
+            Effect.bind("pubsub", () => Effect.acquireRelease(PubSub.unbounded<A>(), PubSub.shutdown)),
+            Effect.let("stream", ({ latest, pubsub }) => Ref.get(latest).pipe(
+                Effect.flatMap(a => Effect.map(
+                    Stream.fromPubSub(pubsub, { scoped: true }),
+                    s => Stream.concat(Stream.make(a), s),
+                )),
+                Stream.unwrapScoped,
+            )),
+            Effect.provide(scopeLayer),
+        ), [scopeLayer], { doNotReExecuteOnRuntimeOrContextChange: true })
+
+        this.useEffect(() => Ref.set(latest, values).pipe(
+            Effect.andThen(PubSub.publish(pubsub, values)),
+            Effect.unlessEffect(PubSub.isShutdown(pubsub)),
+        ), values, { doNotReExecuteOnRuntimeOrContextChange: true })
 
         return stream
     }
 
-    useSubscribeStream<A, InitialA extends A | undefined, E, R>(
+    useSubscribeStream<A, E, R>(
         this: ReffuseNamespace<R>,
         stream: Stream.Stream<A, E, R>,
-        initialValue?: InitialA,
+    ): Option.Option<A>
-    ): InitialA extends A ? Option.Some<A> : Option.Option<A> {
+    useSubscribeStream<A, E, IE, R>(
-        const [reactStateValue, setReactStateValue] = React.useState<Option.Option<A>>(Option.fromNullable(initialValue))
+        this: ReffuseNamespace<R>,
+        stream: Stream.Stream<A, E, R>,
+        initialValue: () => Effect.Effect<A, IE, R>,
+    ): Option.Some<A>
+    useSubscribeStream<A, E, IE, R>(
+        this: ReffuseNamespace<R>,
+        stream: Stream.Stream<A, E, R>,
+        initialValue?: () => Effect.Effect<A, IE, R>,
+    ): Option.Option<A> {
+        const [reactStateValue, setReactStateValue] = React.useState(this.useMemo(
+            () => initialValue
+                ? Effect.map(initialValue(), Option.some)
+                : Effect.succeed(Option.none()),
+            [],
+            { doNotReExecuteOnRuntimeOrContextChange: true },
+        ))
 
         this.useFork(() => Stream.runForEach(
             Stream.changesWith(stream, (x, y) => x === y),
             v => Effect.sync(() => setReactStateValue(Option.some(v))),
         ), [stream])
 
-        return reactStateValue as InitialA extends A ? Option.Some<A> : Option.Option<A>
+        return reactStateValue
     }
 
 
-    SubRef<B, const P extends PropertyPath.Paths<B>, R>(
+    SubRefFromGetSet<A, B, R>(
+        this: ReffuseNamespace<R>,
+        props: {
+            readonly parent: SubscriptionRef.SubscriptionRef<B>,
+            readonly getter: (parentValue: B) => A,
+            readonly setter: (parentValue: B, value: A) => B,
+            readonly children: (subRef: SubscriptionSubRef.SubscriptionSubRef<A, B>) => React.ReactNode
+        },
+    ): React.ReactNode {
+        return props.children(this.useSubRefFromGetSet(props.parent, props.getter, props.setter))
+    }
+
+    SubRefFromPath<B, const P extends PropertyPath.Paths<B>, R>(
         this: ReffuseNamespace<R>,
         props: {
             readonly parent: SubscriptionRef.SubscriptionRef<B>,
@@ -492,7 +563,7 @@ export abstract class ReffuseNamespace<R> {
             readonly children: (subRef: SubscriptionSubRef.SubscriptionSubRef<PropertyPath.ValueFromPath<B, P>, B>) => React.ReactNode
         },
     ): React.ReactNode {
-        return props.children(this.useSubRef(props.parent, props.path))
+        return props.children(this.useSubRefFromPath(props.parent, props.path))
     }
 
     SubscribeRefs<
@@ -518,15 +589,30 @@ export abstract class ReffuseNamespace<R> {
         return props.children(this.useRefState(props.ref))
     }
 
-    SubscribeStream<A, InitialA extends A | undefined, E, R>(
+    SubscribeStream<A, E, R>(
         this: ReffuseNamespace<R>,
         props: {
             readonly stream: Stream.Stream<A, E, R>
-            readonly initialValue?: InitialA
+            readonly children: (latestValue: Option.Option<A>) => React.ReactNode
-            readonly children: (latestValue: InitialA extends A ? Option.Some<A> : Option.Option<A>) => React.ReactNode
+        },
+    ): React.ReactNode
+    SubscribeStream<A, E, IE, R>(
+        this: ReffuseNamespace<R>,
+        props: {
+            readonly stream: Stream.Stream<A, E, R>
+            readonly initialValue: () => Effect.Effect<A, IE, R>
+            readonly children: (latestValue: Option.Some<A>) => React.ReactNode
+        },
+    ): React.ReactNode
+    SubscribeStream<A, E, IE, R>(
+        this: ReffuseNamespace<R>,
+        props: {
+            readonly stream: Stream.Stream<A, E, R>
+            readonly initialValue?: () => Effect.Effect<A, IE, R>
+            readonly children: (latestValue: Option.Some<A>) => React.ReactNode
         },
     ): React.ReactNode {
-        return props.children(this.useSubscribeStream(props.stream, props.initialValue))
+        return props.children(this.useSubscribeStream(props.stream, props.initialValue as () => Effect.Effect<A, IE, R>))
     }
 }
 

packages/reffuse/src/types/SubscriptionSubRef.ts

@@ -52,7 +52,7 @@ class SubscriptionSubRefImpl<in out A, in out B> extends Effectable.Class<A> imp
         readonly setter: (parentValue: B, value: A) => B,
     ) {
         super()
-        this.get = Ref.get(this.parent).pipe(Effect.map(this.getter))
+        this.get = Effect.map(Ref.get(this.parent), this.getter)
     }
 
     commit() {