import { Cause, Deferred, Effect, Exit, Fiber, Latch, Schema, Scope, SynchronizedRef } from "effect" export interface Runner { readonly state: State readonly busy: boolean readonly ensureRunning: (work: Effect.Effect) => Effect.Effect readonly startShell: (work: Effect.Effect, ready?: Latch.Latch) => Effect.Effect readonly cancel: Effect.Effect } export class Cancelled extends Schema.TaggedErrorClass()("RunnerCancelled", {}) {} export class Busy extends Schema.TaggedErrorClass()("RunnerBusy", {}) {} interface RunHandle { id: number done: Deferred.Deferred fiber: Fiber.Fiber } interface ShellHandle { id: number cancelled: Deferred.Deferred ready?: Latch.Latch fiber: Fiber.Fiber } interface PendingHandle { id: number done: Deferred.Deferred work: Effect.Effect } export type State = | { readonly _tag: "Idle" } | { readonly _tag: "Running"; readonly run: RunHandle } | { readonly _tag: "Shell"; readonly shell: ShellHandle } | { readonly _tag: "ShellThenRun"; readonly shell: ShellHandle; readonly run: PendingHandle } export const make = ( scope: Scope.Scope, opts?: { onIdle?: Effect.Effect onBusy?: Effect.Effect onInterrupt?: Effect.Effect }, ): Runner => { const ref = SynchronizedRef.makeUnsafe>({ _tag: "Idle" }) const idle = opts?.onIdle ?? Effect.void const onBusy = opts?.onBusy ?? Effect.void const onInterrupt = opts?.onInterrupt let ids = 0 const state = () => SynchronizedRef.getUnsafe(ref) const next = () => { ids += 1 return ids } const complete = (done: Deferred.Deferred, exit: Exit.Exit) => Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause) ? Deferred.fail(done, new Cancelled()).pipe(Effect.asVoid) : Deferred.done(done, exit).pipe(Effect.asVoid) const awaitDone = (done: Deferred.Deferred) => Deferred.await(done).pipe(Effect.catchTag("RunnerCancelled", (e) => onInterrupt ?? Effect.die(e))) const idleIfCurrent = () => SynchronizedRef.modify(ref, (st) => [st._tag === "Idle" ? idle : Effect.void, st] as const).pipe(Effect.flatten) const finishRun = (id: number, done: Deferred.Deferred, exit: Exit.Exit) => SynchronizedRef.modify( ref, (st) => [ Effect.gen(function* () { if (st._tag === "Running" && st.run.id === id) yield* idle yield* complete(done, exit) }), st._tag === "Running" && st.run.id === id ? ({ _tag: "Idle" } as const) : st, ] as const, ).pipe(Effect.flatten) const startRun = (work: Effect.Effect, done: Deferred.Deferred) => Effect.gen(function* () { const id = next() const fiber = yield* work.pipe( Effect.onExit((exit) => finishRun(id, done, exit)), Effect.forkIn(scope), ) return { id, done, fiber } satisfies RunHandle }) const finishShell = (id: number) => SynchronizedRef.modifyEffect( ref, Effect.fnUntraced(function* (st) { if (st._tag === "Shell" && st.shell.id === id) { return [idle, { _tag: "Idle" }] as const } if (st._tag === "ShellThenRun" && st.shell.id === id) { const run = yield* startRun(st.run.work, st.run.done) return [Effect.void, { _tag: "Running", run }] as const } return [Effect.void, st] as const }), ).pipe(Effect.flatten) const stopShell = (shell: ShellHandle) => Effect.gen(function* () { if (shell.ready) yield* shell.ready.await.pipe(Effect.exit, Effect.asVoid) yield* Deferred.succeed(shell.cancelled, undefined).pipe(Effect.asVoid) yield* Fiber.interrupt(shell.fiber) }) const ensureRunning = (work: Effect.Effect) => SynchronizedRef.modifyEffect( ref, Effect.fnUntraced(function* (st) { switch (st._tag) { case "Running": case "ShellThenRun": return [awaitDone(st.run.done), st] as const case "Shell": { const run = { id: next(), done: yield* Deferred.make(), work, } satisfies PendingHandle return [awaitDone(run.done), { _tag: "ShellThenRun", shell: st.shell, run }] as const } case "Idle": { const done = yield* Deferred.make() const run = yield* startRun(work, done) return [awaitDone(done), { _tag: "Running", run }] as const } } }), ).pipe(Effect.flatten) const startShell = (work: Effect.Effect, ready?: Latch.Latch): Effect.Effect => SynchronizedRef.modifyEffect( ref, Effect.fnUntraced(function* (st) { if (st._tag !== "Idle") { const reject: Effect.Effect = Effect.fail(new Busy()) return [reject, st] as const } yield* onBusy const id = next() const cancelled = yield* Deferred.make() const fiber = yield* work.pipe(Effect.ensuring(finishShell(id)), Effect.forkChild) const shell = { id, cancelled, ready, fiber } satisfies ShellHandle return [ Effect.gen(function* () { const exit = yield* Fiber.await(fiber) if (Exit.isSuccess(exit)) return exit.value if ( Cause.hasInterruptsOnly(exit.cause) || ((yield* Deferred.isDone(cancelled)) && Cause.hasInterrupts(exit.cause) && !Cause.hasDies(exit.cause)) ) { if (onInterrupt) return yield* onInterrupt return yield* Effect.die(new Cancelled()) } return yield* Effect.failCause(exit.cause) }), { _tag: "Shell", shell }, ] as const }), ).pipe(Effect.flatten) const cancel = SynchronizedRef.modify(ref, (st) => { switch (st._tag) { case "Idle": return [Effect.void, st] as const case "Running": return [ Effect.gen(function* () { yield* Fiber.interrupt(st.run.fiber) yield* Deferred.fail(st.run.done, new Cancelled()).pipe(Effect.asVoid) yield* idleIfCurrent() }), { _tag: "Idle" } as const, ] as const case "Shell": return [ Effect.gen(function* () { yield* stopShell(st.shell) yield* idleIfCurrent() }), { _tag: "Idle" } as const, ] as const case "ShellThenRun": return [ Effect.gen(function* () { yield* stopShell(st.shell) yield* Deferred.fail(st.run.done, new Cancelled()).pipe(Effect.asVoid) yield* idleIfCurrent() }), { _tag: "Idle" } as const, ] as const } }).pipe(Effect.flatten) return { get state() { return state() }, get busy() { return state()._tag !== "Idle" }, ensureRunning, startShell, cancel, } } export * as Runner from "./runner"