1import * as IAM from "@/AWS/IAM";
2import * as Lambda from "@/AWS/Lambda";
3import * as MWAAServerless from "@/AWS/MWAAServerless";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Stream from "effect/Stream";
8import { HttpServerRequest } from "effect/http/HttpServerRequest";
9import * as HttpServerResponse from "effect/http/HttpServerResponse";
10import path from "pathe";
11
12const main = path.resolve(import.meta.dirname, "handler.ts");
13
14// A well-formed-but-nonexistent run id. Probes must answer with a typed
15// service error (ResourceNotFoundException or ValidationException), which
16// proves the IAM grant passed authorization AND the workflow ARN was
17// injected from the binding.
18const FAKE_RUN_ID = "00000000-0000-4000-8000-000000000000";
19
20/** Deterministic names shared with the test for out-of-band verification. */
21export const BINDINGS_WORKFLOW_NAME = "alchemy-test-mwaas-bind";
22export const BINDINGS_ROLE_NAME = "alchemy-test-mwaas-bind-role";
23// S3 bucket names are globally unique — keep a stable random-ish suffix.
24export const BINDINGS_BUCKET_NAME = "alchemy-test-mwaas-bind-defs-8k2f";
25export const DEFINITION_KEY = "workflows/bindings.yaml";
26
27/**
28 * The checked-in workflow definition the test uploads to S3 before the
29 * fixture deploys: a single S3ListOperator task listing the definitions
30 * bucket, schedule `null` so runs are on-demand only.
31 */
32export const WORKFLOW_DEFINITION = [
33 "alchemy-mwaa-serverless-bindings-test:",
34 " default_args:",
35 " owner: alchemy",
36 " schedule: null",
37 " tasks:",
38 " list_definitions:",
39 " operator: airflow.providers.amazon.aws.operators.s3.S3ListOperator",
40 ` bucket: ${BINDINGS_BUCKET_NAME}`,
41 " prefix: workflows/",
42 "",
43].join("\n");
44
45export class MwaaServerlessTestFunction extends Lambda.Function<Lambda.Function>()(
46 "MwaaServerlessTestFunction",
47) {}
48
49export default MwaaServerlessTestFunction.make(
50 {
51 main,
52 functionUrl: true,
53 timeout: Duration.seconds(30),
54 },
55 Effect.gen(function* () {
56 // The execution role the workflow's tasks assume — read access to the
57 // definitions bucket so the single S3ListOperator task can succeed.
58 const role = yield* IAM.Role("WorkflowRole", {
59 roleName: BINDINGS_ROLE_NAME,
60 assumeRolePolicyDocument: {
61 Version: "2012-10-17",
62 Statement: [
63 {
64 Effect: "Allow",
65 Principal: { Service: "airflow-serverless.amazonaws.com" },
66 Action: ["sts:AssumeRole"],
67 },
68 ],
69 },
70 inlinePolicies: {
71 definitions: {
72 Version: "2012-10-17",
73 Statement: [
74 {
75 Effect: "Allow",
76 Action: ["s3:GetObject*", "s3:GetBucket*", "s3:List*"],
77 Resource: [
78 `arn:aws:s3:::${BINDINGS_BUCKET_NAME}`,
79 `arn:aws:s3:::${BINDINGS_BUCKET_NAME}/*`,
80 ],
81 },
82 ],
83 },
84 },
85 });
86
87 // The workflow the bindings are bound to. The definition object is
88 // uploaded out-of-band by the test BEFORE this fixture deploys.
89 const workflow = yield* MWAAServerless.Workflow("BindingWorkflow", {
90 name: BINDINGS_WORKFLOW_NAME,
91 definitionS3Location: {
92 bucket: BINDINGS_BUCKET_NAME,
93 objectKey: DEFINITION_KEY,
94 },
95 roleArn: role.roleArn,
96 description: "alchemy mwaa-serverless bindings fixture",
97 tags: { purpose: "alchemy-test" },
98 });
99
100 // Event source: subscribe the host to workflow-run state changes. The
101 // deploy proves the EventBridge rule + invoke permission.
102 yield* MWAAServerless.consumeWorkflowRunEvents(
103 { runStates: ["Succeeded", "Failed", "Stopped"] },
104 (events) =>
105 Stream.runForEach(events, (event) =>
106 Effect.log(
107 `mwaa-serverless run ${event.detail.runId} -> ${event["detail-type"]}`,
108 ),
109 ),
110 );
111
112 const startWorkflowRun = yield* MWAAServerless.StartWorkflowRun(workflow);
113 const getWorkflowRun = yield* MWAAServerless.GetWorkflowRun(workflow);
114 const stopWorkflowRun = yield* MWAAServerless.StopWorkflowRun(workflow);
115 const listWorkflowRuns = yield* MWAAServerless.ListWorkflowRuns(workflow);
116 const getTaskInstance = yield* MWAAServerless.GetTaskInstance(workflow);
117 const listTaskInstances = yield* MWAAServerless.ListTaskInstances(workflow);
118 const listWorkflowVersions =
119 yield* MWAAServerless.ListWorkflowVersions(workflow);
120
121 const bound = {
122 startWorkflowRun,
123 getWorkflowRun,
124 stopWorkflowRun,
125 listWorkflowRuns,
126 getTaskInstance,
127 listTaskInstances,
128 listWorkflowVersions,
129 };
130
131 // Run an operation and answer with its typed outcome: `ok` on success,
132 // the error's `_tag` otherwise. Probes assert the tag — a typed service
133 // error (not AccessDeniedException) proves the IAM grant and the
134 // injected workflow ARN both work.
135 const probe = <A, E extends { _tag: string }>(
136 effect: Effect.Effect<A, E>,
137 ) =>
138 effect.pipe(
139 Effect.map(() => ({ tag: "ok", detail: "" }) as const),
140 Effect.catch((error) =>
141 Effect.succeed({ tag: error._tag, detail: String(error) } as const),
142 ),
143 Effect.flatMap((outcome) => HttpServerResponse.json(outcome)),
144 );
145
146 return {
147 fetch: Effect.gen(function* () {
148 const request = yield* HttpServerRequest;
149 const url = new URL(request.originalUrl);
150 const pathname = url.pathname;
151
152 if (request.method === "GET" && pathname === "/bindings") {
153 return yield* HttpServerResponse.json({ bound: Object.keys(bound) });
154 }
155
156 // Real reads: workflow ARN injected from the binding.
157 if (request.method === "GET" && pathname === "/runs") {
158 const { WorkflowRuns } = yield* listWorkflowRuns().pipe(Effect.orDie);
159 return yield* HttpServerResponse.json({
160 ids: (WorkflowRuns ?? []).map((run) => run.RunId),
161 });
162 }
163 if (request.method === "GET" && pathname === "/versions") {
164 const { WorkflowVersions } = yield* listWorkflowVersions().pipe(
165 Effect.orDie,
166 );
167 return yield* HttpServerResponse.json({
168 versions: (WorkflowVersions ?? []).map((v) => v.WorkflowVersion),
169 });
170 }
171
172 // Typed not-found probes on a nonexistent run.
173 if (request.method === "GET" && pathname === "/run-fake") {
174 return yield* probe(getWorkflowRun({ RunId: FAKE_RUN_ID }));
175 }
176 if (request.method === "GET" && pathname === "/tasks-fake") {
177 return yield* probe(listTaskInstances({ RunId: FAKE_RUN_ID }));
178 }
179 if (request.method === "GET" && pathname === "/task-fake") {
180 return yield* probe(
181 getTaskInstance({
182 RunId: FAKE_RUN_ID,
183 TaskInstanceId: FAKE_RUN_ID,
184 }),
185 );
186 }
187 if (request.method === "POST" && pathname === "/run-stop-fake") {
188 return yield* probe(stopWorkflowRun({ RunId: FAKE_RUN_ID }));
189 }
190
191 // Real run round trip: start an on-demand run; the test observes it
192 // and stops it immediately after.
193 if (request.method === "POST" && pathname === "/run-start") {
194 const started = yield* startWorkflowRun().pipe(Effect.orDie);
195 return yield* HttpServerResponse.json({
196 runId: started.RunId,
197 status: started.Status,
198 });
199 }
200 if (request.method === "GET" && pathname === "/run-detail") {
201 const id = url.searchParams.get("id");
202 if (!id) {
203 return yield* HttpServerResponse.json(
204 { error: "id query parameter required" },
205 { status: 400 },
206 );
207 }
208 const run = yield* getWorkflowRun({ RunId: id }).pipe(Effect.orDie);
209 return yield* HttpServerResponse.json({
210 runId: run.RunId,
211 status: run.RunDetail?.RunState,
212 runType: run.RunType,
213 });
214 }
215 if (request.method === "POST" && pathname === "/run-stop") {
216 const id = url.searchParams.get("id");
217 if (!id) {
218 return yield* HttpServerResponse.json(
219 { error: "id query parameter required" },
220 { status: 400 },
221 );
222 }
223 return yield* probe(stopWorkflowRun({ RunId: id }));
224 }
225
226 return yield* HttpServerResponse.json(
227 { error: "Not found", method: request.method, pathname },
228 { status: 404 },
229 );
230 }).pipe(Effect.orDie),
231 };
232 }).pipe(
233 Effect.provide(
234 Layer.mergeAll(
235 Lambda.EventSource,
236 MWAAServerless.StartWorkflowRunHttp,
237 MWAAServerless.GetWorkflowRunHttp,
238 MWAAServerless.StopWorkflowRunHttp,
239 MWAAServerless.ListWorkflowRunsHttp,
240 MWAAServerless.GetTaskInstanceHttp,
241 MWAAServerless.ListTaskInstancesHttp,
242 MWAAServerless.ListWorkflowVersionsHttp,
243 ),
244 ),
245 ),
246);