1import * as DMS from "@/AWS/DMS";
2import * as Lambda from "@/AWS/Lambda";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Redacted from "effect/Redacted";
6import * as Stream from "effect/Stream";
7import { HttpServerRequest } from "effect/http/HttpServerRequest";
8import * as HttpServerResponse from "effect/http/HttpServerResponse";
9import path from "pathe";
10
11const main = path.resolve(import.meta.dirname, "handler.ts");
12
13export class DmsTestFunction extends Lambda.Function<Lambda.Function>()(
14 "DmsTestFunction",
15) {}
16
17export default DmsTestFunction.make(
18 {
19 main,
20 functionUrl: true,
21 },
22 Effect.gen(function* () {
23 // A DMS endpoint is metadata-only (free, fast) — the bindings exercise
24 // ARN injection + IAM grants without a live database or replication
25 // instance. (Instance-scoped bindings — TestConnection, RefreshSchemas,
26 // RebootReplicationInstance, DescribeReplicationInstanceTaskLogs — need
27 // a replication instance, which is AWS_TEST_SLOW territory.)
28 const endpoint = yield* DMS.Endpoint("BindingsSource", {
29 endpointType: "source",
30 engineName: "mysql",
31 serverName: "bindings-source-db.example.com",
32 port: 3306,
33 username: "admin",
34 password: Redacted.make("correct-horse-battery-staple"),
35 databaseName: "app",
36 });
37
38 // --- endpoint-scoped bindings ---
39 const describeSchemas = yield* DMS.DescribeSchemas(endpoint);
40 const refreshStatus = yield* DMS.DescribeRefreshSchemasStatus(endpoint);
41
42 // --- account-level bindings ---
43 const describeConnections = yield* DMS.DescribeConnections();
44 const describeEvents = yield* DMS.DescribeEvents();
45 const endpointSettings = yield* DMS.DescribeEndpointSettings();
46 const orderable = yield* DMS.DescribeOrderableReplicationInstances();
47
48 // --- task-orchestration bindings (account-level: task/config ARNs are
49 // runtime data — DMS generates the ids, so no resource to bind) ---
50 const describeReplicationTasks = yield* DMS.DescribeReplicationTasks();
51 const startReplicationTask = yield* DMS.StartReplicationTask();
52 const stopReplicationTask = yield* DMS.StopReplicationTask();
53 const describeTableStatistics = yield* DMS.DescribeTableStatistics();
54 const reloadTables = yield* DMS.ReloadTables();
55 const describeReplications = yield* DMS.DescribeReplications();
56 const startReplication = yield* DMS.StartReplication();
57 const stopReplication = yield* DMS.StopReplication();
58
59 // Deploy-time: creates the EventBridge rule (default bus, source
60 // aws.dms) targeting this Function. Runtime firing needs a replication
61 // instance/task lifecycle; the test verifies the rule deploys.
62 yield* DMS.consumeReplicationEvents({ kinds: ["task-state"] }, (events) =>
63 Stream.runForEach(events, (event) =>
64 Effect.log(
65 `dms replication event: ${event.detail.sourceId} -> ${event.detail.eventType}`,
66 ),
67 ),
68 );
69
70 const bound = {
71 describeSchemas,
72 refreshStatus,
73 describeConnections,
74 describeEvents,
75 endpointSettings,
76 orderable,
77 describeReplicationTasks,
78 startReplicationTask,
79 stopReplicationTask,
80 describeTableStatistics,
81 reloadTables,
82 describeReplications,
83 startReplication,
84 stopReplication,
85 };
86
87 const EndpointArn = yield* endpoint.endpointArn;
88
89 return {
90 fetch: Effect.gen(function* () {
91 const request = yield* HttpServerRequest;
92 const url = new URL(request.originalUrl);
93 const pathname = url.pathname;
94
95 if (request.method === "GET" && pathname === "/bindings") {
96 return yield* HttpServerResponse.json({
97 bound: Object.keys(bound),
98 });
99 }
100
101 if (request.method === "GET" && pathname === "/schemas") {
102 // The fixture endpoint has never had a schema refresh (that needs
103 // a replication instance), so DMS answers with a typed fault —
104 // which still proves ARN injection + the IAM grant end-to-end (a
105 // grant gap would surface AccessDeniedException and 500 the
106 // route).
107 const result = yield* describeSchemas().pipe(
108 Effect.map((response) => ({
109 schemas: response.Schemas ?? [],
110 fault: null as string | null,
111 })),
112 Effect.catchTag(
113 ["ResourceNotFoundFault", "InvalidResourceStateFault"],
114 (error) =>
115 Effect.succeed({ schemas: [], fault: error._tag as string }),
116 ),
117 );
118 return yield* HttpServerResponse.json(result);
119 }
120
121 if (request.method === "GET" && pathname === "/refresh-status") {
122 const result = yield* refreshStatus().pipe(
123 Effect.map((response) => ({
124 status: response.RefreshSchemasStatus?.Status ?? null,
125 fault: null as string | null,
126 })),
127 Effect.catchTag(
128 ["ResourceNotFoundFault", "InvalidResourceStateFault"],
129 (error) =>
130 Effect.succeed({ status: null, fault: error._tag as string }),
131 ),
132 );
133 return yield* HttpServerResponse.json(result);
134 }
135
136 if (request.method === "GET" && pathname === "/connections") {
137 // No connection test has ever run against the fixture endpoint;
138 // DMS answers a no-match filter with ResourceNotFoundFault.
139 const endpointArn = yield* EndpointArn;
140 const result = yield* describeConnections({
141 Filters: [{ Name: "endpoint-arn", Values: [endpointArn] }],
142 }).pipe(
143 Effect.map((response) => response.Connections ?? []),
144 Effect.catchTag("ResourceNotFoundFault", () => Effect.succeed([])),
145 );
146 return yield* HttpServerResponse.json({ count: result.length });
147 }
148
149 if (request.method === "GET" && pathname === "/events") {
150 const result = yield* describeEvents({
151 SourceType: "replication-instance",
152 Duration: 60,
153 });
154 return yield* HttpServerResponse.json({
155 count: (result.Events ?? []).length,
156 });
157 }
158
159 if (request.method === "GET" && pathname === "/endpoint-settings") {
160 const result = yield* endpointSettings({ EngineName: "mysql" });
161 return yield* HttpServerResponse.json({
162 count: (result.EndpointSettings ?? []).length,
163 });
164 }
165
166 if (request.method === "GET" && pathname === "/orderable") {
167 const result = yield* orderable();
168 return yield* HttpServerResponse.json({
169 count: (result.OrderableReplicationInstances ?? []).length,
170 });
171 }
172
173 // Well-formed-but-nonexistent task/config ARNs in this account —
174 // derived from the fixture endpoint's ARN (same region + account).
175 // Driving the task-orchestration bindings against them proves ARN
176 // plumbing + IAM grants + typed error decoding without paying for a
177 // replication instance: DMS answers with a typed fault (a grant gap
178 // would surface AccessDeniedException and 500 the route instead).
179 const arnPrefix = (yield* EndpointArn).split(":").slice(0, 5).join(":");
180 const NONEXISTENT_TASK_ARN = `${arnPrefix}:task:AAAAAAAAAAAAAAAAAAAAAAAAAA`;
181 const NONEXISTENT_CONFIG_ARN = `${arnPrefix}:replication-config:AAAAAAAAAAAAAAAAAAAAAAAAAA`;
182
183 if (request.method === "GET" && pathname === "/tasks") {
184 // A no-match filter answers with ResourceNotFoundFault.
185 const result = yield* describeReplicationTasks({
186 Filters: [
187 {
188 Name: "replication-task-id",
189 Values: ["alchemy-nonexistent-task"],
190 },
191 ],
192 }).pipe(
193 Effect.map((response) => ({
194 count: (response.ReplicationTasks ?? []).length,
195 fault: null as string | null,
196 })),
197 Effect.catchTag("ResourceNotFoundFault", (error) =>
198 Effect.succeed({ count: 0, fault: error._tag as string }),
199 ),
200 );
201 return yield* HttpServerResponse.json(result);
202 }
203
204 if (request.method === "POST" && pathname === "/task/start") {
205 const fault = yield* startReplicationTask({
206 ReplicationTaskArn: NONEXISTENT_TASK_ARN,
207 StartReplicationTaskType: "start-replication",
208 }).pipe(
209 Effect.map(() => "Started"),
210 Effect.catchTag(
211 [
212 "ResourceNotFoundFault",
213 "InvalidResourceStateFault",
214 "AccessDeniedFault",
215 ],
216 (error) => Effect.succeed(error._tag),
217 ),
218 );
219 return yield* HttpServerResponse.json({ fault });
220 }
221
222 if (request.method === "POST" && pathname === "/task/stop") {
223 const fault = yield* stopReplicationTask({
224 ReplicationTaskArn: NONEXISTENT_TASK_ARN,
225 }).pipe(
226 Effect.map(() => "Stopped"),
227 Effect.catchTag(
228 ["ResourceNotFoundFault", "InvalidResourceStateFault"],
229 (error) => Effect.succeed(error._tag),
230 ),
231 );
232 return yield* HttpServerResponse.json({ fault });
233 }
234
235 if (request.method === "GET" && pathname === "/task/tables") {
236 const fault = yield* describeTableStatistics({
237 ReplicationTaskArn: NONEXISTENT_TASK_ARN,
238 }).pipe(
239 Effect.map(() => "Found"),
240 Effect.catchTag(
241 [
242 "ResourceNotFoundFault",
243 "InvalidResourceStateFault",
244 "AccessDeniedFault",
245 ],
246 (error) => Effect.succeed(error._tag),
247 ),
248 );
249 return yield* HttpServerResponse.json({ fault });
250 }
251
252 if (request.method === "POST" && pathname === "/task/reload") {
253 const fault = yield* reloadTables({
254 ReplicationTaskArn: NONEXISTENT_TASK_ARN,
255 TablesToReload: [
256 { SchemaName: "public", TableName: "nonexistent" },
257 ],
258 }).pipe(
259 Effect.map(() => "Reloaded"),
260 Effect.catchTag(
261 ["ResourceNotFoundFault", "InvalidResourceStateFault"],
262 (error) => Effect.succeed(error._tag),
263 ),
264 );
265 return yield* HttpServerResponse.json({ fault });
266 }
267
268 if (request.method === "GET" && pathname === "/replications") {
269 const result = yield* describeReplications({
270 Filters: [
271 {
272 Name: "replication-config-arn",
273 Values: [NONEXISTENT_CONFIG_ARN],
274 },
275 ],
276 }).pipe(
277 Effect.map((response) => ({
278 count: (response.Replications ?? []).length,
279 fault: null as string | null,
280 })),
281 Effect.catchTag("ResourceNotFoundFault", (error) =>
282 Effect.succeed({ count: 0, fault: error._tag as string }),
283 ),
284 );
285 return yield* HttpServerResponse.json(result);
286 }
287
288 if (request.method === "POST" && pathname === "/replication/start") {
289 const fault = yield* startReplication({
290 ReplicationConfigArn: NONEXISTENT_CONFIG_ARN,
291 StartReplicationType: "start-replication",
292 }).pipe(
293 Effect.map(() => "Started"),
294 Effect.catchTag(
295 [
296 "ResourceNotFoundFault",
297 "InvalidResourceStateFault",
298 "AccessDeniedFault",
299 ],
300 (error) => Effect.succeed(error._tag),
301 ),
302 );
303 return yield* HttpServerResponse.json({ fault });
304 }
305
306 if (request.method === "POST" && pathname === "/replication/stop") {
307 const fault = yield* stopReplication({
308 ReplicationConfigArn: NONEXISTENT_CONFIG_ARN,
309 }).pipe(
310 Effect.map(() => "Stopped"),
311 Effect.catchTag(
312 [
313 "ResourceNotFoundFault",
314 "InvalidResourceStateFault",
315 "AccessDeniedFault",
316 ],
317 (error) => Effect.succeed(error._tag),
318 ),
319 );
320 return yield* HttpServerResponse.json({ fault });
321 }
322
323 return yield* HttpServerResponse.json(
324 { error: "Not found", method: request.method, pathname },
325 { status: 404 },
326 );
327 }).pipe(Effect.orDie),
328 };
329 }).pipe(
330 Effect.provide(
331 Layer.mergeAll(
332 Lambda.EventSource,
333 DMS.DescribeSchemasHttp,
334 DMS.DescribeRefreshSchemasStatusHttp,
335 DMS.DescribeConnectionsHttp,
336 DMS.DescribeEventsHttp,
337 DMS.DescribeEndpointSettingsHttp,
338 DMS.DescribeOrderableReplicationInstancesHttp,
339 DMS.DescribeReplicationTasksHttp,
340 DMS.StartReplicationTaskHttp,
341 DMS.StopReplicationTaskHttp,
342 DMS.DescribeTableStatisticsHttp,
343 DMS.ReloadTablesHttp,
344 DMS.DescribeReplicationsHttp,
345 DMS.StartReplicationHttp,
346 DMS.StopReplicationHttp,
347 ),
348 ),
349 ),
350);