Fix

packages/server/src/rpc/routers/todos.ts

@@ -1,7 +1,7 @@
 import { Schema as S } from "@effect/schema"
 import { JsonifiableTodo } from "@todo-tests/common/data"
 import { observable } from "@trpc/server/observable"
-import { Effect, Fiber, Runtime, Stream, flow } from "effect"
+import { Effect, Fiber, Stream, flow } from "effect"
 import { TodoRepository } from "../../todo/TodoRepository"
 import { TRPCBuilder } from "../../trpc/TRPCBuilder"
 import { RPCProcedureBuilder } from "../procedures/RPCProcedureBuilder"
@@ -28,6 +28,10 @@ export const todosRouter = Effect.gen(function*() {
                     const watcher = ctx.fork(Effect.gen(function*() {
                         const todos = yield* TodoRepository
 
+                        // emit.next(
+                        //     yield* encodeTodos(yield* todos.todos.get)
+                        // )
+
                         yield* todos.todos.changes.pipe(
                             Stream.runForEach(flow(
                                 encodeTodos,
@@ -38,8 +42,8 @@ export const todosRouter = Effect.gen(function*() {
                         )
                     }))
 
-                    return () => Runtime.runSync(ctx.runtime)(
+                    return () => ctx.fork(
-                        Fiber.interruptFork(watcher)
+                        Fiber.interrupt(watcher)
                     )
                 })
             ),
@@ -47,29 +51,27 @@ export const todosRouter = Effect.gen(function*() {
         getByID: procedure
             .input(S.decodeUnknownPromise(S.String))
             .query(({ ctx, input }) => ctx.run(Effect.gen(function*() {
-                const todos = yield* TodoRepository
-
                 return yield* encodeOptionalTodo(
-                    yield* todos.getByID(input)
+                    yield* (yield* TodoRepository).getByID(input)
                 )
             }))),
 
         add: procedure
             .input(S.decodeUnknownPromise(JsonifiableTodo))
-            .mutation(({ ctx, input }) => ctx.run(TodoRepository.pipe(
+            .mutation(({ ctx, input }) => ctx.run(Effect.gen(function*() {
-                Effect.flatMap(todos => todos.add(input))
+                return yield* (yield* TodoRepository).add(input)
-            ))),
+            }))),
 
         update: procedure
             .input(S.decodeUnknownPromise(JsonifiableTodo))
-            .mutation(({ ctx, input }) => ctx.run(TodoRepository.pipe(
+            .mutation(({ ctx, input }) => ctx.run(Effect.gen(function*() {
-                Effect.flatMap(todos => todos.update(input))
+                return yield* (yield* TodoRepository).update(input)
-            ))),
+            }))),
 
         remove: procedure
             .input(S.decodeUnknownPromise(JsonifiableTodo))
-            .mutation(({ ctx, input }) => ctx.run(TodoRepository.pipe(
+            .mutation(({ ctx, input }) => ctx.run(Effect.gen(function*() {
-                Effect.flatMap(todos => todos.remove(input))
+                return yield* (yield* TodoRepository).remove(input)
-            ))),
+            }))),
     })
 })

packages/server/src/todo/TodoRepository.ts

@@ -58,9 +58,9 @@ export class TodoRepositoryService {
     update(todo: Todo) {
         return Effect.gen(this, function*() {
             if (Option.isNone(todo.id))
-                return yield* Effect.fail(new TodoHasNoID({ todo }))
+                return yield* new TodoHasNoID({ todo })
 
-            yield* Ref.update(this.todos, flow(
+            return yield* Ref.update(this.todos, flow(
                 Chunk.replace(
                     yield* yield* this.getIndexByID(todo.id.value),
                     todo as IdentifiedTodo,
@@ -75,9 +75,9 @@ export class TodoRepositoryService {
     remove(todo: Todo) {
         return Effect.gen(this, function*() {
             if (Option.isNone(todo.id))
-                return yield* Effect.fail(new TodoHasNoID({ todo }))
+                return yield* new TodoHasNoID({ todo })
 
-            yield* Ref.update(this.todos, flow(
+            return yield* Ref.update(this.todos, flow(
                 Chunk.remove(
                     yield* yield* this.getIndexByID(todo.id.value)
                 ),