1import * as Effect from "effect/Effect";
2import * as FileSystem from "effect/FileSystem";
3import * as Layer from "effect/Layer";
4import * as Path from "effect/Path";
5import { dotAlchemyDirectory } from "../AlchemyContext.ts";
6import type { PlatformError } from "effect/PlatformError";
7import { existsSync } from "node:fs";
8import { decodeFqn, encodeFqn } from "../FQN.ts";
9import { recordStateStoreInit } from "../Telemetry/Metrics.ts";
10import { initialCwd } from "../Util/Node.ts";
11import { writeFileAtomic } from "../Util/AtomicFile.ts";
12import { STATE_STORE_VERSION } from "./HttpStateApi.ts";
13import { State, StateStoreError, type StateService } from "./State.ts";
14import { encodeState, reviveState } from "./StateEncoding.ts";
15
16export const localState = () =>
17 Layer.effect(
18 State,
19 Effect.gen(function* () {
20 const context = yield* Effect.context<
21 FileSystem.FileSystem | Path.Path
22 >();
23
24 const make = makeLocalState().pipe(
25 recordStateStoreInit,
26 Effect.provideContext(context),
27 );
28
29 return yield* Effect.cached(make);
30 }),
31 );
32
33export const makeLocalState = () =>
34 Effect.gen(function* () {
35 const fs = yield* FileSystem.FileSystem;
36 const path = yield* Path.Path;
37 const dotAlchemy = yield* dotAlchemyDirectory;
38 // Keep the store anchored across deploy/destroy even if another task changes cwd.
39 const stateDir = path.resolve(initialCwd, dotAlchemy, "state");
40
41 const fail = (err: PlatformError) =>
42 Effect.fail(
43 new StateStoreError({
44 message: err.message,
45 cause: err,
46 }),
47 );
48
49 const recover = <T>(effect: Effect.Effect<T, PlatformError, never>) =>
50 effect.pipe(
51 Effect.catchTag("PlatformError", (e) =>
52 e.reason._tag === "NotFound" ? Effect.void : fail(e),
53 ),
54 );
55
56 // Directory-level NotFound recovery with a trust-but-verify twist.
57 //
58 // `list` and `deleteStack` treat a missing stage/stack directory as
59 // "no state" — the legitimate shape for a never-deployed (or fully
60 // destroyed) stack. But a FALSE NotFound here is catastrophic: a
61 // destroy that cannot see its state plans "no changes" and silently
62 // leaks every cloud resource of the stack. So before recovering, the
63 // async result is cross-checked with a synchronous `existsSync` — a
64 // deliberately independent code path (`node:fs`, not the FileSystem
65 // service) so a misbehaving async fs answer cannot vouch for itself.
66 // If the directory actually exists, fail loudly instead of degrading
67 // to an empty listing. A directory cannot legitimately reappear
68 // between the two checks: nothing recreates a stage dir concurrently
69 // with the session that is listing or deleting it.
70 const recoverMissingDir = <T>(
71 dir: string,
72 effect: Effect.Effect<T, PlatformError, never>,
73 ) =>
74 effect.pipe(
75 Effect.catchTag("PlatformError", (e) => {
76 if (e.reason._tag !== "NotFound") return fail(e);
77 return Effect.flatMap(
78 Effect.sync(() => existsSync(dir)),
79 (exists) =>
80 exists
81 ? Effect.fail(
82 new StateStoreError({
83 message:
84 `state store reported NotFound for '${dir}', but the directory exists — ` +
85 `refusing to treat the stack's state as empty (a destroy acting on this ` +
86 `answer would leak every resource of the stack)`,
87 cause: e,
88 }),
89 )
90 : Effect.void,
91 );
92 }),
93 );
94
95 const stageDir = ({ stack, stage }: { stack: string; stage: string }) =>
96 path.join(stateDir, stack, stage);
97
98 const resource = ({
99 stack,
100 stage,
101 fqn,
102 }: {
103 stack: string;
104 stage: string;
105 fqn: string;
106 }) => path.join(stateDir, stack, stage, `${encodeFqn(fqn)}.json`);
107
108 const outputFile = ({ stack, stage }: { stack: string; stage: string }) =>
109 path.join(stateDir, stack, stage, `__stack_output__.json`);
110
111 // Write state files atomically so a concurrent `get` (e.g. a parallel
112 // test reading shared `.alchemy/state`) never observes a truncated,
113 // mid-write file — which would otherwise surface as `JSON.parse("")` →
114 // "Unexpected end of JSON input".
115 const writeAtomic = (file: string, contents: string) =>
116 writeFileAtomic(fs, file, contents);
117
118 // Parse a state file, tolerating an empty read. A zero-length file can
119 // linger from a write that was interrupted before this atomic-write change
120 // (or any non-atomic external writer); treat it as "absent" rather than
121 // throwing a JSON parse error that would abort the whole operation.
122 const parseState = (contents: string) =>
123 contents.trim().length === 0
124 ? undefined
125 : JSON.parse(contents, reviveState);
126
127 const created = new Set<string>();
128
129 const ensure = (dir: string) =>
130 created.has(dir)
131 ? Effect.succeed(void 0)
132 : fs
133 .makeDirectory(dir, { recursive: true })
134 .pipe(Effect.tap(() => Effect.sync(() => created.add(dir))));
135
136 const state: StateService = {
137 id: "local",
138 getVersion: () => Effect.succeed(STATE_STORE_VERSION),
139 listStacks: () =>
140 fs.readDirectory(stateDir).pipe(
141 recover,
142 Effect.map((files) => files ?? []),
143 ),
144 listStages: (stack: string) =>
145 fs.readDirectory(path.join(stateDir, stack)).pipe(
146 recover,
147 Effect.map((files) => files ?? []),
148 ),
149 get: (request) =>
150 fs.readFile(resource(request)).pipe(
151 Effect.map((file) => parseState(file.toString())),
152 recover,
153 ),
154 getReplacedResources: Effect.fn(function* (request) {
155 return (yield* Effect.all(
156 (yield* state.list(request)).map((fqn) =>
157 state.get({
158 stack: request.stack,
159 stage: request.stage,
160 fqn,
161 }),
162 ),
163 )).filter((r) => r?.status === "replaced");
164 }),
165 set: (request) =>
166 ensure(stageDir(request)).pipe(
167 Effect.flatMap(() =>
168 writeAtomic(
169 resource(request),
170 JSON.stringify(encodeState(request.value), null, 2),
171 ),
172 ),
173 recover,
174 Effect.map(() => request.value),
175 ),
176 delete: (request) => fs.remove(resource(request)).pipe(recover),
177 deleteStack: ({ stack, stage }) =>
178 Effect.suspend(() => {
179 const dir =
180 stage === undefined
181 ? path.join(stateDir, stack)
182 : stageDir({ stack, stage });
183 return fs.remove(dir, { recursive: true }).pipe(
184 (eff) => recoverMissingDir(dir, eff),
185 // Drop cached `ensure`d directories under the removed tree, or a
186 // later `set` for the same (stack, stage) skips makeDirectory and
187 // its write fails with NotFound — silently swallowed by `recover`.
188 Effect.tap(() =>
189 Effect.sync(() => {
190 for (const cached of created) {
191 if (cached === dir || cached.startsWith(dir + path.sep)) {
192 created.delete(cached);
193 }
194 }
195 }),
196 ),
197 // Deleting the last stage leaves an empty `{stack}/` husk that
198 // `listStacks` would keep reporting forever (and durable
199 // per-test scratch stacks would accumulate one per test). Prune
200 // it when no stages remain. The read-then-remove is not racy:
201 // nothing recreates a stage dir concurrently with the session
202 // that is deleting it (same invariant as recoverMissingDir).
203 Effect.tap(() => {
204 if (stage === undefined) return Effect.void;
205 const stackDir = path.join(stateDir, stack);
206 return fs.readDirectory(stackDir).pipe(
207 Effect.flatMap((entries) =>
208 entries.length === 0
209 ? fs
210 .remove(stackDir, { recursive: true })
211 .pipe(
212 Effect.tap(() =>
213 Effect.sync(() => created.delete(stackDir)),
214 ),
215 )
216 : Effect.void,
217 ),
218 Effect.ignore,
219 );
220 }),
221 );
222 }),
223 list: (request) =>
224 fs.readDirectory(stageDir(request)).pipe(
225 (eff) => recoverMissingDir(stageDir(request), eff),
226 Effect.map((files) =>
227 (files ?? [])
228 // Only decode committed state files. Exclude:
229 // - the `__stack_output__.json` bookkeeping file — `decodeFqn`
230 // turns `__` into `/`, which would slip the literal name past
231 // a bare-name filter and make the engine look up a
232 // non-existent resource;
233 // - in-flight `*.tmp` files written by `writeAtomic` (and any
234 // other non-`.json` entry), which are not resources.
235 .filter(
236 (file) =>
237 file.endsWith(".json") && file !== "__stack_output__.json",
238 )
239 .map((file) => decodeFqn(file.replace(/\.json$/, ""))),
240 ),
241 ),
242 getOutput: (request) =>
243 fs.readFile(outputFile(request)).pipe(
244 Effect.map((file) => parseState(file.toString())),
245 recover,
246 ),
247 setOutput: (request) =>
248 ensure(stageDir(request)).pipe(
249 Effect.flatMap(() =>
250 writeAtomic(
251 outputFile(request),
252 JSON.stringify(encodeState(request.value as any), null, 2),
253 ),
254 ),
255 recover,
256 Effect.map(() => request.value),
257 ),
258 };
259 return state;
260 });