1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as emr from "@distilled.cloud/aws/emr-serverless";
5import * as eventbridge from "@distilled.cloud/aws/eventbridge";
6import * as sts from "@distilled.cloud/aws/sts";
7import { describe, expect } from "alchemy-test";
8import * as Data from "effect/Data";
9import * as Effect from "effect/Effect";
10import * as Option from "effect/Option";
11import * as Schedule from "effect/Schedule";
12import * as Stream from "effect/Stream";
13import * as HttpClient from "effect/http/HttpClient";
14import * as HttpClientRequest from "effect/http/HttpClientRequest";
15import EmrServerlessTestFunctionLive, {
16 BINDINGS_APP_NAME,
17 BINDINGS_ROLE_NAME,
18 EmrServerlessTestFunction,
19} from "./handler";
20
21const testOptions = { providers: AWS.providers() };
22const { test, beforeAll, afterAll } = Test.make(testOptions);
23const sharedStack = Core.scratchStack(testOptions, "EMRServerlessBindings");
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
40class JobRunNotTerminal extends Data.TaggedError("JobRunNotTerminal")<{
41 readonly jobRunId: string;
42 readonly state: string;
43 readonly stateDetails?: string;
44}> {
45 override get message() {
46 return `EMR Serverless job ${this.jobRunId} is still ${this.state}: ${this.stateDetails ?? "no state details"}`;
47 }
48}
49
50class ApplicationNotStarted extends Data.TaggedError("ApplicationNotStarted")<{
51 readonly applicationId: string;
52 readonly state: string;
53}> {}
54
55const terminalJobRunStates = ["SUCCESS", "CANCELLED", "FAILED"] as const;
56
57// The shared Lambda fixture occasionally answers a transient 5xx under load
58// (cold re-init, IAM propagation on the freshly attached policy). Retry only
59// 5xx; a genuine 4xx/assertion failure surfaces immediately.
60const send = (request: HttpClientRequest.HttpClientRequest) =>
61 HttpClient.execute(request).pipe(
62 Effect.flatMap((response) =>
63 response.status >= 500
64 ? response.text.pipe(
65 Effect.flatMap((body) =>
66 Effect.fail(
67 new TransientUpstream({ status: response.status, body }),
68 ),
69 ),
70 )
71 : Effect.succeed(response),
72 ),
73 Effect.retry({
74 while: (e) => e._tag === "TransientUpstream",
75 schedule: Schedule.max([
76 Schedule.exponential("500 millis"),
77 Schedule.recurs(6),
78 ]),
79 }),
80 );
81
82const getJson = (path: string) =>
83 send(HttpClientRequest.get(`${baseUrl}${path}`)).pipe(
84 Effect.flatMap((r) => r.json),
85 );
86
87const postJson = (path: string) =>
88 send(HttpClientRequest.post(`${baseUrl}${path}`)).pipe(
89 Effect.flatMap((r) => r.json),
90 );
91
92const waitForTerminalJobRun = (jobRunId: string) =>
93 getJson(`/jobrun-detail?id=${jobRunId}`).pipe(
94 Effect.flatMap((value) => {
95 const detail = value as {
96 jobRunId: string;
97 state: string;
98 stateDetails?: string;
99 };
100 return terminalJobRunStates.includes(
101 detail.state as (typeof terminalJobRunStates)[number],
102 )
103 ? Effect.succeed(detail)
104 : Effect.fail(
105 new JobRunNotTerminal({
106 jobRunId: detail.jobRunId,
107 state: detail.state,
108 stateDetails: detail.stateDetails,
109 }),
110 );
111 }),
112 Effect.tapError((error) =>
113 error._tag === "JobRunNotTerminal"
114 ? Effect.logInfo(error.message)
115 : Effect.void,
116 ),
117 Effect.retry({
118 while: (error) => error._tag === "JobRunNotTerminal",
119 schedule: Schedule.max([
120 Schedule.spaced("3 seconds"),
121 Schedule.recurs(40),
122 ]),
123 }),
124 );
125
126const waitForApplicationStarted = (applicationId: string) =>
127 emr.getApplication({ applicationId }).pipe(
128 Effect.flatMap(({ application }) =>
129 application.state === "STARTED"
130 ? Effect.succeed(application)
131 : Effect.fail(
132 new ApplicationNotStarted({
133 applicationId,
134 state: application.state,
135 }),
136 ),
137 ),
138 Effect.retry({
139 while: (error) => error._tag === "ApplicationNotStarted",
140 schedule: Schedule.max([
141 Schedule.spaced("3 seconds"),
142 Schedule.recurs(15),
143 ]),
144 }),
145 );
146
147/** Find the fixture application out-of-band via its deterministic name. */
148const findApplication = Effect.gen(function* () {
149 const summary = yield* emr.listApplications.items({}).pipe(
150 Stream.filter(
151 (s) => s.name === BINDINGS_APP_NAME && s.state !== "TERMINATED",
152 ),
153 Stream.runHead,
154 Effect.map(Option.getOrUndefined),
155 );
156 expect(summary).toBeDefined();
157 return summary!;
158});
159
160describe.sequential(
161 "EMRServerless Bindings",
162 {
163 tags: [
164 "provider:aws",
165 "provider:aws:emrserverless",
166 "provider:aws:iam",
167 "provider:aws:lambda",
168 "live",
169 ],
170 },
171 () => {
172 beforeAll(
173 Effect.gen(function* () {
174 yield* Effect.logInfo(
175 "EMRServerless test setup: destroying previous resources",
176 );
177 yield* sharedStack.destroy();
178
179 yield* Effect.logInfo("EMRServerless test setup: deploying fixture");
180 const attrs = yield* sharedStack.deploy(
181 Effect.gen(function* () {
182 return yield* EmrServerlessTestFunction;
183 }).pipe(Effect.provide(EmrServerlessTestFunctionLive)),
184 );
185
186 expect(attrs.functionUrl).toBeTruthy();
187 baseUrl = attrs.functionUrl!.replace(/\/+$/, "");
188 functionArn = attrs.functionArn;
189
190 const readinessUrl = `${baseUrl}/bindings`;
191 yield* Effect.logInfo(
192 `EMRServerless test setup: probing readiness at ${readinessUrl}`,
193 );
194 yield* HttpClient.get(readinessUrl).pipe(
195 Effect.flatMap((response) =>
196 response.status === 200
197 ? Effect.succeed(response)
198 : Effect.fail(
199 new Error(`Function not ready: ${response.status}`),
200 ),
201 ),
202 Effect.tapError((error) =>
203 Effect.logWarning(
204 `EMRServerless test setup: fixture not ready yet (${String(error)})`,
205 ),
206 ),
207 Effect.retry({ schedule: readinessPolicy }),
208 );
209 }),
210 { timeout: 300_000 },
211 );
212
213 afterAll(sharedStack.destroy(), { timeout: 240_000 });
214
215 describe("binding registration", () => {
216 test.provider("all 14 capabilities initialize in the runtime", (_stack) =>
217 Effect.gen(function* () {
218 const response = (yield* getJson("/bindings")) as { bound: string[] };
219 expect(response.bound).toHaveLength(14);
220 }),
221 );
222 });
223
224 describe("ListJobRuns", () => {
225 test.provider(
226 "reads job runs through the injected application id",
227 (_stack) =>
228 Effect.gen(function* () {
229 const response = (yield* getJson("/jobruns")) as { ids: string[] };
230 expect(Array.isArray(response.ids)).toBe(true);
231 }),
232 );
233 });
234
235 describe("ListSessions", () => {
236 test.provider("lists sessions on the bound application", (_stack) =>
237 Effect.gen(function* () {
238 const response = (yield* getJson("/sessions")) as { tag: string };
239 expect(response.tag).toBe("ok");
240 }),
241 );
242 });
243
244 // Typed not-found probes: a ResourceNotFoundException (not
245 // AccessDeniedException) proves the IAM grant covers the sub-resource ARN
246 // and the application id was injected from the binding.
247 describe("typed not-found probes", () => {
248 // The session APIs validate the session id format before lookup, so a
249 // well-formed-but-fake job-run-shaped id answers with the typed
250 // ValidationException instead — either tag proves the IAM grant passed
251 // authorization and the application id was injected.
252 const probes: ReadonlyArray<
253 readonly [
254 name: string,
255 method: "GET" | "POST",
256 path: string,
257 tags: readonly string[],
258 ]
259 > = [
260 ["GetJobRun", "GET", "/jobrun", ["ResourceNotFoundException"]],
261 [
262 "GetDashboardForJobRun",
263 "GET",
264 "/jobrun-dashboard",
265 ["ResourceNotFoundException"],
266 ],
267 [
268 "ListJobRunAttempts",
269 "GET",
270 "/jobrun-attempts",
271 ["ResourceNotFoundException"],
272 ],
273 [
274 "CancelJobRun",
275 "POST",
276 "/jobrun-cancel-fake",
277 ["ResourceNotFoundException"],
278 ],
279 [
280 "GetSession",
281 "GET",
282 "/session",
283 ["ResourceNotFoundException", "ValidationException"],
284 ],
285 [
286 "GetSessionEndpoint",
287 "GET",
288 "/session-endpoint",
289 ["ResourceNotFoundException", "ValidationException"],
290 ],
291 [
292 "TerminateSession",
293 "POST",
294 "/session-terminate",
295 ["ResourceNotFoundException", "ValidationException"],
296 ],
297 ] as const;
298
299 for (const [name, method, path, tags] of probes) {
300 test.provider(
301 `${name} answers with a typed error (not AccessDenied)`,
302 (_stack) =>
303 Effect.gen(function* () {
304 const response = (yield* method === "GET"
305 ? getJson(path)
306 : postJson(path)) as { tag: string; detail: string };
307 expect(response.detail).not.toContain("not authorized");
308 expect(tags).toContain(response.tag);
309 }),
310 );
311 }
312 });
313
314 describe("GetResourceDashboard", () => {
315 test.provider(
316 "answers with the typed service-gate tag (see probe.test.ts)",
317 (_stack) =>
318 Effect.gen(function* () {
319 // The service currently denies emr-serverless:GetResourceDashboard
320 // for every caller (even Action:"*" admins) — the operation backs
321 // the console dashboards and has not launched for API callers.
322 // probe.test.ts pins that platform gate; here we assert the
323 // binding surfaces the same typed tag end-to-end from the Lambda.
324 const response = (yield* getJson("/resource-dashboard")) as {
325 tag: string;
326 };
327 expect(response.tag).toBe("AccessDeniedException");
328 }),
329 );
330 });
331
332 describe("StartSession", () => {
333 test.provider(
334 "grant + PassRole reach the service (typed validation error)",
335 (_stack) =>
336 Effect.gen(function* () {
337 // The fixture application has no interactive configuration, so the
338 // service must answer with a typed validation error — reaching it
339 // proves both the StartSession grant and the PassRole statement.
340 const { Account } = yield* sts.getCallerIdentity({});
341 const roleArn = `arn:aws:iam::${Account}:role/${BINDINGS_ROLE_NAME}`;
342 const response = (yield* postJson(
343 `/session-start?roleArn=${encodeURIComponent(roleArn)}`,
344 )) as { tag: string; detail: string };
345 expect(response.detail).not.toContain("not authorized");
346 expect(response.tag).toBe("ValidationException");
347 }),
348 );
349 });
350
351 describe("application control + job run round trip", () => {
352 test.provider(
353 "start application, submit + cancel a job run, stop application",
354 (_stack) =>
355 Effect.gen(function* () {
356 // 1. StartApplication (free — the fixture has no pre-init capacity).
357 const started = (yield* postJson("/app-start")) as { tag: string };
358 expect(started.tag).toBe("ok");
359
360 // 2. Submit a job run (SparkPi from the EMR image).
361 const { Account } = yield* sts.getCallerIdentity({});
362 const roleArn = `arn:aws:iam::${Account}:role/${BINDINGS_ROLE_NAME}`;
363 const run = (yield* postJson(
364 `/jobrun-run?roleArn=${encodeURIComponent(roleArn)}`,
365 )) as { jobRunId: string; arn: string };
366 expect(run.jobRunId).toBeTruthy();
367 expect(run.arn).toContain("/jobruns/");
368
369 // 3. Cancel it immediately (no workers ever run).
370 const cancelled = (yield* postJson(
371 `/jobrun-cancel?id=${run.jobRunId}`,
372 )) as { jobRunId: string };
373 expect(cancelled.jobRunId).toBe(run.jobRunId);
374
375 // 4. GetJobRun observes the cancellation through a terminal state.
376 // StopApplication rejects while any run is still CANCELLING, so the
377 // cancellation response alone is not a sufficient readiness signal.
378 const detail = yield* waitForTerminalJobRun(run.jobRunId);
379 expect(detail.jobRunId).toBe(run.jobRunId);
380 expect(terminalJobRunStates).toContain(detail.state);
381
382 // 5. The submitted run shows up in ListJobRuns.
383 const listed = (yield* getJson("/jobruns")) as { ids: string[] };
384 expect(listed.ids).toContain(run.jobRunId);
385
386 // 6. Converge the application back to STARTED. The fixture's
387 // one-minute auto-stop can legitimately win while the cancelled
388 // job settles under load, but StopApplication is only valid from
389 // STARTED.
390 const app = yield* findApplication;
391 const { application } = yield* emr.getApplication({
392 applicationId: app.id,
393 });
394 if (
395 application.state === "CREATED" ||
396 application.state === "STOPPED"
397 ) {
398 const restarted = (yield* postJson("/app-start")) as {
399 tag: string;
400 };
401 expect(restarted.tag).toBe("ok");
402 }
403 const startedApplication = yield* waitForApplicationStarted(app.id);
404 expect(startedApplication.state).toBe("STARTED");
405
406 // 7. StopApplication.
407 const stopped = (yield* postJson("/app-stop")) as { tag: string };
408 expect(stopped.tag).toBe("ok");
409 }),
410 { timeout: 240_000 },
411 );
412 });
413
414 describe(
415 "consumeJobRunEvents",
416 { tags: ["provider:aws:eventbridge"] },
417 () => {
418 test.provider(
419 "the deploy created an EventBridge rule targeting the function",
420 (_stack) =>
421 Effect.gen(function* () {
422 // Out-of-band via distilled: the fixture's consumeJobRunEvents must
423 // have materialized as a rule on the default bus with the Lambda as
424 // target.
425 const { RuleNames } = yield* eventbridge.listRuleNamesByTarget({
426 TargetArn: functionArn,
427 });
428 expect((RuleNames ?? []).length).toBeGreaterThanOrEqual(1);
429 }),
430 );
431 },
432 );
433 },
434);