1import * as ecs from "@distilled.cloud/aws/ecs";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags, diffTags } from "../../Tags.ts";
11import { AWSEnvironment, type AccountID } from "../Environment.ts";
12import type { Providers } from "../Providers.ts";
13import type { RegionID } from "../Region.ts";
14
15export type ClusterName = string;
16export type ClusterArn =
17 `arn:aws:ecs:${RegionID}:${AccountID}:cluster/${ClusterName}`;
18
19export interface ClusterProps {
20 /**
21 * Cluster name. If omitted, a unique name is generated.
22 */
23 clusterName?: string;
24 /**
25 * ECS cluster settings such as container insights.
26 */
27 settings?: ecs.ClusterSetting[];
28 /**
29 * Cluster configuration such as execute command logging.
30 */
31 configuration?: ecs.ClusterConfiguration;
32 /**
33 * Optional capacity providers associated with the cluster.
34 */
35 capacityProviders?: string[];
36 /**
37 * Default capacity provider strategy for the cluster.
38 */
39 defaultCapacityProviderStrategy?: ecs.CapacityProviderStrategyItem[];
40 /**
41 * Optional Service Connect defaults for the cluster.
42 */
43 serviceConnectDefaults?: ecs.ClusterServiceConnectDefaultsRequest;
44 /**
45 * User-defined tags to apply to the cluster.
46 */
47 tags?: Record<string, string>;
48}
49
50export interface Cluster extends Resource<
51 "AWS.ECS.Cluster",
52 ClusterProps,
53 {
54 /** The ARN of the cluster. */
55 clusterArn: ClusterArn;
56 /** The name of the cluster. */
57 clusterName: ClusterName;
58 /** The current status of the cluster, e.g. `ACTIVE`. */
59 status: string;
60 /** The cluster settings, e.g. Container Insights. */
61 settings: ecs.ClusterSetting[];
62 /** The execute-command configuration of the cluster. */
63 configuration?: ecs.ClusterConfiguration;
64 /** The capacity providers associated with the cluster. */
65 capacityProviders: string[];
66 /** The default capacity provider strategy for the cluster. */
67 defaultCapacityProviderStrategy: ecs.CapacityProviderStrategyItem[];
68 /** The default Service Connect namespace. */
69 serviceConnectDefaults?: ecs.ClusterServiceConnectDefaultsRequest;
70 /** The tags attached to the cluster. */
71 tags: Record<string, string>;
72 },
73 never,
74 Providers
75> {}
76
77/**
78 * An Amazon ECS cluster for running tasks and services.
79 * ### Creating Clusters
80 * **Example:** Default Cluster
81 * ```typescript
82 * const cluster = yield* Cluster("AppCluster", {});
83 * ```
84 *
85 * @resource
86 */
87export const Cluster = Resource<Cluster>("AWS.ECS.Cluster");
88
89class ClusterStillActive extends Data.TaggedError("ClusterStillActive")<{
90 readonly cluster: string;
91 readonly status: string | undefined;
92}> {}
93
94class ClusterNotActive extends Data.TaggedError("ClusterNotActive")<{
95 readonly cluster: string;
96 readonly status: string | undefined;
97}> {}
98
99export const ClusterProvider = () =>
100 Provider.effect(
101 Cluster,
102 Effect.gen(function* () {
103 const toEcsTags = (tags: Record<string, string>): ecs.Tag[] =>
104 Object.entries(tags).map(([key, value]) => ({
105 key,
106 value,
107 }));
108
109 const toClusterName = (
110 id: string,
111 props: { clusterName?: string } = {},
112 ) =>
113 props.clusterName
114 ? Effect.succeed(props.clusterName)
115 : createPhysicalName({ id, maxLength: 255, lowercase: true });
116
117 const applyCapacityProviders = Effect.fn(function* ({
118 cluster,
119 capacityProviders,
120 defaultCapacityProviderStrategy,
121 }: {
122 cluster: string;
123 capacityProviders?: string[];
124 defaultCapacityProviderStrategy?: ecs.CapacityProviderStrategyItem[];
125 }) {
126 if (
127 capacityProviders !== undefined ||
128 defaultCapacityProviderStrategy !== undefined
129 ) {
130 yield* ecs.putClusterCapacityProviders({
131 cluster,
132 capacityProviders: capacityProviders ?? [],
133 defaultCapacityProviderStrategy:
134 defaultCapacityProviderStrategy ?? [],
135 });
136 }
137 });
138
139 return {
140 stables: ["clusterArn", "clusterName"],
141 diff: Effect.fn(function* ({ id, olds, news }) {
142 if (!isResolved(news)) return;
143 if (
144 (yield* toClusterName(id, olds ?? {})) !==
145 (yield* toClusterName(id, news ?? {}))
146 ) {
147 return { action: "replace" } as const;
148 }
149 }),
150 read: Effect.fn(function* ({ id, olds, output }) {
151 const clusterName =
152 output?.clusterName ?? (yield* toClusterName(id, olds ?? {}));
153 const described = yield* ecs.describeClusters({
154 clusters: [output?.clusterArn ?? clusterName],
155 include: ["SETTINGS", "TAGS", "CONFIGURATIONS"],
156 });
157 const cluster = described.clusters?.[0];
158 // ECS deletion is a transition to INACTIVE. AWS may continue to
159 // return an inactive cluster from DescribeClusters for a while, but
160 // it is no longer a usable resource and must not be resurrected in
161 // state during refresh.
162 if (!cluster?.clusterArn || cluster.status === "INACTIVE") {
163 return undefined;
164 }
165 return {
166 clusterArn: cluster.clusterArn as ClusterArn,
167 clusterName: cluster.clusterName!,
168 status: cluster.status ?? "ACTIVE",
169 settings: cluster.settings ?? [],
170 configuration: cluster.configuration,
171 capacityProviders: cluster.capacityProviders ?? [],
172 defaultCapacityProviderStrategy:
173 cluster.defaultCapacityProviderStrategy ?? [],
174 serviceConnectDefaults: cluster.serviceConnectDefaults?.namespace
175 ? { namespace: cluster.serviceConnectDefaults.namespace }
176 : undefined,
177 tags: output?.tags ?? {},
178 };
179 }),
180 list: () =>
181 Effect.gen(function* () {
182 // Enumerate every cluster ARN in the account/region, paginating
183 // listClusters exhaustively.
184 const arns = yield* ecs.listClusters.pages({}).pipe(
185 Stream.runCollect,
186 Effect.map((chunk) =>
187 Array.from(chunk).flatMap((page) => page.clusterArns ?? []),
188 ),
189 );
190 if (arns.length === 0) {
191 return [];
192 }
193 // describeClusters accepts at most 100 clusters per call; batch.
194 const batches: string[][] = [];
195 for (let i = 0; i < arns.length; i += 100) {
196 batches.push(arns.slice(i, i + 100));
197 }
198 const described = yield* Effect.forEach(
199 batches,
200 (clusters) =>
201 ecs
202 .describeClusters({
203 clusters,
204 include: ["SETTINGS", "TAGS", "CONFIGURATIONS"],
205 })
206 .pipe(Effect.map((res) => res.clusters ?? [])),
207 { concurrency: 5 },
208 );
209 return described.flat().flatMap((cluster) => {
210 // DeleteCluster does not immediately erase a cluster. Inactive
211 // clusters can remain discoverable according to the ECS API,
212 // so exclude that terminal state from nuke/provider inventory.
213 if (!cluster.clusterArn || cluster.status === "INACTIVE") {
214 return [];
215 }
216 const tags = Object.fromEntries(
217 (cluster.tags ?? [])
218 .filter(
219 (t): t is { key: string; value: string } =>
220 typeof t.key === "string" && typeof t.value === "string",
221 )
222 .map((t) => [t.key, t.value]),
223 );
224 return [
225 {
226 clusterArn: cluster.clusterArn as ClusterArn,
227 clusterName: cluster.clusterName!,
228 status: cluster.status ?? "ACTIVE",
229 settings: cluster.settings ?? [],
230 configuration: cluster.configuration,
231 capacityProviders: cluster.capacityProviders ?? [],
232 defaultCapacityProviderStrategy:
233 cluster.defaultCapacityProviderStrategy ?? [],
234 serviceConnectDefaults: cluster.serviceConnectDefaults
235 ?.namespace
236 ? { namespace: cluster.serviceConnectDefaults.namespace }
237 : undefined,
238 tags,
239 },
240 ];
241 });
242 }),
243 reconcile: Effect.fn(function* ({ id, news: rawNews, session }) {
244 // Every ClusterProps field is optional, so `Cluster("Id")` (no
245 // props object at all) is a legal instantiation — normalize.
246 const news = rawNews ?? {};
247 const { accountId, region } = yield* AWSEnvironment.current;
248 const clusterName = yield* toClusterName(id, news);
249 const clusterArn =
250 `arn:aws:ecs:${region}:${accountId}:cluster/${clusterName}` as ClusterArn;
251 const internalTags = yield* createInternalTags(id);
252 const desiredTags = { ...internalTags, ...news.tags };
253
254 // Observe — fetch live cloud state.
255 let described = yield* ecs.describeClusters({
256 clusters: [clusterArn],
257 include: ["SETTINGS", "TAGS", "CONFIGURATIONS"],
258 });
259 let cluster = described.clusters?.find(
260 (c) =>
261 c.clusterName === clusterName &&
262 (c.status === "ACTIVE" || c.status === "PROVISIONING"),
263 );
264
265 // Ensure — create if missing. ECS createCluster is idempotent for
266 // identical params and returns the existing cluster on conflict;
267 // we always sync below regardless.
268 if (!cluster?.clusterArn) {
269 const created = yield* ecs.createCluster({
270 clusterName,
271 settings: news.settings,
272 configuration: news.configuration,
273 serviceConnectDefaults: news.serviceConnectDefaults,
274 tags: toEcsTags(desiredTags),
275 });
276 cluster = created.cluster;
277 }
278
279 // CreateCluster may return before the cluster is ready for updates.
280 // Re-read its status before configuration, capacity-provider or tag writes.
281 cluster = yield* ecs
282 .describeClusters({
283 clusters: [clusterArn],
284 include: ["SETTINGS", "TAGS", "CONFIGURATIONS"],
285 })
286 .pipe(
287 Effect.flatMap((response) => {
288 const observed = response.clusters?.find(
289 (candidate) => candidate.clusterArn === clusterArn,
290 );
291 return observed?.status === "ACTIVE"
292 ? Effect.succeed(observed)
293 : Effect.fail(
294 new ClusterNotActive({
295 cluster: clusterArn,
296 status: observed?.status,
297 }),
298 );
299 }),
300 Effect.retry({
301 while: (error) => error._tag === "ClusterNotActive",
302 schedule: Schedule.spaced("2 seconds"),
303 times: 10,
304 }),
305 );
306
307 // Sync cluster config — call updateCluster to converge settings,
308 // configuration, and serviceConnectDefaults to desired state.
309 yield* ecs.updateCluster({
310 cluster: clusterArn,
311 settings: news.settings,
312 configuration: news.configuration,
313 serviceConnectDefaults: news.serviceConnectDefaults,
314 });
315
316 // Sync capacity providers — observed ↔ desired.
317 yield* applyCapacityProviders({
318 cluster: clusterArn,
319 capacityProviders: news.capacityProviders,
320 defaultCapacityProviderStrategy:
321 news.defaultCapacityProviderStrategy,
322 });
323
324 // Sync tags — diff observed cloud tags against desired.
325 const observedTags = Object.fromEntries(
326 (cluster?.tags ?? [])
327 .filter(
328 (t): t is { key: string; value: string } =>
329 typeof t.key === "string" && typeof t.value === "string",
330 )
331 .map((t) => [t.key, t.value]),
332 );
333 const { removed, upsert } = diffTags(observedTags, desiredTags);
334 if (upsert.length > 0) {
335 yield* ecs.tagResource({
336 resourceArn: clusterArn,
337 tags: upsert.map((tag) => ({ key: tag.Key, value: tag.Value })),
338 });
339 }
340 if (removed.length > 0) {
341 yield* ecs.untagResource({
342 resourceArn: clusterArn,
343 tagKeys: removed,
344 });
345 }
346
347 yield* session.note(clusterArn);
348 return {
349 clusterArn,
350 clusterName,
351 status: cluster?.status ?? "ACTIVE",
352 settings: news.settings ?? [],
353 configuration: news.configuration,
354 capacityProviders: news.capacityProviders ?? [],
355 defaultCapacityProviderStrategy:
356 news.defaultCapacityProviderStrategy ?? [],
357 serviceConnectDefaults: news.serviceConnectDefaults,
358 tags: desiredTags,
359 };
360 }),
361 delete: Effect.fn(function* ({ output }) {
362 const cluster = output.clusterArn;
363
364 // Observe mutable associations instead of trusting persisted output:
365 // capacity providers can be attached out of band or output can be
366 // stale after a prior interrupted reconcile.
367 const observedCluster = (yield* ecs.describeClusters({
368 clusters: [cluster],
369 })).clusters?.find((candidate) => candidate.clusterArn === cluster);
370
371 // A cluster cannot be deleted while it still contains services,
372 // running tasks, or registered container instances — empty it
373 // first so deletion actually converges instead of silently
374 // leaving the cluster behind.
375
376 // 1. Delete services (force skips the scale-to-zero dance).
377 const serviceArns = yield* ecs.listServices.pages({ cluster }).pipe(
378 Stream.runCollect,
379 Effect.map((chunk) =>
380 Array.from(chunk).flatMap((page) => page.serviceArns ?? []),
381 ),
382 Effect.catchTag("ClusterNotFoundException", () =>
383 Effect.succeed([] as string[]),
384 ),
385 );
386 yield* Effect.forEach(
387 serviceArns,
388 (service) =>
389 ecs.deleteService({ cluster, service, force: true }).pipe(
390 Effect.catchTag(
391 ["ServiceNotFoundException", "ClusterNotFoundException"],
392 () => Effect.succeed(undefined),
393 ),
394 Effect.asVoid,
395 ),
396 { discard: true },
397 );
398
399 // 2. Stop any remaining standalone tasks.
400 const taskArns = yield* ecs.listTasks.pages({ cluster }).pipe(
401 Stream.runCollect,
402 Effect.map((chunk) =>
403 Array.from(chunk).flatMap((page) => page.taskArns ?? []),
404 ),
405 Effect.catchTag("ClusterNotFoundException", () =>
406 Effect.succeed([] as string[]),
407 ),
408 );
409 yield* Effect.forEach(
410 taskArns,
411 (task) =>
412 ecs.stopTask({ cluster, task, reason: "alchemy delete" }).pipe(
413 Effect.catchTag("ClusterNotFoundException", () =>
414 Effect.succeed(undefined),
415 ),
416 Effect.asVoid,
417 ),
418 { discard: true },
419 );
420
421 // 3. Deregister container instances (EC2 launch type).
422 const instanceArns = yield* ecs.listContainerInstances
423 .pages({ cluster })
424 .pipe(
425 Stream.runCollect,
426 Effect.map((chunk) =>
427 Array.from(chunk).flatMap(
428 (page) => page.containerInstanceArns ?? [],
429 ),
430 ),
431 Effect.catchTag("ClusterNotFoundException", () =>
432 Effect.succeed([] as string[]),
433 ),
434 );
435 yield* Effect.forEach(
436 instanceArns,
437 (containerInstance) =>
438 ecs
439 .deregisterContainerInstance({
440 cluster,
441 containerInstance,
442 force: true,
443 })
444 .pipe(
445 Effect.catchTag("ClusterNotFoundException", () =>
446 Effect.succeed(undefined),
447 ),
448 Effect.asVoid,
449 ),
450 { discard: true },
451 );
452
453 // 4. Remove custom capacity-provider associations. A cluster with
454 // an associated provider cannot be deleted even after its
455 // services, tasks, and container instances have drained.
456 if (
457 (observedCluster?.capacityProviders ?? output.capacityProviders)
458 .length > 0
459 ) {
460 yield* ecs
461 .putClusterCapacityProviders({
462 cluster,
463 capacityProviders: [],
464 defaultCapacityProviderStrategy: [],
465 })
466 .pipe(
467 Effect.retry({
468 while: (e) =>
469 e._tag === "UpdateInProgressException" ||
470 e._tag === "ResourceInUseException",
471 schedule: Schedule.max([
472 Schedule.fixed("3 seconds"),
473 Schedule.recurs(10),
474 ]),
475 }),
476 Effect.catchTag("ClusterNotFoundException", () => Effect.void),
477 );
478 }
479
480 // 5. Delete the (now empty) cluster. Draining services/tasks is
481 // asynchronous, so retry the contains-* rejections briefly.
482 yield* ecs
483 .deleteCluster({
484 cluster,
485 })
486 .pipe(
487 Effect.retry({
488 while: (e): boolean =>
489 e._tag === "ClusterContainsServicesException" ||
490 e._tag === "ClusterContainsTasksException" ||
491 e._tag === "ClusterContainsContainerInstancesException" ||
492 e._tag === "ClusterContainsCapacityProviderException" ||
493 e._tag === "UpdateInProgressException",
494 schedule: Schedule.max([
495 Schedule.fixed("3 seconds"),
496 Schedule.recurs(15),
497 ]),
498 }),
499 Effect.catchTag("ClusterNotFoundException", () => Effect.void),
500 );
501
502 // DeleteCluster's successful response only means that ECS accepted
503 // the transition. Observe the terminal INACTIVE state (or absence)
504 // before reporting deletion so dependent teardown and a following
505 // nuke pass do not race the cluster lifecycle.
506 yield* ecs.describeClusters({ clusters: [cluster] }).pipe(
507 Effect.flatMap((response) => {
508 const observed = response.clusters?.find(
509 (candidate) => candidate.clusterArn === cluster,
510 );
511 return !observed || observed.status === "INACTIVE"
512 ? Effect.void
513 : Effect.fail(
514 new ClusterStillActive({
515 cluster,
516 status: observed.status,
517 }),
518 );
519 }),
520 Effect.retry({
521 while: (error) => error instanceof ClusterStillActive,
522 schedule: Schedule.max([
523 Schedule.fixed("2 seconds"),
524 Schedule.recurs(15),
525 ]),
526 }),
527 );
528 }),
529 };
530 }),
531 );