1import * as Context from "effect/Context";
2import * as Effect from "effect/Effect";
3import * as Exit from "effect/Exit";
4import * as Layer from "effect/Layer";
5import type * as Redacted from "effect/Redacted";
6import * as Scope from "effect/Scope";
7import * as Semaphore from "effect/Semaphore";
8import type * as SqlClient from "effect/sql/SqlClient";
9import type * as SqlError from "effect/sql/SqlError";
10import { resolveConnectionOptions } from "../SQL/PostgresTls.ts";
11import { recordStateStoreInit } from "../Telemetry/Metrics.ts";
12import { STATE_STORE_VERSION } from "./HttpStateApi.ts";
13import type { ReplacedResourceState } from "./ResourceState.ts";
14import {
15 State,
16 StateStoreError,
17 type PersistedState,
18 type StateService,
19} from "./State.ts";
20import { encodeState, reviveStateRecursive } from "./StateEncoding.ts";
21
22// `@effect/sql-pg` is loaded lazily rather than imported statically (as
23// `SQL/Postgres.ts` does) so that `postgresState({ client })` — which takes
24// any `@effect/sql` client — works without the optional peer installed at
25// all. Only `postgresState({ url })`, which builds a pool, needs it.
26// `@effect/sql-pg` depends on `pg` itself, so `pg` is never imported here.
27const importPgClient = () =>
28 import("@effect/sql-pg/PgClient").catch((cause) => {
29 throw new Error(
30 "Failed to load '@effect/sql-pg'. Install the optional peer dependency '@effect/sql-pg' to use postgresState with a `url`.",
31 { cause },
32 );
33 });
34
35export interface PostgresStateOptions<E = never, R = never> {
36 /**
37 * An existing `@effect/sql` client. The caller owns its lifecycle — the
38 * store only issues queries against it. Exactly one of `client` and `url`
39 * must be provided.
40 *
41 * The client must be **pool-backed** and able to hand out at least two
42 * concurrent connections: the store keeps one connection reserved for the
43 * advisory lock and verifies that lock from another one. A
44 * single-connection client (`PgClient.makeClient`) is refused at the first
45 * lease check, and a pool capped at `maxConnections: 1` blocks forever
46 * waiting for a free connection.
47 */
48 client?: SqlClient.SqlClient;
49 /**
50 * Postgres connection URL, as a `Redacted` value or an Effect yielding one
51 * — `Config.Redacted("STATE_DATABASE_URL")` is itself an Effect, so it can
52 * be passed directly. The store creates its own `@effect/sql-pg` pool from
53 * the URL and closes that pool when the state layer is released.
54 */
55 url?:
56 | Redacted.Redacted<string>
57 | Effect.Effect<Redacted.Redacted<string>, E, R>;
58 /**
59 * Prefix for the advisory-lock key. The full key for a stack/stage is
60 * `{lockKeyPrefix}:{stack}/{stage}`.
61 *
62 * @default "alchemy"
63 */
64 lockKeyPrefix?: string;
65 /**
66 * State-store id reported in telemetry (`alchemy.state_store.id`).
67 *
68 * @default "postgres"
69 */
70 id?: string;
71 /**
72 * How long (in milliseconds) a passing lease check is trusted before the
73 * next state operation re-verifies the advisory lock. A lease lost inside
74 * the window is detected within this many milliseconds rather than
75 * instantly; in exchange, a burst of state operations does one lock
76 * round-trip instead of one per operation.
77 *
78 * @default 5000
79 */
80 leaseCheckTtlMs?: number;
81}
82
83const DEFAULT_LEASE_CHECK_TTL_MS = 5_000;
84
85/**
86 * Amortizes a lease check over a short TTL: a passing check is trusted for
87 * `ttlMs`, so a burst of operations inside the window does one round-trip,
88 * not one per operation. A failing check is never cached — it propagates
89 * immediately and leaves the last-good timestamp untouched, so the very
90 * next operation re-checks.
91 */
92const amortizeCheck = (
93 checkLive: Effect.Effect<void, StateStoreError>,
94 ttlMs: number,
95): Effect.Effect<void, StateStoreError> => {
96 let lastOkAt: number | undefined;
97 return Effect.suspend(() => {
98 if (lastOkAt !== undefined && Date.now() - lastOkAt < ttlMs) {
99 return Effect.void;
100 }
101 return checkLive.pipe(
102 Effect.tap(() =>
103 Effect.sync(() => {
104 lastOkAt = Date.now();
105 }),
106 ),
107 );
108 });
109};
110
111interface Lease {
112 readonly checkLive: Effect.Effect<void, StateStoreError>;
113}
114
115/**
116 * State store backed by any Postgres database.
117 *
118 * Stack state lives in two tables — `alchemy_resource_state` and
119 * `alchemy_stack_output` — created on first use with
120 * `create table if not exists`.
121 *
122 * Concurrent deploys are serialized with a session-scoped Postgres advisory
123 * lock per `(stack, stage)`: the lock is taken on a reserved connection
124 * (`SqlClient.reserve`) when the first operation for the pair runs, and
125 * contention fails immediately instead of queueing. Every subsequent
126 * operation first re-verifies — from a *different* connection, by inspecting
127 * `pg_locks` — that the backend which took the lock still holds it, so a
128 * dropped lock connection fails loudly instead of letting operations run
129 * unlocked. That check reports the backend it ran on as well, so a
130 * single-connection client, which would have the lock holder vouch for
131 * itself, is refused instead of silently trusted. If the process crashes,
132 * Postgres releases the session lock when the connection drops; no recovery
133 * bookkeeping is needed.
134 *
135 * ### Using the Postgres State Store
136 * **Example:** Connection URL from configuration
137 * ```typescript
138 * import * as Alchemy from "alchemy";
139 * import { postgresState } from "alchemy/State/PostgresState";
140 * import * as Config from "effect/Config";
141 *
142 * const Stack = Alchemy.Stack(
143 * "my-stack",
144 * {
145 * providers: myProviders(),
146 * state: postgresState({ url: Config.Redacted("STATE_DATABASE_URL") }),
147 * },
148 * Effect.gen(function* () {
149 * // ...
150 * }),
151 * );
152 * ```
153 *
154 * **Example:** Caller-owned pool
155 * ```typescript
156 * import * as PgClient from "@effect/sql-pg/PgClient";
157 * import * as Config from "effect/Config";
158 *
159 * // A pool, not `PgClient.makeClient`: the store needs a second connection
160 * // to verify the advisory lock held on the reserved one.
161 * const sql = yield* PgClient.make({
162 * url: yield* Config.Redacted("STATE_DATABASE_URL"),
163 * });
164 * const state = postgresState({
165 * client: sql,
166 * lockKeyPrefix: yield* Config.String("STATE_LOCK_PREFIX"),
167 * });
168 * ```
169 */
170export const postgresState = <E = never, R = never>(
171 options: PostgresStateOptions<E, R>,
172) =>
173 Layer.effect(
174 State,
175 Effect.gen(function* () {
176 const scope = yield* Effect.scope;
177 // The layer carries the `url` Effect's requirements; the service it
178 // builds must not, so they are provided here, once.
179 const context = yield* Effect.context<R>();
180
181 const make = makePostgresState(options, scope).pipe(
182 recordStateStoreInit,
183 Effect.provideContext(context),
184 );
185
186 return yield* Effect.cached(make);
187 }),
188 );
189
190/**
191 * Construct a Postgres-backed {@link StateService}.
192 *
193 * Construction itself never touches the database — pool creation (when a
194 * `url` was given), schema migration, and advisory-lock acquisition are all
195 * deferred to the first state operation. Finalizers for the advisory locks
196 * and any store-owned pool are registered on `scope`.
197 */
198export const makePostgresState = <E = never, R = never>(
199 options: PostgresStateOptions<E, R>,
200 scope: Scope.Scope,
201) =>
202 Effect.gen(function* () {
203 const prefix = options.lockKeyPrefix ?? "alchemy";
204 const ttlMs = options.leaseCheckTtlMs ?? DEFAULT_LEASE_CHECK_TTL_MS;
205 // Captured once, so a `url` Effect that needs services (a custom
206 // ConfigProvider, say) can still be resolved later — from inside the
207 // service methods, whose own requirements are fixed at `never`.
208 const context = yield* Effect.context<R>();
209
210 const toError = (cause: unknown): StateStoreError =>
211 cause instanceof StateStoreError
212 ? cause
213 : new StateStoreError({
214 message:
215 cause instanceof Error
216 ? cause.message
217 : `Postgres state store error: ${String(cause)}`,
218 cause: cause instanceof Error ? cause : undefined,
219 });
220
221 const stateError = <A, E2, R2>(
222 effect: Effect.Effect<A, E2, R2>,
223 ): Effect.Effect<A, StateStoreError, R2> =>
224 Effect.mapError(effect, toError);
225
226 /**
227 * Idempotent schema migration. Concurrent `create table if not exists`
228 * statements can still collide inside Postgres on the shared catalog
229 * rows (duplicate pg_type/pg_class key errors), so the migration runs
230 * in a transaction — one connection, pinned by `withTransaction` — that
231 * first takes a transaction-scoped advisory lock on a fixed migration
232 * key. The lock releases automatically at commit or rollback.
233 */
234 const migrate = (sql: SqlClient.SqlClient) =>
235 sql
236 .withTransaction(
237 Effect.gen(function* () {
238 yield* sql`select pg_advisory_xact_lock(hashtextextended(${`${prefix}:schema`}, 0))`;
239 yield* sql`
240 create table if not exists alchemy_resource_state (
241 stack text not null,
242 stage text not null,
243 fqn text not null,
244 value jsonb not null,
245 updated_at timestamptz not null default now(),
246 primary key (stack, stage, fqn)
247 )
248 `;
249 yield* sql`
250 create table if not exists alchemy_stack_output (
251 stack text not null,
252 stage text not null,
253 value jsonb not null,
254 updated_at timestamptz not null default now(),
255 primary key (stack, stage)
256 )
257 `;
258 }),
259 )
260 .pipe(stateError);
261
262 /**
263 * Builds a store-owned `@effect/sql-pg` pool. Its `end` finalizer is
264 * registered on `scope`, so the pool closes with the state layer.
265 */
266 const openPool = (
267 url: NonNullable<PostgresStateOptions<E, R>["url"]>,
268 ): Effect.Effect<SqlClient.SqlClient, StateStoreError> =>
269 Effect.gen(function* () {
270 const PgClient = yield* Effect.tryPromise({
271 try: importPgClient,
272 catch: toError,
273 });
274 const resolved = Effect.isEffect(url)
275 ? yield* Effect.provideContext(url, context).pipe(stateError)
276 : url;
277 const built = yield* Layer.build(
278 PgClient.layer(resolveConnectionOptions(resolved)),
279 ).pipe(Scope.provide(scope), stateError);
280 return Context.get(built, PgClient.PgClient);
281 });
282
283 // Nothing touches the database at layer construction time. The client
284 // is resolved (or created from the URL) and the schema migrated inside
285 // this cached Effect, which runs once on the first state operation.
286 const ready = yield* Effect.cached(
287 Effect.gen(function* () {
288 const { client, url } = options;
289 const resolved: SqlClient.SqlClient =
290 client !== undefined && url === undefined
291 ? client
292 : url !== undefined && client === undefined
293 ? yield* openPool(url)
294 : yield* Effect.fail(
295 new StateStoreError({
296 message:
297 "postgresState requires exactly one of `client` or `url`",
298 }),
299 );
300 // The store reads columns by the exact names written below, so a
301 // client configured with name transforms must not rewrite them.
302 const sql = resolved.withoutTransforms();
303 yield* migrate(sql);
304 return sql;
305 }),
306 );
307
308 const run = <A>(
309 f: (sql: SqlClient.SqlClient) => Effect.Effect<A, SqlError.SqlError>,
310 ): Effect.Effect<A, StateStoreError> =>
311 ready.pipe(Effect.flatMap((sql) => stateError(f(sql))));
312
313 const lockKey = (stack: string, stage: string) =>
314 `${prefix}:${stack}/${stage}`;
315
316 /**
317 * Acquires the session-scoped advisory lock for `key` on a connection
318 * reserved for the store's lifetime. Session (not transaction) scope,
319 * because a deploy spans many commits. Contention fails immediately
320 * rather than queueing.
321 */
322 const acquireLease = (key: string): Effect.Effect<Lease, StateStoreError> =>
323 Effect.uninterruptibleMask((restore) =>
324 Effect.gen(function* () {
325 const sql = yield* restore(ready);
326 // Forked from the store's scope, so the reserved connection is
327 // reclaimed with the store even if nothing below closes it. From
328 // here to the unlock finalizer nothing is interruptible: handing
329 // a connection back to the pool while it still holds a session
330 // lock would strand that lock until the process exits, and every
331 // later deploy of this stack/stage would fail on it.
332 const lockScope = yield* Scope.fork(scope);
333 const releaseReserved = Scope.close(lockScope, Exit.void);
334 const reserved = yield* restore(
335 sql.reserve.pipe(Scope.provide(lockScope), stateError),
336 ).pipe(Effect.onError(() => releaseReserved));
337
338 // Pins a statement to the reserved connection exactly the way
339 // `withTransaction` does, so spans, span attributes and row
340 // handling stay identical to every other statement.
341 const onReserved = <A>(
342 effect: Effect.Effect<A, SqlError.SqlError>,
343 ): Effect.Effect<A, SqlError.SqlError> =>
344 Effect.provideService(effect, sql.transactionService, [
345 reserved,
346 0,
347 ]);
348
349 const acquired = yield* onReserved(
350 sql`select pg_try_advisory_lock(hashtextextended(${key}, 0)) as acquired, pg_backend_pid() as pid`,
351 ).pipe(
352 stateError,
353 Effect.map((rows) => rows[0]),
354 Effect.onError(() => releaseReserved),
355 );
356 if (acquired?.acquired !== true) {
357 yield* releaseReserved;
358 return yield* Effect.fail(
359 new StateStoreError({
360 message: `another deploy holds the Postgres state lock '${key}'`,
361 }),
362 );
363 }
364 const lockPid = Number(acquired.pid);
365
366 yield* Scope.addFinalizer(
367 scope,
368 onReserved(
369 sql`select pg_advisory_unlock(hashtextextended(${key}, 0))`,
370 ).pipe(
371 // If the connection already dropped, Postgres auto-released
372 // the session-scoped lock; there is nothing left to unlock.
373 Effect.ignore,
374 Effect.andThen(releaseReserved),
375 ),
376 );
377
378 // Deliberately verifies from a DIFFERENT connection, never the
379 // reserved one: once the reserved connection's backend has been
380 // killed server-side, querying it cannot answer reliably. Asking
381 // another connection whether the backend pid captured at acquire
382 // time still holds this advisory lock in `pg_locks` gives the same
383 // answer (a dead or reused backend cannot hold the lock) without
384 // touching the connection that might be dead. The check reports
385 // its own backend pid too, because a single-connection client
386 // routes it straight back to the lock holder, which cannot vouch
387 // for itself.
388 const checkLive = run(
389 (poolSql) => poolSql`
390 select
391 exists (
392 select 1 from pg_locks
393 where locktype = 'advisory'
394 and pid = ${lockPid}
395 and objsubid = 1
396 and granted
397 -- pg_locks splits the 64-bit advisory key into two
398 -- int4 halves; reassembling them with << 32 relies on
399 -- bigint wraparound matching hashtextextended's signed
400 -- 64-bit result, which is exact for all inputs.
401 and ((classid::bigint << 32) | (objid::bigint & 4294967295))
402 = hashtextextended(${key}, 0)
403 ) as live,
404 pg_backend_pid() as checker_pid
405 `,
406 ).pipe(
407 Effect.flatMap((rows) => {
408 const row = rows[0];
409 if (Number(row?.checker_pid) === lockPid) {
410 return Effect.fail(
411 new StateStoreError({
412 message: `the Postgres state lock '${key}' cannot be verified: the check ran on the same backend (pid ${lockPid}) that holds the lock, so this client is not pool-backed; pass postgresState a connection pool`,
413 }),
414 );
415 }
416 return row?.live === true
417 ? Effect.void
418 : Effect.fail(
419 new StateStoreError({
420 message: `the Postgres state lock '${key}' was lost mid-run; refusing to continue unlocked`,
421 }),
422 );
423 }),
424 );
425
426 return { checkLive: amortizeCheck(checkLive, ttlMs) };
427 }),
428 );
429
430 // One lease per (stack, stage), acquired lazily on the first operation
431 // that touches the pair and cached for the store's lifetime. The mutex
432 // makes acquisition single-flight so two concurrent first operations
433 // cannot race each other into a self-inflicted contention failure.
434 const leaseMutex = Semaphore.makeUnsafe(1);
435 const leases = new Map<string, Effect.Effect<Lease, StateStoreError>>();
436
437 const leaseFor = (
438 stack: string,
439 stage: string,
440 ): Effect.Effect<Lease, StateStoreError> =>
441 Semaphore.withPermits(
442 leaseMutex,
443 1,
444 )(
445 Effect.gen(function* () {
446 const key = lockKey(stack, stage);
447 const existing = leases.get(key);
448 if (existing !== undefined) return existing;
449 const cached = yield* Effect.cached(acquireLease(key));
450 leases.set(key, cached);
451 return cached;
452 }),
453 ).pipe(Effect.flatMap((lease) => lease));
454
455 /**
456 * Every storage operation on a `(stack, stage)` first ensures this
457 * store holds the pair's advisory lock and that the lease is still
458 * live. Reads are guarded too, not just writes: a lost lease means a
459 * concurrent deploy may already be mutating these rows, so a read could
460 * return stale or conflicting data.
461 */
462 const guarded = <A>(
463 request: { stack: string; stage: string },
464 op: Effect.Effect<A, StateStoreError>,
465 ): Effect.Effect<A, StateStoreError> =>
466 leaseFor(request.stack, request.stage).pipe(
467 Effect.flatMap((lease) => lease.checkLive),
468 Effect.andThen(op),
469 );
470
471 // Operations without a stage cannot name a single lease, so they
472 // re-verify every lease this store currently holds instead.
473 const verifyHeldLeases: Effect.Effect<void, StateStoreError> =
474 Effect.suspend(() =>
475 Effect.forEach(
476 Array.from(leases.values()),
477 (lease) => lease.pipe(Effect.flatMap((held) => held.checkLive)),
478 { discard: true },
479 ),
480 );
481
482 const jsonParam = (value: unknown) => JSON.stringify(encodeState(value));
483
484 const deleteStage = (stack: string, stage: string) =>
485 run(
486 (sql) =>
487 sql`delete from alchemy_resource_state where stack = ${stack} and stage = ${stage}`,
488 ).pipe(
489 Effect.andThen(
490 run(
491 (sql) =>
492 sql`delete from alchemy_stack_output where stack = ${stack} and stage = ${stage}`,
493 ),
494 ),
495 Effect.asVoid,
496 );
497
498 const service: StateService = {
499 id: options.id ?? "postgres",
500 getVersion: () => Effect.succeed(STATE_STORE_VERSION),
501 listStacks: () =>
502 verifyHeldLeases.pipe(
503 Effect.andThen(
504 run(
505 (sql) =>
506 sql`select stack from alchemy_resource_state union select stack from alchemy_stack_output order by stack`,
507 ),
508 ),
509 Effect.map((rows) => rows.map((row) => String(row.stack))),
510 ),
511 listStages: (stack) =>
512 verifyHeldLeases.pipe(
513 Effect.andThen(
514 run(
515 (sql) =>
516 sql`select stage from alchemy_resource_state where stack = ${stack} union select stage from alchemy_stack_output where stack = ${stack} order by stage`,
517 ),
518 ),
519 Effect.map((rows) => rows.map((row) => String(row.stage))),
520 ),
521 get: (request) =>
522 guarded(
523 request,
524 run(
525 (sql) =>
526 sql`select value from alchemy_resource_state where stack = ${request.stack} and stage = ${request.stage} and fqn = ${request.fqn}`,
527 ).pipe(
528 Effect.map((rows) => {
529 const row = rows[0];
530 // Every row was written by `set` through `encodeState`, so
531 // reviving it recovers a PersistedState by construction.
532 return row === undefined
533 ? undefined
534 : (reviveStateRecursive(row.value) as PersistedState);
535 }),
536 ),
537 ),
538 // Filters by status directly in SQL rather than listing FQNs and
539 // re-fetching each one — same semantics as LocalState, without the
540 // N+1 round-trips.
541 getReplacedResources: (request) =>
542 guarded(
543 request,
544 run(
545 (sql) =>
546 sql`select value from alchemy_resource_state where stack = ${request.stack} and stage = ${request.stage} and value ->> 'status' = 'replaced'`,
547 ).pipe(
548 Effect.map((rows) =>
549 rows.map(
550 (row) =>
551 reviveStateRecursive(row.value) as ReplacedResourceState,
552 ),
553 ),
554 ),
555 ),
556 set: (request) =>
557 guarded(
558 request,
559 run(
560 (sql) => sql`
561 insert into alchemy_resource_state (stack, stage, fqn, value, updated_at)
562 values (${request.stack}, ${request.stage}, ${request.fqn}, ${jsonParam(request.value)}::jsonb, now())
563 on conflict (stack, stage, fqn) do update
564 set value = excluded.value, updated_at = excluded.updated_at
565 `,
566 ).pipe(Effect.map(() => request.value)),
567 ),
568 delete: (request) =>
569 guarded(
570 request,
571 run(
572 (sql) =>
573 sql`delete from alchemy_resource_state where stack = ${request.stack} and stage = ${request.stage} and fqn = ${request.fqn}`,
574 ).pipe(Effect.asVoid),
575 ),
576 // Deleting a whole stack first acquires the advisory lock for every
577 // stage it is about to remove, so a concurrent deploy of any stage
578 // fails the lock instead of racing the delete. The stack-wide sweep
579 // that follows runs while all of those stage locks are still held.
580 deleteStack: ({ stack, stage }) =>
581 stage === undefined
582 ? verifyHeldLeases.pipe(
583 Effect.andThen(service.listStages(stack)),
584 Effect.flatMap((stages) =>
585 Effect.forEach(
586 stages,
587 (found) =>
588 guarded({ stack, stage: found }, deleteStage(stack, found)),
589 { discard: true },
590 ),
591 ),
592 Effect.andThen(
593 run(
594 (sql) =>
595 sql`delete from alchemy_resource_state where stack = ${stack}`,
596 ),
597 ),
598 Effect.andThen(
599 run(
600 (sql) =>
601 sql`delete from alchemy_stack_output where stack = ${stack}`,
602 ),
603 ),
604 Effect.asVoid,
605 )
606 : guarded({ stack, stage }, deleteStage(stack, stage)),
607 list: (request) =>
608 guarded(
609 request,
610 run(
611 (sql) =>
612 sql`select fqn from alchemy_resource_state where stack = ${request.stack} and stage = ${request.stage} order by fqn`,
613 ).pipe(Effect.map((rows) => rows.map((row) => String(row.fqn)))),
614 ),
615 getOutput: (request) =>
616 guarded(
617 request,
618 run(
619 (sql) =>
620 sql`select value from alchemy_stack_output where stack = ${request.stack} and stage = ${request.stage}`,
621 ).pipe(
622 Effect.map((rows) => {
623 const row = rows[0];
624 return row === undefined
625 ? undefined
626 : reviveStateRecursive(row.value);
627 }),
628 ),
629 ),
630 setOutput: (request) =>
631 guarded(
632 request,
633 run(
634 (sql) => sql`
635 insert into alchemy_stack_output (stack, stage, value, updated_at)
636 values (${request.stack}, ${request.stage}, ${jsonParam(request.value)}::jsonb, now())
637 on conflict (stack, stage) do update
638 set value = excluded.value, updated_at = excluded.updated_at
639 `,
640 ).pipe(Effect.map(() => request.value)),
641 ),
642 };
643 return service;
644 });