1import * as datasync from "@distilled.cloud/aws/datasync";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { createInternalTags } from "../../Tags.ts";
10import type { Providers } from "../Providers.ts";
11import { readObservedTags, syncTags } from "./internal.ts";
12
13export interface TaskProps {
14 /**
15 * ARN of the source location data is transferred from. Cannot be changed
16 * after creation (replacement).
17 */
18 sourceLocationArn: string;
19 /**
20 * ARN of the destination location data is transferred to. Cannot be
21 * changed after creation (replacement).
22 */
23 destinationLocationArn: string;
24 /**
25 * Name of the task. If omitted, a unique name is generated from the app,
26 * stage, and logical id.
27 */
28 name?: string;
29 /**
30 * Transfer options — verification mode, overwrite behaviour, metadata
31 * preservation, bandwidth throttle, logging, etc. Updatable in place.
32 */
33 options?: datasync.Options;
34 /**
35 * Filter rules excluding files/folders from the transfer. Updatable in
36 * place.
37 */
38 excludes?: datasync.FilterRule[];
39 /**
40 * Filter rules including only matching files/folders. Updatable in place.
41 */
42 includes?: datasync.FilterRule[];
43 /**
44 * Schedule on which the task runs automatically (cron expression).
45 * Updatable in place.
46 */
47 schedule?: datasync.TaskSchedule;
48 /**
49 * ARN of the CloudWatch log group DataSync publishes task logs to.
50 * Updatable in place.
51 */
52 cloudWatchLogGroupArn?: string;
53 /**
54 * Task execution mode. `BASIC` (default) or `ENHANCED` (higher scale,
55 * S3-only). Cannot be changed after creation (replacement).
56 */
57 taskMode?: datasync.TaskMode;
58 /**
59 * Tags to apply to the task. Merged with internal Alchemy tags.
60 */
61 tags?: Record<string, string>;
62}
63
64export interface Task extends Resource<
65 "AWS.DataSync.Task",
66 TaskProps,
67 {
68 /** ARN of the DataSync task. */
69 taskArn: string;
70 /** Current task status (e.g. `AVAILABLE`, `RUNNING`). */
71 taskStatus: datasync.TaskStatus;
72 /** ARN of the source location. */
73 sourceLocationArn: string;
74 /** ARN of the destination location. */
75 destinationLocationArn: string;
76 },
77 {},
78 Providers
79> {}
80
81/**
82 * A DataSync task — the transfer definition binding a source location to a
83 * destination location, together with the filters, schedule, and transfer
84 * options that govern each run.
85 *
86 * Creating the task does not move any data; it defines the transfer.
87 * Start a run with `StartTaskExecution` (or attach a `schedule`). Source and
88 * destination locations and the task mode are immutable; everything else
89 * (name, options, filters, schedule, log group, tags) is updated in place.
90 *
91 * ### Creating Tasks
92 * **Example:** S3 → S3 transfer
93 * ```typescript
94 * import * as AWS from "alchemy/AWS";
95 *
96 * const task = yield* AWS.DataSync.Task("Backup", {
97 * sourceLocationArn: source.locationArn,
98 * destinationLocationArn: dest.locationArn,
99 * });
100 * ```
101 *
102 * **Example:** With verification and a schedule
103 * ```typescript
104 * const task = yield* AWS.DataSync.Task("Nightly", {
105 * sourceLocationArn: source.locationArn,
106 * destinationLocationArn: dest.locationArn,
107 * options: { VerifyMode: "ONLY_FILES_TRANSFERRED" },
108 * schedule: { ScheduleExpression: "cron(0 2 * * ? *)" },
109 * });
110 * ```
111 *
112 * @resource
113 */
114export const Task = Resource<Task>("AWS.DataSync.Task");
115
116const canonical = (value: unknown): string => JSON.stringify(value ?? null);
117
118export const TaskProvider = () =>
119 Provider.effect(
120 Task,
121 Effect.gen(function* () {
122 const createName = Effect.fn(function* (id: string, props: TaskProps) {
123 return (
124 props.name ?? (yield* createPhysicalName({ id, maxLength: 128 }))
125 );
126 });
127
128 const describe = Effect.fn(function* (taskArn: string) {
129 return yield* datasync
130 .describeTask({ TaskArn: taskArn })
131 .pipe(
132 Effect.catchTag("TaskNotFound", () => Effect.succeed(undefined)),
133 );
134 });
135
136 const findByName = Effect.fn(function* (name: string) {
137 const tasks = yield* datasync.listTasks.pages({}).pipe(
138 Stream.runCollect,
139 Effect.map((chunk) =>
140 Array.from(chunk).flatMap((p) => p.Tasks ?? []),
141 ),
142 );
143 return tasks.find((t) => t.Name === name)?.TaskArn;
144 });
145
146 const attrsOf = (t: datasync.DescribeTaskResponse) => ({
147 taskArn: t.TaskArn!,
148 taskStatus: t.Status ?? "AVAILABLE",
149 sourceLocationArn: t.SourceLocationArn!,
150 destinationLocationArn: t.DestinationLocationArn!,
151 });
152
153 return Task.Provider.of({
154 stables: ["taskArn", "sourceLocationArn", "destinationLocationArn"],
155
156 list: () =>
157 Effect.gen(function* () {
158 const tasks = yield* datasync.listTasks.pages({}).pipe(
159 Stream.runCollect,
160 Effect.map((chunk) =>
161 Array.from(chunk).flatMap((p) => p.Tasks ?? []),
162 ),
163 );
164 return yield* Effect.forEach(tasks, (t) =>
165 Effect.gen(function* () {
166 const full = yield* describe(t.TaskArn!);
167 return full ? attrsOf(full) : undefined;
168 }),
169 ).pipe(
170 Effect.map((items) =>
171 items.filter(
172 (i): i is ReturnType<typeof attrsOf> => i !== undefined,
173 ),
174 ),
175 );
176 }),
177
178 read: Effect.fn(function* ({ id, olds, output }) {
179 const arn =
180 output?.taskArn ?? (yield* findByName(yield* createName(id, olds)));
181 if (arn === undefined) return undefined;
182 const t = yield* describe(arn);
183 return t === undefined ? undefined : attrsOf(t);
184 }),
185
186 diff: Effect.fn(function* ({ news, olds }) {
187 if (!isResolved(news)) return undefined;
188 const replaced =
189 news.sourceLocationArn !== olds.sourceLocationArn ||
190 news.destinationLocationArn !== olds.destinationLocationArn ||
191 (news.taskMode ?? "BASIC") !== (olds.taskMode ?? "BASIC");
192 if (replaced) return { action: "replace" } as const;
193 }),
194
195 reconcile: Effect.fn(function* ({ id, news, output, session }) {
196 const internalTags = yield* createInternalTags(id);
197 const desiredTags = { ...news.tags, ...internalTags };
198 const name = yield* createName(id, news);
199
200 // 1. OBSERVE — output cache, else re-discover by deterministic name.
201 let arn = output?.taskArn ?? (yield* findByName(name));
202 let t = arn ? yield* describe(arn) : undefined;
203
204 // 2. ENSURE — create if missing.
205 if (t === undefined) {
206 const created = yield* datasync
207 .createTask({
208 SourceLocationArn: news.sourceLocationArn,
209 DestinationLocationArn: news.destinationLocationArn,
210 Name: name,
211 Options: news.options,
212 Excludes: news.excludes,
213 Includes: news.includes,
214 Schedule: news.schedule,
215 CloudWatchLogGroupArn: news.cloudWatchLogGroupArn,
216 TaskMode: news.taskMode,
217 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
218 Key,
219 Value,
220 })),
221 })
222 .pipe(
223 // createTask synchronously probes the locations' S3 access,
224 // and a freshly-created location role's IAM policy can lag
225 // (typed `LocationAccessTestFailed`). Retry bounded through
226 // the propagation window.
227 Effect.retry({
228 while: (e) => e._tag === "LocationAccessTestFailed",
229 schedule: Schedule.max([
230 Schedule.spaced("5 seconds"),
231 Schedule.recurs(9),
232 ]),
233 }),
234 );
235 arn = created.TaskArn!;
236 t = yield* describe(arn);
237 } else {
238 // 3a. SYNC updatable fields — only on a real delta.
239 const changed =
240 t.Name !== name ||
241 (t.CloudWatchLogGroupArn ?? "") !==
242 (news.cloudWatchLogGroupArn ?? "") ||
243 canonical(t.Options) !== canonical(news.options) ||
244 canonical(t.Excludes) !== canonical(news.excludes) ||
245 canonical(t.Includes) !== canonical(news.includes) ||
246 canonical(t.Schedule) !== canonical(news.schedule);
247 if (changed) {
248 yield* datasync.updateTask({
249 TaskArn: arn!,
250 Name: name,
251 Options: news.options,
252 Excludes: news.excludes,
253 Includes: news.includes,
254 Schedule: news.schedule,
255 CloudWatchLogGroupArn: news.cloudWatchLogGroupArn,
256 });
257 t = yield* describe(arn!);
258 }
259 }
260
261 // 3b. SYNC tags against observed cloud state.
262 const observed = yield* readObservedTags(arn!);
263 yield* syncTags(arn!, observed, desiredTags);
264
265 yield* session.note(arn!);
266 return attrsOf(t!);
267 }),
268
269 delete: Effect.fn(function* ({ output }) {
270 yield* datasync
271 .deleteTask({ TaskArn: output.taskArn })
272 .pipe(Effect.catchTag("TaskNotFound", () => Effect.void));
273 }),
274 });
275 }),
276 );