1import * as logs from "@distilled.cloud/aws/cloudwatch-logs";
2import * as mwaa from "@distilled.cloud/aws/mwaa-serverless";
3import * as Effect from "effect/Effect";
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
14export interface WorkflowDefinitionS3Location {
15 /**
16 * Name of the S3 bucket that holds the workflow definition YAML file.
17 */
18 bucket: string;
19 /**
20 * Key of the workflow definition YAML object within the bucket.
21 */
22 objectKey: string;
23 /**
24 * Specific S3 object version of the definition to use. Omit to use the
25 * latest version.
26 */
27 versionId?: string;
28}
29
30export interface WorkflowEncryptionConfiguration {
31 /**
32 * How workflow data is encrypted — `"AWS_MANAGED_KEY"` (the default) or
33 * `"CUSTOMER_MANAGED_KEY"`.
34 *
35 * Encryption cannot be changed in place — changing it replaces the
36 * workflow.
37 * @default "AWS_MANAGED_KEY"
38 */
39 type: mwaa.EncryptionType;
40 /**
41 * ID or ARN of the customer managed KMS key to encrypt workflow data
42 * with. Required when `type` is `"CUSTOMER_MANAGED_KEY"`.
43 */
44 kmsKeyId?: string;
45}
46
47export interface WorkflowLoggingConfiguration {
48 /**
49 * Name of the CloudWatch log group that receives task logs for the
50 * workflow's runs.
51 */
52 logGroupName: string;
53}
54
55export interface WorkflowNetworkConfiguration {
56 /**
57 * Security group IDs applied to tasks that access resources inside your
58 * VPC.
59 */
60 securityGroupIds?: string[];
61 /**
62 * Subnet IDs the workflow's tasks run in when accessing resources inside
63 * your VPC.
64 */
65 subnetIds?: string[];
66}
67
68export interface WorkflowProps {
69 /**
70 * Name of the workflow. If omitted, a deterministic physical name is
71 * generated from the app, stage, and logical ID.
72 *
73 * Changing the name replaces the workflow.
74 * @default ${app}-${stage}-${id}
75 */
76 name?: string;
77 /**
78 * S3 location of the workflow definition — a YAML file that defines the
79 * DAG structure using supported AWS operators.
80 *
81 * Updating the definition creates a new workflow version and disables
82 * scheduling on all previous versions.
83 */
84 definitionS3Location: WorkflowDefinitionS3Location;
85 /**
86 * ARN of the IAM execution role that Amazon MWAA Serverless assumes to
87 * run the workflow's tasks. The role's trust policy must allow the
88 * `airflow-serverless.amazonaws.com` service principal to assume it.
89 */
90 roleArn: string;
91 /**
92 * Human-readable description of the workflow.
93 */
94 description?: string;
95 /**
96 * Encryption configuration for workflow data. Omit to use an AWS managed
97 * key. Changing encryption replaces the workflow (it cannot be updated
98 * in place).
99 * @default AWS managed key
100 */
101 encryptionConfiguration?: WorkflowEncryptionConfiguration;
102 /**
103 * CloudWatch logging configuration for the workflow's task logs.
104 */
105 loggingConfiguration?: WorkflowLoggingConfiguration;
106 /**
107 * Version of the workflow definition engine.
108 * @default 1
109 */
110 engineVersion?: mwaa.EngineVersion;
111 /**
112 * Network configuration for tasks that access resources inside your VPC.
113 */
114 networkConfiguration?: WorkflowNetworkConfiguration;
115 /**
116 * How workflow runs are triggered (for example on a schedule defined in
117 * the workflow definition, or only on demand).
118 */
119 triggerMode?: string;
120 /**
121 * Tags to apply to the workflow. Merged with internal Alchemy tags.
122 */
123 tags?: Record<string, string>;
124}
125
126export interface Workflow extends Resource<
127 "AWS.MWAAServerless.Workflow",
128 WorkflowProps,
129 {
130 /** Name of the workflow. */
131 name: string;
132 /** ARN of the workflow. */
133 workflowArn: string;
134 /** Latest workflow version (a new version is published on each update). */
135 workflowVersion: string | undefined;
136 /** Current lifecycle status of the workflow. */
137 workflowStatus: mwaa.WorkflowStatus | undefined;
138 /** ARN of the IAM role the workflow's tasks assume. */
139 roleArn: string | undefined;
140 /** How runs are triggered (e.g. scheduled or on-demand). */
141 triggerMode: string | undefined;
142 },
143 never,
144 Providers
145> {}
146
147/**
148 * An Amazon Managed Workflows for Apache Airflow **Serverless** workflow —
149 * a serverless Airflow DAG defined by a YAML file in S3 and executed by
150 * AWS-managed, multi-tenant Airflow infrastructure without provisioning an
151 * environment.
152 *
153 * Each update to the definition or configuration creates a new workflow
154 * version; MWAA Serverless keeps only the latest version actively
155 * scheduled.
156 * ### Creating a Workflow
157 * **Example:** Basic Workflow
158 * ```typescript
159 * import * as MWAAServerless from "alchemy/AWS/MWAAServerless";
160 * import * as IAM from "alchemy/AWS/IAM";
161 *
162 * const role = yield* IAM.Role("WorkflowRole", {
163 * assumeRolePolicyDocument: {
164 * Version: "2012-10-17",
165 * Statement: [{
166 * Effect: "Allow",
167 * Principal: { Service: "airflow-serverless.amazonaws.com" },
168 * Action: ["sts:AssumeRole"],
169 * }],
170 * },
171 * });
172 *
173 * const workflow = yield* MWAAServerless.Workflow("Etl", {
174 * definitionS3Location: {
175 * bucket: "my-dag-bucket",
176 * objectKey: "workflows/etl.yaml",
177 * },
178 * roleArn: role.roleArn,
179 * });
180 * ```
181 *
182 * **Example:** Workflow with Logging and Tags
183 * ```typescript
184 * const workflow = yield* MWAAServerless.Workflow("Etl", {
185 * definitionS3Location: {
186 * bucket: "my-dag-bucket",
187 * objectKey: "workflows/etl.yaml",
188 * },
189 * roleArn: role.roleArn,
190 * description: "nightly ETL",
191 * loggingConfiguration: { logGroupName: "/mwaa-serverless/etl" },
192 * tags: { team: "data" },
193 * });
194 * ```
195 *
196 * @resource
197 */
198export const Workflow = Resource<Workflow>("AWS.MWAAServerless.Workflow");
199
200/**
201 * Bounded retry for `createWorkflow` while a freshly created IAM execution
202 * role propagates — the service validates that it can assume the role at
203 * create time, which surfaces as a ValidationException/AccessDeniedException
204 * mentioning the role for the first seconds of the role's life.
205 *
206 * Explicitly annotated so the conditional `Retry.Return` type never leaks
207 * into declaration emit (it would widen `AWS.providers()` for consumers).
208 */
209const retryWhileRolePropagates = <A, R>(
210 self: Effect.Effect<A, mwaa.CreateWorkflowError, R>,
211): Effect.Effect<A, mwaa.CreateWorkflowError, R> =>
212 Effect.retry(self, {
213 while: (e) =>
214 (e._tag === "ValidationException" ||
215 e._tag === "AccessDeniedException") &&
216 /role/i.test(e.message ?? ""),
217 schedule: Schedule.max([Schedule.fixed("3 seconds"), Schedule.recurs(8)]),
218 });
219
220export const WorkflowProvider = () =>
221 Provider.effect(
222 Workflow,
223 Effect.gen(function* () {
224 const createName = Effect.fn(function* (
225 id: string,
226 props: Pick<WorkflowProps, "name">,
227 ) {
228 return props.name ?? (yield* createPhysicalName({ id, maxLength: 64 }));
229 });
230
231 const observeByArn = (arn: string) =>
232 mwaa
233 .getWorkflow({ WorkflowArn: arn })
234 .pipe(
235 Effect.catchTag("ResourceNotFoundException", () =>
236 Effect.succeed(undefined),
237 ),
238 );
239
240 const findArnByName = (name: string) =>
241 mwaa.listWorkflows.items({}).pipe(
242 Stream.runCollect,
243 Effect.map(
244 (items) =>
245 Array.from(items).find((w) => w.Name === name)?.WorkflowArn,
246 ),
247 );
248
249 const observe = Effect.fn(function* (
250 name: string,
251 arnHint: string | undefined,
252 ) {
253 if (arnHint !== undefined) {
254 const found = yield* observeByArn(arnHint);
255 if (found !== undefined) return found;
256 }
257 const arn = yield* findArnByName(name);
258 return arn === undefined ? undefined : yield* observeByArn(arn);
259 });
260
261 const observedTags = (arn: string) =>
262 mwaa.listTagsForResource({ ResourceArn: arn }).pipe(
263 Effect.map((r) => {
264 const tags: Record<string, string> = {};
265 for (const [key, value] of Object.entries(r.Tags ?? {})) {
266 if (value !== undefined) tags[key] = value;
267 }
268 return tags;
269 }),
270 Effect.catchTag("ResourceNotFoundException", () =>
271 Effect.succeed({} as Record<string, string>),
272 ),
273 );
274
275 const toDefinitionS3Location = (
276 location: WorkflowDefinitionS3Location,
277 ): mwaa.DefinitionS3Location => ({
278 Bucket: location.bucket,
279 ObjectKey: location.objectKey,
280 VersionId: location.versionId,
281 });
282
283 const sameDefinition = (
284 live: mwaa.DefinitionS3Location | undefined,
285 desired: WorkflowDefinitionS3Location,
286 ) =>
287 live !== undefined &&
288 live.Bucket === desired.bucket &&
289 live.ObjectKey === desired.objectKey &&
290 (desired.versionId === undefined ||
291 live.VersionId === desired.versionId);
292
293 const sameNetwork = (
294 live: mwaa.NetworkConfiguration | undefined,
295 desired: WorkflowNetworkConfiguration,
296 ) =>
297 JSON.stringify({
298 securityGroupIds: live?.SecurityGroupIds ?? [],
299 subnetIds: live?.SubnetIds ?? [],
300 }) ===
301 JSON.stringify({
302 securityGroupIds: desired.securityGroupIds ?? [],
303 subnetIds: desired.subnetIds ?? [],
304 });
305
306 const toAttributes = (live: mwaa.GetWorkflowResponse, name: string) => ({
307 name: live.Name ?? name,
308 workflowArn: live.WorkflowArn,
309 workflowVersion: live.WorkflowVersion,
310 workflowStatus: live.WorkflowStatus,
311 roleArn: live.RoleArn,
312 triggerMode: live.TriggerMode,
313 });
314
315 return Workflow.Provider.of({
316 stables: ["name", "workflowArn"],
317 list: () =>
318 Effect.gen(function* () {
319 const items = yield* mwaa.listWorkflows
320 .items({})
321 .pipe(Stream.runCollect);
322 const workflows: {
323 name: string;
324 workflowArn: string;
325 workflowVersion: string | undefined;
326 workflowStatus: mwaa.WorkflowStatus | undefined;
327 roleArn: string | undefined;
328 triggerMode: string | undefined;
329 }[] = [];
330 for (const item of Array.from(items)) {
331 if (item.WorkflowArn === undefined || item.Name === undefined) {
332 continue;
333 }
334 // tolerate delete races between list and get
335 const live = yield* observeByArn(item.WorkflowArn);
336 if (live !== undefined && live.Name !== undefined) {
337 workflows.push(toAttributes(live, live.Name));
338 }
339 }
340 return workflows;
341 }),
342 read: Effect.fn(function* ({ id, olds, output }) {
343 const name = output?.name ?? (yield* createName(id, olds ?? {}));
344 const live = yield* observe(name, output?.workflowArn);
345 if (live === undefined || live.Name === undefined) {
346 return undefined;
347 }
348 const attrs = toAttributes(live, name);
349 const tags = yield* observedTags(live.WorkflowArn);
350 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
351 }),
352 diff: Effect.fn(function* ({ id, news, olds }) {
353 if (!isResolved(news)) return undefined;
354 const oldName = yield* createName(id, olds ?? {});
355 const newName = yield* createName(id, news ?? {});
356 if (oldName !== newName) {
357 return { action: "replace" } as const;
358 }
359 // Encryption cannot be updated in place — UpdateWorkflow does not
360 // accept EncryptionConfiguration.
361 const oldEncryption = olds?.encryptionConfiguration;
362 const newEncryption = news.encryptionConfiguration;
363 if (
364 (oldEncryption?.type ?? "AWS_MANAGED_KEY") !==
365 (newEncryption?.type ?? "AWS_MANAGED_KEY") ||
366 oldEncryption?.kmsKeyId !== newEncryption?.kmsKeyId
367 ) {
368 return { action: "replace" } as const;
369 }
370 // fall through: engine default update logic for mutable fields
371 }),
372 reconcile: Effect.fn(function* ({ id, news, output, session }) {
373 const name = output?.name ?? (yield* createName(id, news));
374 const internalTags = yield* createInternalTags(id);
375 const desiredTags = { ...news.tags, ...internalTags };
376
377 // 1. OBSERVE — cloud state is authoritative; output.workflowArn is
378 // only a cache of the identifier.
379 let live = yield* observe(name, output?.workflowArn);
380
381 // 2. ENSURE — create when missing. A ConflictException means a
382 // concurrent create (or a previous run's create whose state
383 // persistence failed) raced us: converge on the workflow with
384 // OUR name if one now exists, otherwise propagate.
385 if (live === undefined) {
386 live = yield* mwaa
387 .createWorkflow({
388 Name: name,
389 DefinitionS3Location: toDefinitionS3Location(
390 news.definitionS3Location,
391 ),
392 RoleArn: news.roleArn,
393 Description: news.description,
394 EncryptionConfiguration:
395 news.encryptionConfiguration === undefined
396 ? undefined
397 : {
398 Type: news.encryptionConfiguration.type,
399 KmsKeyId: news.encryptionConfiguration.kmsKeyId,
400 },
401 LoggingConfiguration:
402 news.loggingConfiguration === undefined
403 ? undefined
404 : {
405 LogGroupName: news.loggingConfiguration.logGroupName,
406 },
407 EngineVersion: news.engineVersion,
408 NetworkConfiguration:
409 news.networkConfiguration === undefined
410 ? undefined
411 : {
412 SecurityGroupIds:
413 news.networkConfiguration.securityGroupIds,
414 SubnetIds: news.networkConfiguration.subnetIds,
415 },
416 TriggerMode: news.triggerMode,
417 Tags: desiredTags,
418 })
419 .pipe(
420 retryWhileRolePropagates,
421 Effect.flatMap((created) => observeByArn(created.WorkflowArn)),
422 Effect.catchTag("ConflictException", (error) =>
423 observe(name, undefined).pipe(
424 Effect.flatMap((existing) =>
425 existing === undefined
426 ? Effect.fail(error)
427 : Effect.succeed(existing),
428 ),
429 ),
430 ),
431 );
432 }
433
434 const arn = live?.WorkflowArn ?? output?.workflowArn;
435
436 // 3. SYNC — diff each OBSERVED mutable aspect against the desired
437 // state; UpdateWorkflow requires the definition and role, so
438 // issue one full update when any managed aspect drifted (each
439 // update creates a new workflow version — skip on no-op).
440 if (live !== undefined && arn !== undefined) {
441 const drifted =
442 !sameDefinition(
443 live.DefinitionS3Location,
444 news.definitionS3Location,
445 ) ||
446 live.RoleArn !== news.roleArn ||
447 (news.description !== undefined &&
448 live.Description !== news.description) ||
449 (news.loggingConfiguration !== undefined &&
450 live.LoggingConfiguration?.LogGroupName !==
451 news.loggingConfiguration.logGroupName) ||
452 (news.engineVersion !== undefined &&
453 live.EngineVersion !== news.engineVersion) ||
454 (news.networkConfiguration !== undefined &&
455 !sameNetwork(
456 live.NetworkConfiguration,
457 news.networkConfiguration,
458 )) ||
459 (news.triggerMode !== undefined &&
460 live.TriggerMode !== news.triggerMode);
461 if (drifted) {
462 yield* mwaa.updateWorkflow({
463 WorkflowArn: arn,
464 DefinitionS3Location: toDefinitionS3Location(
465 news.definitionS3Location,
466 ),
467 RoleArn: news.roleArn,
468 Description: news.description,
469 LoggingConfiguration:
470 news.loggingConfiguration === undefined
471 ? undefined
472 : {
473 LogGroupName: news.loggingConfiguration.logGroupName,
474 },
475 EngineVersion: news.engineVersion,
476 NetworkConfiguration:
477 news.networkConfiguration === undefined
478 ? undefined
479 : {
480 SecurityGroupIds:
481 news.networkConfiguration.securityGroupIds,
482 SubnetIds: news.networkConfiguration.subnetIds,
483 },
484 TriggerMode: news.triggerMode,
485 });
486 live = (yield* observeByArn(arn)) ?? live;
487 }
488 }
489
490 // 3b. SYNC TAGS — diff against OBSERVED cloud tags so adoption
491 // converges (create-time tags only apply on first create).
492 if (arn !== undefined) {
493 const currentTags = yield* observedTags(arn);
494 const { upsert, removed } = diffTags(currentTags, desiredTags);
495 if (upsert.length > 0) {
496 yield* mwaa.tagResource({
497 ResourceArn: arn,
498 Tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
499 });
500 }
501 if (removed.length > 0) {
502 yield* mwaa.untagResource({
503 ResourceArn: arn,
504 TagKeys: removed,
505 });
506 }
507 }
508
509 yield* session.note(name);
510 return {
511 name: live?.Name ?? name,
512 workflowArn: arn!,
513 workflowVersion: live?.WorkflowVersion,
514 workflowStatus: live?.WorkflowStatus,
515 roleArn: live?.RoleArn ?? news.roleArn,
516 triggerMode: live?.TriggerMode ?? news.triggerMode,
517 };
518 }),
519 delete: Effect.fn(function* ({ output }) {
520 // Omitting WorkflowVersion deletes the workflow and all versions.
521 yield* mwaa
522 .deleteWorkflow({ WorkflowArn: output.workflowArn })
523 .pipe(
524 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
525 );
526
527 // MWAA Serverless auto-creates a per-workflow log group named
528 // `/aws/mwaa-serverless/{name}-{id}/` (the ARN's resource id,
529 // which includes the service-assigned 10-char suffix, plus a
530 // trailing slash) at CREATE time, and deleteWorkflow does NOT
531 // remove it — without this reap every deleted workflow (including
532 // the old workflow of a replacement) leaks an orphaned log group.
533 // A group with no streams or provably-quiescent ingestion (last
534 // ingestion > 2 minutes ago) has no pending flush and deletes in
535 // a single call, so routine deletes of never-run workflows stay
536 // fast. A group with recent ingestion — or one that does not
537 // exist yet — is re-reaped on a short bounded schedule
538 // (t=0s / 20s / 40s) to catch a late log flush from the
539 // workflow's final runs; each attempt is idempotent.
540 const workflowId = output.workflowArn.split(":workflow/")[1];
541 if (workflowId !== undefined) {
542 const logGroupName = `/aws/mwaa-serverless/${workflowId}/`;
543 const reapLogGroup = logs
544 .deleteLogGroup({ logGroupName })
545 .pipe(
546 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
547 );
548 const observed = yield* logs
549 .describeLogStreams({
550 logGroupName,
551 orderBy: "LastEventTime",
552 descending: true,
553 limit: 1,
554 })
555 .pipe(
556 Effect.map((r) => ({
557 exists: true,
558 lastIngestion: r.logStreams?.[0]?.lastIngestionTime,
559 })),
560 Effect.catchTag("ResourceNotFoundException", () =>
561 Effect.succeed({
562 exists: false,
563 lastIngestion: undefined as number | undefined,
564 }),
565 ),
566 );
567 const now = yield* Effect.sync(() => Date.now());
568 const quiescent =
569 observed.exists &&
570 (observed.lastIngestion === undefined ||
571 now - observed.lastIngestion > 120_000);
572 if (quiescent) {
573 yield* reapLogGroup;
574 } else {
575 yield* reapLogGroup.pipe(
576 Effect.repeat({
577 schedule: Schedule.spaced("20 seconds"),
578 times: 2,
579 }),
580 );
581 }
582 }
583 }),
584 });
585 }),
586 );