1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
|
import * as TE from "fp-ts/lib/TaskEither";
import { pipe, identity } from "fp-ts/lib/function";
import type { Job, Schedule } from "../canary";
import * as RA from "fp-ts/lib/ReadonlyArray";
import * as T from "fp-ts/lib/Task";
import type { Separated } from "fp-ts/lib/Separated";
interface ScheduledJob {
id: string;
execute: () => TE.TaskEither<Error, boolean>;
at: Date;
schedule: Schedule;
}
type SchedulerState = ReadonlyArray<ScheduledJob>;
const executeAll = (
jobs: ReadonlyArray<TE.TaskEither<Error, boolean>>,
): T.Task<Separated<ReadonlyArray<Error>, ReadonlyArray<boolean>>> =>
pipe(jobs, RA.wilt(T.ApplicativePar)(identity));
export const schedulerLoop =
(state: SchedulerState): TE.TaskEither<Error, void> =>
() => {
const loop = (
currentState: SchedulerState,
time: Date,
): TE.TaskEither<Error, void> =>
pipe(
currentState,
RA.filter((job) => job.at <= time),
RA.map(({ execute }) => execute()),
executeAll,
//T.delay(1000), // Delay for 1 second
// map((newState) => loop(newState)),
);
return loop(state)();
};
|
