Query work

packages/effect-fc/src/Query.ts

@@ -1,4 +1,4 @@
-import { Effect, Fiber, Option, Pipeable, Predicate, Stream, type Subscribable, SubscriptionRef } from "effect"
+import { Effect, Fiber, Option, Pipeable, Predicate, type Scope, Stream, type Subscribable, SubscriptionRef } from "effect"
 import * as Result from "./Result.js"
 
 
@@ -45,11 +45,11 @@ extends Pipeable.Class() implements Query<K, A, E, R, P> {
     query(key: K): Effect.Effect<
         Result.Result<A, E, P>,
         never,
-        Result.forkEffect.OutputContext<A, E, R, never>
+        Result.unsafeForkEffect.OutputContext<A, E, R, P>
     > {
         return this.fiber.pipe(
             Effect.andThen(this.interrupt()),
-            Effect.andThen(Result.unsafeForkEffect(this.f(key))),
+            Effect.andThen(Result.unsafeForkEffect(this.f(key), { initialProgress: this.initialProgress })),
             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(
@@ -67,24 +67,31 @@ export const isQuery = (u: unknown): u is Query<unknown[], unknown, unknown, unk
 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 f: (key: NoInfer<K>) => Effect.Effect<A, E, Result.forkEffect.InputContext<R, NoInfer<P>>>
         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>
-) {
+): Effect.fn.Return<Query<K, A, E, Result.forkEffect.OutputContext<A, E, R, P>, 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>),
+        yield* SubscriptionRef.make(Result.initial<A, E, P>()),
     )
 })
 
+export const service = <K extends readonly any[], A, E = never, R = never, P = never>(
+    options: make.Options<K, A, E, R, P>
+): Effect.Effect<Query<K, A, E, R, P>, never, Scope.Scope> => Effect.tap(
+    make(options),
+    query => Effect.forkScoped(run(query)),
+)
+
 export const run = <K extends readonly any[], A, E, R, P>(
     self: Query<K, A, E, R, P>
 ) => Stream.runForEach(self.key, key => Effect.andThen(