1import * as Effect from "effect/Effect";
2import * as Option from "effect/Option";
3import * as Queue from "effect/Queue";
4
5// Do not debounce upstream: its consumer can retain an older snapshot during a save.
6export const runCachePersistence = Effect.fn("runCachePersistence")(function* <A, E, R>(
7 queue: Queue.Queue<A>,
8 persist: (value: A) => Effect.Effect<void, E, R>,
9) {
10 let next = yield* Queue.take(queue);
11 yield* Effect.sleep("500 millis");
12 next = Option.getOrElse(yield* Queue.poll(queue), () => next);
13 while (true) {
14 yield* persist(next);
15 yield* Effect.sleep("10 seconds");
16 next = yield* Queue.take(queue);
17 }
18});