Global effect runtime for use with react(-native)

managed_runtime.ts

/**
* Example of a "global managed runtime" for use with react(-native)
* React isn't a traditional zio/effect app that has one entry point
* There may be many effects run but we don't want to reconstruct all our app services each
* time we want to run an effect.
*
* This code builds app services layers just once then provides a global runtime object that can
* be used to run as many effects as you want all with access to the same service instances.
*
* ex: 
* const myRuntime = await setupGlobalEffectRuntime(MyAppServicesLayer)
* ...then anywhere in your app, provide myRuntime via react context or other means
* const foo = await myRuntime.unsafeRunPromise(someEffectToRun)
*/

import * as Cause from "@effect/io/Cause"
import { runtimeDebug } from "@effect/io/Debug"
import * as T from "@effect/io/Effect"
import * as Exit from "@effect/io/Exit"
import type * as Fiber from "@effect/io/Fiber"
import * as L from "@effect/io/Layer"
import * as RT from "@effect/io/Runtime"
import * as SC from "@effect/io/Scope"
import * as Chunk from "@fp-ts/data/Chunk"
import type * as Context from "@fp-ts/data/Context"
import { pipe } from "@fp-ts/data/Function"
import * as M from "@fp-ts/data/Option"


export const namedFiberRef = FiberRef.unsafeMake("anonymous")

export function forkDaemonNamed(name: string) {
    return <R, E, A>(self: T.Effect<R, E, A>) =>
        T.logSpan(`fiber_${name}`)(pipe(self, FiberRef.locally(namedFiberRef)(name), T.forkDaemon))
}
export function forkNamed(name: string) {
    return <R, E, A>(self: T.Effect<R, E, A>) =>
        T.logSpan(`fiber_${name}`)(pipe(self, FiberRef.locally(namedFiberRef)(name), T.fork)) //self.apply(namedFiberRef.locally(name)).fork
}

export function forkInNamed(name: string, scope: S.Scope) {
    return <R, E, A>(self: T.Effect<R, E, A>) =>
        T.logSpan(`fiber_${name}`)(pipe(self, FiberRef.locally(namedFiberRef)(name), T.forkIn(scope))) // self.apply(namedFiberRef.locally(name)).forkIn(scope)
}

export function forkScopedNamed(name: string) {
    return <R, E, A>(self: T.Effect<R, E, A>) =>
        T.logSpan(`fiber_${name}`)(pipe(self, FiberRef.locally(namedFiberRef)(name), T.forkScoped)) //self.apply(namedFiberRef.locally(name)).forkScoped
}

      
/**
 * Builds a "managed runtime" to be shared across the app environment. All running effects will be interrupted when the scope is closed.
 * This allows for starting as many effects as needed in order to interface with other code such as react while only initializing a single
 * instance of services and ensuring everything gets cleaned up.
 * @returns
 */
export async function setupGlobalEffectRuntime(layer: IAppServicesLayer) {

    //build a runtime with the app services impl provided and our custom logger
    const rt = await T.unsafeRunPromiseExit(scopedManagedRuntime(layer))

    if (Exit.isSuccess(rt)) {
        console.log(`success building layers, init all env`)
        setKeystoneContexts(rt.value)
        await initAllEnv(rt.value)
        console.log("done initallenv")

        //set observable value indicating runtime is now ready
        rt.value.runtime.unsafeRunSync("setupGlobalEffectRuntimeLayers")(
            T.serviceWith(LifecycleServiceTag)(l => l.setRuntimeInitCompleted(true))
        )
        return rt.value
    } else {
        const { cause } = rt
        log.error(
            `Error initializing global effect runtime environment!`,
            (pipe(rt.cause, Cause.defects, Chunk.toReadonlyArray)[0] ?? new Error(`no_defect`)) as any as Error,
            { report: true, impactsStability: true },
            {
                failures: pipe(cause, Cause.failures, Chunk.map(serializeError)),
                defects: pipe(cause, Cause.defects, Chunk.map(serializeError)),
            }
        )
        throw Cause.squash(cause)
    }
}

interface NamedRuntimeType<A> {
    unsafeRunPromiseExit: (name: string) => <F, V>(e: T.Effect<A, F, V>) => Promise<Exit.Exit<F, V>>
    unsafeRunPromise: (name: string) => <F, V>(e: T.Effect<A, F, V>) => Promise<V>
    unsafeRunSync: (name: string) => <F, V>(e: T.Effect<A, F, V>) => V
    unsafeRunSyncExit: (name: string) => <F, V>(e: T.Effect<A, F, V>) => Exit.Exit<F, V>
    unsafeFork: (name: string) => <F, V>(e: T.Effect<A, F, V>) => Fiber.RuntimeFiber<F, V>
}

interface ManagedRuntimeType<A> {
    runtime: NamedRuntimeType<A>
    closeNow: () => Promise<Exit.Exit<never, void>>
    close: T.Effect<never, never, void>
    getServiceSync: <T extends A>(t: Context.Tag<T>) => T
}

/**
 * Provide a "global" layer and wrap all the unsafeRun methods from a runtime with a scope so anything started
 * gets interrupted when scope closes.
 *
 * This is useful for interop with other runtimes like React. Pass the runtime reference around to execute as
 * many effects as needed with only a single instance of services in the layer. This is effectively a
 * singleton layer.
 *
 * Attaches any effects started via the runtime to the scope so they will be interrupted properly when closeNow() is called
 */
function scopedManagedRuntime<R, E, A>(layer: L.Layer<R, E, A>): T.Effect<R, E, ManagedRuntimeType<A>> {
    const ret = pipe(
        T.Do(),
        T.bind("sc", () => SC.make()),

        T.bind("env", ({ sc }) => pipe(layer, L.buildWithScope(sc))),
        T.bind("refs", () => T.getFiberRefs()),
        T.map(({ env, refs, sc }) => ({
            runtime: RT.make(env, RT.defaultRuntimeFlags, refs),
            sc,
        })),
        T.map(({ runtime, sc }) => ({
            runtime: {
                unsafeRunPromiseExit:
                    (name: string) =>
                    <F, V>(e: T.Effect<A, F, V>) =>
                        pipe(e, forkInNamed(name, sc), T.fromFiberEffect, runtime.unsafeRunPromiseExit),
                unsafeRunPromise:
                    (name: string) =>
                    <F, V>(e: T.Effect<A, F, V>) =>
                        pipe(e, forkInNamed(name, sc), T.fromFiberEffect, runtime.unsafeRunPromise),
                unsafeRunSync:
                    (name: string) =>
                    <F, V>(e: T.Effect<A, F, V>) =>
                        pipe(e, forkInNamed(name, sc), T.fromFiberEffect, runtime.unsafeRunSync),
                unsafeRunSyncExit:
                    (name: string) =>
                    <F, V>(e: T.Effect<A, F, V>) =>
                        pipe(e, forkInNamed(name, sc), T.fromFiberEffect, runtime.unsafeRunSyncExit),
                unsafeFork:
                    (name: string) =>
                    <F, V>(e: T.Effect<A, F, V>) =>
                        pipe(e, forkInNamed(name, sc), T.fromFiberEffect, v => runtime.unsafeFork<F, V>(v)),
            },
            getServiceSync: <T extends A>(serviceTag: Context.Tag<T>): T =>
                pipe(T.service(serviceTag), runtime.unsafeRunSync),
            closeNow: async () => {
                log.log(`closing managed runtime`)
                try {
                    const e = await runtime.unsafeRunPromiseExit(SC.close(Exit.unit())(sc))
                  
                    return e
                } catch (err: any) {
                    console.error(`error shutting down managed runtime`, err, { report: false })
                    throw err
                }
            },
            close: T.suspendSucceed(() => SC.close(Exit.unit())(sc)),
        })),
        T.tap(() => T.logInfo(`Done creating managed effect runtime`)),

        T.tapErrorCause(c => T.logErrorCauseMessage(`Error building managed_runtime!`, c))
    )
    return ret
}