1import * as Effect from "effect/Effect";
2import { type StateService } from "./State.ts";
3
4/**
5 * Synchronize all state (every stack/stage/resource) from `source` into
6 * `destination` so that `destination` becomes a mirror of `source`.
7 *
8 * For each `{ stack, stage, fqn }` present in `source`, the resource is
9 * written into `destination`, overwriting any existing entry under the same
10 * key. Any keys present in `destination` but absent from `source` are
11 * deleted, ensuring the two stores end up structurally identical.
12 *
13 * Stacks are walked sequentially; stages within a stack and resources
14 * within a stage are processed concurrently for throughput.
15 */
16export const syncState = Effect.fn(function* (
17 source: StateService,
18 destination: StateService,
19 options?: {
20 stacks?: string[];
21 /**
22 * Maximum number of resources to copy in parallel within a single stage.
23 * @default "unbounded".
24 */
25 concurrency?: number | "unbounded";
26 },
27) {
28 const concurrency = options?.concurrency ?? "unbounded";
29 const [sourceStacks, destStacks] = yield* Effect.all([
30 source.listStacks(),
31 destination.listStacks(),
32 ]);
33 const sourceStackSet = new Set(
34 sourceStacks.filter((stack) => options?.stacks?.includes(stack) ?? true),
35 );
36
37 yield* Effect.forEach(
38 sourceStacks,
39 Effect.fn(function* (stack) {
40 const [sourceStages, destStages] = yield* Effect.all([
41 source.listStages(stack),
42 destination.listStages(stack),
43 ]);
44 const stages = union(sourceStages, destStages);
45
46 yield* Effect.forEach(
47 stages,
48 Effect.fn(function* (stage) {
49 const sourceFqns = yield* source.list({ stack, stage });
50 const destFqns = yield* destination.list({ stack, stage });
51
52 const sourceSet = new Set(sourceFqns);
53 const toDelete = destFqns.filter((fqn) => !sourceSet.has(fqn));
54
55 yield* Effect.all(
56 [
57 Effect.forEach(
58 sourceFqns,
59 Effect.fn(function* (fqn) {
60 const value = yield* source.get({ stack, stage, fqn });
61 if (value) {
62 yield* destination.set({ stack, stage, fqn, value });
63 }
64 }),
65 { concurrency },
66 ),
67 Effect.forEach(
68 toDelete,
69 (fqn) => destination.delete({ stack, stage, fqn }),
70 { concurrency },
71 ),
72 ],
73 { concurrency: "unbounded" },
74 );
75 }),
76 { concurrency: "unbounded" },
77 );
78 }),
79 );
80
81 yield* Effect.forEach(
82 destStacks.filter((stack) => !sourceStackSet.has(stack)),
83 (stack) => destination.deleteStack({ stack }),
84 );
85});
86
87const union = <T>(left: Iterable<T>, right: Iterable<T>) => [
88 ...new Set([...left, ...right]),
89];