Query work

packages/effect-fc/src/Query.ts

@@ -1,4 +1,4 @@
-import { Effect, Fiber, Option, Pipeable, Predicate, Stream, type Subscribable, type SubscriptionRef } from "effect"
+import { Effect, Fiber, Option, Pipeable, Predicate, Stream, type Subscribable, SubscriptionRef } from "effect"
 import * as Result from "./Result.js"
 
 
@@ -13,7 +13,7 @@ extends Pipeable.Pipeable {
     readonly f: (key: K) => Effect.Effect<A, E, R>
     readonly initialProgress: P
 
-    readonly fiber: Subscribable.Subscribable<Option.Option<Fiber.Fiber<Result.Result<A, E, P>>>>
+    readonly fiber: Subscribable.Subscribable<Option.Option<Fiber.Fiber<A, E>>>
     readonly result: Subscribable.Subscribable<Result.Result<A, E, P>>
 }
 
@@ -26,40 +26,68 @@ extends Pipeable.Class() implements Query<K, A, E, R, P> {
         readonly f: (key: K) => Effect.Effect<A, E, R>,
         readonly initialProgress: P,
 
-        readonly fiber: SubscriptionRef.SubscriptionRef<Option.Option<Fiber.Fiber<Result.Result<A, E, P>>>>,
+        readonly fiber: SubscriptionRef.SubscriptionRef<Option.Option<Fiber.Fiber<A, E>>>,
         readonly result: SubscriptionRef.SubscriptionRef<Result.Result<A, E, P>>,
     ) {
         super()
     }
 
-    query(key: K) {
+    interrupt(): Effect.Effect<void> {
-        return this.fiber.pipe(
+        return Effect.andThen(this.fiber, Option.match({
-            Effect.andThen(Option.match({
+            onSome: fiber => Effect.andThen(
-                onSome: Fiber.interrupt,
+                Fiber.interrupt(fiber),
+                SubscriptionRef.set(this.fiber, Option.none()),
+            ),
             onNone: () => Effect.void,
-            })),
+        }))
-            Effect.andThen(Result.forkEffect(this.f(key))),
+    }
-            Effect.tap(([result, fiber]) => ),
+
+    query(key: K): Effect.Effect<
+        Result.Result<A, E, P>,
+        never,
+        Result.forkEffect.OutputContext<A, E, R, never>
+    > {
+        return this.fiber.pipe(
+            Effect.andThen(this.interrupt()),
+            Effect.andThen(Result.unsafeForkEffect(this.f(key))),
+            Effect.tap(([, fiber]) => SubscriptionRef.set(this.fiber, Option.some(fiber))),
+            Effect.andThen(([sub]) => Effect.all([Effect.succeed(sub), sub.get])),
+            Effect.andThen(([sub, initial]) => Stream.runFoldEffect(
+                sub.changes,
+                initial,
+                (_, result) => Effect.as(SubscriptionRef.set(this.result, result), result),
+            )),
+            Effect.tap(SubscriptionRef.set(this.fiber, Option.none())),
         )
     }
 }
 
 export const isQuery = (u: unknown): u is Query<unknown[], unknown, unknown, unknown, unknown> => Predicate.hasProperty(u, QueryTypeId)
 
+export declare namespace make {
+    export interface Options<K extends readonly any[], A, E = never, R = never, P = never> {
+        readonly key: Stream.Stream<K>
+        readonly f: (key: NoInfer<K>) => Effect.Effect<A, E, R>
+        readonly initialProgress?: P
+    }
+}
+
+export const make = Effect.fnUntraced(function* <K extends readonly any[], A, E = never, R = never, P = never>(
+    options: make.Options<K, A, E, R, P>
+) {
+    return new QueryImpl(
+        options.key,
+        options.f,
+        options.initialProgress as P,
+
+        yield* SubscriptionRef.make(Option.none<Fiber.Fiber<A, E>>()),
+        yield* SubscriptionRef.make(Result.initial() as Result.Result<A, E, P>),
+    )
+})
+
 export const run = <K extends readonly any[], A, E, R, P>(
     self: Query<K, A, E, R, P>
-) => Stream.runForEach(self.key,
+) => Stream.runForEach(self.key, key => Effect.andThen(
-)
+    (self as QueryImpl<K, A, E, R, P>).interrupt(),
-
+    Effect.forkScoped((self as QueryImpl<K, A, E, R, P>).query(key)),
-export const query = <K extends readonly any[], A, E, R, P>(
+))
-    self: Query<K, A, E, R, P>,
-    key: K,
-) => self.fiberRef.pipe(
-    Effect.andThen(Option.match({
-        onSome: Fiber.interrupt,
-        onNone: () => Effect.void,
-    })),
-    Effect.andThen(Effect.forkScoped(
-
-    ))
-)