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 DmsTestFunctionLive, { DmsTestFunction } from "./handler";
12
13const testOptions = { providers: AWS.providers() };
14const { test, beforeAll, afterAll } = Test.make(testOptions);
15const sharedStack = Core.scratchStack(testOptions, "DmsBindings");
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;
25
26class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
27 readonly status: number;
28 readonly body: string;
29}> {}
30
31// The shared Lambda fixture occasionally answers a transient 5xx under load
32// (cold re-init, IAM propagation on the freshly attached policy that the
33// handler's `Effect.orDie` surfaces as a 500). Retry only 5xx; a genuine
34// 4xx/assertion failure surfaces immediately.
35const send = (request: HttpClientRequest.HttpClientRequest) =>
36 HttpClient.execute(request).pipe(
37 Effect.flatMap((response) =>
38 response.status >= 500
39 ? response.text.pipe(
40 Effect.flatMap((body) =>
41 Effect.fail(
42 new TransientUpstream({ status: response.status, body }),
43 ),
44 ),
45 )
46 : Effect.succeed(response),
47 ),
48 Effect.retry({
49 while: (e) => e._tag === "TransientUpstream",
50 schedule: Schedule.max([
51 Schedule.exponential("500 millis"),
52 Schedule.recurs(6),
53 ]),
54 }),
55 );
56
57describe.sequential(
58 "DMS Bindings",
59 { tags: ["provider:aws", "provider:aws:dms", "provider:aws:lambda", "live"] },
60 () => {
61 beforeAll(
62 Effect.gen(function* () {
63 yield* Effect.logInfo("DMS test setup: destroying previous resources");
64 yield* sharedStack.destroy();
65
66 yield* Effect.logInfo("DMS test setup: deploying fixture");
67 const { functionUrl } = yield* sharedStack.deploy(
68 Effect.gen(function* () {
69 return yield* DmsTestFunction;
70 }).pipe(Effect.provide(DmsTestFunctionLive)),
71 );
72
73 expect(functionUrl).toBeTruthy();
74 baseUrl = functionUrl!.replace(/\/+$/, "");
75
76 const readinessUrl = `${baseUrl}/bindings`;
77 yield* Effect.logInfo(
78 `DMS test setup: probing readiness at ${readinessUrl}`,
79 );
80 yield* HttpClient.get(readinessUrl).pipe(
81 Effect.flatMap((response) =>
82 response.status === 200
83 ? Effect.succeed(response)
84 : Effect.fail(
85 new Error(`Function not ready: ${response.status}`),
86 ),
87 ),
88 Effect.tapError((error) =>
89 Effect.logWarning(
90 `DMS test setup: fixture not ready yet (${String(error)})`,
91 ),
92 ),
93 Effect.retry({ schedule: readinessPolicy }),
94 );
95 }),
96 { timeout: 240_000 },
97 );
98
99 afterAll(sharedStack.destroy(), { timeout: 120_000 });
100
101 describe("binding registration", () => {
102 test.provider("all capabilities initialize in the runtime", (_stack) =>
103 Effect.gen(function* () {
104 const response = yield* send(
105 HttpClientRequest.get(`${baseUrl}/bindings`),
106 ).pipe(Effect.flatMap((r) => r.json));
107 expect((response as any).bound).toHaveLength(14);
108 }),
109 );
110 });
111
112 describe("DescribeSchemas", () => {
113 test.provider(
114 "answers with a typed fault for a never-refreshed endpoint",
115 (_stack) =>
116 Effect.gen(function* () {
117 const response = (yield* send(
118 HttpClientRequest.get(`${baseUrl}/schemas`),
119 ).pipe(Effect.flatMap((r) => r.json))) as {
120 schemas: unknown[];
121 fault: string | null;
122 };
123 // The fixture endpoint has never had a schema refresh (that needs
124 // a replication instance): DMS answers with a typed fault, or an
125 // empty schema list. An IAM grant gap would 500 the route instead.
126 if (response.fault !== null) {
127 expect([
128 "ResourceNotFoundFault",
129 "InvalidResourceStateFault",
130 ]).toContain(response.fault);
131 } else {
132 expect(response.schemas).toEqual([]);
133 }
134 }),
135 { timeout: 60_000 },
136 );
137 });
138
139 describe("DescribeRefreshSchemasStatus", () => {
140 test.provider(
141 "answers with a typed fault for a never-refreshed endpoint",
142 (_stack) =>
143 Effect.gen(function* () {
144 const response = (yield* send(
145 HttpClientRequest.get(`${baseUrl}/refresh-status`),
146 ).pipe(Effect.flatMap((r) => r.json))) as {
147 status: string | null;
148 fault: string | null;
149 };
150 if (response.fault !== null) {
151 expect([
152 "ResourceNotFoundFault",
153 "InvalidResourceStateFault",
154 ]).toContain(response.fault);
155 } else {
156 expect(response.status).toBeNull();
157 }
158 }),
159 { timeout: 60_000 },
160 );
161 });
162
163 describe("DescribeConnections", () => {
164 test.provider(
165 "reports zero connection tests for the fresh endpoint",
166 (_stack) =>
167 Effect.gen(function* () {
168 const response = (yield* send(
169 HttpClientRequest.get(`${baseUrl}/connections`),
170 ).pipe(Effect.flatMap((r) => r.json))) as { count: number };
171 expect(response.count).toBe(0);
172 }),
173 { timeout: 60_000 },
174 );
175 });
176
177 describe("DescribeEvents", () => {
178 test.provider(
179 "reads recent replication-instance events (empty on a quiet account)",
180 (_stack) =>
181 Effect.gen(function* () {
182 const response = (yield* send(
183 HttpClientRequest.get(`${baseUrl}/events`),
184 ).pipe(Effect.flatMap((r) => r.json))) as { count: number };
185 expect(response.count).toBeGreaterThanOrEqual(0);
186 }),
187 { timeout: 60_000 },
188 );
189 });
190
191 describe("DescribeEndpointSettings", () => {
192 test.provider(
193 "enumerates the settings the mysql engine accepts",
194 (_stack) =>
195 Effect.gen(function* () {
196 const response = (yield* send(
197 HttpClientRequest.get(`${baseUrl}/endpoint-settings`),
198 ).pipe(Effect.flatMap((r) => r.json))) as { count: number };
199 expect(response.count).toBeGreaterThan(0);
200 }),
201 { timeout: 60_000 },
202 );
203 });
204
205 describe("DescribeOrderableReplicationInstances", () => {
206 test.provider(
207 "enumerates orderable replication instance classes",
208 (_stack) =>
209 Effect.gen(function* () {
210 const response = (yield* send(
211 HttpClientRequest.get(`${baseUrl}/orderable`),
212 ).pipe(Effect.flatMap((r) => r.json))) as { count: number };
213 expect(response.count).toBeGreaterThan(0);
214 }),
215 { timeout: 60_000 },
216 );
217 });
218
219 describe("DescribeReplicationTasks", () => {
220 test.provider(
221 "answers a no-match filter with zero tasks or the typed fault",
222 (_stack) =>
223 Effect.gen(function* () {
224 const response = (yield* send(
225 HttpClientRequest.get(`${baseUrl}/tasks`),
226 ).pipe(Effect.flatMap((r) => r.json))) as {
227 count: number;
228 fault: string | null;
229 };
230 expect(response.count).toBe(0);
231 if (response.fault !== null) {
232 expect(response.fault).toBe("ResourceNotFoundFault");
233 }
234 }),
235 { timeout: 60_000 },
236 );
237 });
238
239 describe("StartReplicationTask", () => {
240 test.provider(
241 "surfaces the typed not-found fault for a nonexistent task",
242 (_stack) =>
243 Effect.gen(function* () {
244 const response = (yield* send(
245 HttpClientRequest.post(`${baseUrl}/task/start`),
246 ).pipe(Effect.flatMap((r) => r.json))) as { fault: string };
247 expect([
248 "ResourceNotFoundFault",
249 "InvalidResourceStateFault",
250 "AccessDeniedFault",
251 ]).toContain(response.fault);
252 }),
253 { timeout: 60_000 },
254 );
255 });
256
257 describe("StopReplicationTask", () => {
258 test.provider(
259 "surfaces the typed not-found fault for a nonexistent task",
260 (_stack) =>
261 Effect.gen(function* () {
262 const response = (yield* send(
263 HttpClientRequest.post(`${baseUrl}/task/stop`),
264 ).pipe(Effect.flatMap((r) => r.json))) as { fault: string };
265 expect([
266 "ResourceNotFoundFault",
267 "InvalidResourceStateFault",
268 ]).toContain(response.fault);
269 }),
270 { timeout: 60_000 },
271 );
272 });
273
274 describe("DescribeTableStatistics", () => {
275 test.provider(
276 "surfaces the typed not-found fault for a nonexistent task",
277 (_stack) =>
278 Effect.gen(function* () {
279 const response = (yield* send(
280 HttpClientRequest.get(`${baseUrl}/task/tables`),
281 ).pipe(Effect.flatMap((r) => r.json))) as { fault: string };
282 expect([
283 "ResourceNotFoundFault",
284 "InvalidResourceStateFault",
285 "AccessDeniedFault",
286 ]).toContain(response.fault);
287 }),
288 { timeout: 60_000 },
289 );
290 });
291
292 describe("ReloadTables", () => {
293 test.provider(
294 "surfaces the typed not-found fault for a nonexistent task",
295 (_stack) =>
296 Effect.gen(function* () {
297 const response = (yield* send(
298 HttpClientRequest.post(`${baseUrl}/task/reload`),
299 ).pipe(Effect.flatMap((r) => r.json))) as { fault: string };
300 expect([
301 "ResourceNotFoundFault",
302 "InvalidResourceStateFault",
303 ]).toContain(response.fault);
304 }),
305 { timeout: 60_000 },
306 );
307 });
308
309 describe("DescribeReplications", () => {
310 test.provider(
311 "answers a no-match filter with zero replications or the typed fault",
312 (_stack) =>
313 Effect.gen(function* () {
314 const response = (yield* send(
315 HttpClientRequest.get(`${baseUrl}/replications`),
316 ).pipe(Effect.flatMap((r) => r.json))) as {
317 count: number;
318 fault: string | null;
319 };
320 expect(response.count).toBe(0);
321 if (response.fault !== null) {
322 expect(response.fault).toBe("ResourceNotFoundFault");
323 }
324 }),
325 { timeout: 60_000 },
326 );
327 });
328
329 describe("StartReplication", () => {
330 test.provider(
331 "surfaces the typed not-found fault for a nonexistent config",
332 (_stack) =>
333 Effect.gen(function* () {
334 const response = (yield* send(
335 HttpClientRequest.post(`${baseUrl}/replication/start`),
336 ).pipe(Effect.flatMap((r) => r.json))) as { fault: string };
337 expect([
338 "ResourceNotFoundFault",
339 "InvalidResourceStateFault",
340 "AccessDeniedFault",
341 ]).toContain(response.fault);
342 }),
343 { timeout: 60_000 },
344 );
345 });
346
347 describe("StopReplication", () => {
348 test.provider(
349 "surfaces the typed not-found fault for a nonexistent config",
350 (_stack) =>
351 Effect.gen(function* () {
352 const response = (yield* send(
353 HttpClientRequest.post(`${baseUrl}/replication/stop`),
354 ).pipe(Effect.flatMap((r) => r.json))) as { fault: string };
355 expect([
356 "ResourceNotFoundFault",
357 "InvalidResourceStateFault",
358 "AccessDeniedFault",
359 ]).toContain(response.fault);
360 }),
361 { timeout: 60_000 },
362 );
363 });
364
365 describe(
366 "consumeReplicationEvents",
367 { tags: ["provider:aws:eventbridge"] },
368 () => {
369 test.provider(
370 "created the EventBridge rule for DMS replication task state changes",
371 () =>
372 Effect.gen(function* () {
373 // The rule's physical name embeds the fixture's logical id
374 // (`DmsTestFunction-DmsReplicationEvents`); find it on the default
375 // bus with bounded manual pagination.
376 let rule: eventbridge.Rule | undefined;
377 let nextToken: string | undefined;
378 for (let page = 0; page < 10 && !rule; page++) {
379 const result = yield* eventbridge.listRules({
380 NextToken: nextToken,
381 });
382 rule = (result.Rules ?? []).find((candidate) =>
383 candidate.Name?.includes("DmsReplicationEvents"),
384 );
385 nextToken = result.NextToken;
386 if (!nextToken) break;
387 }
388 expect(rule).toBeDefined();
389 expect(rule?.EventPattern).toContain("aws.dms");
390 expect(rule?.EventPattern).toContain(
391 "DMS Replication Task State Change",
392 );
393 }),
394 { timeout: 60_000 },
395 );
396 },
397 );
398 },
399);