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