1import * as EMRServerless from "@/AWS/EMRServerless";
2import * as IAM from "@/AWS/IAM";
3import * as Lambda from "@/AWS/Lambda";
4import type * as emr from "@distilled.cloud/aws/emr-serverless";
5import * as Duration from "effect/Duration";
6import * as Effect from "effect/Effect";
7import * as Layer from "effect/Layer";
8import * as Schedule from "effect/Schedule";
9import * as Stream from "effect/Stream";
10import { HttpServerRequest } from "effect/http/HttpServerRequest";
11import * as HttpServerResponse from "effect/http/HttpServerResponse";
12import path from "pathe";
13
14const main = path.resolve(import.meta.dirname, "handler.ts");
15
16// Well-formed (16 chars of [0-9a-z]) but nonexistent ids — probes must answer
17// with the typed ResourceNotFoundException, which proves the IAM grant passed
18// authorization AND the application id was injected from the binding.
19const FAKE_JOB_RUN_ID = "00abcdefabcdef01";
20const FAKE_SESSION_ID = "00abcdefabcdef01";
21
22// Session authorization can lag behind other EMR APIs on a fresh Lambda role.
23const sessionAuthorizationPolicy = {
24 while: (
25 error:
26 | emr.GetSessionEndpointError
27 | emr.StartSessionError
28 | emr.ListSessionsError,
29 ) => error._tag === "AccessDeniedException",
30 schedule: Schedule.spaced("3 seconds"),
31 times: 8,
32};
33
34/** Deterministic names shared with the test for out-of-band verification. */
35const stageSuffix = process.env.ALCHEMY_TEST_STAGE
36 ? `-${process.env.ALCHEMY_TEST_STAGE}`
37 : "";
38export const BINDINGS_APP_NAME = `alchemy-test-emrs-bind${stageSuffix}`;
39export const BINDINGS_ROLE_NAME = `alchemy-test-emrs-bind-role${stageSuffix}`;
40
41export class EmrServerlessTestFunction extends Lambda.Function<Lambda.Function>()(
42 "EmrServerlessTestFunction",
43) {}
44
45export default EmrServerlessTestFunction.make(
46 {
47 main,
48 functionUrl: true,
49 timeout: Duration.seconds(30),
50 },
51 Effect.gen(function* () {
52 // The Spark application the bindings are bound to. Auto-start lets a job
53 // submission start it; auto-stop reaps it after a minute idle. In the
54 // CREATED/STOPPED state it costs nothing.
55 const app = yield* EMRServerless.Application("BindingApp", {
56 applicationName: BINDINGS_APP_NAME,
57 releaseLabel: "emr-7.5.0",
58 autoStartConfiguration: { enabled: true },
59 autoStopConfiguration: { enabled: true, idleTimeout: "1 minute" },
60 tags: { purpose: "alchemy-test" },
61 });
62
63 // The execution role submitted job runs assume (no data permissions —
64 // the fixture cancels jobs immediately after submission).
65 yield* IAM.Role("JobRole", {
66 roleName: BINDINGS_ROLE_NAME,
67 assumeRolePolicyDocument: {
68 Version: "2012-10-17",
69 Statement: [
70 {
71 Effect: "Allow",
72 Principal: { Service: "emr-serverless.amazonaws.com" },
73 Action: ["sts:AssumeRole"],
74 },
75 ],
76 },
77 });
78
79 // Event source: subscribe the host to EMR Serverless job-run state
80 // changes. The deploy proves the EventBridge rule + invoke permission.
81 yield* EMRServerless.consumeJobRunEvents(
82 { kinds: ["job-run-state-change"] },
83 (events) =>
84 Stream.runForEach(events, (event) =>
85 Effect.log(
86 `emr-serverless job ${event.detail.jobRunId} -> ${event.detail.state}`,
87 ),
88 ),
89 );
90
91 const startJobRun = yield* EMRServerless.StartJobRun(app);
92 const getJobRun = yield* EMRServerless.GetJobRun(app);
93 const cancelJobRun = yield* EMRServerless.CancelJobRun(app);
94 const listJobRuns = yield* EMRServerless.ListJobRuns(app);
95 const listJobRunAttempts = yield* EMRServerless.ListJobRunAttempts(app);
96 const getDashboardForJobRun =
97 yield* EMRServerless.GetDashboardForJobRun(app);
98 const getResourceDashboard = yield* EMRServerless.GetResourceDashboard(app);
99 const startApplication = yield* EMRServerless.StartApplication(app);
100 const stopApplication = yield* EMRServerless.StopApplication(app);
101 const startSession = yield* EMRServerless.StartSession(app);
102 const getSession = yield* EMRServerless.GetSession(app);
103 const terminateSession = yield* EMRServerless.TerminateSession(app);
104 const listSessions = yield* EMRServerless.ListSessions(app);
105 const getSessionEndpoint = yield* EMRServerless.GetSessionEndpoint(app);
106
107 const bound = {
108 startJobRun,
109 getJobRun,
110 cancelJobRun,
111 listJobRuns,
112 listJobRunAttempts,
113 getDashboardForJobRun,
114 getResourceDashboard,
115 startApplication,
116 stopApplication,
117 startSession,
118 getSession,
119 terminateSession,
120 listSessions,
121 getSessionEndpoint,
122 };
123
124 // Run an operation and answer with its typed outcome: `ok` on success,
125 // the error's `_tag` otherwise. Probes assert the tag — a typed
126 // service error (not AccessDeniedException) proves the IAM grant and
127 // the injected application id both work.
128 const probe = <A, E extends { _tag: string }>(
129 effect: Effect.Effect<A, E>,
130 ) =>
131 effect.pipe(
132 Effect.map(() => ({ tag: "ok", detail: "" }) as const),
133 Effect.catch((error) =>
134 Effect.succeed({ tag: error._tag, detail: String(error) } as const),
135 ),
136 Effect.flatMap((outcome) => HttpServerResponse.json(outcome)),
137 );
138
139 return {
140 fetch: Effect.gen(function* () {
141 const request = yield* HttpServerRequest;
142 const url = new URL(request.originalUrl);
143 const pathname = url.pathname;
144
145 if (request.method === "GET" && pathname === "/bindings") {
146 return yield* HttpServerResponse.json({ bound: Object.keys(bound) });
147 }
148
149 // Real reads: application id injected from the binding.
150 if (request.method === "GET" && pathname === "/jobruns") {
151 const { jobRuns } = yield* listJobRuns().pipe(Effect.orDie);
152 return yield* HttpServerResponse.json({
153 ids: jobRuns.map((run) => run.id),
154 });
155 }
156 if (request.method === "GET" && pathname === "/sessions") {
157 return yield* probe(
158 listSessions().pipe(Effect.retry(sessionAuthorizationPolicy)),
159 );
160 }
161
162 // Typed not-found probes on nonexistent sub-resources.
163 if (request.method === "GET" && pathname === "/jobrun") {
164 return yield* probe(getJobRun({ jobRunId: FAKE_JOB_RUN_ID }));
165 }
166 if (request.method === "GET" && pathname === "/jobrun-dashboard") {
167 return yield* probe(
168 getDashboardForJobRun({ jobRunId: FAKE_JOB_RUN_ID }),
169 );
170 }
171 if (request.method === "GET" && pathname === "/jobrun-attempts") {
172 return yield* probe(
173 listJobRunAttempts({ jobRunId: FAKE_JOB_RUN_ID }),
174 );
175 }
176 if (request.method === "POST" && pathname === "/jobrun-cancel-fake") {
177 return yield* probe(cancelJobRun({ jobRunId: FAKE_JOB_RUN_ID }));
178 }
179 if (request.method === "GET" && pathname === "/session") {
180 return yield* probe(getSession({ sessionId: FAKE_SESSION_ID }));
181 }
182 if (request.method === "GET" && pathname === "/session-endpoint") {
183 return yield* probe(
184 getSessionEndpoint({ sessionId: FAKE_SESSION_ID }).pipe(
185 Effect.retry(sessionAuthorizationPolicy),
186 ),
187 );
188 }
189 if (request.method === "POST" && pathname === "/session-terminate") {
190 return yield* probe(terminateSession({ sessionId: FAKE_SESSION_ID }));
191 }
192 if (request.method === "GET" && pathname === "/resource-dashboard") {
193 return yield* probe(
194 getResourceDashboard({
195 resourceId: FAKE_JOB_RUN_ID,
196 resourceType: "SPARK_DRIVER",
197 }),
198 );
199 }
200 // startSession is registration-covered; the runtime probe proves the
201 // grant + PassRole path end-to-end (the fixture application has no
202 // interactive configuration, so the service answers with a typed
203 // validation error rather than starting a billable session).
204 if (request.method === "POST" && pathname === "/session-start") {
205 const roleArn = url.searchParams.get("roleArn");
206 if (!roleArn) {
207 return yield* HttpServerResponse.json(
208 { error: "roleArn query parameter required" },
209 { status: 400 },
210 );
211 }
212 return yield* probe(
213 startSession({ executionRoleArn: roleArn }).pipe(
214 Effect.retry(sessionAuthorizationPolicy),
215 ),
216 );
217 }
218
219 // Application control.
220 if (request.method === "POST" && pathname === "/app-start") {
221 return yield* probe(startApplication());
222 }
223 if (request.method === "POST" && pathname === "/app-stop") {
224 return yield* probe(stopApplication());
225 }
226
227 // Real job-run round trip: submit (SparkPi from the EMR image — no
228 // S3 assets), then the test cancels it immediately.
229 if (request.method === "POST" && pathname === "/jobrun-run") {
230 const roleArn = url.searchParams.get("roleArn");
231 if (!roleArn) {
232 return yield* HttpServerResponse.json(
233 { error: "roleArn query parameter required" },
234 { status: 400 },
235 );
236 }
237 const started = yield* startJobRun({
238 executionRoleArn: roleArn,
239 jobDriver: {
240 sparkSubmit: {
241 entryPoint:
242 "local:///usr/lib/spark/examples/src/main/python/pi.py",
243 },
244 },
245 executionTimeoutMinutes: 10,
246 }).pipe(Effect.orDie);
247 return yield* HttpServerResponse.json({
248 jobRunId: started.jobRunId,
249 arn: started.arn,
250 });
251 }
252 if (request.method === "POST" && pathname === "/jobrun-cancel") {
253 const id = url.searchParams.get("id");
254 if (!id) {
255 return yield* HttpServerResponse.json(
256 { error: "id query parameter required" },
257 { status: 400 },
258 );
259 }
260 const cancelled = yield* cancelJobRun({ jobRunId: id }).pipe(
261 Effect.orDie,
262 );
263 return yield* HttpServerResponse.json({
264 jobRunId: cancelled.jobRunId,
265 });
266 }
267 if (request.method === "GET" && pathname === "/jobrun-detail") {
268 const id = url.searchParams.get("id");
269 if (!id) {
270 return yield* HttpServerResponse.json(
271 { error: "id query parameter required" },
272 { status: 400 },
273 );
274 }
275 const { jobRun } = yield* getJobRun({ jobRunId: id }).pipe(
276 Effect.orDie,
277 );
278 return yield* HttpServerResponse.json({
279 jobRunId: jobRun.jobRunId,
280 state: jobRun.state,
281 stateDetails: jobRun.stateDetails,
282 });
283 }
284
285 return yield* HttpServerResponse.json(
286 { error: "Not found", method: request.method, pathname },
287 { status: 404 },
288 );
289 }).pipe(Effect.orDie),
290 };
291 }).pipe(
292 Effect.provide(
293 Layer.mergeAll(
294 Lambda.EventSource,
295 EMRServerless.StartJobRunHttp,
296 EMRServerless.GetJobRunHttp,
297 EMRServerless.CancelJobRunHttp,
298 EMRServerless.ListJobRunsHttp,
299 EMRServerless.ListJobRunAttemptsHttp,
300 EMRServerless.GetDashboardForJobRunHttp,
301 EMRServerless.GetResourceDashboardHttp,
302 EMRServerless.StartApplicationHttp,
303 EMRServerless.StopApplicationHttp,
304 EMRServerless.StartSessionHttp,
305 EMRServerless.GetSessionHttp,
306 EMRServerless.TerminateSessionHttp,
307 EMRServerless.ListSessionsHttp,
308 EMRServerless.GetSessionEndpointHttp,
309 ),
310 ),
311 ),
312);