1import * as EMRContainers from "@/AWS/EMRContainers";
2import { Role } 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 * as Schedule from "effect/Schedule";
8import * as Stream from "effect/Stream";
9import { HttpServerRequest } from "effect/http/HttpServerRequest";
10import * as HttpServerResponse from "effect/http/HttpServerResponse";
11import path from "pathe";
12
13const main = path.resolve(import.meta.dirname, "handler.ts");
14
15export class EmrcTestFunction extends Lambda.Function<Lambda.Function>()(
16 "EmrcTestFunction",
17) {}
18
19export default EmrcTestFunction.make(
20 {
21 main,
22 functionUrl: true,
23 timeout: Duration.seconds(30),
24 },
25 Effect.gen(function* () {
26 // The job template the template-scoped binding is bound to. Templates
27 // are free, account-level, and never run anything by themselves.
28 const jobRole = yield* Role("JobRole", {
29 assumeRolePolicyDocument: {
30 Version: "2012-10-17",
31 Statement: [
32 {
33 Effect: "Allow",
34 Principal: { Service: "pods.eks.amazonaws.com" },
35 Action: ["sts:AssumeRole", "sts:TagSession"],
36 },
37 ],
38 },
39 });
40 const template = yield* EMRContainers.JobTemplate("BindingTemplate", {
41 jobTemplateData: {
42 executionRoleArn: jobRole.roleArn,
43 releaseLabel: "emr-7.5.0-latest",
44 jobDriver: {
45 sparkSubmitJobDriver: {
46 entryPoint: "s3://alchemy-test-emrc/scripts/etl.py",
47 },
48 },
49 },
50 });
51
52 // Event source: subscribe the host to EMR on EKS job run state changes.
53 // The deploy proves the EventBridge rule + invoke permission wiring.
54 yield* EMRContainers.consumeJobRunEvents(
55 { states: ["FAILED", "COMPLETED"] },
56 (events) =>
57 Stream.runForEach(events, (event) =>
58 Effect.log(`job run ${event.detail.id} -> ${event.detail.state}`),
59 ),
60 );
61
62 const describeJobTemplate =
63 yield* EMRContainers.DescribeJobTemplate(template);
64 const listJobTemplates = yield* EMRContainers.ListJobTemplates();
65 const listVirtualClusters = yield* EMRContainers.ListVirtualClusters();
66 const templateId = yield* template.jobTemplateId;
67
68 const bound = {
69 describeJobTemplate,
70 listJobTemplates,
71 listVirtualClusters,
72 };
73
74 // Drain a bounded number of pages on every observation. AWS list APIs may
75 // return a short (or empty) page with a continuation token while the
76 // account is being mutated by other test files, so repeatedly reading only
77 // page one makes the full-concurrency sweep flaky.
78 const listAllJobTemplates = Effect.fn(function* () {
79 const templates: Array<{ readonly id?: string }> = [];
80 let nextToken: string | undefined;
81 for (let page = 0; page < 10; page++) {
82 const response = yield* listJobTemplates({
83 maxResults: 100,
84 nextToken,
85 });
86 templates.push(...(response.templates ?? []));
87 nextToken = response.nextToken;
88 if (nextToken === undefined) break;
89 }
90 return templates;
91 });
92
93 return {
94 fetch: Effect.gen(function* () {
95 const request = yield* HttpServerRequest;
96 const url = new URL(request.originalUrl);
97 const pathname = url.pathname;
98
99 if (request.method === "GET" && pathname === "/bindings") {
100 return yield* HttpServerResponse.json({
101 bound: Object.keys(bound),
102 });
103 }
104
105 // Template-scoped read: the template id is injected by the binding.
106 if (request.method === "GET" && pathname === "/template") {
107 const { jobTemplate } = yield* describeJobTemplate();
108 return yield* HttpServerResponse.json({
109 id: jobTemplate?.id,
110 name: jobTemplate?.name,
111 releaseLabel: jobTemplate?.jobTemplateData.releaseLabel,
112 });
113 }
114
115 // Account-level list.
116 if (request.method === "GET" && pathname === "/templates") {
117 const expectedTemplateId = yield* templateId;
118 const templates = yield* Effect.repeat(listAllJobTemplates(), {
119 schedule: Schedule.fixed("2 seconds"),
120 until: (items) =>
121 items.some((item) => item.id === expectedTemplateId),
122 times: 10,
123 });
124 return yield* HttpServerResponse.json({
125 ids: templates.map((template) => template.id),
126 });
127 }
128
129 // Account-level virtual cluster list (proves IAM grant + call path
130 // without needing an EKS-backed virtual cluster to exist).
131 if (request.method === "GET" && pathname === "/virtual-clusters") {
132 const { virtualClusters } = yield* listVirtualClusters({
133 states: ["RUNNING"],
134 });
135 return yield* HttpServerResponse.json({
136 ids: (virtualClusters ?? []).map((vc) => vc.id),
137 });
138 }
139
140 return yield* HttpServerResponse.json(
141 { error: "Not found", method: request.method, pathname },
142 { status: 404 },
143 );
144 }).pipe(Effect.orDie),
145 };
146 }).pipe(
147 Effect.provide(
148 Layer.mergeAll(
149 Lambda.EventSource,
150 EMRContainers.DescribeJobTemplateHttp,
151 EMRContainers.ListJobTemplatesHttp,
152 EMRContainers.ListVirtualClustersHttp,
153 ),
154 ),
155 ),
156);