/** * DrainableWorker - A queue-based worker that exposes a `drain()` effect. * * Wraps the common `Queue.unbounded` + `Effect.forever` pattern and adds * a signal that resolves when the queue is empty **and** the current item * has finished processing. This lets tests replace timing-sensitive * `Effect.sleep` calls with deterministic `drain()`. * * @module DrainableWorker */import { Deferred, Effect, Queue, Ref } from "effect";import type { Scope } from "effect"; export interface DrainableWorker<A> { /** * Enqueue a work item and track it for `drain()`. * * This wraps `Queue.offer` so drain state is updated atomically with the * enqueue path instead of inferring it from queue internals. */ readonly enqueue: (item: A) => Effect.Effect<void>; /** * Resolves when the queue is empty and the worker is idle (not processing). */ readonly drain: Effect.Effect<void>;} /** * Create a drainable worker that processes items from an unbounded queue. * * The worker is forked into the current scope and will be interrupted when * the scope closes. A finalizer shuts down the queue. * * @param process - The effect to run for each queued item. * @returns A `DrainableWorker` with `queue` and `drain`. */export const makeDrainableWorker = <A, E, R>( process: (item: A) => Effect.Effect<void, E, R>,): Effect.Effect<DrainableWorker<A>, never, Scope.Scope | R> => Effect.gen(function* () { const queue = yield* Queue.unbounded<A>(); const initialIdle = yield* Deferred.make<void>(); yield* Deferred.succeed(initialIdle, undefined).pipe(Effect.orDie); const state = yield* Ref.make({ outstanding: 0, idle: initialIdle, }); yield* Effect.addFinalizer(() => Queue.shutdown(queue).pipe(Effect.asVoid)); const finishOne = Ref.modify(state, (current) => { const remaining = Math.max(0, current.outstanding - 1); return [ remaining === 0 ? current.idle : null, { outstanding: remaining, idle: current.idle, }, ] as const; }).pipe( Effect.flatMap((idle) => idle === null ? Effect.void : Deferred.succeed(idle, undefined).pipe(Effect.orDie), ), ); yield* Effect.forkScoped( Effect.forever( Queue.take(queue).pipe( Effect.flatMap((item) => process(item).pipe(Effect.ensuring(finishOne))), ), ), ); const enqueue: DrainableWorker<A>["enqueue"] = (item) => Effect.gen(function* () { const nextIdle = yield* Deferred.make<void>(); yield* Ref.update(state, (current) => current.outstanding === 0 ? { outstanding: 1, idle: nextIdle, } : { outstanding: current.outstanding + 1, idle: current.idle, }, ); const accepted = yield* Queue.offer(queue, item); if (!accepted) { yield* finishOne; } }); const drain: DrainableWorker<A>["drain"] = Ref.get(state).pipe( Effect.flatMap(({ idle }) => Deferred.await(idle)), ); return { enqueue, drain } satisfies DrainableWorker<A>; });
/**
* DrainableWorker - A queue-based worker that exposes a `drain()` effect.
*
* Wraps the common `Queue.unbounded` + `Effect.forever` pattern and adds
* a signal that resolves when the queue is empty **and** the current item
* has finished processing. This lets tests replace timing-sensitive
* `Effect.sleep` calls with deterministic `drain()`.
*
* @module DrainableWorker
*/
import { Deferred, Effect, Queue, Ref } from "effect";
import type { Scope } from "effect";
export interface DrainableWorker<A> {
/**
* Enqueue a work item and track it for `drain()`.
*
* This wraps `Queue.offer` so drain state is updated atomically with the
* enqueue path instead of inferring it from queue internals.
*/
readonly enqueue: (item: A) => Effect.Effect<void>;
/**
* Resolves when the queue is empty and the worker is idle (not processing).
*/
readonly drain: Effect.Effect<void>;
}
/**
* Create a drainable worker that processes items from an unbounded queue.
*
* The worker is forked into the current scope and will be interrupted when
* the scope closes. A finalizer shuts down the queue.
*
* @param process - The effect to run for each queued item.
* @returns A `DrainableWorker` with `queue` and `drain`.
*/
export const makeDrainableWorker = <A, E, R>(
process: (item: A) => Effect.Effect<void, E, R>,
): Effect.Effect<DrainableWorker<A>, never, Scope.Scope | R> =>
Effect.gen(function* () {
const queue = yield* Queue.unbounded<A>();
const initialIdle = yield* Deferred.make<void>();
yield* Deferred.succeed(initialIdle, undefined).pipe(Effect.orDie);
const state = yield* Ref.make({
outstanding: 0,
idle: initialIdle,
});
yield* Effect.addFinalizer(() => Queue.shutdown(queue).pipe(Effect.asVoid));
const finishOne = Ref.modify(state, (current) => {
const remaining = Math.max(0, current.outstanding - 1);
return [
remaining === 0 ? current.idle : null,
{
outstanding: remaining,
idle: current.idle,
},
] as const;
}).pipe(
Effect.flatMap((idle) =>
idle === null ? Effect.void : Deferred.succeed(idle, undefined).pipe(Effect.orDie),
),
);
yield* Effect.forkScoped(
Effect.forever(
Queue.take(queue).pipe(
Effect.flatMap((item) => process(item).pipe(Effect.ensuring(finishOne))),
),
),
);
const enqueue: DrainableWorker<A>["enqueue"] = (item) =>
Effect.gen(function* () {
const nextIdle = yield* Deferred.make<void>();
yield* Ref.update(state, (current) =>
current.outstanding === 0
? {
outstanding: 1,
idle: nextIdle,
}
: {
outstanding: current.outstanding + 1,
idle: current.idle,
},
);
const accepted = yield* Queue.offer(queue, item);
if (!accepted) {
yield* finishOne;
}
});
const drain: DrainableWorker<A>["drain"] = Ref.get(state).pipe(
Effect.flatMap(({ idle }) => Deferred.await(idle)),
);
return { enqueue, drain } satisfies DrainableWorker<A>;
});