1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as s3 from "@distilled.cloud/aws/s3";
5import { describe, expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as HttpClient from "effect/http/HttpClient";
10import * as HttpClientRequest from "effect/http/HttpClientRequest";
11import DataBrewTestFunctionLive, {
12 DataBrewTestFunction,
13 foundation,
14 SOURCE_KEY,
15} from "./handler";
16
17const testOptions = { providers: AWS.providers() };
18const { test, beforeAll, afterAll } = Test.make(testOptions);
19const sharedStack = Core.scratchStack(testOptions, "DataBrewBindings");
20
21// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
22// over 60s on a fresh deploy under parallel-suite load.
23const readinessPolicy = Schedule.max([
24 Schedule.fixed("2 seconds"),
25 Schedule.recurs(75),
26]);
27
28let baseUrl: string;
29
30class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
31 readonly status: number;
32 readonly body: string;
33}> {}
34
35// Retry transient 5xx from the shared Lambda fixture (cold re-init, IAM
36// propagation on the freshly attached databrew policy surfaced as a 500 by
37// the handler's `Effect.orDie`). Genuine 4xx/assertion failures return
38// immediately.
39const send = (request: HttpClientRequest.HttpClientRequest) =>
40 HttpClient.execute(request).pipe(
41 Effect.flatMap((response) =>
42 response.status >= 500
43 ? response.text.pipe(
44 Effect.flatMap((body) =>
45 Effect.fail(
46 new TransientUpstream({ status: response.status, body }),
47 ),
48 ),
49 )
50 : Effect.succeed(response),
51 ),
52 Effect.retry({
53 while: (e) => e._tag === "TransientUpstream",
54 schedule: Schedule.max([
55 Schedule.exponential("500 millis"),
56 Schedule.recurs(6),
57 ]),
58 }),
59 );
60
61const getJson = (url: string) =>
62 send(HttpClientRequest.get(url)).pipe(Effect.flatMap((r) => r.json));
63
64const postJson = (url: string, body: unknown) =>
65 send(
66 HttpClientRequest.post(url).pipe(HttpClientRequest.bodyJsonUnsafe(body)),
67 ).pipe(Effect.flatMap((r) => r.json));
68
69const post = (url: string) =>
70 send(HttpClientRequest.post(url)).pipe(Effect.flatMap((r) => r.json));
71
72/**
73 * A route answered with a typed error tag. The tag being present proves the
74 * binding produced a typed failure (untyped errors crash into a 500), and it
75 * not being an authorization tag proves the IAM grant covered the call.
76 */
77const expectTypedNonAuthz = (body: any) => {
78 expect(typeof body.errorTag).toBe("string");
79 expect(body.errorTag).not.toBe("AccessDeniedException");
80};
81
82/** A route that either succeeded or failed with a typed, authorized tag. */
83const expectAuthorized = (body: any) => {
84 if (body.errorTag !== undefined) expectTypedNonAuthz(body);
85};
86
87class RunStillActive extends Data.TaggedError("RunStillActive") {}
88
89/** Poll a run until it leaves STARTING/RUNNING (bounded). */
90const waitForInactive = (runId: string) =>
91 getJson(`${baseUrl}/run/get?id=${encodeURIComponent(runId)}`).pipe(
92 Effect.flatMap((body: any) =>
93 body.state === undefined ||
94 body.state === "STARTING" ||
95 body.state === "RUNNING"
96 ? Effect.fail(new RunStillActive())
97 : Effect.succeed(body.state as string),
98 ),
99 Effect.retry({
100 while: (e): boolean => e._tag === "RunStillActive",
101 schedule: Schedule.max([
102 Schedule.fixed("5 seconds"),
103 Schedule.recurs(12),
104 ]),
105 }),
106 );
107
108describe.sequential(
109 "DataBrew Bindings",
110 {
111 tags: [
112 "provider:aws",
113 "provider:aws:databrew",
114 "provider:aws:iam",
115 "provider:aws:lambda",
116 "provider:aws:s3",
117 "live",
118 ],
119 },
120 () => {
121 beforeAll(
122 Effect.gen(function* () {
123 yield* Effect.logInfo("DataBrew test setup: destroying previous");
124 yield* sharedStack.destroy();
125
126 // Stage 1: foundation only — the CSV must exist in the bucket before
127 // the job/project are created (DataBrew validates the dataset's
128 // source with the job role at Create{Profile}Job/CreateProject time).
129 yield* Effect.logInfo("DataBrew test setup: deploying foundation");
130 const base = yield* sharedStack.deploy(foundation);
131 // `beforeAll` doesn't run inside `test.provider`'s environment, so
132 // provide the AWS providers (Credentials/Region) explicitly for the
133 // out-of-band distilled call.
134 yield* Core.withProviders(
135 s3.putObject({
136 Bucket: base.bucket.bucketName,
137 Key: SOURCE_KEY,
138 Body: new TextEncoder().encode("id,name\n1,alice\n2,bob\n"),
139 ContentType: "text/csv",
140 }),
141 testOptions,
142 "DataBrewBindingsSeed",
143 );
144
145 // Stage 2: the full fixture (job, project, Lambda + bindings).
146 yield* Effect.logInfo("DataBrew test setup: deploying fixture");
147 const { functionUrl } = yield* sharedStack.deploy(
148 Effect.gen(function* () {
149 return yield* DataBrewTestFunction;
150 }).pipe(Effect.provide(DataBrewTestFunctionLive)),
151 );
152
153 expect(functionUrl).toBeTruthy();
154 baseUrl = functionUrl!.replace(/\/+$/, "");
155 const readinessUrl = `${baseUrl}/run/list`;
156
157 yield* Effect.logInfo(
158 `DataBrew test setup: probing readiness at ${readinessUrl}`,
159 );
160 // Ready = the function answers 200 AND the freshly attached databrew
161 // policy has propagated (an AccessDeniedException errorTag means IAM
162 // is still converging — keep probing).
163 yield* HttpClient.get(readinessUrl).pipe(
164 Effect.flatMap((response) =>
165 response.status === 200
166 ? response.json
167 : Effect.fail(
168 new Error(`Function not ready: ${response.status}`),
169 ),
170 ),
171 Effect.flatMap((body: any) =>
172 body.errorTag === undefined
173 ? Effect.succeed(body)
174 : Effect.fail(new Error(`IAM not propagated: ${body.errorTag}`)),
175 ),
176 Effect.retry({ schedule: readinessPolicy }),
177 );
178 }),
179 { timeout: 480_000 },
180 );
181
182 afterAll.skipIf(!!process.env.NO_DESTROY)(sharedStack.destroy(), {
183 timeout: 240_000,
184 });
185
186 describe("StartJobRun + ListJobRuns + DescribeJobRun + StopJobRun", () => {
187 test.provider(
188 "starts a profile run, observes it, stops it",
189 (_stack) =>
190 Effect.gen(function* () {
191 // StartJobRun
192 const started = (yield* post(`${baseUrl}/run/start`)) as any;
193 expect(started.errorTag).toBeUndefined();
194 expect(typeof started.runId).toBe("string");
195 const runId = started.runId as string;
196
197 // ListJobRuns sees it
198 const listed = (yield* getJson(`${baseUrl}/run/list`)) as any;
199 expect(listed.errorTag).toBeUndefined();
200 expect(listed.runIds).toContain(runId);
201
202 // DescribeJobRun reports a state
203 const described = (yield* getJson(
204 `${baseUrl}/run/get?id=${encodeURIComponent(runId)}`,
205 )) as any;
206 expect(described.errorTag).toBeUndefined();
207 expect(typeof described.state).toBe("string");
208
209 // StopJobRun — don't leave managed Spark capacity spinning.
210 const stopped = (yield* postJson(`${baseUrl}/run/stop`, {
211 id: runId,
212 })) as any;
213 expectAuthorized(stopped);
214
215 // The run leaves STARTING/RUNNING (bounded poll).
216 const state = yield* waitForInactive(runId);
217 expect(["STOPPING", "STOPPED", "FAILED", "TIMEOUT"]).toContain(
218 state,
219 );
220 }),
221 { timeout: 240_000 },
222 );
223 });
224
225 describe("PublishRecipe", () => {
226 test.provider("publishes a new numbered version", (_stack) =>
227 Effect.gen(function* () {
228 const body = (yield* post(`${baseUrl}/recipe/publish`)) as any;
229 expect(body.errorTag).toBeUndefined();
230 expect(typeof body.name).toBe("string");
231 }),
232 );
233 });
234
235 describe("StartProjectSession + SendProjectSessionAction", () => {
236 test.provider(
237 "opens an interactive session and sends a preview action",
238 (_stack) =>
239 Effect.gen(function* () {
240 const body = (yield* post(`${baseUrl}/session/run`)) as any;
241 // Session provisioning is asynchronous — success and a typed
242 // rejection both prove the IAM grants and the Redacted
243 // ClientSessionId round-trip.
244 expectAuthorized(body.started ?? body);
245 if (body.started?.hasSessionId === true) {
246 expect(body.started.name).toBeTruthy();
247 expectAuthorized(body.action);
248 }
249 }),
250 { timeout: 120_000 },
251 );
252 });
253 },
254);