BackupHEADmain

1 files changed, 21 insertions, 7 deletions

diff --git a/src/util/scheduler.ts b/src/util/scheduler.ts
index 9f9f594..476aca0 100644
--- a/src/util/scheduler.ts
+++ b/src/util/scheduler.ts
@@ -1,24 +1,38 @@
 import * as TE from "fp-ts/lib/TaskEither";
-import { pipe } from "fp-ts/lib/function";
-import type { Schedule } from "../canary";
+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>;
+  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): TE.TaskEither<Error, void> =>
+    const loop = (
+      currentState: SchedulerState,
+      time: Date,
+    ): TE.TaskEither<Error, void> =>
       pipe(
-        executeDueJobs(currentState),
-        delay(1000), // Delay for 1 second
-        map((newState) => loop(newState)),
+        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)();