1import * as mwaa from "@distilled.cloud/aws/mwaa";
2import * as Effect from "effect/Effect";
3import * as Redacted from "effect/Redacted";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
12import type { Providers } from "../Providers.ts";
13
14/**
15 * Per-module CloudWatch logging configuration for an MWAA {@link Environment}.
16 */
17export interface ModuleLoggingConfig {
18 /**
19 * Whether this Airflow log group is published to CloudWatch Logs.
20 */
21 enabled: boolean;
22 /**
23 * Airflow log level — one of `CRITICAL`, `ERROR`, `WARNING`, `INFO`, `DEBUG`.
24 */
25 logLevel: string;
26}
27
28/**
29 * CloudWatch logging configuration for the five Airflow log groups of an MWAA
30 * {@link Environment}. Any group left unset is disabled.
31 */
32export interface EnvironmentLoggingConfig {
33 /** DAG processing logs. */
34 dagProcessingLogs?: ModuleLoggingConfig;
35 /** Scheduler logs. */
36 schedulerLogs?: ModuleLoggingConfig;
37 /** Webserver logs. */
38 webserverLogs?: ModuleLoggingConfig;
39 /** Worker logs. */
40 workerLogs?: ModuleLoggingConfig;
41 /** Task logs. */
42 taskLogs?: ModuleLoggingConfig;
43}
44
45export interface EnvironmentProps {
46 /**
47 * Name of the environment (1-80 chars, must start with a letter). If omitted,
48 * a deterministic physical name is generated. Changing the name replaces the
49 * environment.
50 */
51 environmentName?: string;
52 /**
53 * ARN of the IAM execution role Amazon MWAA and Apache Airflow assume to
54 * access AWS resources (the DAGs bucket, CloudWatch, SQS, KMS). Updateable
55 * in place.
56 */
57 executionRoleArn: string;
58 /**
59 * ARN of the S3 bucket that holds your DAG code, `requirements.txt`, plugins,
60 * and startup script. The bucket must block public access and have
61 * versioning enabled. Updateable in place.
62 */
63 sourceBucketArn: string;
64 /**
65 * Relative path (within the source bucket) to the folder containing your DAG
66 * files, e.g. `"dags"`. Updateable in place.
67 */
68 dagS3Path: string;
69 /**
70 * The private subnet IDs (exactly two, in distinct Availability Zones) the
71 * environment's webserver, scheduler, and workers run in. Changing the
72 * subnets replaces the environment.
73 */
74 subnetIds: string[];
75 /**
76 * VPC security group IDs applied to the environment's network interfaces.
77 * Updateable in place.
78 * @default a security group created for the environment
79 */
80 securityGroupIds?: string[];
81 /**
82 * Apache Airflow version, e.g. `"2.10.3"`. Downgrades are not permitted.
83 * @default the latest version supported by MWAA
84 */
85 airflowVersion?: string;
86 /**
87 * Environment class (sizing), e.g. `"mw1.small"`, `"mw1.medium"`,
88 * `"mw1.large"`. Updateable in place.
89 * @default "mw1.small"
90 */
91 environmentClass?: string;
92 /**
93 * Maximum number of Airflow workers to scale up to. Updateable in place.
94 * @default 10
95 */
96 maxWorkers?: number;
97 /**
98 * Minimum number of Airflow workers. Updateable in place.
99 * @default 1
100 */
101 minWorkers?: number;
102 /**
103 * Maximum number of Airflow web servers (Airflow 2.2.2+). Updateable in
104 * place.
105 */
106 maxWebservers?: number;
107 /**
108 * Minimum number of Airflow web servers (Airflow 2.2.2+). Updateable in
109 * place.
110 */
111 minWebservers?: number;
112 /**
113 * Number of Airflow schedulers. Updateable in place.
114 */
115 schedulers?: number;
116 /**
117 * Apache Airflow web server access mode — `PRIVATE_ONLY` (VPC-only endpoint)
118 * or `PUBLIC_ONLY` (internet-accessible). Updateable in place.
119 * @default "PRIVATE_ONLY"
120 */
121 webserverAccessMode?: string;
122 /**
123 * Day and time of the weekly 30-minute maintenance window in
124 * `DAY:HH:MM` (UTC), e.g. `"MON:03:30"`. Updateable in place.
125 */
126 weeklyMaintenanceWindowStart?: string;
127 /**
128 * ARN or key ID of the customer-managed KMS key used to encrypt data.
129 * Changing the key replaces the environment.
130 * @default an AWS-owned key
131 */
132 kmsKey?: string;
133 /**
134 * Whether the VPC endpoints for the environment are managed by MWAA
135 * (`SERVICE`) or by you (`CUSTOMER`). Changing this replaces the environment.
136 * @default "SERVICE"
137 */
138 endpointManagement?: string;
139 /**
140 * Apache Airflow configuration overrides, keyed by
141 * `section.option`, e.g. `{ "core.default_task_retries": "3" }`. Values
142 * carrying secrets (SMTP passwords, connection URIs, …) may be passed as
143 * `Redacted.Redacted<string>` so they never appear in logs. Updateable in
144 * place.
145 */
146 airflowConfigurationOptions?: Record<
147 string,
148 string | Redacted.Redacted<string>
149 >;
150 /**
151 * Relative path to the plugins `.zip` in the source bucket. Updateable in
152 * place.
153 */
154 pluginsS3Path?: string;
155 /**
156 * S3 object version of the plugins `.zip`. Updateable in place.
157 */
158 pluginsS3ObjectVersion?: string;
159 /**
160 * Relative path to `requirements.txt` in the source bucket. Updateable in
161 * place.
162 */
163 requirementsS3Path?: string;
164 /**
165 * S3 object version of `requirements.txt`. Updateable in place.
166 */
167 requirementsS3ObjectVersion?: string;
168 /**
169 * Relative path to the startup shell script in the source bucket. Updateable
170 * in place.
171 */
172 startupScriptS3Path?: string;
173 /**
174 * S3 object version of the startup script. Updateable in place.
175 */
176 startupScriptS3ObjectVersion?: string;
177 /**
178 * CloudWatch logging configuration for the five Airflow log groups.
179 * Updateable in place.
180 */
181 loggingConfiguration?: EnvironmentLoggingConfig;
182 /**
183 * User-defined tags for the environment.
184 */
185 tags?: Record<string, string>;
186}
187
188export interface Environment extends Resource<
189 "AWS.MWAA.Environment",
190 EnvironmentProps,
191 {
192 /** Name of the environment. */
193 environmentName: string;
194 /** ARN of the environment. */
195 arn: string;
196 /** Current lifecycle status (e.g. `CREATING`, `AVAILABLE`). */
197 status: string;
198 /** Host name of the Airflow web server UI. */
199 webserverUrl: string | undefined;
200 /** ARN of the execution role Airflow tasks run as. */
201 executionRoleArn: string | undefined;
202 /** ARN of the service-linked role MWAA uses to manage the environment. */
203 serviceRoleArn: string | undefined;
204 /** Running Apache Airflow version. */
205 airflowVersion: string | undefined;
206 /** Environment size class (e.g. `mw1.small`). */
207 environmentClass: string | undefined;
208 /** ARN of the S3 bucket holding DAG code. */
209 sourceBucketArn: string | undefined;
210 /** Relative S3 path to the DAGs folder. */
211 dagS3Path: string | undefined;
212 /** Celery executor queue used by the environment's workers. */
213 celeryExecutorQueue: string | undefined;
214 /** VPC endpoint service name for the Airflow metadata database. */
215 databaseVpcEndpointService: string | undefined;
216 /** VPC endpoint service name for the Airflow web server. */
217 webserverVpcEndpointService: string | undefined;
218 /** Tags on the environment (user + internal Alchemy tags). */
219 tags: Record<string, string>;
220 },
221 never,
222 Providers
223> {}
224
225/**
226 * An Amazon Managed Workflows for Apache Airflow (MWAA) environment — a fully
227 * managed Apache Airflow deployment that runs your DAGs.
228 *
229 * Environments take roughly 20-30 minutes to create and are billed hourly for
230 * the environment, workers, and schedulers while they exist. Each environment
231 * needs an S3 bucket for DAG code (versioned, public access blocked), an IAM
232 * execution role, and two private subnets in distinct Availability Zones with
233 * outbound internet access (via NAT gateway or VPC endpoints). Destroy
234 * environments you are not using.
235 * ### Creating an Environment
236 * **Example:** Basic Environment
237 * ```typescript
238 * const environment = yield* Environment("Airflow", {
239 * executionRoleArn: role.roleArn,
240 * sourceBucketArn: bucket.bucketArn,
241 * dagS3Path: "dags",
242 * subnetIds: [privateSubnetA.subnetId, privateSubnetB.subnetId],
243 * airflowVersion: "2.10.3",
244 * environmentClass: "mw1.small",
245 * maxWorkers: 5,
246 * });
247 * ```
248 *
249 * **Example:** Public Webserver with Logging
250 * ```typescript
251 * const environment = yield* Environment("Airflow", {
252 * executionRoleArn: role.roleArn,
253 * sourceBucketArn: bucket.bucketArn,
254 * dagS3Path: "dags",
255 * subnetIds: [privateSubnetA.subnetId, privateSubnetB.subnetId],
256 * webserverAccessMode: "PUBLIC_ONLY",
257 * loggingConfiguration: {
258 * schedulerLogs: { enabled: true, logLevel: "INFO" },
259 * workerLogs: { enabled: true, logLevel: "INFO" },
260 * taskLogs: { enabled: true, logLevel: "INFO" },
261 * },
262 * airflowConfigurationOptions: {
263 * "core.default_task_retries": "3",
264 * },
265 * });
266 * ```
267 *
268 * @resource
269 */
270export const Environment = Resource<Environment>("AWS.MWAA.Environment");
271
272// Terminal states: environment stopped converging.
273const FAILED_STATES = new Set([
274 "CREATE_FAILED",
275 "UPDATE_FAILED",
276 "UNAVAILABLE",
277 "ROLLBACK_FAILED",
278]);
279
280const toModuleInput = (
281 config: ModuleLoggingConfig | undefined,
282): mwaa.ModuleLoggingConfigurationInput | undefined =>
283 config === undefined
284 ? undefined
285 : { Enabled: config.enabled, LogLevel: config.logLevel };
286
287const toLoggingInput = (
288 config: EnvironmentLoggingConfig | undefined,
289): mwaa.LoggingConfigurationInput | undefined => {
290 if (config === undefined) return undefined;
291 return {
292 DagProcessingLogs: toModuleInput(config.dagProcessingLogs),
293 SchedulerLogs: toModuleInput(config.schedulerLogs),
294 WebserverLogs: toModuleInput(config.webserverLogs),
295 WorkerLogs: toModuleInput(config.workerLogs),
296 TaskLogs: toModuleInput(config.taskLogs),
297 };
298};
299
300const sameStringSet = (a: string[], b: string[]): boolean => {
301 if (a.length !== b.length) return false;
302 const setB = new Set(b);
303 return a.every((v) => setB.has(v));
304};
305
306const configValue = (
307 v: string | Redacted.Redacted<string> | undefined,
308): string | undefined => (Redacted.isRedacted(v) ? Redacted.value(v) : v);
309
310// Compare desired Airflow configuration options against the observed ones.
311// Both sides may carry Redacted values (the props by choice, the observed
312// options because distilled decodes them as SensitiveString) — unwrap for the
313// comparison only.
314const sameConfigOptions = (
315 a: Record<string, string | Redacted.Redacted<string>> | undefined,
316 b:
317 | { [key: string]: string | Redacted.Redacted<string> | undefined }
318 | undefined,
319): boolean => {
320 const ea = Object.entries(a ?? {});
321 const eb = Object.entries(b ?? {}).filter(([, v]) => v !== undefined);
322 if (ea.length !== eb.length) return false;
323 const bMap = b ?? {};
324 return ea.every(([k, v]) => configValue(bMap[k]) === configValue(v));
325};
326
327export const EnvironmentProvider = () =>
328 Provider.effect(
329 Environment,
330 Effect.gen(function* () {
331 const toName = (id: string, props: Partial<EnvironmentProps>) =>
332 props.environmentName
333 ? Effect.succeed(props.environmentName)
334 : createPhysicalName({ id, maxLength: 80 });
335
336 const readEnvironment = Effect.fn(function* (name: string) {
337 const response = yield* mwaa
338 .getEnvironment({ Name: name })
339 .pipe(
340 Effect.catchTag("ResourceNotFoundException", () =>
341 Effect.succeed(undefined),
342 ),
343 );
344 return response?.Environment;
345 });
346
347 const readTags = Effect.fn(function* (arn: string) {
348 const response = yield* mwaa
349 .listTagsForResource({ ResourceArn: arn })
350 .pipe(Effect.catch(() => Effect.succeed(undefined)));
351 const tags: Record<string, string> = {};
352 for (const [key, value] of Object.entries(response?.Tags ?? {})) {
353 if (value !== undefined) tags[key] = value;
354 }
355 return tags;
356 });
357
358 // Bounded readiness wait. MWAA environment provisioning/modification
359 // typically completes in 20-30 minutes; budget ~45 min (90 * 30s). A
360 // FAILED terminal state stops the wait immediately (non-retryable).
361 const waitForAvailable = Effect.fn(function* (name: string) {
362 const policy = Schedule.max([
363 Schedule.fixed("30 seconds"),
364 Schedule.recurs(90),
365 ]);
366 return yield* readEnvironment(name).pipe(
367 Effect.flatMap((env) => {
368 if (!env?.Arn) {
369 return Effect.fail(new Error(`Environment '${name}' not found`));
370 }
371 const status = env.Status ?? "CREATING";
372 if (FAILED_STATES.has(status)) {
373 return Effect.fail(
374 new Error(
375 `Environment '${name}' reached terminal state '${status}'`,
376 ),
377 );
378 }
379 if (status !== "AVAILABLE") {
380 return Effect.fail(
381 new Error(
382 `Environment '${name}' not available (status: ${status})`,
383 ),
384 );
385 }
386 return Effect.succeed(env);
387 }),
388 Effect.retry({
389 while: (e) =>
390 e instanceof Error && !e.message.includes("terminal state"),
391 schedule: policy,
392 }),
393 );
394 });
395
396 const toAttrs = Effect.fn(function* (env: mwaa.Environment) {
397 if (!env.Name || !env.Arn) {
398 return yield* Effect.fail(
399 new Error(`Environment '${env.Name}' is missing its ARN`),
400 );
401 }
402 return {
403 environmentName: env.Name,
404 arn: env.Arn,
405 status: env.Status ?? "CREATING",
406 webserverUrl: env.WebserverUrl,
407 executionRoleArn: env.ExecutionRoleArn,
408 serviceRoleArn: env.ServiceRoleArn,
409 airflowVersion: env.AirflowVersion,
410 environmentClass: env.EnvironmentClass,
411 sourceBucketArn: env.SourceBucketArn,
412 dagS3Path: env.DagS3Path,
413 celeryExecutorQueue: env.CeleryExecutorQueue,
414 databaseVpcEndpointService: env.DatabaseVpcEndpointService,
415 webserverVpcEndpointService: env.WebserverVpcEndpointService,
416 tags: yield* readTags(env.Arn),
417 };
418 });
419
420 return {
421 stables: ["environmentName", "arn"],
422
423 diff: Effect.fn(function* ({ id, olds, news }) {
424 if (!isResolved(news)) return undefined;
425 const n = news;
426 const o = olds;
427 if (n === undefined || o === undefined) return undefined;
428 if ((yield* toName(id, o)) !== (yield* toName(id, n))) {
429 return { action: "replace" } as const;
430 }
431 // Create-only properties force a replacement.
432 if (!sameStringSet(n.subnetIds ?? [], o.subnetIds ?? [])) {
433 return { action: "replace" } as const;
434 }
435 if ((n.kmsKey ?? undefined) !== (o.kmsKey ?? undefined)) {
436 return { action: "replace" } as const;
437 }
438 if (
439 (n.endpointManagement ?? undefined) !==
440 (o.endpointManagement ?? undefined)
441 ) {
442 return { action: "replace" } as const;
443 }
444 }),
445
446 read: Effect.fn(function* ({ id, olds, output }) {
447 const name =
448 output?.environmentName ?? (yield* toName(id, olds ?? {}));
449 const env = yield* readEnvironment(name);
450 if (!env?.Arn) return undefined;
451 const attrs = yield* toAttrs(env);
452 return (yield* hasAlchemyTags(id, attrs.tags))
453 ? attrs
454 : Unowned(attrs);
455 }),
456
457 reconcile: Effect.fn(function* ({ id, news, output, session }) {
458 const props = news!;
459 const name = output?.environmentName ?? (yield* toName(id, props));
460 const internalTags = yield* createInternalTags(id);
461 const desiredTags = { ...internalTags, ...props.tags };
462
463 // 1. Observe — cloud state is authoritative.
464 let observed = yield* readEnvironment(name);
465
466 // 2. Ensure — create if missing. Create errors (invalid subnets,
467 // role, etc.) propagate so misconfiguration surfaces directly
468 // rather than as a confusing "not found" after the wait below.
469 if (observed === undefined) {
470 yield* mwaa.createEnvironment({
471 Name: name,
472 ExecutionRoleArn: props.executionRoleArn,
473 SourceBucketArn: props.sourceBucketArn,
474 DagS3Path: props.dagS3Path,
475 NetworkConfiguration: {
476 SubnetIds: props.subnetIds,
477 SecurityGroupIds: props.securityGroupIds,
478 },
479 AirflowVersion: props.airflowVersion,
480 EnvironmentClass: props.environmentClass,
481 MaxWorkers: props.maxWorkers,
482 MinWorkers: props.minWorkers,
483 MaxWebservers: props.maxWebservers,
484 MinWebservers: props.minWebservers,
485 Schedulers: props.schedulers,
486 WebserverAccessMode: props.webserverAccessMode,
487 WeeklyMaintenanceWindowStart: props.weeklyMaintenanceWindowStart,
488 KmsKey: props.kmsKey,
489 EndpointManagement: props.endpointManagement,
490 AirflowConfigurationOptions: props.airflowConfigurationOptions,
491 PluginsS3Path: props.pluginsS3Path,
492 PluginsS3ObjectVersion: props.pluginsS3ObjectVersion,
493 RequirementsS3Path: props.requirementsS3Path,
494 RequirementsS3ObjectVersion: props.requirementsS3ObjectVersion,
495 StartupScriptS3Path: props.startupScriptS3Path,
496 StartupScriptS3ObjectVersion: props.startupScriptS3ObjectVersion,
497 LoggingConfiguration: toLoggingInput(props.loggingConfiguration),
498 Tags: desiredTags,
499 });
500 }
501
502 // Provisioning and in-flight modifications both surface as a
503 // non-AVAILABLE status; wait for availability (bounded) so update
504 // calls do not hit a conflict.
505 observed = yield* waitForAvailable(name);
506
507 // 3. Sync — compute an update delta from OBSERVED state.
508 const update: mwaa.UpdateEnvironmentInput = { Name: name };
509 let mutated = false;
510 if (
511 props.executionRoleArn !== undefined &&
512 props.executionRoleArn !== observed.ExecutionRoleArn
513 ) {
514 update.ExecutionRoleArn = props.executionRoleArn;
515 mutated = true;
516 }
517 if (
518 props.sourceBucketArn !== undefined &&
519 props.sourceBucketArn !== observed.SourceBucketArn
520 ) {
521 update.SourceBucketArn = props.sourceBucketArn;
522 mutated = true;
523 }
524 if (
525 props.dagS3Path !== undefined &&
526 props.dagS3Path !== observed.DagS3Path
527 ) {
528 update.DagS3Path = props.dagS3Path;
529 mutated = true;
530 }
531 if (
532 props.airflowVersion !== undefined &&
533 props.airflowVersion !== observed.AirflowVersion
534 ) {
535 update.AirflowVersion = props.airflowVersion;
536 mutated = true;
537 }
538 if (
539 props.environmentClass !== undefined &&
540 props.environmentClass !== observed.EnvironmentClass
541 ) {
542 update.EnvironmentClass = props.environmentClass;
543 mutated = true;
544 }
545 if (
546 props.maxWorkers !== undefined &&
547 props.maxWorkers !== observed.MaxWorkers
548 ) {
549 update.MaxWorkers = props.maxWorkers;
550 mutated = true;
551 }
552 if (
553 props.minWorkers !== undefined &&
554 props.minWorkers !== observed.MinWorkers
555 ) {
556 update.MinWorkers = props.minWorkers;
557 mutated = true;
558 }
559 if (
560 props.maxWebservers !== undefined &&
561 props.maxWebservers !== observed.MaxWebservers
562 ) {
563 update.MaxWebservers = props.maxWebservers;
564 mutated = true;
565 }
566 if (
567 props.minWebservers !== undefined &&
568 props.minWebservers !== observed.MinWebservers
569 ) {
570 update.MinWebservers = props.minWebservers;
571 mutated = true;
572 }
573 if (
574 props.schedulers !== undefined &&
575 props.schedulers !== observed.Schedulers
576 ) {
577 update.Schedulers = props.schedulers;
578 mutated = true;
579 }
580 if (
581 props.webserverAccessMode !== undefined &&
582 props.webserverAccessMode !== observed.WebserverAccessMode
583 ) {
584 update.WebserverAccessMode = props.webserverAccessMode;
585 mutated = true;
586 }
587 if (
588 props.weeklyMaintenanceWindowStart !== undefined &&
589 props.weeklyMaintenanceWindowStart !==
590 observed.WeeklyMaintenanceWindowStart
591 ) {
592 update.WeeklyMaintenanceWindowStart =
593 props.weeklyMaintenanceWindowStart;
594 mutated = true;
595 }
596 const observedSg = observed.NetworkConfiguration?.SecurityGroupIds;
597 if (
598 props.securityGroupIds !== undefined &&
599 !sameStringSet(props.securityGroupIds, observedSg ?? [])
600 ) {
601 update.NetworkConfiguration = {
602 SecurityGroupIds: props.securityGroupIds,
603 };
604 mutated = true;
605 }
606 if (
607 props.pluginsS3Path !== undefined &&
608 props.pluginsS3Path !== observed.PluginsS3Path
609 ) {
610 update.PluginsS3Path = props.pluginsS3Path;
611 update.PluginsS3ObjectVersion = props.pluginsS3ObjectVersion;
612 mutated = true;
613 }
614 if (
615 props.requirementsS3Path !== undefined &&
616 props.requirementsS3Path !== observed.RequirementsS3Path
617 ) {
618 update.RequirementsS3Path = props.requirementsS3Path;
619 update.RequirementsS3ObjectVersion =
620 props.requirementsS3ObjectVersion;
621 mutated = true;
622 }
623 if (
624 props.startupScriptS3Path !== undefined &&
625 props.startupScriptS3Path !== observed.StartupScriptS3Path
626 ) {
627 update.StartupScriptS3Path = props.startupScriptS3Path;
628 update.StartupScriptS3ObjectVersion =
629 props.startupScriptS3ObjectVersion;
630 mutated = true;
631 }
632 if (
633 props.airflowConfigurationOptions !== undefined &&
634 !sameConfigOptions(
635 props.airflowConfigurationOptions,
636 observed.AirflowConfigurationOptions,
637 )
638 ) {
639 update.AirflowConfigurationOptions =
640 props.airflowConfigurationOptions;
641 mutated = true;
642 }
643 if (props.loggingConfiguration !== undefined) {
644 update.LoggingConfiguration = toLoggingInput(
645 props.loggingConfiguration,
646 );
647 mutated = true;
648 }
649
650 if (mutated) {
651 yield* mwaa.updateEnvironment(update);
652 observed = yield* waitForAvailable(name);
653 }
654
655 // 3b. Sync tags — diff against OBSERVED cloud tags.
656 const arn = observed.Arn;
657 if (arn) {
658 const observedTags = yield* readTags(arn);
659 const { removed, upsert } = diffTags(observedTags, desiredTags);
660 if (upsert.length > 0) {
661 yield* mwaa.tagResource({
662 ResourceArn: arn,
663 Tags: Object.fromEntries(
664 upsert.map(({ Key, Value }) => [Key, Value]),
665 ),
666 });
667 }
668 if (removed.length > 0) {
669 yield* mwaa.untagResource({ ResourceArn: arn, tagKeys: removed });
670 }
671 }
672
673 yield* session.note(name);
674 return yield* toAttrs(observed);
675 }),
676
677 delete: Effect.fn(function* ({ output }) {
678 yield* mwaa
679 .deleteEnvironment({ Name: output.environmentName })
680 .pipe(
681 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
682 );
683 }),
684
685 list: () =>
686 mwaa.listEnvironments.pages({}).pipe(
687 Stream.runCollect,
688 Effect.map((chunk) =>
689 Array.from(chunk).flatMap((page) => page.Environments ?? []),
690 ),
691 Effect.flatMap(
692 Effect.forEach((name) => readEnvironment(name), {
693 concurrency: 4,
694 }),
695 ),
696 Effect.map((envs) =>
697 envs.filter(
698 (env): env is mwaa.Environment =>
699 env !== undefined && env.Name !== undefined && !!env.Arn,
700 ),
701 ),
702 Effect.flatMap(
703 Effect.forEach((env) => toAttrs(env), { concurrency: 4 }),
704 ),
705 ),
706 };
707 }),
708 );