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