1import * as redshift from "@distilled.cloud/aws/redshift";
2import * as Data from "effect/Data";
3import type * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import type * as Redacted from "effect/Redacted";
6import * as Schedule from "effect/Schedule";
7import * as Stream from "effect/Stream";
8import { Unowned } from "../../AdoptPolicy.ts";
9import { isResolved } from "../../Diff.ts";
10import { createPhysicalName } from "../../PhysicalName.ts";
11import * as Provider from "../../Provider.ts";
12import { Resource } from "../../Resource.ts";
13import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
14import { AWSEnvironment } from "../Environment.ts";
15import { durationToDays, unwrapRedactedString } from "../IAM/common.ts";
16import type { Providers } from "../Providers.ts";
17import {
18 applyRedshiftTagDelta,
19 redshiftArn,
20 sameStringSet,
21 toTagRecord,
22} from "./internal.ts";
23
24/**
25 * Creating a provisioned Redshift cluster requires either a
26 * `masterUserPassword` or `manageMasterPassword: true`.
27 */
28export class RedshiftMasterPasswordRequired extends Data.TaggedError(
29 "RedshiftMasterPasswordRequired",
30)<{
31 readonly clusterIdentifier: string;
32}> {}
33
34export interface ClusterProps {
35 /**
36 * Unique identifier of the cluster. Must be 1-63 lowercase alphanumeric
37 * characters or hyphens, starting with a letter. If omitted, a
38 * deterministic physical name is generated. Changing the identifier
39 * replaces the cluster.
40 */
41 clusterIdentifier?: string;
42 /**
43 * Node type of the cluster, e.g. `ra3.large`, `ra3.xlplus` or
44 * `dc2.large`. Changing the node type triggers an in-place resize.
45 * @default "ra3.large"
46 */
47 nodeType?: string;
48 /**
49 * Number of compute nodes. `1` provisions a single-node cluster; any
50 * larger value provisions a multi-node cluster. Changing the count
51 * triggers an in-place resize.
52 * @default 1
53 */
54 numberOfNodes?: number;
55 /**
56 * Admin username for the cluster database. Changing the username replaces
57 * the cluster.
58 * @default "awsuser"
59 */
60 masterUsername?: string;
61 /**
62 * Admin password. Must be 8-64 characters with at least one uppercase
63 * letter, one lowercase letter and one number. Provide this or set
64 * `manageMasterPassword`.
65 */
66 masterUserPassword?: Redacted.Redacted<string>;
67 /**
68 * Let Amazon Redshift manage the admin password in Secrets Manager
69 * instead of supplying `masterUserPassword`. The secret ARN is surfaced
70 * as the `masterPasswordSecretArn` attribute.
71 * @default false
72 */
73 manageMasterPassword?: boolean;
74 /**
75 * Name of the initial database created in the cluster. Changing the
76 * database name replaces the cluster.
77 * @default "dev"
78 */
79 dbName?: string;
80 /**
81 * Name of the {@link ClusterSubnetGroup} the cluster's nodes are placed
82 * into. Changing the subnet group replaces the cluster.
83 * @default the default cluster subnet group
84 */
85 clusterSubnetGroupName?: string;
86 /**
87 * VPC security group IDs that control network access to the cluster
88 * endpoint.
89 * @default the VPC's default security group
90 */
91 vpcSecurityGroupIds?: string[];
92 /**
93 * Name of the {@link ClusterParameterGroup} applied to the cluster.
94 * @default the family's default parameter group
95 */
96 clusterParameterGroupName?: string;
97 /**
98 * Whether the cluster endpoint is reachable from the public internet.
99 * @default false
100 */
101 publiclyAccessible?: boolean;
102 /**
103 * Whether data at rest is encrypted. RA3 node types are always encrypted.
104 * @default true
105 */
106 encrypted?: boolean;
107 /**
108 * Customer-managed KMS key used for encryption at rest.
109 * @default AWS-owned Redshift key
110 */
111 kmsKeyId?: string;
112 /**
113 * Port the cluster database accepts connections on. Changing the port
114 * replaces the cluster.
115 * @default 5439
116 */
117 port?: number;
118 /**
119 * Availability Zone the cluster is provisioned in. Changing the AZ
120 * replaces the cluster.
121 * @default chosen by Redshift
122 */
123 availabilityZone?: string;
124 /**
125 * Weekly maintenance window, e.g. `sun:05:00-sun:05:30` (UTC).
126 */
127 preferredMaintenanceWindow?: string;
128 /**
129 * How long automated snapshots are retained, e.g. `"7 days"` or
130 * `Duration.days(7)` (a bare number is milliseconds). Rounded to whole
131 * days on the wire; zero disables automated snapshots.
132 */
133 automatedSnapshotRetentionPeriod?: Duration.Input;
134 /**
135 * Whether major engine upgrades may be applied during the maintenance
136 * window.
137 * @default true
138 */
139 allowVersionUpgrade?: boolean;
140 /**
141 * Whether enhanced VPC routing forces COPY/UNLOAD traffic through the
142 * VPC.
143 * @default false
144 */
145 enhancedVpcRouting?: boolean;
146 /**
147 * IAM role ARNs the cluster can assume for COPY/UNLOAD and federated
148 * queries (max 50).
149 */
150 iamRoles?: string[];
151 /**
152 * User-defined tags for the cluster.
153 */
154 tags?: Record<string, string>;
155}
156
157export interface Cluster extends Resource<
158 "AWS.Redshift.Cluster",
159 ClusterProps,
160 {
161 /**
162 * Unique identifier of the cluster.
163 */
164 clusterIdentifier: string;
165 /**
166 * ARN of the cluster.
167 */
168 clusterArn: string;
169 /**
170 * ARN of the cluster's namespace (used by datashares and the Data API).
171 */
172 clusterNamespaceArn: string | undefined;
173 /**
174 * Current cluster status (e.g. `"available"`).
175 */
176 clusterStatus: string;
177 /**
178 * Node type of the cluster (e.g. `"ra3.large"`).
179 */
180 nodeType: string;
181 /**
182 * Number of compute nodes in the cluster.
183 */
184 numberOfNodes: number;
185 /**
186 * Name of the initial database.
187 */
188 dbName: string;
189 /**
190 * Admin (master) user name.
191 */
192 masterUsername: string | undefined;
193 /**
194 * DNS address of the cluster endpoint (pgwire host).
195 */
196 endpointAddress: string | undefined;
197 /**
198 * Port of the cluster endpoint (5439 by default).
199 */
200 endpointPort: number | undefined;
201 /**
202 * ID of the VPC the cluster runs in.
203 */
204 vpcId: string | undefined;
205 /**
206 * Availability zone the cluster is placed in.
207 */
208 availabilityZone: string | undefined;
209 /**
210 * Name of the cluster subnet group the cluster is placed in, if any.
211 */
212 clusterSubnetGroupName: string | undefined;
213 /**
214 * Whether the cluster endpoint is reachable from the public internet.
215 */
216 publiclyAccessible: boolean | undefined;
217 /**
218 * Whether the cluster's data is encrypted at rest.
219 */
220 encrypted: boolean | undefined;
221 /**
222 * KMS key encrypting the cluster, if any.
223 */
224 kmsKeyId: string | undefined;
225 /**
226 * ARN of the Secrets Manager secret holding the admin password when
227 * `manageMasterPassword` is enabled.
228 */
229 masterPasswordSecretArn: string | undefined;
230 /**
231 * Tags on the cluster (including internal Alchemy tags).
232 */
233 tags: Record<string, string>;
234 },
235 never,
236 Providers
237> {}
238
239/**
240 * A provisioned Amazon Redshift data-warehouse cluster.
241 *
242 * Clusters take roughly 5-10 minutes to provision and are billed hourly per
243 * node while they exist (`ra3.large` and `dc2.large` are the smallest node
244 * types). For serverless data warehousing see the `RedshiftServerless`
245 * namespace instead. Destroy clusters you are not using.
246 * ### Creating a Cluster
247 * **Example:** Single-Node Cluster
248 * ```typescript
249 * const cluster = yield* Redshift.Cluster("Warehouse", {
250 * nodeType: "ra3.large",
251 * numberOfNodes: 1,
252 * masterUsername: "admin",
253 * masterUserPassword: warehousePassword,
254 * dbName: "analytics",
255 * });
256 * ```
257 * **Example:** Cluster in a VPC Subnet Group
258 * ```typescript
259 * const subnetGroup = yield* Redshift.ClusterSubnetGroup("WarehouseSubnets", {
260 * subnetIds: [subnetA.subnetId, subnetB.subnetId],
261 * });
262 * const cluster = yield* Redshift.Cluster("Warehouse", {
263 * nodeType: "ra3.large",
264 * numberOfNodes: 2,
265 * masterUsername: "admin",
266 * manageMasterPassword: true,
267 * clusterSubnetGroupName: subnetGroup.clusterSubnetGroupName,
268 * publiclyAccessible: false,
269 * encrypted: true,
270 * });
271 * ```
272 *
273 * @resource
274 */
275export const Cluster = Resource<Cluster>("AWS.Redshift.Cluster");
276
277const DEFAULT_NODE_TYPE = "ra3.large";
278const DEFAULT_MASTER_USERNAME = "awsuser";
279const DEFAULT_DB_NAME = "dev";
280
281/**
282 * Retry an effect while the cluster is mid-transition
283 * (`InvalidClusterStateFault`), bounded to ~5 minutes.
284 *
285 * Expressed as an explicitly-typed module-scope helper: inlining
286 * `Effect.retry` in a lifecycle operation leaves `Retry.Return`'s
287 * conditional type unresolved in the provider's inferred layer type, which
288 * declaration emit widens to an `unknown` R — poisoning `AWS.providers()`
289 * for every downstream consumer.
290 */
291const retryWhileClusterTransitioning = <A, E extends { _tag: string }, R>(
292 self: Effect.Effect<A, E, R>,
293): Effect.Effect<A, E, R> =>
294 Effect.retry(self, {
295 while: (e) => e._tag === "InvalidClusterStateFault",
296 schedule: Schedule.max([Schedule.fixed("15 seconds"), Schedule.recurs(20)]),
297 });
298
299/**
300 * Bounded status polling (~20 minutes; cluster provisioning/resizing
301 * typically completes in 5-10). Same explicit-annotation rationale as
302 * {@link retryWhileClusterTransitioning}.
303 */
304const retryUntilSettled = <A, E, R>(
305 self: Effect.Effect<A, E, R>,
306): Effect.Effect<A, E, R> =>
307 Effect.retry(self, {
308 schedule: Schedule.max([Schedule.fixed("15 seconds"), Schedule.recurs(80)]),
309 });
310
311export const ClusterProvider = () =>
312 Provider.effect(
313 Cluster,
314 Effect.gen(function* () {
315 const toName = (id: string, props: Partial<ClusterProps>) =>
316 props.clusterIdentifier
317 ? Effect.succeed(props.clusterIdentifier)
318 : createPhysicalName({ id, maxLength: 63, lowercase: true });
319
320 const readCluster = Effect.fn(function* (identifier: string) {
321 const response = yield* redshift
322 .describeClusters({ ClusterIdentifier: identifier })
323 .pipe(
324 Effect.catchTag("ClusterNotFoundFault", () =>
325 Effect.succeed(undefined),
326 ),
327 );
328 return response?.Clusters?.[0];
329 });
330
331 const waitForAvailable = Effect.fn(function* (identifier: string) {
332 return yield* retryUntilSettled(
333 readCluster(identifier).pipe(
334 Effect.flatMap((cluster) => {
335 if (cluster === undefined) {
336 return Effect.fail(
337 new Error(`Cluster '${identifier}' not found`),
338 );
339 }
340 if (cluster.ClusterStatus !== "available") {
341 return Effect.fail(
342 new Error(
343 `Cluster '${identifier}' not available (status: ${cluster.ClusterStatus})`,
344 ),
345 );
346 }
347 return Effect.succeed(cluster);
348 }),
349 ),
350 );
351 });
352
353 // Wait for a cluster to leave a transitional state before delete.
354 // Ends when the cluster is available, deleting, or gone.
355 const waitUntilSettled = Effect.fn(function* (identifier: string) {
356 return yield* retryUntilSettled(
357 readCluster(identifier).pipe(
358 Effect.flatMap((cluster) => {
359 if (
360 cluster !== undefined &&
361 cluster.ClusterStatus !== "available" &&
362 cluster.ClusterStatus !== "deleting"
363 ) {
364 return Effect.fail(
365 new Error(
366 `Cluster '${identifier}' still settling (status: ${cluster.ClusterStatus})`,
367 ),
368 );
369 }
370 return Effect.succeed(cluster);
371 }),
372 ),
373 );
374 });
375
376 const toAttrs = Effect.fn(function* (cluster: redshift.Cluster) {
377 const { accountId, region } = yield* AWSEnvironment.current;
378 if (!cluster.ClusterIdentifier) {
379 return yield* Effect.fail(
380 new Error("Cluster is missing its identifier"),
381 );
382 }
383 return {
384 clusterIdentifier: cluster.ClusterIdentifier,
385 clusterArn: redshiftArn(
386 region,
387 accountId,
388 "cluster",
389 cluster.ClusterIdentifier,
390 ),
391 clusterNamespaceArn: cluster.ClusterNamespaceArn,
392 clusterStatus: cluster.ClusterStatus ?? "available",
393 nodeType: cluster.NodeType ?? DEFAULT_NODE_TYPE,
394 numberOfNodes: cluster.NumberOfNodes ?? 1,
395 dbName: cluster.DBName ?? DEFAULT_DB_NAME,
396 masterUsername: cluster.MasterUsername,
397 endpointAddress: cluster.Endpoint?.Address,
398 endpointPort: cluster.Endpoint?.Port,
399 vpcId: cluster.VpcId,
400 availabilityZone: cluster.AvailabilityZone,
401 clusterSubnetGroupName: cluster.ClusterSubnetGroupName,
402 publiclyAccessible: cluster.PubliclyAccessible,
403 encrypted: cluster.Encrypted,
404 kmsKeyId: cluster.KmsKeyId,
405 masterPasswordSecretArn: cluster.MasterPasswordSecretArn,
406 tags: toTagRecord(cluster.Tags),
407 };
408 });
409
410 return {
411 stables: ["clusterIdentifier", "clusterArn", "clusterNamespaceArn"],
412
413 diff: Effect.fn(function* ({ id, olds, news }) {
414 if (!isResolved(news)) return undefined;
415 const n = news ?? {};
416 const o = olds ?? {};
417 if ((yield* toName(id, o)) !== (yield* toName(id, n))) {
418 return { action: "replace" } as const;
419 }
420 // Create-only properties force a replacement.
421 if ((n.dbName ?? DEFAULT_DB_NAME) !== (o.dbName ?? DEFAULT_DB_NAME)) {
422 return { action: "replace" } as const;
423 }
424 if (
425 (n.masterUsername ?? DEFAULT_MASTER_USERNAME) !==
426 (o.masterUsername ?? DEFAULT_MASTER_USERNAME)
427 ) {
428 return { action: "replace" } as const;
429 }
430 if (
431 n.clusterSubnetGroupName !== undefined &&
432 o.clusterSubnetGroupName !== undefined &&
433 n.clusterSubnetGroupName !== o.clusterSubnetGroupName
434 ) {
435 return { action: "replace" } as const;
436 }
437 if ((n.port ?? undefined) !== (o.port ?? undefined)) {
438 return { action: "replace" } as const;
439 }
440 if (
441 (n.availabilityZone ?? undefined) !==
442 (o.availabilityZone ?? undefined)
443 ) {
444 return { action: "replace" } as const;
445 }
446 }),
447
448 read: Effect.fn(function* ({ id, olds, output }) {
449 const identifier =
450 output?.clusterIdentifier ?? (yield* toName(id, olds ?? {}));
451 const cluster = yield* readCluster(identifier);
452 if (cluster === undefined) return undefined;
453 const attrs = yield* toAttrs(cluster);
454 return (yield* hasAlchemyTags(id, attrs.tags))
455 ? attrs
456 : Unowned(attrs);
457 }),
458
459 reconcile: Effect.fn(function* ({ id, news, olds, output, session }) {
460 const identifier =
461 output?.clusterIdentifier ?? (yield* toName(id, news));
462 const internalTags = yield* createInternalTags(id);
463 const desiredTags = { ...internalTags, ...news.tags };
464 const desiredNodes = news.numberOfNodes ?? 1;
465 const desiredNodeType = news.nodeType ?? DEFAULT_NODE_TYPE;
466 const clusterType = desiredNodes > 1 ? "multi-node" : "single-node";
467
468 // 1. Observe — cloud state is authoritative.
469 let observed = yield* readCluster(identifier);
470
471 // 2. Ensure — create if missing; tolerate AlreadyExists as a race.
472 if (observed === undefined) {
473 if (!news.masterUserPassword && !news.manageMasterPassword) {
474 return yield* Effect.fail(
475 new RedshiftMasterPasswordRequired({
476 clusterIdentifier: identifier,
477 }),
478 );
479 }
480 yield* redshift
481 .createCluster({
482 ClusterIdentifier: identifier,
483 NodeType: desiredNodeType,
484 ClusterType: clusterType,
485 NumberOfNodes: desiredNodes > 1 ? desiredNodes : undefined,
486 MasterUsername: news.masterUsername ?? DEFAULT_MASTER_USERNAME,
487 MasterUserPassword: news.manageMasterPassword
488 ? undefined
489 : news.masterUserPassword,
490 ManageMasterPassword: news.manageMasterPassword,
491 DBName: news.dbName ?? DEFAULT_DB_NAME,
492 ClusterSubnetGroupName: news.clusterSubnetGroupName,
493 VpcSecurityGroupIds: news.vpcSecurityGroupIds,
494 ClusterParameterGroupName: news.clusterParameterGroupName,
495 PubliclyAccessible: news.publiclyAccessible ?? false,
496 Encrypted: news.encrypted ?? true,
497 KmsKeyId: news.kmsKeyId,
498 Port: news.port,
499 AvailabilityZone: news.availabilityZone,
500 PreferredMaintenanceWindow: news.preferredMaintenanceWindow,
501 AutomatedSnapshotRetentionPeriod: durationToDays(
502 news.automatedSnapshotRetentionPeriod,
503 ),
504 AllowVersionUpgrade: news.allowVersionUpgrade,
505 EnhancedVpcRouting: news.enhancedVpcRouting,
506 IamRoles: news.iamRoles,
507 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
508 Key,
509 Value,
510 })),
511 })
512 .pipe(
513 Effect.catchTag("ClusterAlreadyExistsFault", () =>
514 Effect.succeed(undefined),
515 ),
516 );
517 }
518
519 // Provisioning and in-flight modifications both surface as a
520 // non-available status; wait (bounded) so modify calls do not hit
521 // InvalidClusterStateFault.
522 observed = yield* waitForAvailable(identifier);
523
524 // 3. Sync — compute the update delta from OBSERVED state.
525 const update: redshift.ModifyClusterMessage = {
526 ClusterIdentifier: identifier,
527 };
528 let mutated = false;
529 if (
530 desiredNodeType !== observed.NodeType ||
531 desiredNodes !== (observed.NumberOfNodes ?? 1)
532 ) {
533 // Resizes must specify node type, count and cluster type
534 // together even when only one of them changes.
535 update.NodeType = desiredNodeType;
536 update.NumberOfNodes = desiredNodes;
537 update.ClusterType = clusterType;
538 mutated = true;
539 }
540 const observedSecurityGroups = (
541 observed.VpcSecurityGroups ?? []
542 ).flatMap((group) =>
543 group.VpcSecurityGroupId ? [group.VpcSecurityGroupId] : [],
544 );
545 if (
546 news.vpcSecurityGroupIds !== undefined &&
547 !sameStringSet(news.vpcSecurityGroupIds, observedSecurityGroups)
548 ) {
549 update.VpcSecurityGroupIds = news.vpcSecurityGroupIds;
550 mutated = true;
551 }
552 if (
553 news.clusterParameterGroupName !== undefined &&
554 !(observed.ClusterParameterGroups ?? []).some(
555 (group) =>
556 group.ParameterGroupName === news.clusterParameterGroupName,
557 )
558 ) {
559 update.ClusterParameterGroupName = news.clusterParameterGroupName;
560 mutated = true;
561 }
562 if (
563 news.publiclyAccessible !== undefined &&
564 news.publiclyAccessible !== observed.PubliclyAccessible
565 ) {
566 update.PubliclyAccessible = news.publiclyAccessible;
567 mutated = true;
568 }
569 if (
570 news.encrypted !== undefined &&
571 news.encrypted !== observed.Encrypted
572 ) {
573 update.Encrypted = news.encrypted;
574 update.KmsKeyId = news.kmsKeyId;
575 mutated = true;
576 } else if (
577 news.kmsKeyId !== undefined &&
578 news.kmsKeyId !== observed.KmsKeyId
579 ) {
580 update.Encrypted = news.encrypted ?? true;
581 update.KmsKeyId = news.kmsKeyId;
582 mutated = true;
583 }
584 if (
585 news.preferredMaintenanceWindow !== undefined &&
586 news.preferredMaintenanceWindow !==
587 observed.PreferredMaintenanceWindow
588 ) {
589 update.PreferredMaintenanceWindow = news.preferredMaintenanceWindow;
590 mutated = true;
591 }
592 const desiredRetentionDays = durationToDays(
593 news.automatedSnapshotRetentionPeriod,
594 );
595 if (
596 desiredRetentionDays !== undefined &&
597 desiredRetentionDays !== observed.AutomatedSnapshotRetentionPeriod
598 ) {
599 update.AutomatedSnapshotRetentionPeriod = desiredRetentionDays;
600 mutated = true;
601 }
602 if (
603 news.allowVersionUpgrade !== undefined &&
604 news.allowVersionUpgrade !== observed.AllowVersionUpgrade
605 ) {
606 update.AllowVersionUpgrade = news.allowVersionUpgrade;
607 mutated = true;
608 }
609 if (
610 news.enhancedVpcRouting !== undefined &&
611 news.enhancedVpcRouting !== observed.EnhancedVpcRouting
612 ) {
613 update.EnhancedVpcRouting = news.enhancedVpcRouting;
614 mutated = true;
615 }
616 // The live password is unobservable — apply it only when the
617 // declared value actually changed (olds is a hint, never truth).
618 if (
619 olds !== undefined &&
620 news.masterUserPassword !== undefined &&
621 (olds.masterUserPassword === undefined ||
622 unwrapRedactedString(news.masterUserPassword) !==
623 unwrapRedactedString(olds.masterUserPassword))
624 ) {
625 update.MasterUserPassword = news.masterUserPassword;
626 mutated = true;
627 }
628 if (
629 news.manageMasterPassword === true &&
630 observed.MasterPasswordSecretArn === undefined
631 ) {
632 update.ManageMasterPassword = true;
633 mutated = true;
634 }
635
636 if (mutated) {
637 yield* retryWhileClusterTransitioning(
638 redshift.modifyCluster(update),
639 );
640 observed = yield* waitForAvailable(identifier);
641 }
642
643 // 3b. Sync tags — diff against OBSERVED cloud tags (describe
644 // surfaces them inline).
645 const { accountId, region } = yield* AWSEnvironment.current;
646 const arn = redshiftArn(region, accountId, "cluster", identifier);
647 const { removed, upsert } = diffTags(
648 toTagRecord(observed.Tags),
649 desiredTags,
650 );
651 yield* applyRedshiftTagDelta({ arn, upsert, removed });
652
653 yield* session.note(arn);
654 const attrs = yield* toAttrs(observed);
655 return { ...attrs, tags: desiredTags };
656 }),
657
658 delete: Effect.fn(function* ({ output }) {
659 const identifier = output.clusterIdentifier;
660 // A cluster mid-create/modify rejects deletion with
661 // InvalidClusterStateFault — wait (bounded) for it to settle
662 // first. A cluster already deleting (or gone) is success.
663 yield* waitUntilSettled(identifier).pipe(
664 Effect.catch(() => Effect.succeed(undefined)),
665 );
666 yield* retryWhileClusterTransitioning(
667 redshift
668 .deleteCluster({
669 ClusterIdentifier: identifier,
670 SkipFinalClusterSnapshot: true,
671 })
672 .pipe(
673 Effect.catchTag("ClusterNotFoundFault", () =>
674 Effect.succeed(undefined),
675 ),
676 ),
677 ).pipe(
678 Effect.catchTag("InvalidClusterStateFault", () => Effect.void),
679 );
680 }),
681
682 list: () =>
683 // Top-level account/region collection: exhaustively paginate
684 // describeClusters; tags come inline on each cluster.
685 redshift.describeClusters.pages({}).pipe(
686 Stream.runCollect,
687 Effect.map((chunk) =>
688 Array.from(chunk).flatMap((page) =>
689 (page.Clusters ?? []).filter(
690 (cluster) => cluster.ClusterIdentifier !== undefined,
691 ),
692 ),
693 ),
694 Effect.flatMap(
695 Effect.forEach((cluster) => toAttrs(cluster), {
696 concurrency: 4,
697 }),
698 ),
699 ),
700 };
701 }),
702 );