1/** @effect-diagnostics anyUnknownInErrorContext:off */
2
3/**
4 * INTERNAL — the shared skeleton for `alchemy dev` AWS providers that deploy
5 * into the floci emulator and keep a long-lived watch loop hot-swapping the
6 * resource on file change ([Lambda FlociFunctionProvider](../Lambda/FlociFunctionProvider.ts),
7 * [ECS FlociTaskProvider](../ECS/FlociTaskProvider.ts),
8 * [ECS FlociServiceProvider](../ECS/FlociServiceProvider.ts)).
9 *
10 * What the helper owns (identically for every consumer):
11 *
12 * - **RPC-sidecar hosting** via {@link RpcProvider.effect} with the shared
13 * AWS provider group ([Local.ts](./Local.ts)), so the provider and its
14 * watch fibers survive exec-process hot reloads during `alchemy dev`.
15 * - **Live-provider delegation** — builds the caller's LIVE provider layer
16 * inside the floci override context ({@link flociServices} + optional
17 * caller-supplied extra service layers) and wraps every lifecycle method
18 * with {@link withProviderContext}, so the exact same reconcile/read/
19 * delete/list logic runs against the emulator gateway with the dummy
20 * account.
21 * - **The instanceId-guarded watch registry** — one watch fiber per logical
22 * id; a config change interrupts and restarts it; a create-first
23 * replacement's old-generation delete cannot kill the successor's watcher.
24 * - **The dev diff policy** — the diff NEVER bundles or builds:
25 * resource-specific replacement rules run first (spec `replaceOn`), a
26 * fresh session with state rows but no registered watcher forces an
27 * `update` (converges the emulator and restarts the watcher), and
28 * structurally-equal props (spec `normalizeProps`) are a `noop` because
29 * content-only changes are the watch loop's job.
30 * - **Serialized reconciles** — engine reconciles, watcher-triggered
31 * re-reconciles ({@link DevWatchContext.rerunReconcile}), and deletes are
32 * serialized per logical id, so a rebuild-and-swap never interleaves with
33 * an engine-driven converge.
34 */
35
36import * as Context from "effect/Context";
37import * as Effect from "effect/Effect";
38import * as Fiber from "effect/Fiber";
39import * as Layer from "effect/Layer";
40import * as Scope from "effect/Scope";
41import * as Semaphore from "effect/Semaphore";
42import { havePropsChanged, isResolved, stripEffects } from "../../Diff.ts";
43import { canonicalHash } from "../../Local/LocalProvider.ts";
44import * as RpcProvider from "../../Local/RpcProvider.ts";
45import type { Platform } from "../../Platform.ts";
46import type { ProviderService } from "../../Provider.ts";
47import type { ResourceClassLike, ResourceLike } from "../../Resource.ts";
48import { flociServices } from "./FlociServices.ts";
49import { withProviderContext } from "./ProviderContext.ts";
50import { moduleExtension } from "../../Util/Node.ts";
51
52/**
53 * The AWS provider group module ([Local.ts](./Local.ts)) — every
54 * {@link makeDevWatchProvider} consumer passes this so all floci dev
55 * providers are served from the same group by the dev sidecar.
56 */
57export const flociProvidersUrl = () =>
58 import.meta.resolve(
59 `./Local${moduleExtension(import.meta.url)}`,
60 import.meta.url,
61 );
62
63/** A diff decision produced by {@link DevWatchSpec.replaceOn}. */
64export interface DevReplaceDecision {
65 readonly action: "replace";
66 readonly deleteFirst?: boolean;
67}
68
69/**
70 * The context handed to {@link DevWatchSpec.startWatch}. Runs inside the
71 * sidecar with the floci override services provided.
72 */
73export interface DevWatchContext<Props, Attrs> {
74 readonly id: string;
75 readonly instanceId: string;
76 /** The resolved props of the reconcile that started this watcher. */
77 readonly news: Props;
78 /** Attributes at watch start. */
79 readonly attrs: Attrs;
80 /** The freshest attributes (updated by {@link rerunReconcile}). */
81 readonly currentAttrs: Effect.Effect<Attrs>;
82 /**
83 * Re-runs the wrapped LIVE reconcile with the latest engine input
84 * (props/bindings) and the current attributes as `output`, records and
85 * returns the fresh attributes. Serialized with engine reconciles/deletes
86 * on the same logical id.
87 */
88 readonly rerunReconcile: Effect.Effect<Attrs, unknown>;
89}
90
91/**
92 * The per-resource contract for {@link makeDevWatchProvider}: everything the
93 * shared skeleton cannot know about the resource.
94 */
95export interface DevWatchSpec<Props, Attrs> {
96 /**
97 * The LIVE provider layer (e.g. `() => FunctionProvider()`). Built inside
98 * the floci override context and endpoint-wrapped, exactly like the plain
99 * `flociDual` resources.
100 */
101 readonly liveProvider: () => Layer.Layer<any, any, any>;
102 /**
103 * Extra service layers built alongside {@link flociServices} inside the
104 * override context (e.g. Lambda's fresh `AssetsLive`, whose cached bucket
105 * lookup must resolve against the emulator and never share a cache with
106 * the live arm's instance).
107 */
108 readonly services?: Layer.Layer<any, any, any>;
109 /**
110 * The restart surface of the watch loop: everything that changes WHAT the
111 * watcher builds or WHERE it publishes. Plain, canonically-hashable data
112 * only — closures degrade to stable placeholders under
113 * {@link canonicalHash}. `news` arrives {@link stripEffects}-stripped.
114 */
115 readonly watchConfigOf: (news: Props, attrs: Attrs) => unknown;
116 /**
117 * Transform the resolved props before EVERY delegated reconcile —
118 * engine-driven and watcher-driven alike — inside the per-id lock, with
119 * the floci override services provided. This is where a provider swaps a
120 * build-here artifact form for a build-locally one (e.g. the MicroVM dev
121 * provider docker-builds the image on the HOST and hands the live
122 * reconcile a `codeArtifact: { uri: "docker://…" }` instead of `main`/
123 * `context`, so the emulator never zips, uploads, or builds anything).
124 * The untransformed props stay in the watch registry, so restarts and
125 * watch-config comparisons see the user's real inputs.
126 */
127 readonly transformReconcileNews?: (input: {
128 id: string;
129 news: Props;
130 }) => Effect.Effect<Props, unknown, any>;
131 /**
132 * Called after EVERY successful reconcile — engine-driven (a prop
133 * change, a fresh session's converge) and watcher-driven
134 * ({@link DevWatchContext.rerunReconcile}) alike — inside the per-id
135 * lock, with the floci override services provided. This is where a
136 * provider makes the emulator's RUNNING state follow the new revision
137 * (e.g. ECS restarting tasks): the watch trigger only fires for FILE
138 * changes, so restart logic that lives only in the watch loop silently
139 * skips prop-driven updates (an inline dockerfile edit, a new env var)
140 * and running tasks keep serving the old revision until the next source
141 * edit.
142 *
143 * `previous` is the attrs before this reconcile — `undefined` on a
144 * greenfield create.
145 */
146 readonly onReconciled?: (input: {
147 id: string;
148 news: Props;
149 previous: Attrs | undefined;
150 attrs: Attrs;
151 }) => Effect.Effect<void, unknown, any>;
152 /**
153 * Resource-specific replacement rules, mirroring the LIVE diff's cheap
154 * checks (never bundling/building). Checked before the generic dev policy.
155 */
156 readonly replaceOn?: (input: {
157 id: string;
158 olds: Props;
159 news: Props;
160 output: Attrs;
161 }) => Effect.Effect<DevReplaceDecision | undefined, never, any>;
162 /**
163 * Normalization applied to both sides before the structural code-only
164 * `noop` equivalence check (e.g. Lambda resolves layer references to ARNs).
165 * @default identity
166 */
167 readonly normalizeProps?: (props: Props) => unknown;
168 /**
169 * Pin `precreate`'s news to a stub-compatible shape (Lambda pins the
170 * handler-affecting props to the 503 stub's own single-module shape).
171 * Only consulted when the live provider implements `precreate`.
172 */
173 readonly transformPrecreateNews?: (news: Props) => Props;
174 /**
175 * The watch loop, forked (detached, in the provider scope) after every
176 * successful reconcile that changes the watch config. Runs with the floci
177 * override services provided; failures are logged, never propagated. A
178 * spec with nothing to watch for a given props shape simply returns.
179 */
180 readonly startWatch: (
181 ctx: DevWatchContext<Props, Attrs>,
182 ) => Effect.Effect<void, any, any>;
183}
184
185interface WatchEntry {
186 readonly instanceId: string;
187 readonly configHash: string;
188 fiber: Fiber.Fiber<void, never> | undefined;
189 /** The latest engine reconcile input (news/bindings/…) for re-runs. */
190 lastInput: any;
191 /** The freshest attributes. */
192 attrs: any;
193}
194
195/**
196 * Build an RPC-sidecar-hosted `alchemy dev` provider that delegates its
197 * lifecycle to the LIVE provider inside the floci override context and owns
198 * a per-resource watch loop. See the module doc for the exact split between
199 * the shared skeleton and the {@link DevWatchSpec}.
200 */
201export const makeDevWatchProvider = <
202 R extends ResourceLike,
203 Props = R["Props"],
204 Attrs = R["Attributes"],
205>(
206 cls: ResourceClassLike<R> | Platform<R, any, any, any, any, any>,
207 providersUrl: string,
208 spec: DevWatchSpec<Props, Attrs>,
209) =>
210 RpcProvider.effect(
211 cls,
212 providersUrl,
213 Effect.gen(function* () {
214 const scope = yield* Effect.scope;
215 const ambient = yield* Effect.context<never>();
216
217 // The floci override context (emulator endpoint, dummy credentials,
218 // account 000000000000) plus any caller-supplied extra services.
219 const overrides = yield* Layer.buildWithScope(
220 Layer.mergeAll(flociServices(), spec.services ?? Layer.empty).pipe(
221 Layer.provide(Layer.succeedContext(ambient)),
222 ) as Layer.Layer<any, any, never>,
223 scope,
224 ).pipe(Effect.orDie);
225 const services = Layer.succeedContext(
226 Context.merge(ambient, overrides),
227 ) as Layer.Layer<any, never, never>;
228
229 // The live provider machinery, every lifecycle method endpoint-wrapped
230 // to the emulator — same F1 combinator the plain `flociDual` resources
231 // use, applied to a provider instance we build ourselves.
232 const liveCtx = yield* Layer.buildWithScope(
233 spec
234 .liveProvider()
235 .pipe(Layer.provide(services)) as unknown as Layer.Layer<
236 any,
237 any,
238 never
239 >,
240 scope,
241 ).pipe(Effect.orDie);
242 const live = liveCtx.mapUnsafe.get(cls.Type) as ProviderService<R>;
243 const wrapped = withProviderContext(live, services);
244
245 // Sidecar-process watch registry, keyed by logical id. `instanceId`
246 // guards a create-first replacement's old-generation delete against
247 // killing the successor's watcher.
248 const watches = new Map<string, WatchEntry>();
249
250 // Per-id serialization: engine reconciles, watcher re-reconciles, and
251 // deletes never interleave for the same logical resource.
252 const locks = new Map<string, Semaphore.Semaphore>();
253 const withLock = (id: string) => {
254 let lock = locks.get(id);
255 if (lock === undefined) {
256 lock = Semaphore.makeUnsafe(1);
257 locks.set(id, lock);
258 }
259 return <A, E, Req>(effect: Effect.Effect<A, E, Req>) =>
260 Semaphore.withPermits(lock, 1)(effect);
261 };
262
263 // Plan-status callbacks don't cross the RPC boundary (functions
264 // serialize to `null`) — the live machinery calls `session.note`, so
265 // substitute a log-backed session when the wire stripped it.
266 const withSafeSession = (input: any) =>
267 typeof input?.session?.note === "function"
268 ? input
269 : {
270 ...input,
271 session: {
272 ...input?.session,
273 note: (note: string) => Effect.logDebug(note),
274 },
275 };
276
277 const ensureWatch = Effect.fn(function* (input: any, attrs: any) {
278 const strippedNews = stripEffects(input.news);
279 const configHash = yield* canonicalHash(
280 spec.watchConfigOf(strippedNews as Props, attrs as Attrs),
281 );
282 const existing = watches.get(input.id);
283 if (
284 existing !== undefined &&
285 existing.configHash === configHash &&
286 existing.instanceId === input.instanceId
287 ) {
288 // Same watcher keeps running — just refresh the re-run input.
289 existing.lastInput = input;
290 existing.attrs = attrs;
291 return;
292 }
293 if (existing?.fiber !== undefined) {
294 yield* Fiber.interrupt(existing.fiber);
295 }
296 const entry: WatchEntry = {
297 instanceId: input.instanceId,
298 configHash,
299 fiber: undefined,
300 lastInput: input,
301 attrs,
302 };
303 watches.set(input.id, entry);
304 const ctx: DevWatchContext<Props, Attrs> = {
305 id: input.id,
306 instanceId: input.instanceId,
307 news: strippedNews as Props,
308 attrs: attrs as Attrs,
309 currentAttrs: Effect.sync(() => entry.attrs as Attrs),
310 rerunReconcile: withLock(input.id)(
311 Effect.gen(function* () {
312 const previous = entry.attrs as Attrs | undefined;
313 const rerunNews =
314 spec.transformReconcileNews !== undefined
315 ? yield* spec
316 .transformReconcileNews({
317 id: input.id,
318 news: stripEffects(entry.lastInput.news) as Props,
319 })
320 .pipe(Effect.provide(services))
321 : entry.lastInput.news;
322 const fresh = yield* wrapped.reconcile(
323 withSafeSession({
324 ...entry.lastInput,
325 news: rerunNews,
326 olds: entry.lastInput.news,
327 output: entry.attrs,
328 }) as any,
329 );
330 entry.attrs = fresh;
331 if (spec.onReconciled !== undefined) {
332 yield* spec
333 .onReconciled({
334 id: input.id,
335 news: stripEffects(entry.lastInput.news) as Props,
336 previous,
337 attrs: fresh as Attrs,
338 })
339 .pipe(Effect.provide(services));
340 }
341 return fresh as Attrs;
342 }),
343 ) as Effect.Effect<Attrs, unknown>,
344 };
345 const fiber = yield* spec.startWatch(ctx).pipe(
346 Effect.provide(services),
347 Effect.catchCause((cause) =>
348 Effect.logWarning(
349 `[alchemy dev] ${input.id}: watch loop exited`,
350 cause,
351 ),
352 ),
353 Effect.forkDetach,
354 Scope.provide(scope),
355 );
356 entry.fiber = fiber as Fiber.Fiber<void, never>;
357 });
358
359 const stopWatch = Effect.fn(function* (id: string, instanceId: string) {
360 const existing = watches.get(id);
361 if (existing !== undefined && existing.instanceId === instanceId) {
362 watches.delete(id);
363 if (existing.fiber !== undefined) {
364 yield* Fiber.interrupt(existing.fiber);
365 }
366 }
367 });
368
369 const normalize = (props: Props | undefined): object | undefined =>
370 props === undefined
371 ? undefined
372 : ((spec.normalizeProps?.(props) ?? props) as object);
373
374 const provider: any = {
375 // read / precreate / list / tail / logs / stables delegate to the
376 // endpoint-wrapped live provider (spreading the Proxy materializes
377 // each member through the wrapping `get`).
378 ...(wrapped as any),
379 diff: Effect.fn(function* ({
380 id,
381 olds,
382 news: rawNews,
383 output,
384 }: {
385 id: string;
386 olds: Props;
387 news: Props;
388 output: Attrs | undefined;
389 }) {
390 // Runtime props (an Effect-native Function's handler, a Layer)
391 // arrive as Effects, which `isResolved` classifies as unresolved.
392 // They are irrelevant to the dev diff, so strip them BEFORE the
393 // resolved check — otherwise this diff bails to the engine default
394 // (`havePropsChanged` → noop) and the "fresh session with a state
395 // row but no watcher" branch below never runs: `alchemy dev`
396 // resumed over an existing stage would start no watch loop and
397 // hot reload would silently be dead.
398 const news = stripEffects(rawNews);
399 if (!isResolved(news)) return;
400 if (!output) return undefined;
401 if (spec.replaceOn !== undefined) {
402 const decision = yield* spec.replaceOn({ id, olds, news, output });
403 if (decision !== undefined) return decision;
404 }
405 // A fresh dev session has state rows but no running watcher (and
406 // possibly a wiped emulator) — force a reconcile to converge floci
407 // and (re)start the watch loop. A DEAD watcher (its loop exited or
408 // was interrupted) counts as none: nothing is handling content
409 // changes for this id anymore.
410 const entry = watches.get(id);
411 if (
412 entry === undefined ||
413 (entry.fiber !== undefined &&
414 entry.fiber.pollUnsafe() !== undefined)
415 ) {
416 return { action: "update" as const };
417 }
418 // The watcher may have swapped content SINCE state was last
419 // persisted (its rerunReconcile updates the emulator and the
420 // sidecar's in-memory attrs, not the state store). If the
421 // persisted attrs drifted from the watcher's current attrs, plan
422 // an update — the content-addressed reconcile is cheap, the
423 // engine persists the fresh attrs, and downstream consumers
424 // (stack outputs, Actions keyed on `code.hash`) see the swap
425 // instead of replaying stale state forever.
426 const persistedHash = yield* canonicalHash(output as object);
427 const watcherHash = yield* canonicalHash(entry.attrs as object);
428 if (persistedHash !== watcherHash) {
429 return { action: "update" as const };
430 }
431 if (!havePropsChanged(normalize(olds), normalize(news)!)) {
432 // Content-only changes are the watch loop's job — the dev diff
433 // never bundles or builds.
434 return { action: "noop" as const };
435 }
436 return { action: "update" as const };
437 }),
438 reconcile: Effect.fn(function* (input: any) {
439 return yield* withLock(input.id)(
440 Effect.gen(function* () {
441 const previous = (input.output ??
442 watches.get(input.id)?.attrs) as Attrs | undefined;
443 const reconcileNews =
444 spec.transformReconcileNews !== undefined
445 ? yield* spec
446 .transformReconcileNews({
447 id: input.id,
448 news: stripEffects(input.news) as Props,
449 })
450 .pipe(Effect.provide(services))
451 : input.news;
452 const attrs = yield* wrapped.reconcile(
453 withSafeSession({ ...input, news: reconcileNews }) as any,
454 );
455 yield* ensureWatch(input, attrs);
456 if (spec.onReconciled !== undefined) {
457 yield* spec
458 .onReconciled({
459 id: input.id,
460 news: stripEffects(input.news) as Props,
461 previous,
462 attrs: attrs as Attrs,
463 })
464 .pipe(Effect.provide(services));
465 }
466 return attrs;
467 }),
468 );
469 }),
470 delete: Effect.fn(function* (input: any) {
471 yield* stopWatch(input.id, input.instanceId);
472 yield* withLock(input.id)(
473 wrapped.delete(withSafeSession(input) as any),
474 );
475 }),
476 };
477 if (typeof (live as any).precreate === "function") {
478 provider.precreate = (input: any) =>
479 (wrapped as any).precreate(
480 withSafeSession({
481 ...input,
482 news:
483 spec.transformPrecreateNews !== undefined
484 ? spec.transformPrecreateNews(input.news)
485 : input.news,
486 }),
487 );
488 }
489 return provider;
490 }),
491 );