1import * as batch from "@distilled.cloud/aws/batch";
2import * as logs from "@distilled.cloud/aws/cloudwatch-logs";
3import * as ecr from "@distilled.cloud/aws/ecr";
4import * as ecs from "@distilled.cloud/aws/ecs";
5import * as iam from "@distilled.cloud/aws/iam";
6import type { Credentials } from "../Credentials.ts";
7import type { Region } from "@distilled.cloud/aws/Region";
8import * as Data from "effect/Data";
9import type * as Duration from "effect/Duration";
10import * as Effect from "effect/Effect";
11import * as Schedule from "effect/Schedule";
12import type { Scope } from "effect/Scope";
13import * as Stream from "effect/Stream";
14import type * as rolldown from "rolldown";
15import { AlchemyContext } from "../../AlchemyContext.ts";
16import * as Bundle from "../../Bundle/Bundle.ts";
17import {
18 findCwdForBundle,
19 getStableContextDir,
20 resolveMainPath,
21} from "../../Bundle/TempRoot.ts";
22import { isResolved } from "../../Diff.ts";
23import { Docker } from "../../Docker/Docker.ts";
24import { createPhysicalName } from "../../PhysicalName.ts";
25import {
26 Platform,
27 type PlatformProps,
28 type PlatformServices,
29} from "../../Platform.ts";
30import * as Provider from "../../Provider.ts";
31import { Resource, type ResourceBinding } from "../../Resource.ts";
32import type { RuntimeContext } from "../../RuntimeContext.ts";
33import {
34 createHostRuntimeContext,
35 type HostRuntimeContext,
36 type ServerHost,
37} from "../../Server/Process.ts";
38import { Stack } from "../../Stack.ts";
39import {
40 createInternalTags,
41 createTagsList,
42 diffTags,
43 hasTags,
44} from "../../Tags.ts";
45import { toWireSeconds } from "../../Util/Duration.ts";
46import { AWSEnvironment, type AccountID } from "../Environment.ts";
47import type { PolicyStatement } from "../IAM/Policy.ts";
48import { buildAndPushEcrImage } from "../ECR/Image.ts";
49import type { Providers } from "../Providers.ts";
50import type { RegionID } from "../Region.ts";
51
52export type JobDefinitionName = string;
53/** Revision-qualified job definition ARN (`.../job-definition/{name}:{revision}`). */
54export type JobDefinitionArn =
55 `arn:aws:batch:${RegionID}:${AccountID}:job-definition/${JobDefinitionName}:${number}`;
56
57/**
58 * Raised when the props don't select exactly one of the two forms: the
59 * low-level container form (`image` + `executionRoleArn`) or the
60 * Effect-native form (`main`).
61 */
62export class JobDefinitionConfigError extends Data.TaggedError(
63 "JobDefinitionConfigError",
64)<{
65 readonly message: string;
66}> {}
67
68class JobDefinitionStillVisible extends Data.TaggedError(
69 "JobDefinitionStillVisible",
70)<{
71 readonly family: string;
72 readonly service: "Batch" | "ECS";
73 readonly arns: readonly string[];
74}> {}
75
76export interface JobDefinitionProps extends PlatformProps {
77 /**
78 * Name (family) of the job definition. If omitted, a unique name is
79 * generated. Content changes register a new revision under the same name.
80 */
81 jobDefinitionName?: string;
82 /**
83 * Container image the job runs (low-level form, e.g.
84 * `public.ecr.aws/docker/library/busybox:latest`). Mutually exclusive
85 * with `main`.
86 */
87 image?: string;
88 /**
89 * Module entrypoint for an Effect-native run-to-completion job (typically
90 * `import.meta.url` from an inline Effect program). Alchemy bundles the
91 * program, builds a container image, pushes it to a managed ECR
92 * repository, and provisions the job/execution IAM roles — mutually
93 * exclusive with a caller-supplied `image`.
94 */
95 main?: string;
96 /**
97 * Named export to load from `main`.
98 * @default "default"
99 */
100 handler?: string;
101 /**
102 * Command to run in the container (low-level form). Overridable
103 * per-submission via `containerOverrides.command`. The Effect-native form
104 * bakes its entrypoint into the image instead.
105 */
106 command?: string[];
107 /**
108 * Fargate vCPUs for the job. Must be a valid Fargate size (0.25, 0.5, 1,
109 * 2, 4, 8, 16) compatible with `memory`.
110 * @default 0.25
111 */
112 vcpus?: number;
113 /**
114 * Memory (MiB) for the job. Must be compatible with `vcpus` per the
115 * Fargate size matrix.
116 * @default 512
117 */
118 memory?: number;
119 /**
120 * Environment variables set in the job container.
121 */
122 environment?: Record<string, string>;
123 /**
124 * Additional environment variables for the Effect-native container.
125 * Non-string values are JSON-encoded. (Capability bindings also inject
126 * their variables here automatically.)
127 */
128 env?: Record<string, any>;
129 /**
130 * IAM role assumed by the job's application code. For the Effect-native
131 * form (`main`), Alchemy provisions and manages this role automatically
132 * (binding policy statements attach to it).
133 */
134 jobRoleArn?: string;
135 /**
136 * Execution role used by ECS/Fargate to pull the image and write logs.
137 * Required for the low-level form; provisioned automatically for the
138 * Effect-native form.
139 */
140 executionRoleArn?: string;
141 /**
142 * Compute platforms that can run this definition. Use `EC2` for unmanaged
143 * ECS compute environments.
144 * @default ["FARGATE"]
145 */
146 platformCapabilities?: ("EC2" | "FARGATE")[];
147 /**
148 * Whether the Fargate task ENI gets a public IP. Required for image pulls
149 * from public registries when running in public subnets.
150 * @default "ENABLED"
151 */
152 assignPublicIp?: "ENABLED" | "DISABLED";
153 /**
154 * Default parameter substitutions for `Ref::` placeholders in `command`.
155 */
156 parameters?: Record<string, string>;
157 /**
158 * Number of times a failed job is retried (1-10).
159 * @default 1
160 */
161 retryAttempts?: number;
162 /**
163 * Job execution timeout (minimum 60 seconds), e.g. `"15 minutes"` or
164 * `Duration.minutes(15)`.
165 */
166 timeout?: Duration.Input;
167 /**
168 * Propagate job definition tags to the ECS task.
169 * @default false
170 */
171 propagateTags?: boolean;
172 /**
173 * Bundler configuration for the Effect-native entrypoint. Unused
174 * code is tree-shaken. `effect`, alchemy, and `@distilled.cloud` are
175 * marked pure so unused parts prune more aggressively. List extra
176 * packages with `pure.packages`, or disable with `pure: false`.
177 */
178 build?: Bundle.BundleConfig;
179 /**
180 * Docker image build for the Effect-native form: optional full
181 * `dockerfile`. When omitted, Alchemy generates a Dockerfile for the
182 * bundled `index.mjs`.
183 */
184 docker?: {
185 /**
186 * Base image when Alchemy generates the Dockerfile.
187 * @default public.ecr.aws/docker/library/bun:1
188 */
189 base?: string;
190 /** Full Dockerfile content (replaces generated Dockerfile). */
191 dockerfile?: string;
192 };
193 /**
194 * Additional managed policy ARNs for the managed job role
195 * (Effect-native form only).
196 */
197 jobRoleManagedPolicyArns?: string[];
198 /**
199 * User-defined tags to apply to the job definition.
200 */
201 tags?: Record<string, string>;
202}
203
204export interface JobDefinition extends Resource<
205 "AWS.Batch.JobDefinition",
206 JobDefinitionProps,
207 {
208 jobDefinitionName: JobDefinitionName;
209 jobDefinitionArn: JobDefinitionArn;
210 revision: number;
211 status: string;
212 tags: Record<string, string>;
213 /** The full URI of the built container image (Effect-native form only). */
214 imageUri: string | undefined;
215 /** The managed ECR repository name (Effect-native form only). */
216 repositoryName: string | undefined;
217 /** The managed ECR repository URI (Effect-native form only). */
218 repositoryUri: string | undefined;
219 /** The ARN of the managed job role (Effect-native form only). */
220 jobRoleArn: string | undefined;
221 /** The name of the managed job role (Effect-native form only). */
222 jobRoleName: string | undefined;
223 /** The ARN of the execution role the job definition uses. */
224 executionRoleArn: string | undefined;
225 /** The name of the managed execution role (Effect-native form only). */
226 executionRoleName: string | undefined;
227 /** Content hash of the bundled program (Effect-native form only). */
228 codeHash: string | undefined;
229 },
230 {
231 /** Environment variables injected into the job container. */
232 env?: Record<string, any>;
233 /** IAM policy statements attached to the managed job role. */
234 policyStatements?: PolicyStatement[];
235 },
236 Providers
237> {}
238
239export type JobDefinitionServices =
240 | Credentials
241 | Region
242 | ServerHost
243 | AWSEnvironment;
244
245/**
246 * The shape an Effect-native job implementation returns: a single `run`
247 * Effect that is the run-to-completion body of the job. The container exits
248 * 0 when it succeeds (job `SUCCEEDED`) and 1 when it fails (job `FAILED`,
249 * subject to the definition's `retryAttempts`).
250 */
251export type JobDefinitionShape = void | {
252 run: Effect.Effect<
253 void,
254 unknown,
255 JobDefinitionServices | PlatformServices | RuntimeContext | Scope
256 >;
257};
258
259export interface JobDefinitionRuntimeContext extends HostRuntimeContext {
260 readonly Type: "AWS.Batch.JobDefinition";
261}
262
263const createJobDefinitionRuntimeContext = (
264 id: string,
265): JobDefinitionRuntimeContext => {
266 const base = createHostRuntimeContext("AWS.Batch.JobDefinition")(id);
267 return {
268 ...base,
269 // A Batch job is run-to-completion, not a server. `Platform` hands any
270 // non-`fetch` impl shape to `serve`; register the shape's `run` Effect
271 // as a host runner (collected into `exports.program`, executed by the
272 // generated container entrypoint) instead of booting an HTTP server.
273 serve: ((_handler, options) => {
274 const run = (options?.shape as { run?: unknown } | undefined)?.run;
275 return Effect.isEffect(run)
276 ? base.run(run as Effect.Effect<void, never, any>)
277 : Effect.void;
278 }) as HostRuntimeContext["serve"],
279 } as JobDefinitionRuntimeContext;
280};
281
282/**
283 * An AWS Batch job definition for Fargate container jobs. Job definitions are
284 * immutable revisions — changing the container configuration registers a new
285 * revision under the same name (like ECS task definitions); destroying the
286 * resource deregisters every active revision.
287 *
288 * `JobDefinition` is a Platform: alongside the low-level container form
289 * (`image` + `executionRoleArn`), it supports Effect-native run-to-completion
290 * implementations — an inline Effect program that Alchemy bundles,
291 * containerizes as the job container's command, pushes to a managed ECR
292 * repository, and registers, provisioning the job and execution roles
293 * automatically. Capability bindings (e.g. S3 `GetObject`) attach IAM policy
294 * statements to the managed job role and inject their environment variables
295 * into the container.
296 *
297 * ### Creating Job Definitions
298 * **Example:** Busybox echo job (low-level container form)
299 * ```typescript
300 * const jobDef = yield* Batch.JobDefinition("EchoJob", {
301 * image: "public.ecr.aws/docker/library/busybox:latest",
302 * command: ["echo", "hello from batch"],
303 * executionRoleArn: executionRole.roleArn,
304 * });
305 * ```
306 *
307 * **Example:** Sized job with environment
308 * ```typescript
309 * const jobDef = yield* Batch.JobDefinition("EtlJob", {
310 * image: image.imageUri,
311 * vcpus: 1,
312 * memory: 2048,
313 * environment: { STAGE: "prod" },
314 * jobRoleArn: jobRole.roleArn,
315 * executionRoleArn: executionRole.roleArn,
316 * retryAttempts: 3,
317 * timeout: "15 minutes",
318 * });
319 * ```
320 *
321 * ### Effect-Native Jobs
322 * **Example:** Tagged class with an inline run-to-completion Effect
323 * ```typescript
324 * export default class Nightly extends Batch.JobDefinition<Nightly>()(
325 * "Nightly",
326 * { main: import.meta.url, vcpus: 1, memory: 2048 },
327 * Effect.gen(function* () {
328 * const getObject = yield* AWS.S3.GetObject(bucket);
329 * return {
330 * run: Effect.gen(function* () {
331 * const data = yield* getObject({ key: "input.csv" });
332 * yield* Effect.log("processed nightly batch");
333 * }),
334 * };
335 * }),
336 * ) {}
337 * ```
338 *
339 * **Example:** Eager inline job
340 * ```typescript
341 * export default Batch.JobDefinition(
342 * "Reindex",
343 * { main: import.meta.url },
344 * Effect.succeed({
345 * run: Effect.log("reindex complete"),
346 * }),
347 * );
348 * ```
349 *
350 * **Example:** Plain external script (bundled as-is)
351 * ```typescript
352 * // ./job.ts runs top-level and exits; Alchemy bundles + containerizes it.
353 * const jobDef = yield* Batch.JobDefinition("Script", {
354 * main: path.join(import.meta.dirname, "job.ts"),
355 * });
356 * ```
357 *
358 * ### Bundling & Tree-shaking
359 * `main` is bundled with rolldown at deploy time. Unused code is
360 * tree-shaken. `effect`, alchemy, and `@distilled.cloud` are marked
361 * pure so unused parts prune more aggressively. Your app is not
362 * marked pure.
363 *
364 * **Example:** Mark additional packages as pure
365 * Only list packages with no top-level side effects.
366 * ```typescript
367 * {
368 * main: import.meta.url,
369 * build: {
370 * pure: { packages: ["my-lib", "@my-scope/*"] },
371 * },
372 * }
373 * ```
374 *
375 * **Example:** Turn it off
376 * ```typescript
377 * {
378 * main: import.meta.url,
379 * build: { pure: false },
380 * }
381 * ```
382 *
383 * @resource
384 */
385export const JobDefinition: Platform<
386 JobDefinition,
387 JobDefinitionServices,
388 JobDefinitionShape,
389 JobDefinitionRuntimeContext
390> = Platform("AWS.Batch.JobDefinition", {
391 createRuntimeContext: createJobDefinitionRuntimeContext,
392});
393
394const observedTagsOf = (d: { tags?: { [key: string]: string | undefined } }) =>
395 Object.fromEntries(
396 Object.entries(d.tags ?? {}).filter(
397 (e): e is [string, string] => typeof e[1] === "string",
398 ),
399 );
400
401/** Attributes only produced by the Effect-native (`main`) form. */
402interface PlatformAttributes {
403 imageUri: string | undefined;
404 repositoryName: string | undefined;
405 repositoryUri: string | undefined;
406 jobRoleArn: string | undefined;
407 jobRoleName: string | undefined;
408 executionRoleName: string | undefined;
409 codeHash: string | undefined;
410}
411
412const emptyPlatformAttributes: PlatformAttributes = {
413 imageUri: undefined,
414 repositoryName: undefined,
415 repositoryUri: undefined,
416 jobRoleArn: undefined,
417 jobRoleName: undefined,
418 executionRoleName: undefined,
419 codeHash: undefined,
420};
421
422const toAttributes = (
423 d: batch.JobDefinition,
424 tags: Record<string, string>,
425 platform: PlatformAttributes = emptyPlatformAttributes,
426) => ({
427 jobDefinitionName: d.jobDefinitionName!,
428 jobDefinitionArn: d.jobDefinitionArn as JobDefinitionArn,
429 revision: d.revision ?? 1,
430 status: d.status ?? "ACTIVE",
431 tags,
432 executionRoleArn: d.containerProperties?.executionRoleArn,
433 ...platform,
434});
435
436/**
437 * The effective (resolved) container configuration a reconcile wants
438 * registered — identical for both forms once the Effect-native form has
439 * built its image and roles.
440 */
441interface EffectiveContainer {
442 image: string;
443 command: string[] | undefined;
444 vcpus: string;
445 memory: string;
446 environment: Record<string, string>;
447 jobRoleArn: string | undefined;
448 executionRoleArn: string;
449 assignPublicIp: "ENABLED" | "DISABLED";
450 parameters: Record<string, string> | undefined;
451 retryAttempts: number | undefined;
452 timeoutSeconds: number | undefined;
453 propagateTags: boolean | undefined;
454}
455
456/**
457 * Normalized content fingerprint of a job definition — used to decide whether
458 * a new revision must be registered. Only covers the aspects this resource
459 * manages.
460 */
461const fingerprint = (input: {
462 image?: string;
463 command?: string[];
464 vcpus?: string;
465 memory?: string;
466 environment?: Record<string, string>;
467 jobRoleArn?: string;
468 executionRoleArn?: string;
469 assignPublicIp?: string;
470 parameters?: Record<string, string>;
471 retryAttempts?: number;
472 timeoutSeconds?: number;
473 propagateTags?: boolean;
474 platformCapabilities?: ("EC2" | "FARGATE")[];
475}) =>
476 JSON.stringify({
477 image: input.image,
478 command: input.command ?? [],
479 vcpus: input.vcpus,
480 memory: input.memory,
481 environment: Object.entries(input.environment ?? {}).sort(([a], [b]) =>
482 a.localeCompare(b),
483 ),
484 jobRoleArn: input.jobRoleArn,
485 executionRoleArn: input.executionRoleArn,
486 assignPublicIp: input.assignPublicIp ?? "ENABLED",
487 parameters: Object.entries(input.parameters ?? {}).sort(([a], [b]) =>
488 a.localeCompare(b),
489 ),
490 retryAttempts: input.retryAttempts ?? 1,
491 timeoutSeconds: input.timeoutSeconds,
492 propagateTags: input.propagateTags ?? false,
493 platformCapabilities: [
494 ...(input.platformCapabilities ?? ["FARGATE"]),
495 ].sort(),
496 });
497
498const observedFingerprint = (d: batch.JobDefinition) => {
499 const c = d.containerProperties ?? {};
500 const req = (type: string) =>
501 c.resourceRequirements?.find((r) => r.type === type)?.value;
502 return fingerprint({
503 image: c.image,
504 command: [...(c.command ?? [])],
505 vcpus: req("VCPU"),
506 memory: req("MEMORY"),
507 environment: Object.fromEntries(
508 (c.environment ?? []).flatMap((kv) =>
509 kv.name !== undefined ? [[kv.name, kv.value ?? ""]] : [],
510 ),
511 ),
512 jobRoleArn: c.jobRoleArn,
513 executionRoleArn: c.executionRoleArn,
514 assignPublicIp: c.networkConfiguration?.assignPublicIp,
515 parameters: Object.fromEntries(
516 Object.entries(d.parameters ?? {}).flatMap(([k, v]) =>
517 v !== undefined ? [[k, v]] : [],
518 ),
519 ),
520 retryAttempts: d.retryStrategy?.attempts,
521 timeoutSeconds: d.timeout?.attemptDurationSeconds,
522 propagateTags: d.propagateTags,
523 platformCapabilities: d.platformCapabilities as
524 | ("EC2" | "FARGATE")[]
525 | undefined,
526 });
527};
528
529const desiredFingerprint = (
530 effective: EffectiveContainer,
531 news: JobDefinitionProps,
532) =>
533 fingerprint({
534 image: effective.image,
535 command: effective.command,
536 vcpus: effective.vcpus,
537 memory: effective.memory,
538 environment: effective.environment,
539 jobRoleArn: effective.jobRoleArn,
540 executionRoleArn: effective.executionRoleArn,
541 assignPublicIp: news.assignPublicIp,
542 parameters: effective.parameters,
543 retryAttempts: effective.retryAttempts,
544 timeoutSeconds: effective.timeoutSeconds,
545 propagateTags: effective.propagateTags,
546 platformCapabilities: news.platformCapabilities,
547 });
548
549export const JobDefinitionProvider = () =>
550 Provider.effect(
551 JobDefinition,
552 Effect.gen(function* () {
553 const stack = yield* Stack;
554 const docker = yield* Docker;
555 const { dotAlchemy } = yield* AlchemyContext;
556 const virtualEntryPlugin = yield* Bundle.virtualEntryPlugin;
557
558 const alchemyEnv = {
559 ALCHEMY_STACK_NAME: stack.name,
560 ALCHEMY_STAGE: stack.stage,
561 ALCHEMY_PHASE: "runtime",
562 };
563
564 const toName = (
565 id: string,
566 props: { jobDefinitionName?: string } = {},
567 ) =>
568 props.jobDefinitionName
569 ? Effect.succeed(props.jobDefinitionName)
570 : createPhysicalName({ id, maxLength: 128 });
571
572 const createRoleName = (id: string, suffix: string) =>
573 createPhysicalName({ id: `${id}-${suffix}`, maxLength: 64 });
574
575 // Batch caps `containerProperties.image` at 255 characters, and the
576 // image ref is `<account>.dkr.ecr.<region>.amazonaws.com/<repo>:<tag>`
577 // — up to 52 characters of registry prefix plus the 16-character
578 // content tag — so the repository name (ECR itself allows 256) must
579 // stop at 180 to fit in every region.
580 const createRepositoryName = (id: string) =>
581 createPhysicalName({
582 id: `${id}-repo`,
583 maxLength: 180,
584 lowercase: true,
585 });
586
587 /** Every ACTIVE revision of the family, ascending by revision. */
588 const activeRevisions = (name: string) =>
589 batch.describeJobDefinitions
590 .pages({ jobDefinitionName: name, status: "ACTIVE" })
591 .pipe(
592 Stream.runCollect,
593 Effect.map((pages) =>
594 Array.from(pages)
595 .flatMap((page) => page.jobDefinitions ?? [])
596 .sort((a, b) => (a.revision ?? 0) - (b.revision ?? 0)),
597 ),
598 );
599
600 const latestRevision = (name: string) =>
601 activeRevisions(name).pipe(Effect.map((defs) => defs.at(-1)));
602
603 const waitUntilNoActiveBatchRevisions = (family: string) =>
604 Effect.gen(function* () {
605 const active = yield* activeRevisions(family);
606 if (active.length > 0) {
607 return yield* Effect.fail(
608 new JobDefinitionStillVisible({
609 family,
610 service: "Batch",
611 arns: active.flatMap((revision) =>
612 revision.jobDefinitionArn ? [revision.jobDefinitionArn] : [],
613 ),
614 }),
615 );
616 }
617 }).pipe(
618 Effect.retry({
619 while: (error) => error._tag === "JobDefinitionStillVisible",
620 schedule: Schedule.max([
621 Schedule.fixed("2 seconds"),
622 Schedule.recurs(15),
623 ]),
624 }),
625 );
626
627 const listBackingTaskDefinitions = (
628 family: string,
629 status: "ACTIVE" | "INACTIVE",
630 ) =>
631 ecs.listTaskDefinitions.items({ familyPrefix: family, status }).pipe(
632 Stream.filter((arn) => {
633 const suffix = arn.split("/").at(-1);
634 return suffix?.slice(0, suffix.lastIndexOf(":")) === family;
635 }),
636 Stream.runCollect,
637 Effect.map((chunk) => Array.from(chunk)),
638 );
639
640 const waitUntilBackingRevisionDeletionStarted = (
641 family: string,
642 arn: string,
643 ) =>
644 Effect.gen(function* () {
645 const status = yield* ecs
646 .describeTaskDefinition({ taskDefinition: arn })
647 .pipe(
648 Effect.map((response) => response.taskDefinition?.status),
649 // Not-found means ECS already completed the asynchronous
650 // deletion and is therefore terminal as well.
651 Effect.catchTag("ClientException", () =>
652 Effect.succeed(undefined),
653 ),
654 );
655 if (status !== undefined && status !== "DELETE_IN_PROGRESS") {
656 return yield* Effect.fail(
657 new JobDefinitionStillVisible({
658 family,
659 service: "ECS",
660 arns: [arn],
661 }),
662 );
663 }
664 }).pipe(
665 Effect.retry({
666 while: (error) => error._tag === "JobDefinitionStillVisible",
667 schedule: Schedule.max([
668 Schedule.fixed("2 seconds"),
669 Schedule.recurs(30),
670 ]),
671 }),
672 );
673
674 /**
675 * AWS Batch registers an ECS task definition under the same family when
676 * a job first runs. It is not part of Batch's list API or Alchemy state,
677 * and Batch can leave it ACTIVE after the Batch definition is gone.
678 * Exact-family filtering keeps this cleanup scoped to the resource.
679 */
680 const reapBackingTaskDefinitions = (family: string) =>
681 Effect.gen(function* () {
682 const active = yield* listBackingTaskDefinitions(family, "ACTIVE");
683 yield* Effect.forEach(
684 active,
685 (arn) =>
686 ecs
687 .deregisterTaskDefinition({ taskDefinition: arn })
688 .pipe(Effect.catchTag("ClientException", () => Effect.void)),
689 { concurrency: 4, discard: true },
690 );
691
692 // Observe ACTIVE absence before attempting the hard delete. This is
693 // the transition ECS requires before deleteTaskDefinitions.
694 yield* Effect.gen(function* () {
695 const remaining = yield* listBackingTaskDefinitions(
696 family,
697 "ACTIVE",
698 );
699 if (remaining.length > 0) {
700 return yield* Effect.fail(
701 new JobDefinitionStillVisible({
702 family,
703 service: "ECS",
704 arns: remaining,
705 }),
706 );
707 }
708 }).pipe(
709 Effect.retry({
710 while: (error) => error._tag === "JobDefinitionStillVisible",
711 schedule: Schedule.max([
712 Schedule.fixed("2 seconds"),
713 Schedule.recurs(15),
714 ]),
715 }),
716 );
717
718 const inactive = [
719 ...new Set([
720 ...active,
721 ...(yield* listBackingTaskDefinitions(family, "INACTIVE")),
722 ]),
723 ];
724 for (let i = 0; i < inactive.length; i += 10) {
725 const response = yield* ecs.deleteTaskDefinitions({
726 taskDefinitions: inactive.slice(i, i + 10),
727 });
728 if ((response.failures?.length ?? 0) > 0) {
729 return yield* Effect.fail(
730 new Error(
731 `ECS failed to delete Batch backing task definitions for ${family}: ${response.failures
732 ?.map(
733 (failure) =>
734 `${failure.arn ?? "unknown"}: ${failure.reason ?? "unknown"} (${failure.detail ?? "no detail"})`,
735 )
736 .join(", ")}`,
737 ),
738 );
739 }
740 }
741 yield* Effect.forEach(
742 inactive,
743 (arn) => waitUntilBackingRevisionDeletionStarted(family, arn),
744 { concurrency: 4, discard: true },
745 );
746 });
747
748 /**
749 * Ensure an IAM role trusted by ECS tasks exists (creates on miss,
750 * adopts on race when it carries our tags).
751 */
752 const ensureRole = Effect.fn(function* ({
753 id,
754 roleName,
755 managedPolicyArns,
756 }: {
757 id: string;
758 roleName: string;
759 managedPolicyArns?: string[];
760 }) {
761 const tags = yield* createInternalTags(id);
762 const role = yield* iam
763 .createRole({
764 RoleName: roleName,
765 AssumeRolePolicyDocument: JSON.stringify({
766 Version: "2012-10-17",
767 Statement: [
768 {
769 Effect: "Allow",
770 Principal: { Service: "ecs-tasks.amazonaws.com" },
771 Action: "sts:AssumeRole",
772 },
773 ],
774 }),
775 Tags: createTagsList(tags),
776 })
777 .pipe(
778 Effect.catchTag("EntityAlreadyExistsException", () =>
779 iam.getRole({ RoleName: roleName }).pipe(
780 Effect.filterOrFail(
781 (existing) => hasTags(tags, existing.Role?.Tags),
782 () =>
783 new Error(
784 `Role '${roleName}' already exists and is not managed by alchemy`,
785 ),
786 ),
787 ),
788 ),
789 );
790 for (const policyArn of managedPolicyArns ?? []) {
791 yield* iam
792 .attachRolePolicy({ RoleName: roleName, PolicyArn: policyArn })
793 .pipe(Effect.catchTag("LimitExceededException", () => Effect.void));
794 }
795 return role.Role!.Arn!;
796 });
797
798 /** Ensure the managed ECR repository for the built image exists. */
799 const ensureRepository = Effect.fn(function* ({
800 repositoryName,
801 tags,
802 }: {
803 repositoryName: string;
804 tags: Record<string, string>;
805 }) {
806 const created = yield* ecr
807 .createRepository({
808 repositoryName,
809 imageTagMutability: "MUTABLE",
810 imageScanningConfiguration: { scanOnPush: true },
811 tags: Object.entries(tags).map(([Key, Value]) => ({ Key, Value })),
812 })
813 .pipe(
814 Effect.catchTag("RepositoryAlreadyExistsException", () =>
815 Effect.gen(function* () {
816 const existing = yield* ecr.describeRepositories({
817 repositoryNames: [repositoryName],
818 });
819 return { repository: existing.repositories?.[0] };
820 }),
821 ),
822 );
823 const repository = created.repository;
824 if (!repository?.repositoryUri) {
825 return yield* Effect.die(
826 new Error(`Failed to resolve ECR repository '${repositoryName}'`),
827 );
828 }
829 return { repositoryUri: repository.repositoryUri };
830 });
831
832 /**
833 * Attach binding-declared IAM policy statements to the managed job
834 * role and collect binding-declared environment variables.
835 */
836 const attachBindings = Effect.fn(function* ({
837 roleName,
838 policyName,
839 bindings,
840 }: {
841 roleName: string;
842 policyName: string;
843 bindings: ResourceBinding<JobDefinition["Binding"]>[];
844 }) {
845 const activeBindings = bindings.filter(
846 (
847 binding: ResourceBinding<JobDefinition["Binding"]> & {
848 action?: string;
849 },
850 ) => binding.action !== "delete",
851 );
852
853 const env = activeBindings
854 .map((binding) => binding?.data?.env)
855 .reduce((acc, value) => ({ ...acc, ...value }), {});
856
857 const policyStatements = activeBindings.flatMap(
858 (binding) =>
859 binding?.data?.policyStatements?.map((statement) => ({
860 ...statement,
861 Sid: statement.Sid?.replace(/[^A-Za-z0-9]+/gi, ""),
862 })) ?? [],
863 );
864
865 if (policyStatements.length > 0) {
866 yield* iam.putRolePolicy({
867 RoleName: roleName,
868 PolicyName: policyName,
869 PolicyDocument: JSON.stringify({
870 Version: "2012-10-17",
871 Statement: policyStatements,
872 }),
873 });
874 } else {
875 yield* iam
876 .deleteRolePolicy({ RoleName: roleName, PolicyName: policyName })
877 .pipe(Effect.catchTag("NoSuchEntityException", () => Effect.void));
878 }
879
880 return env;
881 });
882
883 /** Bundle the Effect-native program into container-ready files. */
884 const bundleProgram = Effect.fn(function* (props: JobDefinitionProps) {
885 const handler = props.handler ?? "default";
886 const realMain = yield* resolveMainPath(props.main!);
887 const cwd = yield* findCwdForBundle(realMain);
888
889 const buildBundle = Effect.fn(function* (
890 entry: string,
891 plugins?: rolldown.RolldownPluginOption,
892 ) {
893 return yield* Bundle.build(
894 {
895 ...props.build?.input,
896 input: entry,
897 cwd,
898 platform: "node",
899 // The container runs on `bun`; keep `bun`/`bun:*` external (the
900 // runtime provides them) and resolve the `bun` export condition
901 // so `@effect/platform-bun` picks its Bun implementations.
902 external: [
903 "bun",
904 "bun:*",
905 ...((props.build?.input?.external as string[] | undefined) ??
906 []),
907 ],
908 resolve: {
909 conditionNames: [...Bundle.BUN_CONDITION_NAMES],
910 ...props.build?.input?.resolve,
911 },
912 plugins: [props.build?.input?.plugins, plugins],
913 },
914 {
915 ...props.build?.output,
916 format: "esm",
917 sourcemap: props.build?.output?.sourcemap ?? false,
918 minify: props.build?.output?.minify ?? false,
919 entryFileNames: "index.mjs",
920 },
921 props.build,
922 );
923 });
924
925 const bundleOutput = props.isExternal
926 ? yield* buildBundle(realMain)
927 : yield* buildBundle(
928 realMain,
929 virtualEntryPlugin(
930 (importPath) => `
931import { bootstrap } from "alchemy/Runtime/Bootstrap/Batch";
932import { ${handler} as entrypoint } from ${JSON.stringify(importPath)};
933
934await bootstrap(entrypoint);
935`,
936 ),
937 );
938
939 // Return every emitted file (entry + shared chunks). Dynamic imports
940 // in the AWS SDK split into chunks; dropping any of them crashes the
941 // container with `Cannot find module './chunk-X.js'`.
942 const files = bundleOutput.files.map((file) => ({
943 path: file.path,
944 content:
945 typeof file.content === "string"
946 ? new TextEncoder().encode(file.content)
947 : file.content,
948 }));
949
950 // 16 hex characters of the content hash, like the ECS/App Runner image
951 // sources: the full sha256 as a tag pushed long repository names over
952 // Batch's 255-character `image` limit.
953 return { files, hash: bundleOutput.hash.slice(0, 16) };
954 });
955
956 /** Build + push the container image for the bundled program. */
957 const buildAndPushImage = Effect.fn(function* ({
958 id,
959 repositoryUri,
960 hash,
961 files,
962 props,
963 }: {
964 id: string;
965 repositoryUri: string;
966 hash: string;
967 files: { path: string; content: Uint8Array<ArrayBufferLike> }[];
968 props: JobDefinitionProps;
969 }) {
970 const realMain = yield* resolveMainPath(props.main!);
971 const contextDir = yield* getStableContextDir(
972 realMain,
973 dotAlchemy,
974 `${id}-image`,
975 );
976 const imageUri = `${repositoryUri}:${hash}`;
977
978 const generatedDockerfile = (() => {
979 const base =
980 props.docker?.base ?? "public.ecr.aws/docker/library/bun:1";
981 return [
982 `FROM ${base}`,
983 `WORKDIR /app`,
984 `COPY index.mjs /app/index.mjs`,
985 // Copy any additional rolldown chunks (`chunk-XXX.js`, ...).
986 // Non-trivial bundles always emit at least one; minimal bundles
987 // emit none and the COPY no-ops.
988 `COPY *.js /app/`,
989 `ENTRYPOINT ["bun", "/app/index.mjs"]`,
990 ].join("\n");
991 })();
992
993 const dockerfile = props.docker?.dockerfile ?? generatedDockerfile;
994
995 yield* docker.materialize({
996 context: contextDir,
997 dockerfile,
998 // Entry chunk becomes `index.mjs`; all other chunks keep their
999 // emitted `*.js` names so the entry's relative imports resolve.
1000 files: files.map((file, index) => ({
1001 path: index === 0 ? "index.mjs" : file.path,
1002 content: file.content,
1003 })),
1004 });
1005 // Batch Fargate defaults to X86_64 — always build linux/amd64
1006 // (cross-built via emulation on ARM64 hosts).
1007 return yield* buildAndPushEcrImage(docker, {
1008 imageUri,
1009 context: contextDir,
1010 platform: "linux/amd64",
1011 });
1012 });
1013
1014 /**
1015 * Delete the managed platform resources (ECR repository, job +
1016 * execution roles). Used by `delete` and by a reconcile that switches
1017 * an Effect-native definition back to the low-level form. Idempotent.
1018 */
1019 const cleanupPlatformResources = Effect.fn(function* (platform: {
1020 repositoryName?: string | undefined;
1021 jobRoleName?: string | undefined;
1022 executionRoleName?: string | undefined;
1023 }) {
1024 if (platform.repositoryName) {
1025 yield* ecr
1026 .deleteRepository({
1027 repositoryName: platform.repositoryName,
1028 force: true,
1029 })
1030 .pipe(
1031 Effect.catchTag("RepositoryNotFoundException", () => Effect.void),
1032 );
1033 }
1034 for (const roleName of [
1035 platform.jobRoleName,
1036 platform.executionRoleName,
1037 ]) {
1038 if (!roleName) continue;
1039 yield* iam.listRolePolicies.items({ RoleName: roleName }).pipe(
1040 Stream.mapEffect((policyName) =>
1041 iam
1042 .deleteRolePolicy({
1043 RoleName: roleName,
1044 PolicyName: policyName,
1045 })
1046 .pipe(
1047 Effect.catchTag("NoSuchEntityException", () => Effect.void),
1048 ),
1049 ),
1050 Stream.runDrain,
1051 Effect.catchTag("NoSuchEntityException", () => Effect.void),
1052 );
1053 yield* iam.listAttachedRolePolicies
1054 .items({ RoleName: roleName })
1055 .pipe(
1056 Stream.mapEffect((policy) =>
1057 iam
1058 .detachRolePolicy({
1059 RoleName: roleName,
1060 PolicyArn: policy.PolicyArn!,
1061 })
1062 .pipe(
1063 Effect.catchTag("NoSuchEntityException", () => Effect.void),
1064 ),
1065 ),
1066 Stream.runDrain,
1067 Effect.catchTag("NoSuchEntityException", () => Effect.void),
1068 );
1069 yield* iam
1070 .deleteRole({ RoleName: roleName })
1071 .pipe(Effect.catchTag("NoSuchEntityException", () => Effect.void));
1072 }
1073 });
1074
1075 return {
1076 stables: [
1077 "jobDefinitionName",
1078 "repositoryName",
1079 "repositoryUri",
1080 "jobRoleArn",
1081 "jobRoleName",
1082 "executionRoleName",
1083 ],
1084 diff: Effect.fn(function* ({ id, olds, news }) {
1085 if (!isResolved(news)) return;
1086 if (
1087 (yield* toName(id, olds ?? {})) !== (yield* toName(id, news ?? {}))
1088 ) {
1089 return { action: "replace" } as const;
1090 }
1091 }),
1092 read: Effect.fn(function* ({ id, olds, output }) {
1093 const name =
1094 output?.jobDefinitionName ?? (yield* toName(id, olds ?? {}));
1095 const latest = yield* latestRevision(name);
1096 if (!latest?.jobDefinitionArn) return undefined;
1097 // Managed platform resources (repo, roles) are stable identifiers —
1098 // carry them through from the cached output.
1099 return toAttributes(latest, observedTagsOf(latest), {
1100 ...emptyPlatformAttributes,
1101 ...(output
1102 ? {
1103 imageUri: output.imageUri,
1104 repositoryName: output.repositoryName,
1105 repositoryUri: output.repositoryUri,
1106 jobRoleArn: output.jobRoleArn,
1107 jobRoleName: output.jobRoleName,
1108 executionRoleName: output.executionRoleName,
1109 codeHash: output.codeHash,
1110 }
1111 : {}),
1112 });
1113 }),
1114 list: () =>
1115 Effect.gen(function* () {
1116 // Enumerate the latest ACTIVE revision of every family.
1117 const pages = yield* batch.describeJobDefinitions
1118 .pages({ status: "ACTIVE" })
1119 .pipe(Stream.runCollect);
1120 const latest = new Map<string, batch.JobDefinition>();
1121 for (const page of pages) {
1122 for (const d of page.jobDefinitions ?? []) {
1123 if (!d.jobDefinitionName || !d.jobDefinitionArn) continue;
1124 const prior = latest.get(d.jobDefinitionName);
1125 if (!prior || (prior.revision ?? 0) < (d.revision ?? 0)) {
1126 latest.set(d.jobDefinitionName, d);
1127 }
1128 }
1129 }
1130 return Array.from(latest.values()).map((d) =>
1131 toAttributes(d, observedTagsOf(d)),
1132 );
1133 }),
1134 reconcile: Effect.fn(function* ({
1135 id,
1136 news,
1137 bindings,
1138 output,
1139 session,
1140 }) {
1141 const { region } = yield* AWSEnvironment.current;
1142 const name = output?.jobDefinitionName ?? (yield* toName(id, news));
1143 const internalTags = yield* createInternalTags(id);
1144 const desiredTags = { ...internalTags, ...news.tags };
1145
1146 if (news.main !== undefined && news.image !== undefined) {
1147 return yield* Effect.fail(
1148 new JobDefinitionConfigError({
1149 message:
1150 "`main` (Effect-native form) and `image` (low-level form) are mutually exclusive",
1151 }),
1152 );
1153 }
1154 if (news.main === undefined && news.image === undefined) {
1155 return yield* Effect.fail(
1156 new JobDefinitionConfigError({
1157 message:
1158 "one of `main` (Effect-native form) or `image` (low-level form) is required",
1159 }),
1160 );
1161 }
1162 if (news.main === undefined && news.executionRoleArn === undefined) {
1163 return yield* Effect.fail(
1164 new JobDefinitionConfigError({
1165 message:
1166 "`executionRoleArn` is required with `image` (Fargate jobs pull the image and write logs through it)",
1167 }),
1168 );
1169 }
1170
1171 // Resolve the effective container configuration. The Effect-native
1172 // form first ensures its managed roles + repository, bundles the
1173 // program, and builds + pushes the image; the low-level form uses
1174 // the caller-supplied image and roles directly. Each ensure step is
1175 // independently idempotent.
1176 let platformAttributes = emptyPlatformAttributes;
1177 let effective: EffectiveContainer;
1178 if (news.main !== undefined) {
1179 const jobRoleName =
1180 output?.jobRoleName ?? (yield* createRoleName(id, "job-role"));
1181 const executionRoleName =
1182 output?.executionRoleName ??
1183 (yield* createRoleName(id, "execution-role"));
1184 const repositoryName =
1185 output?.repositoryName ?? (yield* createRepositoryName(id));
1186 const jobPolicyName = yield* createPhysicalName({
1187 id: `${id}-job-policy`,
1188 maxLength: 128,
1189 });
1190
1191 const jobRoleArn = yield* ensureRole({
1192 id,
1193 roleName: jobRoleName,
1194 managedPolicyArns: news.jobRoleManagedPolicyArns,
1195 });
1196 const executionRoleArn = yield* ensureRole({
1197 id,
1198 roleName: executionRoleName,
1199 managedPolicyArns: [
1200 "arn:aws:iam::aws:policy/service-role/AmazonECSTaskExecutionRolePolicy",
1201 ],
1202 });
1203
1204 const bindingEnv = yield* attachBindings({
1205 roleName: jobRoleName,
1206 policyName: jobPolicyName,
1207 bindings,
1208 });
1209
1210 const { repositoryUri } =
1211 output?.repositoryUri && output.repositoryName === repositoryName
1212 ? { repositoryUri: output.repositoryUri }
1213 : yield* ensureRepository({
1214 repositoryName,
1215 tags: desiredTags,
1216 });
1217
1218 const { files, hash } = yield* bundleProgram(news);
1219 const imageUri = yield* buildAndPushImage({
1220 id,
1221 repositoryUri,
1222 hash,
1223 files,
1224 props: news,
1225 });
1226
1227 platformAttributes = {
1228 imageUri,
1229 repositoryName,
1230 repositoryUri,
1231 jobRoleArn,
1232 jobRoleName,
1233 executionRoleName,
1234 codeHash: hash,
1235 };
1236 effective = {
1237 image: imageUri,
1238 command: undefined,
1239 vcpus: String(news.vcpus ?? 0.25),
1240 memory: String(news.memory ?? 512),
1241 environment: {
1242 // Bindings + alchemy runtime env first (JSON-encoded), then
1243 // the caller's `env`, then plain `environment` wins last.
1244 ...Object.fromEntries(
1245 Object.entries({
1246 ...bindingEnv,
1247 ...alchemyEnv,
1248 // Fargate does not inject AWS_REGION — the bundled
1249 // program's `Region.fromEnv()` needs it.
1250 AWS_REGION: region,
1251 ...news.env,
1252 }).map(([key, value]) => [
1253 key,
1254 typeof value === "string" ? value : JSON.stringify(value),
1255 ]),
1256 ),
1257 ...news.environment,
1258 },
1259 jobRoleArn,
1260 executionRoleArn,
1261 assignPublicIp: news.assignPublicIp ?? "ENABLED",
1262 parameters: news.parameters,
1263 retryAttempts: news.retryAttempts,
1264 timeoutSeconds: toWireSeconds(news.timeout),
1265 propagateTags: news.propagateTags,
1266 };
1267 } else {
1268 // Low-level form. If a previous deploy used the Effect-native
1269 // form, its managed repo/roles are no longer referenced — reap
1270 // them so nothing leaks.
1271 if (
1272 output?.repositoryName ||
1273 output?.jobRoleName ||
1274 output?.executionRoleName
1275 ) {
1276 yield* cleanupPlatformResources(output);
1277 }
1278 effective = {
1279 image: news.image!,
1280 command: news.command,
1281 vcpus: String(news.vcpus ?? 0.25),
1282 memory: String(news.memory ?? 512),
1283 environment: { ...news.environment },
1284 jobRoleArn: news.jobRoleArn,
1285 executionRoleArn: news.executionRoleArn!,
1286 assignPublicIp: news.assignPublicIp ?? "ENABLED",
1287 parameters: news.parameters,
1288 retryAttempts: news.retryAttempts,
1289 timeoutSeconds: toWireSeconds(news.timeout),
1290 propagateTags: news.propagateTags,
1291 };
1292 }
1293
1294 // Observe — the latest ACTIVE revision is the live state.
1295 const latest = yield* latestRevision(name);
1296
1297 // Ensure/Sync — register a new revision only when managed content
1298 // differs (revisions are immutable; registration IS the update).
1299 if (
1300 !latest ||
1301 observedFingerprint(latest) !== desiredFingerprint(effective, news)
1302 ) {
1303 const platformCapabilities = news.platformCapabilities ?? [
1304 "FARGATE",
1305 ];
1306 const registered = yield* batch.registerJobDefinition({
1307 jobDefinitionName: name,
1308 type: "container",
1309 platformCapabilities,
1310 containerProperties: {
1311 image: effective.image,
1312 command: effective.command,
1313 jobRoleArn: effective.jobRoleArn,
1314 executionRoleArn: effective.executionRoleArn,
1315 resourceRequirements: [
1316 { type: "VCPU", value: effective.vcpus },
1317 { type: "MEMORY", value: effective.memory },
1318 ],
1319 environment: Object.entries(effective.environment).map(
1320 ([key, value]) => ({ name: key, value }),
1321 ),
1322 networkConfiguration: platformCapabilities.includes("FARGATE")
1323 ? { assignPublicIp: effective.assignPublicIp }
1324 : undefined,
1325 },
1326 parameters: effective.parameters,
1327 retryStrategy:
1328 effective.retryAttempts !== undefined
1329 ? { attempts: effective.retryAttempts }
1330 : undefined,
1331 timeout:
1332 effective.timeoutSeconds !== undefined
1333 ? { attemptDurationSeconds: effective.timeoutSeconds }
1334 : undefined,
1335 propagateTags: effective.propagateTags,
1336 tags: desiredTags,
1337 });
1338 yield* session.note(registered.jobDefinitionArn);
1339 return {
1340 jobDefinitionName: registered.jobDefinitionName,
1341 jobDefinitionArn: registered.jobDefinitionArn as JobDefinitionArn,
1342 revision: registered.revision,
1343 status: "ACTIVE",
1344 tags: desiredTags,
1345 executionRoleArn: effective.executionRoleArn,
1346 ...platformAttributes,
1347 };
1348 }
1349
1350 // Content unchanged — sync tags in place on the revision ARN.
1351 const { upsert, removed } = diffTags(
1352 observedTagsOf(latest),
1353 desiredTags,
1354 );
1355 if (upsert.length > 0) {
1356 yield* batch.tagResource({
1357 resourceArn: latest.jobDefinitionArn!,
1358 tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
1359 });
1360 }
1361 if (removed.length > 0) {
1362 yield* batch.untagResource({
1363 resourceArn: latest.jobDefinitionArn!,
1364 tagKeys: removed,
1365 });
1366 }
1367
1368 yield* session.note(latest.jobDefinitionArn!);
1369 return toAttributes(latest, desiredTags, platformAttributes);
1370 }),
1371 delete: Effect.fn(function* ({ output }) {
1372 // Deregister EVERY active revision of the family. Deregistration is
1373 // idempotent (succeeds when the revision is already gone).
1374 const revisions = yield* activeRevisions(output.jobDefinitionName);
1375 yield* Effect.forEach(
1376 revisions,
1377 (d) =>
1378 batch.deregisterJobDefinition({
1379 jobDefinition: `${d.jobDefinitionName}:${d.revision}`,
1380 }),
1381 { concurrency: 4 },
1382 );
1383 yield* waitUntilNoActiveBatchRevisions(output.jobDefinitionName);
1384
1385 // A submitted Batch job causes AWSServiceRoleForBatch to create a
1386 // backing ECS task-definition revision with the exact same family.
1387 // Batch deregistration does not reliably reclaim it, so delete it
1388 // explicitly only after the Batch family is observed inactive.
1389 yield* reapBackingTaskDefinitions(output.jobDefinitionName);
1390
1391 // Reap the managed platform resources (Effect-native form only).
1392 yield* cleanupPlatformResources(output);
1393
1394 // Reap this family's log streams from the shared, service-managed
1395 // `/aws/batch/job` group. The awslogs driver names streams
1396 // `{jobDefinitionName}/default/{taskId}`, so the family prefix is
1397 // exact; the group itself is account-level AWS infrastructure
1398 // (recreated by the Batch service on the next job) and is left
1399 // alone.
1400 const logGroupName = "/aws/batch/job";
1401 const streams = yield* logs.describeLogStreams
1402 .pages({
1403 logGroupName,
1404 logStreamNamePrefix: `${output.jobDefinitionName}/`,
1405 })
1406 .pipe(
1407 Stream.runCollect,
1408 Effect.map((pages) =>
1409 Array.from(pages).flatMap((p) => p.logStreams ?? []),
1410 ),
1411 Effect.catchTag("ResourceNotFoundException", () =>
1412 Effect.succeed([]),
1413 ),
1414 );
1415 yield* Effect.forEach(
1416 streams,
1417 (s) =>
1418 s.logStreamName === undefined
1419 ? Effect.void
1420 : logs
1421 .deleteLogStream({
1422 logGroupName,
1423 logStreamName: s.logStreamName,
1424 })
1425 .pipe(
1426 Effect.catchTag(
1427 "ResourceNotFoundException",
1428 () => Effect.void,
1429 ),
1430 ),
1431 { concurrency: 4 },
1432 );
1433 }),
1434 };
1435 }),
1436 );