1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as logs from "@distilled.cloud/aws/cloudwatch-logs";
5import * as eventbridge from "@distilled.cloud/aws/eventbridge";
6import * as s3 from "@distilled.cloud/aws/s3";
7import { describe, expect } from "alchemy-test";
8import * as Data from "effect/Data";
9import * as Effect from "effect/Effect";
10import * as Schedule from "effect/Schedule";
11import * as HttpClient from "effect/http/HttpClient";
12import * as HttpClientRequest from "effect/http/HttpClientRequest";
13import MwaaServerlessTestFunctionLive, {
14 BINDINGS_BUCKET_NAME,
15 BINDINGS_WORKFLOW_NAME,
16 DEFINITION_KEY,
17 MwaaServerlessTestFunction,
18 WORKFLOW_DEFINITION,
19} from "./handler";
20
21const testOptions = { providers: AWS.providers() };
22const { test, beforeAll, afterAll } = Test.make(testOptions);
23const sharedStack = Core.scratchStack(testOptions, "MWAAServerlessBindings");
24
25// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
26// over 60s on a fresh deploy.
27const readinessPolicy = Schedule.max([
28 Schedule.fixed("2 seconds"),
29 Schedule.recurs(75),
30]);
31
32let baseUrl: string;
33let functionArn: string;
34
35class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
36 readonly status: number;
37 readonly body: string;
38}> {}
39
40// The shared Lambda fixture occasionally answers a transient 5xx under load
41// (cold re-init, IAM propagation on the freshly attached policy). Retry only
42// 5xx; a genuine 4xx/assertion failure surfaces immediately.
43const send = (request: HttpClientRequest.HttpClientRequest) =>
44 HttpClient.execute(request).pipe(
45 Effect.flatMap((response) =>
46 response.status >= 500
47 ? response.text.pipe(
48 Effect.flatMap((body) =>
49 Effect.fail(
50 new TransientUpstream({ status: response.status, body }),
51 ),
52 ),
53 )
54 : Effect.succeed(response),
55 ),
56 Effect.retry({
57 while: (e) => e._tag === "TransientUpstream",
58 schedule: Schedule.max([
59 Schedule.exponential("500 millis"),
60 Schedule.recurs(6),
61 ]),
62 }),
63 );
64
65const getJson = (path: string) =>
66 send(HttpClientRequest.get(`${baseUrl}${path}`)).pipe(
67 Effect.flatMap((r) => r.json),
68 );
69
70const postJson = (path: string) =>
71 send(HttpClientRequest.post(`${baseUrl}${path}`)).pipe(
72 Effect.flatMap((r) => r.json),
73 );
74
75// Direct distilled calls in beforeAll/afterAll run OUTSIDE test.provider, so
76// the AWS client context (credentials, region) must be provided explicitly.
77const withAws = <A, E>(effect: Effect.Effect<A, E, any>) =>
78 Core.withProviders(effect, testOptions, sharedStack.name);
79
80// The definitions bucket + workflow YAML must exist in S3 BEFORE the fixture
81// deploys — MWAA Serverless validates the definition at createWorkflow time.
82const ensureDefinitionUploaded = Effect.gen(function* () {
83 const region = process.env.AWS_REGION ?? "us-west-2";
84 yield* s3
85 .createBucket({
86 Bucket: BINDINGS_BUCKET_NAME,
87 ...(region === "us-east-1"
88 ? {}
89 : {
90 CreateBucketConfiguration: {
91 LocationConstraint: region,
92 },
93 }),
94 })
95 .pipe(Effect.catchTag("BucketAlreadyOwnedByYou", () => Effect.void));
96 yield* s3.putObject({
97 Bucket: BINDINGS_BUCKET_NAME,
98 Key: DEFINITION_KEY,
99 Body: new TextEncoder().encode(WORKFLOW_DEFINITION),
100 ContentType: "application/yaml",
101 });
102});
103
104// Best-effort cleanup: empty and delete the out-of-band definitions bucket,
105// and reap any orphaned auto-created log group left by an interrupted run
106// (its name embeds a service-assigned random suffix we cannot predict).
107const cleanupOutOfBand = Effect.gen(function* () {
108 yield* s3
109 .deleteObject({ Bucket: BINDINGS_BUCKET_NAME, Key: DEFINITION_KEY })
110 .pipe(Effect.catchTag("NoSuchBucket", () => Effect.void));
111 yield* s3
112 .deleteBucket({ Bucket: BINDINGS_BUCKET_NAME })
113 .pipe(Effect.catchTag("NoSuchBucket", () => Effect.void));
114 const found = yield* logs.describeLogGroups({
115 logGroupNamePrefix: `/aws/mwaa-serverless/${BINDINGS_WORKFLOW_NAME}-`,
116 });
117 for (const group of found.logGroups ?? []) {
118 if (group.logGroupName !== undefined) {
119 yield* logs
120 .deleteLogGroup({ logGroupName: group.logGroupName })
121 .pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void));
122 }
123 }
124});
125
126describe.sequential(
127 "MWAAServerless Bindings",
128 {
129 tags: [
130 "provider:aws",
131 "provider:aws:iam",
132 "provider:aws:lambda",
133 "provider:aws:mwaaserverless",
134 "provider:aws:s3",
135 "live",
136 ],
137 },
138 () => {
139 beforeAll(
140 Effect.gen(function* () {
141 yield* Effect.logInfo(
142 "MWAAServerless test setup: destroying previous resources",
143 );
144 yield* sharedStack.destroy();
145
146 yield* Effect.logInfo(
147 "MWAAServerless test setup: uploading workflow definition",
148 );
149 yield* withAws(ensureDefinitionUploaded);
150
151 yield* Effect.logInfo("MWAAServerless test setup: deploying fixture");
152 const attrs = yield* sharedStack.deploy(
153 Effect.gen(function* () {
154 return yield* MwaaServerlessTestFunction;
155 }).pipe(Effect.provide(MwaaServerlessTestFunctionLive)),
156 );
157
158 expect(attrs.functionUrl).toBeTruthy();
159 baseUrl = attrs.functionUrl!.replace(/\/+$/, "");
160 functionArn = attrs.functionArn;
161
162 const readinessUrl = `${baseUrl}/bindings`;
163 yield* Effect.logInfo(
164 `MWAAServerless test setup: probing readiness at ${readinessUrl}`,
165 );
166 yield* HttpClient.get(readinessUrl).pipe(
167 Effect.flatMap((response) =>
168 response.status === 200
169 ? Effect.succeed(response)
170 : Effect.fail(
171 new Error(`Function not ready: ${response.status}`),
172 ),
173 ),
174 Effect.tapError((error) =>
175 Effect.logWarning(
176 `MWAAServerless test setup: fixture not ready yet (${String(error)})`,
177 ),
178 ),
179 Effect.retry({ schedule: readinessPolicy }),
180 );
181 }),
182 { timeout: 300_000 },
183 );
184
185 afterAll(
186 sharedStack
187 .destroy()
188 .pipe(Effect.andThen(Effect.orDie(withAws(cleanupOutOfBand)))),
189 { timeout: 240_000 },
190 );
191
192 describe("binding registration", () => {
193 test.provider("all 7 capabilities initialize in the runtime", (_stack) =>
194 Effect.gen(function* () {
195 const response = (yield* getJson("/bindings")) as { bound: string[] };
196 expect(response.bound).toHaveLength(7);
197 }),
198 );
199 });
200
201 describe("ListWorkflowRuns", () => {
202 test.provider("reads runs through the injected workflow ARN", (_stack) =>
203 Effect.gen(function* () {
204 const response = (yield* getJson("/runs")) as { ids: string[] };
205 expect(Array.isArray(response.ids)).toBe(true);
206 }),
207 );
208 });
209
210 describe("ListWorkflowVersions", () => {
211 test.provider("lists the workflow's versions", (_stack) =>
212 Effect.gen(function* () {
213 const response = (yield* getJson("/versions")) as {
214 versions: string[];
215 };
216 expect(response.versions.length).toBeGreaterThanOrEqual(1);
217 }),
218 );
219 });
220
221 // Typed probes on a nonexistent run: a typed service error (not
222 // AccessDeniedException) proves the IAM grant covers the workflow ARN and
223 // the ARN was injected from the binding. ListTaskInstances is the
224 // exception — the service answers a nonexistent run with an empty list
225 // (`ok`), which proves the same thing.
226 describe("typed not-found probes", () => {
227 const probes: ReadonlyArray<
228 readonly [
229 name: string,
230 method: "GET" | "POST",
231 path: string,
232 tags: readonly string[],
233 ]
234 > = [
235 [
236 "GetWorkflowRun",
237 "GET",
238 "/run-fake",
239 ["ResourceNotFoundException", "ValidationException"],
240 ],
241 [
242 "ListTaskInstances",
243 "GET",
244 "/tasks-fake",
245 ["ok", "ResourceNotFoundException", "ValidationException"],
246 ],
247 [
248 "GetTaskInstance",
249 "GET",
250 "/task-fake",
251 ["ResourceNotFoundException", "ValidationException"],
252 ],
253 [
254 "StopWorkflowRun",
255 "POST",
256 "/run-stop-fake",
257 ["ResourceNotFoundException", "ValidationException"],
258 ],
259 ] as const;
260
261 for (const [name, method, path, tags] of probes) {
262 test.provider(
263 `${name} answers with a typed outcome (not AccessDenied)`,
264 (_stack) =>
265 Effect.gen(function* () {
266 const response = (yield* method === "GET"
267 ? getJson(path)
268 : postJson(path)) as { tag: string; detail: string };
269 expect(response.detail).not.toContain("not authorized");
270 expect(tags).toContain(response.tag);
271 }),
272 );
273 }
274 });
275
276 describe("workflow run round trip", () => {
277 test.provider(
278 "start a run, observe it, stop it",
279 (_stack) =>
280 Effect.gen(function* () {
281 // 1. StartWorkflowRun — an on-demand run of the fixture workflow.
282 const started = (yield* postJson("/run-start")) as {
283 runId: string;
284 status: string;
285 };
286 expect(started.runId).toBeTruthy();
287
288 // 2. GetWorkflowRun observes the run through the binding.
289 const detail = (yield* getJson(
290 `/run-detail?id=${started.runId}`,
291 )) as { runId: string; status?: string };
292 expect(detail.runId).toBe(started.runId);
293
294 // 3. StopWorkflowRun — a typed outcome either way: `ok` when the
295 // run was still stoppable, or a typed conflict/validation error
296 // if it already reached a terminal state. Reaching the service
297 // (not AccessDenied) is what the binding must prove.
298 const stopped = (yield* postJson(
299 `/run-stop?id=${started.runId}`,
300 )) as { tag: string };
301 expect(stopped.tag).not.toBe("AccessDeniedException");
302
303 // 4. The run shows up in ListWorkflowRuns.
304 const listed = (yield* getJson("/runs")) as { ids: string[] };
305 expect(listed.ids).toContain(started.runId);
306 }),
307 { timeout: 120_000 },
308 );
309 });
310
311 describe(
312 "consumeWorkflowRunEvents",
313 { tags: ["provider:aws:eventbridge"] },
314 () => {
315 test.provider(
316 "the deploy created an EventBridge rule targeting the function",
317 (_stack) =>
318 Effect.gen(function* () {
319 // Out-of-band via distilled: the fixture's consumeWorkflowRunEvents
320 // must have materialized as a rule on the default bus with the
321 // Lambda as target.
322 const { RuleNames } = yield* eventbridge.listRuleNamesByTarget({
323 TargetArn: functionArn,
324 });
325 expect((RuleNames ?? []).length).toBeGreaterThanOrEqual(1);
326 }),
327 );
328 },
329 );
330 },
331);