1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as logs from "@distilled.cloud/aws/cloudwatch-logs";
5import * as dataexchange from "@distilled.cloud/aws/dataexchange";
6import * as eventbridge from "@distilled.cloud/aws/eventbridge";
7import { describe, expect } from "alchemy-test";
8import * as Data from "effect/Data";
9import * as Effect from "effect/Effect";
10import * as Schedule from "effect/Schedule";
11import * as HttpClient from "effect/http/HttpClient";
12import * as HttpClientRequest from "effect/http/HttpClientRequest";
13import DataExchangeTestFunctionLive, {
14 DataExchangeTestFunction,
15} from "./handler";
16
17const testOptions = { providers: AWS.providers() };
18const { test, beforeAll, afterAll } = Test.make(testOptions);
19const sharedStack = Core.scratchStack(testOptions, "DataExchangeBindings");
20
21// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
22// over 60s on a fresh deploy.
23const readinessPolicy = Schedule.max([
24 Schedule.fixed("2 seconds"),
25 Schedule.recurs(75),
26]);
27
28let baseUrl: string;
29let functionArn: string;
30let functionName: string | undefined;
31let fixtureDataSetId: string | undefined;
32
33// beforeAll/afterAll hooks run outside `test.provider`'s layer, so raw
34// distilled calls need the provider layer (credentials, region) supplied
35// explicitly.
36const aws = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
37 Core.withProviders(effect, testOptions, sharedStack.name);
38
39class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
40 readonly status: number;
41 readonly body: string;
42}> {}
43
44// The shared Lambda fixture occasionally answers a transient 5xx under load
45// (cold re-init, IAM propagation on the freshly attached policy that the
46// handler's `Effect.orDie` surfaces as a 500). Retry only 5xx; a genuine
47// 4xx/assertion failure surfaces immediately.
48const send = (request: HttpClientRequest.HttpClientRequest) =>
49 HttpClient.execute(request).pipe(
50 Effect.flatMap((response) =>
51 response.status >= 500
52 ? response.text.pipe(
53 Effect.flatMap((body) =>
54 Effect.fail(
55 new TransientUpstream({ status: response.status, body }),
56 ),
57 ),
58 )
59 : Effect.succeed(response),
60 ),
61 Effect.retry({
62 while: (e) => e._tag === "TransientUpstream",
63 schedule: Schedule.max([
64 Schedule.exponential("500 millis"),
65 Schedule.recurs(6),
66 ]),
67 }),
68 );
69
70const getJson = (path: string) =>
71 send(HttpClientRequest.get(`${baseUrl}${path}`)).pipe(
72 Effect.flatMap((r) => r.json),
73 );
74
75const postJson = (path: string) =>
76 send(HttpClientRequest.post(`${baseUrl}${path}`)).pipe(
77 Effect.flatMap((r) => r.json),
78 );
79
80class IamPropagationLag extends Data.TaggedError("IamPropagationLag")<{
81 readonly path: string;
82}> {}
83
84// A freshly attached IAM policy can lag behind the Lambda's first calls —
85// DataExchange then rejects with AccessDeniedException, which the handler
86// surfaces as `{ error: "AccessDeniedException" }`. Retry the route through
87// the propagation window (bounded ~5.5 minutes — under full-suite load IAM
88// propagation for fresh roles routinely exceeds 4 minutes); any other
89// outcome surfaces immediately.
90const postJsonThroughIamPropagation = (path: string) =>
91 postJson(path).pipe(
92 Effect.flatMap((body) =>
93 (body as { error?: string }).error === "AccessDeniedException"
94 ? Effect.fail(new IamPropagationLag({ path }))
95 : Effect.succeed(body),
96 ),
97 Effect.retry({
98 while: (e) => e._tag === "IamPropagationLag",
99 schedule: Schedule.max([
100 Schedule.spaced("10 seconds"),
101 Schedule.recurs(33),
102 ]),
103 }),
104 );
105
106describe.sequential(
107 "DataExchange Bindings",
108 {
109 tags: [
110 "provider:aws",
111 "provider:aws:dataexchange",
112 "provider:aws:lambda",
113 "provider:aws:s3",
114 "live",
115 ],
116 },
117 () => {
118 beforeAll(
119 Effect.gen(function* () {
120 yield* Effect.logInfo(
121 "DataExchange test setup: destroying previous resources",
122 );
123 yield* sharedStack.destroy();
124
125 yield* Effect.logInfo("DataExchange test setup: deploying fixture");
126 const attrs = yield* sharedStack.deploy(
127 Effect.gen(function* () {
128 return yield* DataExchangeTestFunction;
129 }).pipe(Effect.provide(DataExchangeTestFunctionLive)),
130 );
131
132 expect(attrs.functionUrl).toBeTruthy();
133 baseUrl = attrs.functionUrl!.replace(/\/+$/, "");
134 functionArn = attrs.functionArn;
135 functionName = attrs.functionName;
136
137 const readinessUrl = `${baseUrl}/bindings`;
138 yield* Effect.logInfo(
139 `DataExchange test setup: probing readiness at ${readinessUrl}`,
140 );
141 yield* HttpClient.get(readinessUrl).pipe(
142 Effect.flatMap((response) =>
143 response.status === 200
144 ? Effect.succeed(response)
145 : Effect.fail(
146 new Error(`Function not ready: ${response.status}`),
147 ),
148 ),
149 Effect.tapError((error) =>
150 Effect.logWarning(
151 `DataExchange test setup: fixture not ready yet (${String(error)})`,
152 ),
153 ),
154 Effect.retry({ schedule: readinessPolicy }),
155 );
156
157 // Capture the fixture data set's id so afterAll can assert it is gone
158 // after the final destroy.
159 const dataSet = (yield* getJson("/data-set")) as { id: string };
160 fixtureDataSetId = dataSet.id;
161 }),
162 { timeout: 240_000 },
163 );
164
165 afterAll(
166 Effect.gen(function* () {
167 yield* sharedStack.destroy();
168 // Assert the fixture's data set really is gone (zero orphans).
169 if (fixtureDataSetId !== undefined) {
170 const gone = yield* Effect.flip(
171 aws(dataexchange.getDataSet({ DataSetId: fixtureDataSetId })),
172 );
173 expect(gone._tag).toBe("ResourceNotFoundException");
174 }
175
176 // The Lambda provider reaps /aws/lambda/{name} on delete and watches
177 // ~90s for a flush-driven recreation, but this suite invokes the
178 // function right up until destroy, and the Lambda service occasionally
179 // flushes a final log batch AFTER that watch ends — silently
180 // re-creating the just-deleted group. Watch a bit longer here with a
181 // bounded observe→delete loop (every delete is idempotent); a flush
182 // landing after the final observation is unobservable and the nuke
183 // census is the backstop.
184 if (functionName !== undefined) {
185 const logGroupName = `/aws/lambda/${functionName}`;
186 const reapIfObserved = Effect.gen(function* () {
187 const present = yield* aws(
188 logs.describeLogGroups({
189 logGroupNamePrefix: logGroupName,
190 limit: 1,
191 }),
192 ).pipe(
193 Effect.map((response) =>
194 (response.logGroups ?? []).some(
195 (group) => group.logGroupName === logGroupName,
196 ),
197 ),
198 );
199 if (present) {
200 yield* aws(logs.deleteLogGroup({ logGroupName })).pipe(
201 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
202 );
203 }
204 return present;
205 });
206 // Fixed 6 passes over ~100s — do NOT stop on first absence, since an
207 // absent group can still be recreated by a late flush.
208 yield* reapIfObserved.pipe(
209 Effect.repeat({
210 schedule: Schedule.spaced("20 seconds"),
211 times: 5,
212 }),
213 );
214 const stillPresent = yield* reapIfObserved;
215 expect(stillPresent).toBe(false);
216 }
217 }),
218 { timeout: 420_000 },
219 );
220
221 describe("binding registration", () => {
222 test.provider("all capabilities initialize in the runtime", (_stack) =>
223 Effect.gen(function* () {
224 const response = (yield* getJson("/bindings")) as { bound: string[] };
225 expect(response.bound).toContain("getDataSet");
226 expect(response.bound).toContain("createJob");
227 expect(response.bound).toContain("listReceivedDataGrants");
228 expect(response.bound).toHaveLength(16);
229 }),
230 );
231 });
232
233 describe("GetDataSet", () => {
234 test.provider(
235 "reads the bound data set's detail (injected data set id)",
236 (_stack) =>
237 Effect.gen(function* () {
238 const response = (yield* getJson("/data-set")) as {
239 id: string;
240 name: string;
241 assetType: string;
242 origin: string;
243 };
244 expect(response.id).toBeTruthy();
245 expect(response.assetType).toBe("S3_SNAPSHOT");
246 expect(response.origin).toBe("OWNED");
247 }),
248 );
249 });
250
251 describe("ListDataSetRevisions", () => {
252 test.provider("enumerates the bound data set's revisions", (_stack) =>
253 Effect.gen(function* () {
254 const revision = (yield* getJson("/revision")) as { id: string };
255 const response = (yield* getJson("/revisions")) as { ids: string[] };
256 expect(response.ids).toContain(revision.id);
257 }),
258 );
259 });
260
261 describe("GetRevision", () => {
262 test.provider(
263 "reads the bound revision (injected data set + revision ids)",
264 (_stack) =>
265 Effect.gen(function* () {
266 const response = (yield* getJson("/revision")) as {
267 id: string;
268 finalized: boolean;
269 };
270 expect(response.id).toBeTruthy();
271 expect(response.finalized).toBe(false);
272 }),
273 );
274 });
275
276 describe("ListDataSets", () => {
277 test.provider(
278 "enumerates the account's owned data sets including the fixture's",
279 (_stack) =>
280 Effect.gen(function* () {
281 const dataSet = (yield* getJson("/data-set")) as { id: string };
282 const response = (yield* getJson("/data-sets")) as {
283 ids: string[];
284 };
285 expect(response.ids).toContain(dataSet.id);
286 }),
287 );
288 });
289
290 describe("CreateJob / StartJob / GetJob / ListRevisionAssets / GetAsset", () => {
291 test.provider(
292 "imports an S3 object into the revision via a job and reads it back",
293 (_stack) =>
294 Effect.gen(function* () {
295 const response = (yield* postJsonThroughIamPropagation(
296 "/import",
297 )) as {
298 jobState?: string;
299 jobErrors?: unknown[];
300 assetCount?: number;
301 assetName?: string;
302 error?: string;
303 message?: string;
304 };
305 expect(response.error, response.message).toBeUndefined();
306 expect(JSON.stringify(response.jobErrors ?? [])).toBe("[]");
307 expect(response.jobState).toBe("COMPLETED");
308 expect(response.assetCount).toBeGreaterThanOrEqual(1);
309 expect(response.assetName).toBe("prices.csv");
310 }),
311 // Covers the full IAM propagation retry budget (~5.5 min) plus the
312 // import job's own create → start → poll-to-COMPLETED runtime.
313 { timeout: 480_000 },
314 );
315
316 test.provider("ListJobs sees the import job", (_stack) =>
317 Effect.gen(function* () {
318 const response = (yield* getJson("/jobs")) as { states: string[] };
319 expect(response.states).toContain("COMPLETED");
320 }),
321 );
322 });
323
324 describe("SendDataSetNotification", () => {
325 // Provider-generated notifications only work for data sets attached to
326 // an AWS Marketplace data product, which cannot be provisioned
327 // self-contained. The typed rejection proves the binding's IAM grant,
328 // call path, and error decoding end-to-end.
329 test.provider(
330 "rejects a data set outside a Marketplace product with a typed error",
331 (_stack) =>
332 Effect.gen(function* () {
333 const response = (yield* postJsonThroughIamPropagation(
334 "/notify",
335 )) as {
336 ok: boolean;
337 error: string | undefined;
338 message: string | undefined;
339 };
340 expect(response.ok).toBe(false);
341 expect(response.error).toBe("ValidationException");
342 expect(response.message).toContain(
343 "not configured for AWS Marketplace",
344 );
345 }),
346 // Shares the IAM propagation retry budget with the /import test.
347 { timeout: 420_000 },
348 );
349 });
350
351 describe("ListDataGrants / ListReceivedDataGrants", () => {
352 test.provider("enumerates sent and received data grants", (_stack) =>
353 Effect.gen(function* () {
354 const response = (yield* getJson("/grants")) as {
355 sent: number;
356 received: number;
357 };
358 expect(response.sent).toBeGreaterThanOrEqual(0);
359 expect(response.received).toBeGreaterThanOrEqual(0);
360 }),
361 );
362 });
363
364 describe("ListEventActions", () => {
365 test.provider("enumerates the account's event actions", (_stack) =>
366 Effect.gen(function* () {
367 const response = (yield* getJson("/event-actions")) as {
368 count: number;
369 };
370 expect(response.count).toBeGreaterThanOrEqual(0);
371 }),
372 );
373 });
374
375 describe(
376 "consumeDataSetEvents",
377 { tags: ["provider:aws:eventbridge"] },
378 () => {
379 test.provider(
380 "the deploy created an EventBridge rule targeting the function",
381 (_stack) =>
382 Effect.gen(function* () {
383 // Out-of-band via distilled: the fixture's consumeDataSetEvents
384 // must have materialized as a rule on the default bus with the
385 // Lambda as target.
386 const { RuleNames } = yield* eventbridge.listRuleNamesByTarget({
387 TargetArn: functionArn,
388 });
389 expect((RuleNames ?? []).length).toBeGreaterThanOrEqual(1);
390 }),
391 );
392 },
393 );
394 },
395);