1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as eventbridge from "@distilled.cloud/aws/eventbridge";
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 DataSyncTestFunctionLive, { DataSyncTestFunction } from "./handler";
12
13const testOptions = { providers: AWS.providers() };
14const { test, beforeAll, afterAll } = Test.make(testOptions);
15const sharedStack = Core.scratchStack(testOptions, "DataSyncBindings");
16
17// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
18// over 60s on a fresh deploy.
19const readinessPolicy = Schedule.max([
20 Schedule.fixed("2 seconds"),
21 Schedule.recurs(75),
22]);
23
24let baseUrl: string;
25let functionArn: string;
26
27class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
28 readonly status: number;
29 readonly body: string;
30}> {}
31
32// The shared Lambda fixture occasionally answers a transient 5xx under load
33// (cold re-init, IAM propagation on the freshly attached policy that the
34// handler's `Effect.orDie` surfaces as a 500). Retry only 5xx; a genuine
35// 4xx/assertion failure surfaces immediately.
36const send = (request: HttpClientRequest.HttpClientRequest) =>
37 HttpClient.execute(request).pipe(
38 Effect.flatMap((response) =>
39 response.status >= 500
40 ? response.text.pipe(
41 Effect.flatMap((body) =>
42 Effect.fail(
43 new TransientUpstream({ status: response.status, body }),
44 ),
45 ),
46 )
47 : Effect.succeed(response),
48 ),
49 Effect.retry({
50 while: (e) => e._tag === "TransientUpstream",
51 schedule: Schedule.max([
52 Schedule.exponential("500 millis"),
53 Schedule.recurs(6),
54 ]),
55 }),
56 );
57
58const getJson = (path: string) =>
59 send(HttpClientRequest.get(`${baseUrl}${path}`)).pipe(
60 Effect.flatMap((r) => r.json),
61 );
62
63const postJson = (path: string) =>
64 send(HttpClientRequest.post(`${baseUrl}${path}`)).pipe(
65 Effect.flatMap((r) => r.json),
66 );
67
68describe.sequential(
69 "DataSync Bindings",
70 {
71 tags: [
72 "provider:aws",
73 "provider:aws:datasync",
74 "provider:aws:iam",
75 "provider:aws:lambda",
76 "provider:aws:location",
77 "provider:aws:s3",
78 "live",
79 ],
80 },
81 () => {
82 beforeAll(
83 Effect.gen(function* () {
84 yield* Effect.logInfo(
85 "DataSync test setup: destroying previous resources",
86 );
87 yield* sharedStack.destroy();
88
89 yield* Effect.logInfo("DataSync test setup: deploying fixture");
90 const attrs = yield* sharedStack.deploy(
91 Effect.gen(function* () {
92 return yield* DataSyncTestFunction;
93 }).pipe(Effect.provide(DataSyncTestFunctionLive)),
94 );
95
96 expect(attrs.functionUrl).toBeTruthy();
97 baseUrl = attrs.functionUrl!.replace(/\/+$/, "");
98 functionArn = attrs.functionArn;
99
100 const readinessUrl = `${baseUrl}/bindings`;
101 yield* Effect.logInfo(
102 `DataSync test setup: probing readiness at ${readinessUrl}`,
103 );
104 yield* HttpClient.get(readinessUrl).pipe(
105 Effect.flatMap((response) =>
106 response.status === 200
107 ? Effect.succeed(response)
108 : Effect.fail(
109 new Error(`Function not ready: ${response.status}`),
110 ),
111 ),
112 Effect.tapError((error) =>
113 Effect.logWarning(
114 `DataSync test setup: fixture not ready yet (${String(error)})`,
115 ),
116 ),
117 Effect.retry({ schedule: readinessPolicy }),
118 );
119 }),
120 { timeout: 300_000 },
121 );
122
123 afterAll(sharedStack.destroy(), { timeout: 180_000 });
124
125 describe("binding registration", () => {
126 test.provider(
127 "all six capabilities initialize in the runtime",
128 (_stack) =>
129 Effect.gen(function* () {
130 const response = (yield* getJson("/bindings")) as {
131 bound: string[];
132 };
133 expect(response.bound).toHaveLength(6);
134 expect(response.bound).toContain("startTaskExecution");
135 expect(response.bound).toContain("cancelTaskExecution");
136 }),
137 );
138 });
139
140 describe("DescribeTask", () => {
141 test.provider(
142 "reads the bound task's detail (injected task arn)",
143 (_stack) =>
144 Effect.gen(function* () {
145 const response = (yield* getJson("/task")) as {
146 taskArn: string;
147 status: string;
148 };
149 expect(response.taskArn).toContain(":task/task-");
150 expect(typeof response.status).toBe("string");
151 }),
152 );
153 });
154
155 describe("StartTaskExecution / ListTaskExecutions / DescribeTaskExecution / UpdateTaskExecution / CancelTaskExecution", () => {
156 test.provider(
157 "starts a run, observes it, throttles, cancels, and waits it out",
158 (_stack) =>
159 Effect.gen(function* () {
160 // Start: the freshly-propagated IAM role can transiently fail the
161 // location access test (typed LocationAccessTestFailed, reported
162 // by the route as ok:false) — re-poll bounded.
163 const started = (yield* postJson("/start").pipe(
164 Effect.repeat({
165 schedule: Schedule.spaced("10 seconds"),
166 until: (r): boolean => (r as { ok: boolean }).ok === true,
167 times: 12,
168 }),
169 )) as { ok: boolean; executionArn?: string; tag?: string };
170 expect(started.ok).toBe(true);
171 expect(started.executionArn).toContain("/execution/exec-");
172 const executionArn = started.executionArn!;
173 const query = `?arn=${encodeURIComponent(executionArn)}`;
174
175 // Describe: the run reports a status.
176 const execution = (yield* getJson(`/execution${query}`)) as {
177 status: string;
178 };
179 expect(typeof execution.status).toBe("string");
180
181 // List: the run is enumerated under the bound task.
182 const executions = (yield* getJson("/executions")) as {
183 arns: string[];
184 };
185 expect(executions.arns).toContain(executionArn);
186
187 // Throttle: valid only in launching/preparing/transferring/
188 // verifying — either it lands or DataSync rejects it with the
189 // typed InvalidRequestException the route reports.
190 const throttled = (yield* postJson(`/throttle${query}`)) as {
191 updated: boolean;
192 tag?: string;
193 };
194 if (!throttled.updated) {
195 expect(throttled.tag).toBe("InvalidRequestException");
196 }
197
198 // Cancel: a queued/launching run is cancellable; if the empty
199 // transfer raced to completion, DataSync rejects with the typed
200 // InvalidRequestException instead.
201 const cancelled = (yield* postJson(`/cancel${query}`)) as {
202 cancelled: boolean;
203 tag?: string;
204 };
205 if (!cancelled.cancelled) {
206 expect(cancelled.tag).toBe("InvalidRequestException");
207 }
208
209 // Observe the run after the cancel: a cancelled queued run parks
210 // in CANCELLING (sometimes for minutes) before landing in ERROR —
211 // any of the three proves the round-trip. `stack.destroy()`
212 // deletes the task cleanly even with a CANCELLING execution, so
213 // there is no need to wait for a terminal state.
214 const observed = (yield* getJson(`/execution${query}`)) as {
215 status: string;
216 };
217 expect(["CANCELLING", "ERROR", "SUCCESS"]).toContain(
218 observed.status,
219 );
220 }),
221 { timeout: 240_000 },
222 );
223 });
224
225 describe(
226 "consumeTaskEvents",
227 { tags: ["provider:aws:eventbridge"] },
228 () => {
229 test.provider(
230 "the deploy created an EventBridge rule targeting the function",
231 (_stack) =>
232 Effect.gen(function* () {
233 // Out-of-band via distilled: the fixture's consumeTaskEvents must
234 // have materialized as a rule on the default bus with the Lambda
235 // as target.
236 const { RuleNames } = yield* eventbridge.listRuleNamesByTarget({
237 TargetArn: functionArn,
238 });
239 expect((RuleNames ?? []).length).toBeGreaterThanOrEqual(1);
240 }),
241 );
242 },
243 );
244 },
245);