1import * as AWS from "@/AWS";
2import { Role } from "@/AWS/IAM/Role.ts";
3import { Workflow } from "@/AWS/MWAAServerless";
4import { Bucket } from "@/AWS/S3/Bucket.ts";
5import * as Output from "@/Output";
6import * as Test from "@/Test/Alchemy";
7import * as logs from "@distilled.cloud/aws/cloudwatch-logs";
8import * as mwaa from "@distilled.cloud/aws/mwaa-serverless";
9import * as s3 from "@distilled.cloud/aws/s3";
10import * as sts from "@distilled.cloud/aws/sts";
11import { expect } from "alchemy-test";
12import * as Data from "effect/Data";
13import * as Effect from "effect/Effect";
14import * as Schedule from "effect/Schedule";
15
16const { test } = Test.make({ providers: AWS.providers() });
17
18// Typed-error probe: prove the distilled error union carries the not-found
19// tag this provider's read/delete paths depend on, at near-zero cost.
20test.provider(
21 "getWorkflow on a nonexistent workflow fails with ResourceNotFoundException",
22 () =>
23 Effect.gen(function* () {
24 const identity = yield* sts.getCallerIdentity({});
25 const region = process.env.AWS_REGION ?? "us-west-2";
26 // Workflow ARNs end in `-{10 alnum chars}` (a service-assigned id
27 // suffix) — the API rejects anything else with a ValidationException
28 // before the lookup even runs.
29 const bogusArn = `arn:aws:airflow-serverless:${region}:${identity.Account}:workflow/alchemy-nonexistent-probe-0123456789`;
30 const error = yield* Effect.flip(
31 mwaa.getWorkflow({ WorkflowArn: bogusArn }),
32 );
33 expect(error._tag).toBe("ResourceNotFoundException");
34 }),
35 { tags: ["provider:aws", "provider:aws:mwaaserverless", "live"] },
36);
37
38// A deterministic, checked-in workflow definition (YAML DAG using a
39// supported AWS operator). The bucket name is interpolated because the
40// definition must reference a real bucket the execution role can list.
41const workflowDefinition = (bucket: string) =>
42 [
43 "alchemy-mwaa-serverless-test:",
44 " default_args:",
45 " owner: alchemy",
46 " schedule: null",
47 " tasks:",
48 " list_definitions:",
49 " operator: airflow.providers.amazon.aws.operators.s3.S3ListOperator",
50 ` bucket: ${bucket}`,
51 " prefix: workflows/",
52 "",
53 ].join("\n");
54
55const DEFINITION_KEY = "workflows/test.yaml";
56
57// The definition bucket + execution role, shared by every deploy phase of
58// the gated lifecycle (same logical IDs across incremental deploys).
59const infrastructure = Effect.gen(function* () {
60 const bucket = yield* Bucket("Definitions", { forceDestroy: true });
61 const role = yield* Role("WorkflowRole", {
62 assumeRolePolicyDocument: {
63 Version: "2012-10-17",
64 Statement: [
65 {
66 Effect: "Allow",
67 Principal: { Service: "airflow-serverless.amazonaws.com" },
68 Action: ["sts:AssumeRole"],
69 },
70 ],
71 },
72 inlinePolicies: {
73 definitions: {
74 Version: "2012-10-17",
75 Statement: [
76 {
77 Effect: "Allow",
78 Action: ["s3:GetObject*", "s3:GetBucket*", "s3:List*"],
79 Resource: [
80 bucket.bucketArn,
81 Output.interpolate`${bucket.bucketArn}/*`,
82 ],
83 },
84 ],
85 },
86 },
87 });
88 return { bucket, role };
89});
90
91const workflowPhase = (description: string, name?: string) =>
92 Effect.gen(function* () {
93 const { bucket, role } = yield* infrastructure;
94 const workflow = yield* Workflow("Etl", {
95 name,
96 definitionS3Location: {
97 bucket: bucket.bucketName,
98 objectKey: DEFINITION_KEY,
99 },
100 roleArn: role.roleArn,
101 description,
102 tags: { fixture: "mwaa-serverless-workflow" },
103 });
104 return { bucket, role, workflow };
105 });
106
107const findWorkflow = (workflowArn: string) =>
108 mwaa
109 .getWorkflow({ WorkflowArn: workflowArn })
110 .pipe(
111 Effect.catchTag("ResourceNotFoundException", () =>
112 Effect.succeed(undefined),
113 ),
114 );
115
116class WorkflowStillExists extends Data.TaggedError("WorkflowStillExists")<{
117 readonly workflowArn: string;
118}> {}
119
120class LogGroupStillExists extends Data.TaggedError("LogGroupStillExists")<{
121 readonly logGroupName: string;
122}> {}
123
124// MWAA Serverless auto-creates `/aws/mwaa-serverless/{arn resource id}/`
125// per workflow; the provider's delete reaps it. Derive the exact name from
126// the workflow ARN (the resource id embeds the service-assigned suffix).
127const logGroupNameFor = (workflowArn: string) =>
128 `/aws/mwaa-serverless/${workflowArn.split(":workflow/")[1]}/`;
129
130const assertLogGroupDeleted = (workflowArn: string) =>
131 Effect.gen(function* () {
132 const logGroupName = logGroupNameFor(workflowArn);
133 const found = yield* logs.describeLogGroups({
134 logGroupNamePrefix: logGroupName,
135 });
136 const exists = (found.logGroups ?? []).some(
137 (group) => group.logGroupName === logGroupName,
138 );
139 if (exists) {
140 return yield* Effect.fail(new LogGroupStillExists({ logGroupName }));
141 }
142 }).pipe(
143 Effect.retry({
144 while: (e): boolean => e._tag === "LogGroupStillExists",
145 schedule: Schedule.max([Schedule.fixed("5 seconds"), Schedule.recurs(5)]),
146 }),
147 );
148
149// Crash-resilience sweep: if a run dies between createWorkflow and the
150// provider delete, the auto-created log group (whose name embeds a
151// service-assigned random suffix we cannot predict up front) would orphan.
152// Reap every log group carrying this test's deterministic workflow-name
153// prefixes so an interrupted run is cleaned up by the next one.
154const TEST_LOG_GROUP_PREFIXES = [
155 // engine-generated physical name for the `Etl` logical ID
156 "/aws/mwaa-serverless/create-update-and-delete-a-workflow-Etl-",
157 // explicit rename used by the replacement phase
158 "/aws/mwaa-serverless/alchemy-mwaa-serverless-renamed-",
159];
160
161const reapTestLogGroups = Effect.gen(function* () {
162 for (const prefix of TEST_LOG_GROUP_PREFIXES) {
163 const found = yield* logs.describeLogGroups({
164 logGroupNamePrefix: prefix,
165 });
166 for (const group of found.logGroups ?? []) {
167 if (group.logGroupName !== undefined) {
168 yield* logs
169 .deleteLogGroup({ logGroupName: group.logGroupName })
170 .pipe(
171 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
172 );
173 }
174 }
175 }
176});
177
178const assertWorkflowDeleted = (workflowArn: string) =>
179 findWorkflow(workflowArn).pipe(
180 Effect.flatMap((workflow) =>
181 workflow === undefined || workflow.WorkflowStatus === "DELETING"
182 ? Effect.void
183 : Effect.fail(new WorkflowStillExists({ workflowArn })),
184 ),
185 Effect.retry({
186 while: (e) => e._tag === "WorkflowStillExists",
187 schedule: Schedule.max([
188 Schedule.fixed("5 seconds"),
189 Schedule.recurs(10),
190 ]),
191 }),
192 );
193
194test.provider(
195 "create, update, and delete a workflow",
196 (stack) =>
197 Effect.gen(function* () {
198 yield* stack.destroy();
199
200 // Phase 1 — the definition object must exist in S3 before the
201 // workflow can be created, so deploy the bucket + role first and
202 // upload the definition out-of-band.
203 const infra = yield* stack.deploy(infrastructure);
204 yield* s3.putObject({
205 Bucket: infra.bucket.bucketName,
206 Key: DEFINITION_KEY,
207 Body: new TextEncoder().encode(
208 workflowDefinition(infra.bucket.bucketName),
209 ),
210 ContentType: "application/yaml",
211 });
212
213 // Phase 2 — create the workflow.
214 const { workflow } = yield* stack.deploy(
215 workflowPhase("alchemy mwaa-serverless test workflow"),
216 );
217
218 expect(workflow.workflowArn).toContain(":workflow/");
219 expect(workflow.name).toBeDefined();
220
221 // Out-of-band verification via distilled.
222 const observed = yield* mwaa.getWorkflow({
223 WorkflowArn: workflow.workflowArn,
224 });
225 expect(observed.Name).toBe(workflow.name);
226 expect(observed.RoleArn).toBeDefined();
227 const tags = yield* mwaa.listTagsForResource({
228 ResourceArn: workflow.workflowArn,
229 });
230 expect(tags.Tags?.fixture).toBe("mwaa-serverless-workflow");
231
232 // Phase 3 — update mutable config (description) in place; the
233 // workflow ARN must be stable across the update.
234 const updated = yield* stack.deploy(
235 workflowPhase("alchemy mwaa-serverless test workflow (updated)"),
236 );
237 expect(updated.workflow.workflowArn).toBe(workflow.workflowArn);
238 const reobserved = yield* mwaa.getWorkflow({
239 WorkflowArn: workflow.workflowArn,
240 });
241 expect(reobserved.Description).toBe(
242 "alchemy mwaa-serverless test workflow (updated)",
243 );
244
245 // Phase 4 — changing the name is a replacement: a new workflow (new
246 // ARN) is created and the old one deleted.
247 const replaced = yield* stack.deploy(
248 workflowPhase(
249 "alchemy mwaa-serverless test workflow (replaced)",
250 "alchemy-mwaa-serverless-renamed",
251 ),
252 );
253 expect(replaced.workflow.workflowArn).not.toBe(workflow.workflowArn);
254 expect(replaced.workflow.name).toBe("alchemy-mwaa-serverless-renamed");
255 yield* assertWorkflowDeleted(workflow.workflowArn);
256 // The replaced-away workflow's auto-created log group must be reaped
257 // along with it, or every rename leaks a log group.
258 yield* assertLogGroupDeleted(workflow.workflowArn);
259
260 yield* stack.destroy();
261 yield* assertWorkflowDeleted(replaced.workflow.workflowArn);
262 yield* assertLogGroupDeleted(replaced.workflow.workflowArn);
263 }).pipe(Effect.ensuring(Effect.orDie(reapTestLogGroups))),
264 {
265 tags: [
266 "provider:aws",
267 "provider:aws:iam",
268 "provider:aws:mwaaserverless",
269 "provider:aws:s3",
270 "live",
271 ],
272 timeout: 600_000,
273 },
274);