1import type { Credentials } from "@distilled.cloud/aws/Credentials";
2import type { Region } from "@distilled.cloud/aws/Region";
3import * as s3 from "@distilled.cloud/aws/s3";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Schedule from "effect/Schedule";
7import * as Stream from "effect/Stream";
8import type { HttpClient } from "effect/http/HttpClient";
9import { CredentialsStoreLive } from "../../Auth/Credentials.ts";
10import { decodeFqn, encodeFqn } from "../../FQN.ts";
11import { STATE_STORE_VERSION } from "../../State/HttpStateApi.ts";
12import {
13 State,
14 StateStoreError,
15 type PersistedState,
16 type StateService,
17} from "../../State/State.ts";
18import { encodeState, reviveState } from "../../State/StateEncoding.ts";
19import { recordStateStoreInit } from "../../Telemetry/Metrics.ts";
20import { AwsAuth } from "../AuthProvider.ts";
21import * as AwsCredentials from "../Credentials.ts";
22import * as Endpoint from "../Endpoint.ts";
23import {
24 AWSEnvironment,
25 Default as DefaultEnvironment,
26} from "../Environment.ts";
27import * as AwsRegion from "../Region.ts";
28import { syncBucketEncryption, type BucketEncryption } from "../S3/Bucket.ts";
29
30/**
31 * The bookkeeping object that stores a stack's resolved output. Lives
32 * alongside the resource objects under the same stage prefix, so it
33 * must be filtered out of `list` results before FQN decoding.
34 */
35const OUTPUT_FILE = "__stack_output__.json";
36
37/** Maximum number of keys S3 accepts in a single DeleteObjects call. */
38const DELETE_BATCH_SIZE = 1000;
39
40export interface S3StateOptions {
41 /**
42 * Name of the S3 bucket that holds the state objects. The bucket is
43 * created on first use if it does not exist.
44 *
45 * The bucket is created in the account-regional namespace, so custom
46 * names must follow the `<prefix>-<accountId>-<region>-an` convention.
47 *
48 * @see https://docs.aws.amazon.com/AmazonS3/latest/userguide/gpbucketnamespaces.html#account-regional-gp-buckets
49 * @default `alchemy-state-{accountId}-{region}-an`
50 */
51 bucketName?: string;
52 /**
53 * Key prefix within the bucket under which all state objects are
54 * stored, e.g. `"alchemy"`. A trailing `/` is appended automatically.
55 *
56 * @default "" (bucket root)
57 */
58 prefix?: string;
59 /**
60 * Default encryption enforced on every fresh state-service initialization.
61 * Omission restores AES256, no KMS key, bucket keys disabled, and no blocked
62 * encryption types. Set `blockedEncryptionTypes: ["SSE-C"]` to block
63 * customer-provided keys. Existing encrypted state objects are not rewritten.
64 *
65 * @default `{ sseAlgorithm: "AES256", bucketKeyEnabled: false, blockedEncryptionTypes: [] }`
66 */
67 encryption?: BucketEncryption;
68}
69
70/** Context required by the distilled S3 operations. */
71type S3Deps = Credentials | HttpClient | Region;
72
73/**
74 * State store backed by an AWS S3 bucket.
75 *
76 * Stack state is persisted as JSON objects in an account-regional S3
77 * bucket, laid out exactly like the local state store's file tree with
78 * the bucket (plus optional `prefix`) taking the place of the
79 * `.alchemy/state` directory:
80 *
81 * ```
82 * s3://{bucket}/{prefix}{stack}/{stage}/{fqn}.json
83 * s3://{bucket}/{prefix}{stack}/{stage}/__stack_output__.json
84 * ```
85 *
86 * The bucket is created lazily on the first state operation if it does
87 * not already exist — nothing touches AWS credentials at layer
88 * construction time.
89 *
90 *
91 * ### Using the S3 State Store
92 * Pass `AWS.state()` as the `state` option of a Stack. By default the
93 * state is stored in an account-regional bucket named
94 * `alchemy-state-{accountId}-{region}-an`.
95 *
96 * **Example:** Default bucket
97 * ```typescript
98 * import * as Alchemy from "alchemy";
99 * import * as AWS from "alchemy/AWS";
100 *
101 * const Stack = Alchemy.Stack(
102 * "my-stack",
103 * { providers: AWS.providers(), state: AWS.state() },
104 * Effect.gen(function* () {
105 * // ...
106 * }),
107 * );
108 * ```
109 *
110 * **Example:** Custom bucket and key prefix
111 * ```typescript
112 * const Stack = Alchemy.Stack(
113 * "my-stack",
114 * {
115 * providers: AWS.providers(),
116 * state: AWS.state({
117 * bucketName: "my-company-state",
118 * prefix: "alchemy",
119 * encryption: {
120 * sseAlgorithm: "aws:kms",
121 * kmsMasterKeyId: "alias/alchemy-state",
122 * },
123 * }),
124 * },
125 * Effect.gen(function* () {
126 * // ...
127 * }),
128 * );
129 * ```
130 *
131 * ### Managing SSE-C Restrictions
132 * **Example:** Block customer-provided encryption keys on the state bucket
133 * ```typescript
134 * const stateStore = AWS.state({
135 * encryption: {
136 * sseAlgorithm: "AES256",
137 * blockedEncryptionTypes: ["SSE-C"],
138 * },
139 * });
140 * ```
141 *
142 * Omitted `blockedEncryptionTypes` is equivalent to `[]`: no encryption types
143 * are blocked, so SSE-C writes are permitted via AWS's `NONE` value. Removing
144 * an explicit block clears it. Omitting `encryption` restores AES256, no KMS
145 * key, disabled bucket keys, and no encryption restrictions. Existing objects
146 * are not rewritten.
147 *
148 * @resource
149 */
150export const state = (options: S3StateOptions = {}) =>
151 Layer.effect(
152 State,
153 Effect.gen(function* () {
154 const context = yield* Effect.context<S3Deps | AWSEnvironment>();
155
156 const make = makeS3State(options).pipe(
157 recordStateStoreInit,
158 Effect.orDie,
159 Effect.provideContext(context),
160 );
161
162 return yield* Effect.cached(make);
163 }),
164 ).pipe(
165 Layer.provideMerge(AwsRegion.fromEnvironment),
166 Layer.provideMerge(AwsCredentials.fromEnvironment),
167 Layer.provideMerge(Endpoint.fromEnvironment),
168 Layer.provideMerge(DefaultEnvironment),
169 Layer.provideMerge(AwsAuth),
170 Layer.provideMerge(CredentialsStoreLive),
171 Layer.orDie,
172 );
173
174/**
175 * Construct an S3-backed {@link StateService}.
176 *
177 * Construction itself never touches AWS — environment resolution and
178 * the ensure-bucket-exists check are deferred into a cached Effect
179 * that runs once, on the first state operation.
180 */
181export const makeS3State = (options: S3StateOptions = {}) =>
182 Effect.gen(function* () {
183 const context = yield* Effect.context<S3Deps | AWSEnvironment>();
184
185 const prefix = options.prefix
186 ? `${options.prefix.replace(/\/+$/, "")}/`
187 : "";
188
189 const toError = (cause: unknown) =>
190 new StateStoreError({
191 message:
192 cause instanceof Error
193 ? cause.message
194 : `S3 state store error: ${String(cause)}`,
195 cause: cause instanceof Error ? cause : undefined,
196 });
197
198 // Anything that touches AWS credentials must NOT run at layer
199 // construction time. Resolving the environment (account/region),
200 // deriving the bucket name, and ensuring the bucket exists are all
201 // deferred into this cached Effect: `Effect.cached` memoizes the
202 // result and locks concurrent first callers onto a single in-flight
203 // run, so the bucket is ensured exactly once — lazily, on first use
204 // of the state store.
205 const bucket = yield* Effect.cached(
206 Effect.gen(function* () {
207 const { accountId, region } = yield* AWSEnvironment.current;
208 const bucketName =
209 options.bucketName ?? createStateBucketName(accountId, region);
210 yield* ensureStateBucket(bucketName, region, options);
211 return bucketName;
212 }).pipe(Effect.provideContext(context), Effect.mapError(toError)),
213 );
214
215 // Close over the captured context so every StateService method is
216 // self-contained (`R = never`), matching the StateService contract.
217 const run = <A, E>(
218 f: (bucket: string) => Effect.Effect<A, E, S3Deps>,
219 ): Effect.Effect<A, StateStoreError> =>
220 bucket.pipe(
221 Effect.flatMap((bucket) =>
222 f(bucket).pipe(
223 Effect.provideContext(context),
224 Effect.mapError(toError),
225 ),
226 ),
227 );
228
229 const stagePrefix = ({ stack, stage }: { stack: string; stage: string }) =>
230 `${prefix}${stack}/${stage}/`;
231
232 const resourceKey = (request: {
233 stack: string;
234 stage: string;
235 fqn: string;
236 }) => `${stagePrefix(request)}${encodeFqn(request.fqn)}.json`;
237
238 const outputKey = (request: { stack: string; stage: string }) =>
239 `${stagePrefix(request)}${OUTPUT_FILE}`;
240
241 /** All object keys under `keyPrefix`, across pagination. */
242 const listKeys = (bucket: string, keyPrefix: string) =>
243 s3.listObjectsV2.pages({ Bucket: bucket, Prefix: keyPrefix }).pipe(
244 Stream.flatMap((page) => Stream.fromIterable(page.Contents ?? [])),
245 Stream.map((object) => object.Key),
246 Stream.filter((key): key is string => key !== undefined),
247 Stream.runCollect,
248 Effect.map((keys) => Array.from(keys)),
249 );
250
251 /**
252 * Immediate "subdirectory" names under `keyPrefix`, derived from
253 * S3 CommonPrefixes (delimiter `/`), across pagination.
254 */
255 const listChildren = (bucket: string, keyPrefix: string) =>
256 s3.listObjectsV2
257 .pages({ Bucket: bucket, Prefix: keyPrefix, Delimiter: "/" })
258 .pipe(
259 Stream.flatMap((page) =>
260 Stream.fromIterable(page.CommonPrefixes ?? []),
261 ),
262 // `{keyPrefix}{name}/` -> `{name}`
263 Stream.map((common) => common.Prefix),
264 Stream.filter((p): p is string => p !== undefined),
265 Stream.map((p) => p.slice(keyPrefix.length, -1)),
266 Stream.runCollect,
267 Effect.map((names) => Array.from(names)),
268 );
269
270 /** Read and revive a JSON object; `undefined` when the key is absent. */
271 const readJson = <T>(bucket: string, key: string) =>
272 s3.getObject({ Bucket: bucket, Key: key }).pipe(
273 Effect.flatMap((result) =>
274 result.Body === undefined
275 ? Effect.succeed(undefined)
276 : Stream.mkString(Stream.decodeText(result.Body)).pipe(
277 Effect.flatMap((text) =>
278 Effect.try({
279 try: () => JSON.parse(text, reviveState) as T,
280 catch: (cause) =>
281 new StateStoreError({
282 message: `Failed to parse state object '${key}'`,
283 cause: cause instanceof Error ? cause : undefined,
284 }),
285 }),
286 ),
287 ),
288 ),
289 Effect.catchTag("NoSuchKey", () => Effect.succeed(undefined)),
290 );
291
292 const writeJson = (bucket: string, key: string, value: unknown) =>
293 s3.putObject({
294 Bucket: bucket,
295 Key: key,
296 Body: JSON.stringify(encodeState(value), null, 2),
297 ContentType: "application/json",
298 });
299
300 /** Delete every object under `keyPrefix` in batches. Idempotent. */
301 const deleteAll = (bucket: string, keyPrefix: string) =>
302 Effect.gen(function* () {
303 const keys = yield* listKeys(bucket, keyPrefix);
304 for (let i = 0; i < keys.length; i += DELETE_BATCH_SIZE) {
305 yield* s3.deleteObjects({
306 Bucket: bucket,
307 Delete: {
308 Objects: keys
309 .slice(i, i + DELETE_BATCH_SIZE)
310 .map((Key) => ({ Key })),
311 Quiet: true,
312 },
313 });
314 }
315 });
316
317 const state: StateService = {
318 id: "s3",
319 getVersion: () => Effect.succeed(STATE_STORE_VERSION),
320 listStacks: () => run((bucket) => listChildren(bucket, prefix)),
321 listStages: (stack: string) =>
322 run((bucket) => listChildren(bucket, `${prefix}${stack}/`)),
323 get: (request) =>
324 run((bucket) => readJson<PersistedState>(bucket, resourceKey(request))),
325 getReplacedResources: Effect.fn(function* (request) {
326 return (yield* Effect.all(
327 (yield* state.list(request)).map((fqn) =>
328 state.get({
329 stack: request.stack,
330 stage: request.stage,
331 fqn,
332 }),
333 ),
334 )).filter((r) => r?.status === "replaced");
335 }),
336 set: (request) =>
337 run((bucket) =>
338 writeJson(bucket, resourceKey(request), request.value),
339 ).pipe(Effect.map(() => request.value)),
340 delete: (request) =>
341 run((bucket) =>
342 s3.deleteObject({ Bucket: bucket, Key: resourceKey(request) }),
343 ).pipe(Effect.asVoid),
344 deleteStack: ({ stack, stage }) =>
345 run((bucket) =>
346 deleteAll(
347 bucket,
348 stage === undefined
349 ? `${prefix}${stack}/`
350 : stagePrefix({ stack, stage }),
351 ),
352 ),
353 list: (request) =>
354 run((bucket) => listKeys(bucket, stagePrefix(request))).pipe(
355 Effect.map((keys) =>
356 keys
357 .map((key) => key.slice(stagePrefix(request).length))
358 // Filter the bookkeeping file before decoding — `decodeFqn`
359 // replaces `__` with `/`, which would turn the literal name
360 // `__stack_output__` into `/stack_output/` and slip past
361 // the filter, leaving the engine to look up a non-existent
362 // resource.
363 .filter((file) => file !== OUTPUT_FILE && file.endsWith(".json"))
364 .map((file) => decodeFqn(file.replace(/\.json$/, ""))),
365 ),
366 ),
367 getOutput: (request) =>
368 run((bucket) => readJson(bucket, outputKey(request))),
369 setOutput: (request) =>
370 run((bucket) =>
371 writeJson(bucket, outputKey(request), request.value),
372 ).pipe(Effect.map(() => request.value)),
373 };
374 return state;
375 });
376
377/**
378 * Build the default account-regional state bucket name.
379 *
380 * Account-regional buckets must follow the naming convention:
381 * `<prefix>-<accountId>-<region>-an`
382 *
383 * @see https://docs.aws.amazon.com/AmazonS3/latest/userguide/gpbucketnamespaces.html#account-regional-gp-buckets
384 */
385export const createStateBucketName = (accountId: string, region: string) =>
386 `alchemy-state-${accountId}-${region}-an`.toLowerCase();
387
388/**
389 * Observe-then-ensure the state bucket: head it, create it if missing
390 * (tolerating create races), and wait for it to become available.
391 */
392const ensureStateBucket = (
393 bucket: string,
394 region: string,
395 options: S3StateOptions,
396) =>
397 Effect.gen(function* () {
398 // An absent bucket surfaces as either `NotFound` (the HEAD 404) or
399 // `NoSuchBucket` depending on the namespace/path — treat both as "create
400 // it". Catching only `NotFound` let a deleted state bucket (e.g. after a
401 // nuke) escape as an uncaught `NoSuchBucket` instead of being recreated.
402 const exists = yield* s3.headBucket({ Bucket: bucket }).pipe(
403 Effect.map(() => true),
404 Effect.catchTag(["NotFound", "NoSuchBucket"], () =>
405 Effect.succeed(false),
406 ),
407 );
408 if (!exists) {
409 yield* Effect.logInfo(
410 `S3 state store: creating bucket ${bucket} in ${region}`,
411 );
412 yield* s3
413 .createBucket({
414 Bucket: bucket,
415 // account-regional namespace: the bucket name only needs to be
416 // unique within this account+region, so deterministic default
417 // names can't collide with other AWS customers.
418 BucketNamespace: "account-regional",
419 // us-east-1 rejects an explicit LocationConstraint
420 ...(region === "us-east-1"
421 ? {}
422 : {
423 CreateBucketConfiguration: {
424 LocationConstraint: region as s3.BucketLocationConstraint,
425 },
426 }),
427 })
428 .pipe(
429 // Many callers race to create the shared default state bucket on
430 // first use. The loser sees the create already done
431 // (`BucketAlreadyOwnedByYou`/`BucketAlreadyExists`) or mid-flight
432 // (`OperationAborted`). All mean "someone else is creating it".
433 Effect.catchTag(
434 [
435 "BucketAlreadyOwnedByYou",
436 "BucketAlreadyExists",
437 "OperationAborted",
438 ],
439 () => Effect.void,
440 ),
441 );
442
443 // Wait for the bucket to become available. Under a concurrent create
444 // the bucket is briefly not yet head-able and object/configuration ops
445 // would race ahead of it.
446 yield* s3.headBucket({ Bucket: bucket }).pipe(
447 Effect.retry({
448 while: (error) =>
449 error._tag === "NotFound" || error._tag === "NoSuchBucket",
450 schedule: Schedule.max([
451 Schedule.spaced("1 second"),
452 Schedule.recurs(10),
453 ]),
454 }),
455 );
456 }
457
458 // Secure defaults are reconciled for BOTH newly-created and existing
459 // buckets. This makes AWS.state converge out-of-band drift and explicit
460 // custom buckets rather than only securing the greenfield path.
461 const desiredVersioning = "Enabled";
462 const observedVersioning = yield* s3.getBucketVersioning({
463 Bucket: bucket,
464 });
465 if (observedVersioning.Status !== desiredVersioning) {
466 yield* s3.putBucketVersioning({
467 Bucket: bucket,
468 VersioningConfiguration: { Status: desiredVersioning },
469 });
470 }
471
472 yield* syncBucketEncryption(bucket, options.encryption);
473
474 const desiredPublicAccess: s3.PublicAccessBlockConfiguration = {
475 BlockPublicAcls: true,
476 IgnorePublicAcls: true,
477 BlockPublicPolicy: true,
478 RestrictPublicBuckets: true,
479 };
480 const observedPublicAccess = yield* s3
481 .getPublicAccessBlock({ Bucket: bucket })
482 .pipe(
483 Effect.map((result) => result.PublicAccessBlockConfiguration),
484 Effect.catchTag("NoSuchPublicAccessBlockConfiguration", () =>
485 Effect.succeed<s3.PublicAccessBlockConfiguration | undefined>(
486 undefined,
487 ),
488 ),
489 );
490 const publicAccessFingerprint = (
491 config: s3.PublicAccessBlockConfiguration | undefined,
492 ) =>
493 JSON.stringify({
494 blockAcls: config?.BlockPublicAcls ?? false,
495 ignoreAcls: config?.IgnorePublicAcls ?? false,
496 blockPolicy: config?.BlockPublicPolicy ?? false,
497 restrictBuckets: config?.RestrictPublicBuckets ?? false,
498 });
499 if (
500 publicAccessFingerprint(observedPublicAccess) !==
501 publicAccessFingerprint(desiredPublicAccess)
502 ) {
503 yield* s3.putPublicAccessBlock({
504 Bucket: bucket,
505 PublicAccessBlockConfiguration: desiredPublicAccess,
506 });
507 }
508
509 const desiredOwnership = "BucketOwnerEnforced";
510 const observedOwnership = yield* s3
511 .getBucketOwnershipControls({ Bucket: bucket })
512 .pipe(
513 Effect.map(
514 (result) => result.OwnershipControls?.Rules?.[0]?.ObjectOwnership,
515 ),
516 Effect.catchTag("OwnershipControlsNotFoundError", () =>
517 Effect.succeed<string | undefined>(undefined),
518 ),
519 );
520 if (observedOwnership !== desiredOwnership) {
521 yield* s3.putBucketOwnershipControls({
522 Bucket: bucket,
523 OwnershipControls: {
524 Rules: [{ ObjectOwnership: desiredOwnership }],
525 },
526 });
527 }
528 }).pipe(
529 // The whole observe→create→wait sequence races other first-callers of the
530 // shared bucket; `OperationAborted` (conflicting create) and a transiently
531 // absent bucket (`NoSuchBucket`/`NotFound`) are the faces of that race.
532 // Retry the entire sequence so a concurrent create that is still settling
533 // converges instead of surfacing as a `StateStoreError`.
534 Effect.retry({
535 while: (e) =>
536 e._tag === "OperationAborted" ||
537 e._tag === "NoSuchBucket" ||
538 e._tag === "NotFound",
539 schedule: Schedule.max([
540 Schedule.spaced("2 seconds"),
541 Schedule.recurs(10),
542 ]),
543 }),
544 );