1import * as EMR from "@/AWS/EMR";
2import * as IAM from "@/AWS/IAM";
3import * as Lambda from "@/AWS/Lambda";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import { HttpServerRequest } from "effect/http/HttpServerRequest";
8import * as HttpServerResponse from "effect/http/HttpServerResponse";
9import path from "pathe";
10
11const main = path.resolve(import.meta.dirname, "slow-handler.ts");
12
13export class EmrSlowTestFunction extends Lambda.Function<Lambda.Function>()(
14 "EmrSlowTestFunction",
15) {}
16
17/**
18 * Cluster-scoped binding fixture: deploys a real single-node Spark cluster
19 * (~10-15 minutes to WAITING, billed per instance-hour — gated behind
20 * AWS_TEST_SLOW) and a Lambda bound to it with the steps data plane, the
21 * cluster-inspection reads, and the managed scaling policy ops.
22 *
23 * The cluster launches into an EMR-chosen subnet of the account's default
24 * VPC (no explicit `ec2SubnetId`), keeps itself alive between steps, and
25 * carries a 2-hour auto-termination policy as a cost safety net.
26 */
27export default EmrSlowTestFunction.make(
28 {
29 main,
30 functionUrl: true,
31 timeout: Duration.seconds(60),
32 },
33 Effect.gen(function* () {
34 // EMR service role (legacy managed policy — the v2 policy requires
35 // tag-scoped resources).
36 const serviceRole = yield* IAM.Role("EmrBindingsServiceRole", {
37 assumeRolePolicyDocument: {
38 Version: "2012-10-17",
39 Statement: [
40 {
41 Effect: "Allow",
42 Principal: { Service: ["elasticmapreduce.amazonaws.com"] },
43 Action: ["sts:AssumeRole"],
44 },
45 ],
46 },
47 managedPolicyArns: [
48 "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole",
49 ],
50 });
51 // EC2 instance role + profile (the job-flow role).
52 const ec2Role = yield* IAM.Role("EmrBindingsEc2Role", {
53 assumeRolePolicyDocument: {
54 Version: "2012-10-17",
55 Statement: [
56 {
57 Effect: "Allow",
58 Principal: { Service: ["ec2.amazonaws.com"] },
59 Action: ["sts:AssumeRole"],
60 },
61 ],
62 },
63 managedPolicyArns: [
64 "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceforEC2Role",
65 ],
66 });
67 const instanceProfile = yield* IAM.InstanceProfile("EmrBindingsProfile", {
68 roleName: ec2Role.roleName,
69 });
70
71 // Single-node (master-only) Spark cluster — the cheapest live cluster
72 // that can still run steps.
73 const cluster = yield* EMR.Cluster("BindingsCluster", {
74 releaseLabel: "emr-7.5.0",
75 applications: ["Spark"],
76 serviceRole: serviceRole.roleName,
77 jobFlowRole: instanceProfile.instanceProfileName,
78 instances: {
79 masterInstanceType: "m5.xlarge",
80 coreInstanceCount: 0,
81 keepJobFlowAliveWhenNoSteps: true,
82 },
83 autoTerminationPolicy: { idleTimeout: "2 hours" },
84 tags: { fixture: "emr-bindings" },
85 });
86
87 const addSteps = yield* EMR.AddJobFlowSteps(cluster);
88 const describeStep = yield* EMR.DescribeStep(cluster);
89 const listSteps = yield* EMR.ListSteps(cluster);
90 const cancelSteps = yield* EMR.CancelSteps(cluster);
91 const listInstances = yield* EMR.ListInstances(cluster);
92 const listInstanceGroups = yield* EMR.ListInstanceGroups(cluster);
93 const listBootstrapActions = yield* EMR.ListBootstrapActions(cluster);
94 const getScalingPolicy = yield* EMR.GetManagedScalingPolicy(cluster);
95 const putScalingPolicy = yield* EMR.PutManagedScalingPolicy(cluster);
96 const removeScalingPolicy = yield* EMR.RemoveManagedScalingPolicy(cluster);
97
98 return {
99 fetch: Effect.gen(function* () {
100 const request = yield* HttpServerRequest;
101 const url = new URL(request.originalUrl);
102 const pathname = url.pathname;
103
104 if (request.method === "GET" && pathname === "/bindings") {
105 return yield* HttpServerResponse.json({ bound: 10 });
106 }
107
108 // Submit a trivial shell step — proves AddJobFlowSteps and the
109 // JobFlowId injection.
110 if (request.method === "GET" && pathname === "/steps/add") {
111 const { StepIds } = yield* addSteps({
112 Steps: [
113 {
114 Name: "alchemy-echo",
115 ActionOnFailure: "CONTINUE",
116 HadoopJarStep: {
117 Jar: "command-runner.jar",
118 Args: ["bash", "-c", "echo alchemy"],
119 },
120 },
121 ],
122 });
123 return yield* HttpServerResponse.json({ stepId: StepIds?.[0] });
124 }
125
126 if (request.method === "GET" && pathname === "/step") {
127 const { Step } = yield* describeStep({
128 StepId: url.searchParams.get("id")!,
129 });
130 return yield* HttpServerResponse.json({
131 state: Step?.Status?.State,
132 name: Step?.Name,
133 });
134 }
135
136 if (request.method === "GET" && pathname === "/steps") {
137 const { Steps } = yield* listSteps();
138 return yield* HttpServerResponse.json({
139 count: (Steps ?? []).length,
140 });
141 }
142
143 // Cancellation of a completed step surfaces a per-step error reason
144 // rather than failing the call — either shape proves the grant.
145 if (request.method === "GET" && pathname === "/steps/cancel") {
146 const { CancelStepsInfoList } = yield* cancelSteps({
147 StepIds: [url.searchParams.get("id")!],
148 });
149 return yield* HttpServerResponse.json({
150 status: CancelStepsInfoList?.[0]?.Status,
151 });
152 }
153
154 if (request.method === "GET" && pathname === "/instances") {
155 const { Instances } = yield* listInstances();
156 return yield* HttpServerResponse.json({
157 count: (Instances ?? []).length,
158 });
159 }
160
161 if (request.method === "GET" && pathname === "/groups") {
162 const { InstanceGroups } = yield* listInstanceGroups();
163 return yield* HttpServerResponse.json({
164 types: (InstanceGroups ?? []).map((g) => g.InstanceGroupType),
165 });
166 }
167
168 if (request.method === "GET" && pathname === "/bootstrap") {
169 const { BootstrapActions } = yield* listBootstrapActions();
170 return yield* HttpServerResponse.json({
171 count: (BootstrapActions ?? []).length,
172 });
173 }
174
175 // Managed scaling put -> get -> remove round trip. A master-only
176 // cluster may reject the put with a typed validation tag — either
177 // outcome proves the grants and the ClusterId injection.
178 if (request.method === "GET" && pathname === "/scaling") {
179 const putResult = yield* putScalingPolicy({
180 ManagedScalingPolicy: {
181 ComputeLimits: {
182 UnitType: "Instances",
183 MinimumCapacityUnits: 1,
184 MaximumCapacityUnits: 4,
185 },
186 },
187 }).pipe(
188 Effect.map(() => ({ ok: true as const })),
189 Effect.catchTag("ValidationException", (e) =>
190 Effect.succeed({ ok: false as const, tag: e._tag }),
191 ),
192 );
193 if (!putResult.ok) {
194 return yield* HttpServerResponse.json(putResult);
195 }
196 const { ManagedScalingPolicy } = yield* getScalingPolicy();
197 yield* removeScalingPolicy();
198 return yield* HttpServerResponse.json({
199 ok: true,
200 max: ManagedScalingPolicy?.ComputeLimits?.MaximumCapacityUnits,
201 });
202 }
203
204 return yield* HttpServerResponse.json(
205 { error: "Not found", method: request.method, pathname },
206 { status: 404 },
207 );
208 }).pipe(Effect.orDie),
209 };
210 }).pipe(
211 Effect.provide(
212 Layer.mergeAll(
213 EMR.AddJobFlowStepsHttp,
214 EMR.DescribeStepHttp,
215 EMR.ListStepsHttp,
216 EMR.CancelStepsHttp,
217 EMR.ListInstancesHttp,
218 EMR.ListInstanceGroupsHttp,
219 EMR.ListBootstrapActionsHttp,
220 EMR.GetManagedScalingPolicyHttp,
221 EMR.PutManagedScalingPolicyHttp,
222 EMR.RemoveManagedScalingPolicyHttp,
223 ),
224 ),
225 ),
226);