1import * as DataSync from "@/AWS/DataSync";
2import * as IAM from "@/AWS/IAM";
3import * as Lambda from "@/AWS/Lambda";
4import * as S3 from "@/AWS/S3";
5import * as Output from "@/Output";
6import * as Duration from "effect/Duration";
7import * as Effect from "effect/Effect";
8import * as Layer from "effect/Layer";
9import * as Schedule from "effect/Schedule";
10import * as Stream from "effect/Stream";
11import { HttpServerRequest } from "effect/http/HttpServerRequest";
12import * as HttpServerResponse from "effect/http/HttpServerResponse";
13import path from "pathe";
14
15const main = path.resolve(import.meta.dirname, "handler.ts");
16
17const trustPolicy = {
18 Version: "2012-10-17" as const,
19 Statement: [
20 {
21 Effect: "Allow" as const,
22 Principal: { Service: "datasync.amazonaws.com" },
23 Action: ["sts:AssumeRole"],
24 },
25 ],
26};
27
28const s3Policy = (arn: Output.Output<string>) => ({
29 Version: "2012-10-17" as const,
30 Statement: [
31 {
32 Effect: "Allow" as const,
33 Action: [
34 "s3:GetBucketLocation",
35 "s3:ListBucket",
36 "s3:ListBucketMultipartUploads",
37 ],
38 Resource: [arn],
39 },
40 {
41 Effect: "Allow" as const,
42 Action: [
43 "s3:AbortMultipartUpload",
44 "s3:DeleteObject",
45 "s3:GetObject",
46 "s3:GetObjectTagging",
47 "s3:ListMultipartUploadParts",
48 "s3:PutObject",
49 "s3:PutObjectTagging",
50 ],
51 Resource: [Output.interpolate`${arn}/*`],
52 },
53 ],
54});
55
56export class DataSyncTestFunction extends Lambda.Function<Lambda.Function>()(
57 "DataSyncTestFunction",
58) {}
59
60export default DataSyncTestFunction.make(
61 {
62 main,
63 functionUrl: true,
64 timeout: Duration.seconds(60),
65 },
66 Effect.gen(function* () {
67 // The transfer under test: two S3 locations over empty buckets, so a
68 // started execution moves no data and a cancel closes it out quickly.
69 const src = yield* S3.Bucket("BindingsSrc", { forceDestroy: true });
70 const dst = yield* S3.Bucket("BindingsDst", { forceDestroy: true });
71 const role = yield* IAM.Role("BindingsRole", {
72 assumeRolePolicyDocument: trustPolicy,
73 inlinePolicies: {
74 Src: s3Policy(src.bucketArn),
75 Dst: s3Policy(dst.bucketArn),
76 },
77 });
78 const source = yield* DataSync.LocationS3("BindingsSrcLoc", {
79 s3BucketArn: src.bucketArn,
80 bucketAccessRoleArn: role.roleArn,
81 });
82 const dest = yield* DataSync.LocationS3("BindingsDstLoc", {
83 s3BucketArn: dst.bucketArn,
84 bucketAccessRoleArn: role.roleArn,
85 });
86 const task = yield* DataSync.Task("BindingsTask", {
87 sourceLocationArn: source.locationArn,
88 destinationLocationArn: dest.locationArn,
89 });
90
91 // Event source: subscribe the host to DataSync task/execution state
92 // changes. The deploy proves the EventBridge rule + invoke permission
93 // wiring.
94 yield* DataSync.consumeTaskEvents(
95 { kinds: ["task-state-change", "task-execution-state-change"] },
96 (events) =>
97 Stream.runForEach(events, (event) =>
98 Effect.log(`datasync ${event["detail-type"]}: ${event.detail.State}`),
99 ),
100 );
101
102 const startTaskExecution = yield* DataSync.StartTaskExecution(task);
103 const describeTask = yield* DataSync.DescribeTask(task);
104 const listTaskExecutions = yield* DataSync.ListTaskExecutions(task);
105 const describeTaskExecution = yield* DataSync.DescribeTaskExecution(task);
106 const cancelTaskExecution = yield* DataSync.CancelTaskExecution(task);
107 const updateTaskExecution = yield* DataSync.UpdateTaskExecution(task);
108
109 const bound = {
110 startTaskExecution,
111 describeTask,
112 listTaskExecutions,
113 describeTaskExecution,
114 cancelTaskExecution,
115 updateTaskExecution,
116 };
117
118 return {
119 fetch: Effect.gen(function* () {
120 const request = yield* HttpServerRequest;
121 const url = new URL(request.originalUrl);
122 const pathname = url.pathname;
123 const arn = url.searchParams.get("arn");
124
125 if (request.method === "GET" && pathname === "/bindings") {
126 return yield* HttpServerResponse.json({
127 bound: Object.keys(bound),
128 });
129 }
130
131 // Task-scoped read: the TaskArn is injected from the binding.
132 if (request.method === "GET" && pathname === "/task") {
133 const detail = yield* describeTask();
134 return yield* HttpServerResponse.json({
135 taskArn: detail.TaskArn,
136 status: detail.Status,
137 name: detail.Name,
138 });
139 }
140
141 // Start a run. A freshly-attached IAM role can transiently fail
142 // DataSync's location access test (typed LocationAccessTestFailed);
143 // retry briefly in-route, then report the typed tag so the test can
144 // re-poll instead of treating it as a hard failure.
145 if (request.method === "POST" && pathname === "/start") {
146 const started = yield* startTaskExecution().pipe(
147 Effect.retry({
148 while: (e): boolean => e._tag === "LocationAccessTestFailed",
149 schedule: Schedule.max([
150 Schedule.fixed("5 seconds"),
151 Schedule.recurs(5),
152 ]),
153 }),
154 Effect.result,
155 );
156 return yield* HttpServerResponse.json(
157 started._tag === "Success"
158 ? { ok: true, executionArn: started.success.TaskExecutionArn }
159 : { ok: false, tag: started.failure._tag },
160 );
161 }
162
163 // List the task's runs: the TaskArn is injected from the binding.
164 if (request.method === "GET" && pathname === "/executions") {
165 const { TaskExecutions } = yield* listTaskExecutions();
166 return yield* HttpServerResponse.json({
167 arns: (TaskExecutions ?? []).map((e) => e.TaskExecutionArn),
168 });
169 }
170
171 if (arn === null) {
172 return yield* HttpServerResponse.json(
173 { error: "missing arn" },
174 { status: 400 },
175 );
176 }
177
178 // Execution-addressed reads/writes: the execution ARN comes from
179 // /start; IAM was granted on the bound task's execution pattern.
180 if (request.method === "GET" && pathname === "/execution") {
181 const execution = yield* describeTaskExecution({
182 TaskExecutionArn: arn,
183 });
184 return yield* HttpServerResponse.json({
185 status: execution.Status,
186 bytesTransferred: execution.BytesTransferred,
187 });
188 }
189
190 // Throttling is only valid while the run is launching/preparing/
191 // transferring/verifying — report the typed rejection instead of
192 // failing so the test can assert the binding round-trips.
193 if (request.method === "POST" && pathname === "/throttle") {
194 const updated = yield* updateTaskExecution({
195 TaskExecutionArn: arn,
196 Options: { BytesPerSecond: 1024 * 1024 },
197 }).pipe(Effect.result);
198 return yield* HttpServerResponse.json(
199 updated._tag === "Success"
200 ? { updated: true }
201 : { updated: false, tag: updated.failure._tag },
202 );
203 }
204
205 if (request.method === "POST" && pathname === "/cancel") {
206 const cancelled = yield* cancelTaskExecution({
207 TaskExecutionArn: arn,
208 }).pipe(Effect.result);
209 return yield* HttpServerResponse.json(
210 cancelled._tag === "Success"
211 ? { cancelled: true }
212 : { cancelled: false, tag: cancelled.failure._tag },
213 );
214 }
215
216 return yield* HttpServerResponse.json(
217 { error: "Not found", method: request.method, pathname },
218 { status: 404 },
219 );
220 }).pipe(Effect.orDie),
221 };
222 }).pipe(
223 Effect.provide(
224 Layer.mergeAll(
225 Lambda.EventSource,
226 DataSync.StartTaskExecutionHttp,
227 DataSync.DescribeTaskHttp,
228 DataSync.ListTaskExecutionsHttp,
229 DataSync.DescribeTaskExecutionHttp,
230 DataSync.CancelTaskExecutionHttp,
231 DataSync.UpdateTaskExecutionHttp,
232 ),
233 ),
234 ),
235);