import { type Context, Effect, Equal, type Fiber, Option, Pipeable, Predicate, type Scope, Stream, type Subscribable, SubscriptionRef } from "effect" import * as Result from "./Result.js" export const MutationTypeId: unique symbol = Symbol.for("@effect-fc/Mutation/Mutation") export type MutationTypeId = typeof MutationTypeId export interface Mutation extends Pipeable.Pipeable { readonly [MutationTypeId]: MutationTypeId readonly context: Context.Context readonly f: (key: K) => Effect.Effect readonly initialProgress: P readonly latestKey: Subscribable.Subscribable> readonly fiber: Subscribable.Subscribable>> readonly result: Subscribable.Subscribable> readonly latestFinalResult: Subscribable.Subscribable>> mutate(key: K): Effect.Effect> mutateSubscribable(key: K): Effect.Effect>> } export declare namespace Mutation { export type AnyKey = readonly any[] } export class MutationImpl extends Pipeable.Class() implements Mutation { readonly [MutationTypeId]: MutationTypeId = MutationTypeId constructor( readonly context: Context.Context>, readonly f: (key: K) => Effect.Effect, readonly initialProgress: P, readonly latestKey: SubscriptionRef.SubscriptionRef>, readonly fiber: SubscriptionRef.SubscriptionRef>>, readonly result: SubscriptionRef.SubscriptionRef>, readonly latestFinalResult: SubscriptionRef.SubscriptionRef>>, ) { super() } mutate(key: K): Effect.Effect> { return SubscriptionRef.set(this.latestKey, Option.some(key)).pipe( Effect.andThen(this.start(key)), Effect.andThen(sub => this.watch(sub)), Effect.provide(this.context), ) } mutateSubscribable(key: K): Effect.Effect>> { return SubscriptionRef.set(this.latestKey, Option.some(key)).pipe( Effect.andThen(this.start(key)), Effect.tap(sub => Effect.forkScoped(this.watch(sub))), Effect.provide(this.context), ) } start(key: K): Effect.Effect< Subscribable.Subscribable>, never, Scope.Scope | R > { return this.latestFinalResult.pipe( Effect.andThen(initial => Result.unsafeForkEffect( Effect.onExit(this.f(key), () => Effect.andThen( Effect.all([Effect.fiberId, this.fiber]), ([currentFiberId, fiber]) => Option.match(fiber, { onSome: v => Equal.equals(currentFiberId, v.id()) ? SubscriptionRef.set(this.fiber, Option.none()) : Effect.void, onNone: () => Effect.void, }), )), { initial: Option.isSome(initial) ? Result.willFetch(initial.value) : Result.initial(), initialProgress: this.initialProgress, } as Result.unsafeForkEffect.Options, )), Effect.tap(([, fiber]) => SubscriptionRef.set(this.fiber, Option.some(fiber))), Effect.map(([sub]) => sub), ) } watch( sub: Subscribable.Subscribable> ): Effect.Effect> { return sub.get.pipe( Effect.andThen(initial => Stream.runFoldEffect( sub.changes, initial, (_, result) => Effect.as(SubscriptionRef.set(this.result, result), result), ) as Effect.Effect>), Effect.tap(result => SubscriptionRef.set(this.latestFinalResult, Option.some(result))), ) } } export const isMutation = (u: unknown): u is Mutation => Predicate.hasProperty(u, MutationTypeId) export declare namespace make { export interface Options { readonly f: (key: K) => Effect.Effect>> readonly initialProgress?: P } } export const make = Effect.fnUntraced(function* ( options: make.Options ): Effect.fn.Return< Mutation, P>, never, Scope.Scope | Result.forkEffect.OutputContext > { return new MutationImpl( yield* Effect.context>(), options.f as any, options.initialProgress as P, yield* SubscriptionRef.make(Option.none()), yield* SubscriptionRef.make(Option.none>()), yield* SubscriptionRef.make(Result.initial()), yield* SubscriptionRef.make(Option.none>()), ) })