1/** @effect-diagnostics anyUnknownInErrorContext:off */
2
3import * as Clock from "effect/Clock";
4import * as Effect from "effect/Effect";
5import * as Exit from "effect/Exit";
6import * as Fiber from "effect/Fiber";
7import * as Redacted from "effect/Redacted";
8import * as Scope from "effect/Scope";
9import * as Semaphore from "effect/Semaphore";
10import { isResolved, stripEffects, type Diff } from "../Diff.ts";
11import type { Platform } from "../Platform.ts";
12import type { ProviderService } from "../Provider.ts";
13import type { ScopedPlanStatusSession } from "../Report.ts";
14import type { ResourceBinding, ResourceClassLike, ResourceLike } from "../Resource.ts";
15import { LogAnnotation } from "../Util/LogAnnotation.ts";
16import { sha256 } from "../Util/sha256.ts";
17import * as RpcProvider from "./RpcProvider.ts";
18
19/**
20 * # LocalProvider — the standard shape of a long-running local provider
21 *
22 * A local provider's "physical resource" is a running process (a spawned
23 * dev server, a workerd instance) that lives in the dev sidecar and is
24 * invisible to the state store: a state row can say `created` while the
25 * process is dead (the previous dev session ended), and a redundant
26 * reconcile can arrive while the process is healthy. Every local provider
27 * therefore needs the same machinery:
28 *
29 * - an **instance registry** answering "is something running for this
30 * resource right now?",
31 * - a **config comparison** answering "was it started with the config I'm
32 * being asked for?",
33 * - restart/teardown choreography that keeps the process alive across
34 * deploys (scopes forked from the provider root scope) and kills it on
35 * change or delete.
36 *
37 * {@link make} generates `diff` / `reconcile` / `delete` / `list` from that
38 * machinery; the provider author writes {@link LocalProviderSpec.start}
39 * (how to boot one instance) and optionally
40 * {@link LocalProviderSpec.resolveConfig} (what the restart-relevant config
41 * is) and {@link LocalProviderSpec.stop} (cleanup that outlives instance
42 * scopes).
43 *
44 * The produced layer wraps {@link RpcProvider.effect}, so during
45 * `alchemy dev` the provider (and its registry of running processes) lives
46 * in the sidecar process and survives user-code hot reloads.
47 */
48
49/** Inputs common to every generated lifecycle call. */
50export interface LocalProviderInput<R extends ResourceLike> {
51 id: string;
52 fqn: string;
53 instanceId: string;
54 news: R["Props"];
55 bindings: ResourceBinding<R["Binding"]>[];
56}
57
58export interface StartContext<R extends ResourceLike, Config> extends LocalProviderInput<R> {
59 /**
60 * The value produced by {@link LocalProviderSpec.resolveConfig} for this
61 * reconcile — the same value whose canonical hash decided that a
62 * (re)start was needed, so "what changed?" and "what starts?" can never
63 * drift.
64 */
65 config: Config;
66 session: ScopedPlanStatusSession;
67 /**
68 * Removes this instance from the running registry (a no-op if it has
69 * already been replaced), so the next `diff` reports `update` and the
70 * next reconcile boots a fresh instance. Fork this after the process's
71 * exit — e.g. `child.exitCode.pipe(Effect.exit, Effect.flatMap(() =>
72 * invalidate), Effect.forkScoped)` — for processes that can die on their
73 * own.
74 */
75 invalidate: Effect.Effect<void>;
76}
77
78export interface StopContext {
79 id: string;
80 /**
81 * Fully-qualified name (namespace path + logical id) — the key the
82 * generated instance registry uses. Cross-restart state a provider keeps
83 * outside instance scopes MUST be keyed by this, not `id`: two resources
84 * in different namespaces can share a logical id (two `AWS.Website.*`
85 * sites each declaring a `Command.Dev("Dev")`, ...).
86 */
87 fqn: string;
88 instanceId: string;
89}
90
91export interface StablesContext<R extends ResourceLike, Config> extends LocalProviderInput<R> {
92 config: Config;
93 output: R["Attributes"] | undefined;
94}
95
96/** The default restart-relevant config when `resolveConfig` is omitted. */
97export interface DefaultLocalConfig<R extends ResourceLike> {
98 news: R["Props"];
99 bindings: ResourceBinding<R["Binding"]>[];
100}
101
102export interface LocalProviderSpec<
103 R extends ResourceLike,
104 Config = DefaultLocalConfig<R>,
105 StartR = never,
106> {
107 /**
108 * Derive the restart-relevant desired config from resolved inputs.
109 *
110 * MUST return plain, canonically-serializable data (no closures, no
111 * Effects — `Redacted` values are unwrapped by the hasher) and must be
112 * cheap and side-effect-free: it runs inside `diff` on every plan. The
113 * helper canonically hashes the result for the noop-vs-restart decision,
114 * records it on the registry entry, and passes it to {@link start} so a
115 * restart consumes exactly what was compared.
116 *
117 * Lifecycle-scoped services (`Stack`, `Stage`, `ResourceContext`, `Artifacts`)
118 * are available ambiently (e.g. for `createPhysicalName`); resolve
119 * anything else in the spec effect and close over it.
120 *
121 * @default `{ news: stripEffects(news), bindings }`
122 */
123 resolveConfig?: (ctx: LocalProviderInput<R>) => Effect.Effect<Config, any, any>;
124 /**
125 * Boot one instance: acquire the long-running process in the ambient
126 * `Scope` and return the resource's Attributes once it is *ready* (URL
127 * extracted, first serve succeeded, ...). The scope outlives the
128 * reconcile that started it — it is forked from the provider root scope
129 * and closed only when the instance is restarted (config change) or
130 * deleted. The returned Effect MAY complete after readiness; liveness is
131 * "registry entry with an open scope", not "unfinished fiber".
132 *
133 * Extra requirements beyond `Scope` (`StartR`) become requirements of
134 * the provider layer — they must be present at layer build so the
135 * lifecycle wrapper can hand them to `start` at runtime.
136 */
137 start: (
138 ctx: StartContext<R, Config>,
139 ) => Effect.Effect<R["Attributes"], any, Scope.Scope | StartR>;
140 /**
141 * Extra cleanup on delete, after the instance scope has closed — for
142 * state that intentionally spans restarts and therefore cannot live in
143 * the instance scope (proxy servers kept for URL stability, restart
144 * hooks, ...). NOT called on restarts: a restarting instance keeps that
145 * shared state. Must be idempotent; also called when nothing is running
146 * (e.g. cleaning up a local row during a live deploy).
147 */
148 stop?: (ctx: StopContext) => Effect.Effect<void, any>;
149 /**
150 * Attributes that remain stable across the update the generated `diff`
151 * is about to report (see `Diff.stables`). Called only when the diff is
152 * an `update`.
153 */
154 stables?: (
155 ctx: StablesContext<R, Config>,
156 ) => Effect.Effect<Extract<keyof R["Attributes"], string>[] | undefined, any, any>;
157 precreate?: AnyReqProviderService<R>["precreate"];
158 tail?: AnyReqProviderService<R>["tail"];
159 logs?: AnyReqProviderService<R>["logs"];
160 /**
161 * Override the generated `list` (which joins every registered instance's
162 * fiber to its Attributes).
163 */
164 list?: AnyReqProviderService<R>["list"];
165}
166
167/**
168 * {@link ProviderService} with every requirement channel widened to `any` —
169 * passthrough spec hooks may use lifecycle-scoped services (`Stack`,
170 * `ResourceContext`, ...) that the RpcProvider wrapper provides per call.
171 */
172type AnyReqProviderService<R extends ResourceLike> = ProviderService<
173 R,
174 any,
175 any,
176 any,
177 any,
178 any,
179 any,
180 any,
181 any
182>;
183
184interface Instance<Attributes> {
185 /** Identity guard so a replaced generation's delete cannot kill its successor. */
186 instanceId: string;
187 /** The resolved config the running instance was started with (for diagnostics). */
188 config: unknown;
189 configHash: string;
190 fiber: Fiber.Fiber<Attributes, any>;
191 scope: Scope.Closeable;
192 /** Unique per start; guards invalidate/cleanup against racing restarts. */
193 token: object;
194}
195
196/**
197 * Canonical, collision-free hash of a plain-data value: SHA-256 over a
198 * canonical JSON serialization (sorted keys, unwrapped `Redacted`,
199 * `Uint8Array` and `bigint` encodings, cycle-safe).
200 *
201 * We deliberately do NOT use `Hash.structure`: Effect's structural hash
202 * folds sibling fields together with XOR, so the same value change
203 * appearing in two sibling subtrees cancels out and the hash is unchanged
204 * — exactly the shape of a Worker config that mirrors `env` values into
205 * derived `bindings`. An exact canonical serialization avoids the lossy
206 * fingerprint; SHA-256 keeps the retained signature a fixed 64-char digest.
207 */
208export const canonicalHash = (value: unknown): Effect.Effect<string> => {
209 const seen = new WeakSet<object>();
210 const normalize = (input: unknown): unknown => {
211 if (typeof input === "bigint") return `bigint:${input.toString()}`;
212 if (input === null || typeof input !== "object") return input;
213 if (Redacted.isRedacted(input)) {
214 return { __redacted: input.pipe(Redacted.value, normalize) };
215 }
216 if (seen.has(input)) return "[circular]";
217 seen.add(input);
218 if (input instanceof Uint8Array) return { __bytes: Array.from(input) };
219 if (Array.isArray(input)) return input.map(normalize);
220 return Object.fromEntries(
221 Object.keys(input)
222 .sort()
223 .map((key) => [key, normalize((input as Record<string, unknown>)[key])]),
224 );
225 };
226 return sha256(JSON.stringify(normalize(value)));
227};
228
229/**
230 * Build a local provider layer from a {@link LocalProviderSpec}.
231 *
232 * Generated lifecycle:
233 *
234 * - **diff** — `undefined` while inputs are unresolved; otherwise resolve
235 * the config and compare its canonical hash against the running
236 * instance: running + equal → `noop`, anything else → `update`. This is
237 * what encodes "the process may or may not be running" — state alone can
238 * never answer it.
239 * - **reconcile** — same hash + running instance → join its fiber for the
240 * Attributes (redundant reconciles are free). Otherwise tear the old
241 * instance down and fork {@link LocalProviderSpec.start} in a fresh
242 * scope forked from the provider root scope, so the process outlives the
243 * deploy and dies with the sidecar.
244 * - **delete** — tear down only if the registered instance belongs to the
245 * deleted `instanceId` (create-first replacement safety), then run
246 * {@link LocalProviderSpec.stop}. Idempotent.
247 * - **list** — join every registered instance's fiber to its Attributes.
248 *
249 * All reconcile/delete work for one logical resource is serialized behind
250 * a per-id semaphore so restarts never interleave.
251 *
252 * @param cls - the resource class (or Platform) this provider serves.
253 * @param providersUrl - the provider group module this provider is registered
254 * in (see {@link RpcProvider.effect}).
255 * @param spec - Effect constructing the {@link LocalProviderSpec}; resolve
256 * the services your callbacks need here and close over them.
257 */
258export const make = <
259 R extends ResourceLike,
260 Config = DefaultLocalConfig<R>,
261 StartR = never,
262 Req = never,
263>(
264 cls: ResourceClassLike<R> | Platform<R, any, any, any, any>,
265 providersUrl: string,
266 spec: Effect.Effect<LocalProviderSpec<R, Config, StartR>, never, Req>,
267) =>
268 RpcProvider.effect(
269 cls,
270 providersUrl,
271 // oxlint-disable-next-line effecttsgo/unsafe-effect-type-assertion -- StartR is re-provided by the RpcProvider wrapper, see comment at the cast below
272 Effect.gen(function* () {
273 const {
274 resolveConfig = defaultResolveConfig<R, Config>,
275 start,
276 stop,
277 stables,
278 precreate,
279 tail,
280 logs,
281 list,
282 } = yield* spec;
283
284 // The provider layer scope: in the sidecar this closes at session
285 // shutdown, interrupting every registered instance and running its
286 // finalizers (killing the processes).
287 // Close independent instances concurrently at provider shutdown.
288 const providerScope = yield* Effect.scope;
289 // Registered before the fork, so it runs after every instance closed.
290 let closingAt: number | undefined;
291 yield* Scope.addFinalizer(
292 providerScope,
293 Effect.flatMap(Clock.currentTimeMillis, (now) =>
294 closingAt === undefined
295 ? Effect.void
296 : Effect.logDebug(`shutdown: stopped ${cls.Type} instances in ${now - closingAt}ms`),
297 ),
298 );
299 const rootScope = yield* Scope.fork(providerScope, "parallel");
300 // Registered after the fork, so it runs first: the shutdown start.
301 yield* Scope.addFinalizer(
302 providerScope,
303 Effect.flatMap(Clock.currentTimeMillis, (now) => {
304 if (instances.size === 0) return Effect.void;
305 closingAt = now;
306 return Effect.logDebug(`shutdown: stopping ${instances.size} ${cls.Type} instance(s)`);
307 }),
308 );
309
310 // Keyed by FQN (namespace path + logical id): a restart replaces the
311 // entry, a replacement's new generation overwrites it (and the old
312 // generation's delete is gated on `instanceId` so it cannot kill the
313 // successor). NOT the logical id — that collides across namespaces
314 // (two sites each declaring a `Command.Dev("Dev")` would evict each
315 // other's running process).
316 const instances = new Map<string, Instance<R["Attributes"]>>();
317
318 // Serializes reconcile/delete per resource so a restart can never
319 // interleave with another restart or a delete and leak a scope.
320 const locks = new Map<string, Semaphore.Semaphore>();
321 const lock = (id: string) => {
322 let semaphore = locks.get(id);
323 if (!semaphore) {
324 semaphore = Semaphore.makeUnsafe(1);
325 locks.set(id, semaphore);
326 }
327 return semaphore;
328 };
329 const withLock = <A, E, RR>(key: string, effect: Effect.Effect<A, E, RR>) =>
330 Semaphore.withPermits(lock(key), 1)(effect);
331
332 const resolveDesired = Effect.fn(function* (input: LocalProviderInput<R>) {
333 const config = yield* resolveConfig(input);
334 const configHash = yield* canonicalHash(config);
335 return { config, configHash };
336 });
337
338 const teardown = Effect.fn(function* (fqn: string, instance: Instance<R["Attributes"]>) {
339 const start = yield* Clock.currentTimeMillis;
340 yield* Fiber.interrupt(instance.fiber);
341 yield* Scope.close(instance.scope, Exit.void);
342 const end = yield* Clock.currentTimeMillis;
343 yield* Effect.logDebug(`stopped ${fqn} in ${end - start}ms`).pipe(
344 Effect.annotateLogs(LogAnnotation.fqn, fqn),
345 );
346 });
347
348 const startInstance = Effect.fn(function* (
349 input: LocalProviderInput<R>,
350 session: ScopedPlanStatusSession,
351 config: Config,
352 configHash: string,
353 ) {
354 const token = {};
355 const invalidate = Effect.sync(() => {
356 if (instances.get(input.fqn)?.token === token) {
357 instances.delete(input.fqn);
358 }
359 });
360 const scope = yield* Scope.fork(rootScope);
361 const fiber = yield* start({
362 ...input,
363 config,
364 session,
365 invalidate,
366 }).pipe(Effect.forkDetach, Scope.provide(scope));
367 instances.set(input.fqn, {
368 instanceId: input.instanceId,
369 config,
370 configHash,
371 fiber,
372 scope,
373 token,
374 });
375 return yield* Fiber.join(fiber).pipe(
376 Effect.onExit((exit) =>
377 exit._tag === "Failure"
378 ? // A failed start must not linger as "running" — the next
379 // plan should report `update` and try again — nor keep what
380 // it acquired: the instance scope would otherwise hold a
381 // process that outlived its start (one still waiting for
382 // its URL when the deploy was interrupted, say) until the
383 // provider shut down.
384 invalidate.pipe(Effect.andThen(Scope.close(scope, exit)))
385 : Effect.void,
386 ),
387 );
388 });
389
390 const provider = {
391 list:
392 list ??
393 (() =>
394 Effect.forEach(
395 Array.from(instances.values()),
396 (instance) => Fiber.join(instance.fiber),
397 { concurrency: "unbounded" },
398 )),
399 diff: Effect.fn(function* ({
400 id,
401 fqn,
402 instanceId,
403 news: rawNews,
404 newBindings,
405 output,
406 }: {
407 id: string;
408 fqn: string;
409 instanceId: string;
410 news: any;
411 newBindings: any;
412 output: R["Attributes"] | undefined;
413 }) {
414 // Effect-valued leaves (tagged Worker classes in `env`, ...) never
415 // resolve at plan time; their identity is carried by the resolved
416 // binding data. Strip them so the config-based diff still runs.
417 const news = stripEffects(rawNews);
418 if (!isResolved(news) || !isResolved(newBindings)) return undefined;
419 const input: LocalProviderInput<R> = {
420 id,
421 fqn,
422 instanceId,
423 news: news as R["Props"],
424 bindings: newBindings as ResourceBinding<R["Binding"]>[],
425 };
426 const { config, configHash } = yield* resolveDesired(input);
427 const existing = instances.get(fqn);
428 if (existing && existing.configHash === configHash) {
429 return { action: "noop" } satisfies Diff;
430 }
431 return {
432 action: "update",
433 stables: stables ? yield* stables({ ...input, config, output }) : undefined,
434 } satisfies Diff;
435 }),
436 reconcile: Effect.fn(function* ({
437 id,
438 fqn,
439 instanceId,
440 news,
441 bindings,
442 session,
443 }: {
444 id: string;
445 fqn: string;
446 instanceId: string;
447 news: R["Props"];
448 bindings: ResourceBinding<R["Binding"]>[];
449 session: ScopedPlanStatusSession;
450 }) {
451 return yield* withLock(
452 fqn,
453 Effect.gen(function* () {
454 const input: LocalProviderInput<R> = {
455 id,
456 fqn,
457 instanceId,
458 news: stripEffects(news) as R["Props"],
459 bindings,
460 };
461 const { config, configHash } = yield* resolveDesired(input);
462 const existing = instances.get(fqn);
463 if (existing) {
464 if (existing.configHash === configHash) {
465 yield* Effect.log(`No changes, using existing instance`);
466 // Adopt the newest instanceId so a later replacement's
467 // old-generation delete cannot tear this instance down.
468 existing.instanceId = instanceId;
469 return yield* Fiber.join(existing.fiber);
470 }
471 yield* Effect.log(`Changes detected, restarting instance`);
472 yield* teardown(fqn, existing);
473 instances.delete(fqn);
474 }
475 return yield* startInstance(input, session, config, configHash);
476 }).pipe(Effect.annotateLogs(LogAnnotation.fqn, fqn)),
477 );
478 }),
479 delete: Effect.fn(function* ({
480 id,
481 fqn,
482 instanceId,
483 }: {
484 id: string;
485 fqn: string;
486 instanceId: string;
487 }) {
488 yield* withLock(
489 fqn,
490 Effect.gen(function* () {
491 const existing = instances.get(fqn);
492 if (existing && existing.instanceId !== instanceId) {
493 // A replacement's new generation is registered under this
494 // logical id — the old generation's delete must not touch
495 // it (nor the cross-restart state `stop` would clean up).
496 return;
497 }
498 if (existing) {
499 yield* teardown(fqn, existing);
500 instances.delete(fqn);
501 }
502 // Runs even when nothing is registered: cross-restart state
503 // (proxies, restart hooks) and out-of-session cleanup (e.g.
504 // deleting a local row during a live deploy) still need it.
505 if (stop) {
506 yield* stop({ id, fqn, instanceId });
507 }
508 }),
509 );
510 }),
511 ...(precreate ? { precreate } : {}),
512 ...(tail ? { tail } : {}),
513 ...(logs ? { logs } : {}),
514 };
515
516 return provider as any;
517 // `StartR` (start's requirements beyond Scope) is surfaced as a layer
518 // requirement: the RpcProvider wrapper captures the layer-build
519 // context and provides it to every lifecycle call, so services the
520 // layer demanded at build are available to `start` at runtime.
521 }) as Effect.Effect<ProviderService<R>, never, Req | Exclude<StartR, Scope.Scope>>,
522 );
523
524const defaultResolveConfig = <R extends ResourceLike, Config>(
525 input: LocalProviderInput<R>,
526): Effect.Effect<Config, any> =>
527 Effect.succeed({
528 news: stripEffects(input.news),
529 bindings: input.bindings,
530 } as Config);