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
11export class Scheduler extends Context.Service<
12 Scheduler,
13 {
14 readonly register: <E, R>(
15 name: string,
16 runDueWork: Effect.Effect<void, E, R>,
17 ) => Effect.Effect<void, never, R | Scope.Scope>;
18 }
19>()("t3/scheduling/Scheduler") {}
20
21const make = Effect.gen(function* () {
22 const sources = yield* Ref.make(new Map<symbol, Effect.Effect<void>>());
23 const register: Scheduler["Service"]["register"] = Effect.fn("Scheduler.register")(function* <
24 E,
25 R,
26 >(name: string, runDueWork: Effect.Effect<void, E, R>) {
27 const scope = yield* Effect.scope;
28 const context = yield* Effect.context<R>();
29 const permit = yield* Semaphore.make(1);
30 const run = runDueWork.pipe(
31 Effect.provideContext(context),
32 Effect.catchCause((cause) =>
33 Cause.hasInterruptsOnly(cause)
34 ? Effect.void
35 : Effect.logWarning("Scheduler source failed", { source: name, cause }),
36 ),
37 permit.withPermitsIfAvailable(1),
38 Effect.forkIn(scope),
39 Effect.asVoid,
40 );
41 const id = Symbol(name);
42 yield* Effect.acquireRelease(
43 Ref.update(sources, (current) => new Map(current).set(id, run)),
44 () =>
45 Ref.update(sources, (current) => {
46 const next = new Map(current);
47 next.delete(id);
48 return next;
49 }),
50 );
51 // Due work waits for startup recovery, which would cancel runs it started.
52 yield* forkParked(run);
53 });
54 const tick = Ref.get(sources).pipe(
55 Effect.flatMap((current) => Effect.forEach(current.values(), (run) => run, { discard: true })),
56 );
57 yield* forkParked(Effect.sleep("5 seconds").pipe(Effect.andThen(tick), Effect.forever));
58 return Scheduler.of({ register });
59});
60
61export const layer = Layer.effect(Scheduler, make);