1import * as Cause from "effect/Cause";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Ref from "effect/Ref";
6import * as Scope from "effect/Scope";
7import * as Semaphore from "effect/Semaphore";
8
9import { forkParked } from "../serverActivation.ts";
10
11/** Sources load due work from their durable state; registration owns its execution lifetime. */
12export class Scheduler extends Context.Service<
13 Scheduler,
14 {
15 readonly register: <E, R>(
16 name: string,
17 runDueWork: Effect.Effect<void, E, R>,
18 ) => Effect.Effect<void, never, R | Scope.Scope>;
19 }
20>()("t3/scheduling/Scheduler") {}
21
22const make = Effect.gen(function* () {
23 const sources = yield* Ref.make(new Map<symbol, Effect.Effect<void>>());
24 const register: Scheduler["Service"]["register"] = Effect.fn("Scheduler.register")(function* <
25 E,
26 R,
27 >(name: string, runDueWork: Effect.Effect<void, E, R>) {
28 const scope = yield* Effect.scope;
29 const context = yield* Effect.context<R>();
30 const permit = yield* Semaphore.make(1);
31 const run = runDueWork.pipe(
32 Effect.provideContext(context),
33 Effect.catchCause((cause) =>
34 Cause.hasInterruptsOnly(cause)
35 ? Effect.void
36 : Effect.logWarning("Scheduler source failed", { source: name, cause }),
37 ),
38 permit.withPermitsIfAvailable(1),
39 Effect.forkIn(scope),
40 Effect.asVoid,
41 );
42 const id = Symbol(name);
43 yield* Effect.acquireRelease(
44 Ref.update(sources, (current) => new Map(current).set(id, run)),
45 () =>
46 Ref.update(sources, (current) => {
47 const next = new Map(current);
48 next.delete(id);
49 return next;
50 }),
51 );
52 // Due work waits for startup recovery, which would cancel runs it started.
53 yield* forkParked(run);
54 });
55 const tick = Ref.get(sources).pipe(
56 Effect.flatMap((current) => Effect.forEach(current.values(), (run) => run, { discard: true })),
57 );
58 // One clock for all due-work sources. A slow source cannot block another
59 // source or overlap itself, and no extra missed-tick backlog is queued.
60 yield* forkParked(Effect.sleep("5 seconds").pipe(Effect.andThen(tick), Effect.forever));
61 return Scheduler.of({ register });
62});
63
64export const layer = Layer.effect(Scheduler, make);