1import * as logs from "@distilled.cloud/aws/cloudwatch-logs";
2import * as deadline from "@distilled.cloud/aws/deadline";
3import * as Effect from "effect/Effect";
4import * as Redacted from "effect/Redacted";
5import * as Schedule from "effect/Schedule";
6import * as EffectStream from "effect/Stream";
7import { diffTags } from "../../Tags.ts";
8import { AWSEnvironment } from "../Environment.ts";
9
10/**
11 * Unwrap distilled `SensitiveString` values (`string | Redacted<string>`)
12 * to a plain string for drift comparison.
13 */
14export const asPlain = (
15 value: string | Redacted.Redacted<string> | undefined,
16): string | undefined =>
17 value === undefined
18 ? undefined
19 : Redacted.isRedacted(value)
20 ? Redacted.value(value)
21 : value;
22
23/**
24 * Build a Deadline Cloud ARN factory bound to the ambient account/region.
25 * Paths look like `farm/{farmId}`, `farm/{farmId}/queue/{queueId}`,
26 * `monitor/{monitorId}`.
27 */
28export const deadlineArnOf = Effect.gen(function* () {
29 const { accountId, region } = yield* AWSEnvironment.current;
30 return (path: string) => `arn:aws:deadline:${region}:${accountId}:${path}`;
31});
32
33/**
34 * Read the observed tags of a Deadline resource. A missing resource (or a
35 * resource type without tag support) reads as an empty tag set.
36 */
37export const fetchDeadlineTags = Effect.fn(function* (arn: string) {
38 const response = yield* deadline
39 .listTagsForResource({ resourceArn: arn })
40 .pipe(
41 Effect.catchTag("ResourceNotFoundException", () =>
42 Effect.succeed(undefined),
43 ),
44 );
45 return Object.fromEntries(
46 Object.entries(response?.tags ?? {}).filter(
47 (entry): entry is [string, string] => entry[1] !== undefined,
48 ),
49 );
50});
51
52/**
53 * Converge a Deadline resource's tags to the desired set, diffing against
54 * the OBSERVED cloud tags (never olds/output).
55 */
56export const syncDeadlineTags = Effect.fn(function* (
57 arn: string,
58 desired: Record<string, string>,
59) {
60 const observed = yield* fetchDeadlineTags(arn);
61 const { removed, upsert } = diffTags(observed, desired);
62 if (removed.length > 0) {
63 yield* deadline.untagResource({ resourceArn: arn, tagKeys: removed });
64 }
65 if (upsert.length > 0) {
66 yield* deadline.tagResource({
67 resourceArn: arn,
68 tags: Object.fromEntries(upsert.map(({ Key, Value }) => [Key, Value])),
69 });
70 }
71});
72
73/**
74 * Deadline auto-creates CloudWatch log groups under
75 * `/aws/deadline/{farmId}/{queueId}` (queue job/session logs) and
76 * `/aws/deadline/{farmId}/{fleetId}/...` (fleet worker logs), and neither
77 * `deleteQueue` nor `deleteFarm` removes them — without this reap every
78 * deleted farm strands orphaned log groups. Idempotent: a group already
79 * gone (or never created) is not an error.
80 */
81export const reapDeadlineLogGroups = Effect.fn(function* (prefix: string) {
82 const groups = yield* logs.describeLogGroups
83 .pages({ logGroupNamePrefix: prefix })
84 .pipe(
85 EffectStream.runCollect,
86 Effect.map((chunk) =>
87 Array.from(chunk)
88 .flatMap((page) => page.logGroups ?? [])
89 .map((group) => group.logGroupName)
90 .filter((name): name is string => name != null),
91 ),
92 );
93 yield* Effect.forEach(
94 groups,
95 (logGroupName) =>
96 logs
97 .deleteLogGroup({ logGroupName })
98 .pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void)),
99 { concurrency: 4, discard: true },
100 );
101});
102
103/**
104 * Reap every child resource of a farm so `deleteFarm` can succeed. A normal
105 * stack destroy deletes children before the farm, so this observes nothing —
106 * but an orphan sweep (nuke) or a mid-run crash can leave a farm whose
107 * children were never enumerated, and `deleteFarm` then rejects with
108 * `ConflictException` ("This farm still contains some storage profiles /
109 * queues / fleets ...") until the retry budget runs out and the farm leaks.
110 *
111 * Order matters: queue-fleet/queue-limit associations must be stopped and
112 * deleted before their queues/fleets/limits; storage profiles may be
113 * referenced by queues (`allowedStorageProfileIds`) so they go after the
114 * queues. Every step is idempotent — a child already gone is not an error.
115 */
116export const reapFarmChildren = Effect.fn(function* (farmId: string) {
117 // Every list catches ResourceNotFoundException → [] because the farm
118 // itself may already be gone.
119
120 // 1. Queue-fleet associations — stop scheduling, then delete (the delete
121 // rejects with ConflictException until the stop settles to STOPPED).
122 const queueFleetAssociations = yield* deadline.listQueueFleetAssociations
123 .items({ farmId })
124 .pipe(
125 EffectStream.runCollect,
126 Effect.map((chunk) => Array.from(chunk)),
127 Effect.catchTag("ResourceNotFoundException", () =>
128 Effect.succeed([] as deadline.QueueFleetAssociationSummary[]),
129 ),
130 );
131 yield* Effect.forEach(
132 queueFleetAssociations,
133 (assoc) =>
134 Effect.gen(function* () {
135 if (assoc.status !== "STOPPED") {
136 yield* deadline
137 .updateQueueFleetAssociation({
138 farmId,
139 queueId: assoc.queueId,
140 fleetId: assoc.fleetId,
141 status: "STOP_SCHEDULING_AND_CANCEL_TASKS",
142 })
143 .pipe(
144 // Already gone (the association or its farm).
145 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
146 );
147 }
148 yield* retryWhileConflict(
149 deadline.deleteQueueFleetAssociation({
150 farmId,
151 queueId: assoc.queueId,
152 fleetId: assoc.fleetId,
153 }),
154 ).pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void));
155 }),
156 { concurrency: 4, discard: true },
157 );
158
159 // 2. Queue-limit associations — same stop-then-delete dance.
160 const queueLimitAssociations = yield* deadline.listQueueLimitAssociations
161 .items({ farmId })
162 .pipe(
163 EffectStream.runCollect,
164 Effect.map((chunk) => Array.from(chunk)),
165 Effect.catchTag("ResourceNotFoundException", () =>
166 Effect.succeed([] as deadline.QueueLimitAssociationSummary[]),
167 ),
168 );
169 yield* Effect.forEach(
170 queueLimitAssociations,
171 (assoc) =>
172 Effect.gen(function* () {
173 if (assoc.status !== "STOPPED") {
174 yield* deadline
175 .updateQueueLimitAssociation({
176 farmId,
177 queueId: assoc.queueId,
178 limitId: assoc.limitId,
179 status: "STOP_LIMIT_USAGE_AND_CANCEL_TASKS",
180 })
181 .pipe(
182 // Already gone (the association or its farm).
183 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
184 );
185 }
186 yield* retryWhileConflict(
187 deadline.deleteQueueLimitAssociation({
188 farmId,
189 queueId: assoc.queueId,
190 limitId: assoc.limitId,
191 }),
192 ).pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void));
193 }),
194 { concurrency: 4, discard: true },
195 );
196
197 // 3. Budgets.
198 const budgets = yield* deadline.listBudgets.items({ farmId }).pipe(
199 EffectStream.runCollect,
200 Effect.map((chunk) => Array.from(chunk)),
201 Effect.catchTag("ResourceNotFoundException", () =>
202 Effect.succeed([] as deadline.BudgetSummary[]),
203 ),
204 );
205 yield* Effect.forEach(
206 budgets,
207 (budget) =>
208 deadline
209 .deleteBudget({ farmId, budgetId: budget.budgetId })
210 .pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void)),
211 { concurrency: 4, discard: true },
212 );
213
214 // 4. Queues (deletion is asynchronous; the farm delete below retries
215 // through the drain window). A queue's own log group is reaped too.
216 const queues = yield* deadline.listQueues.items({ farmId }).pipe(
217 EffectStream.runCollect,
218 Effect.map((chunk) => Array.from(chunk)),
219 Effect.catchTag("ResourceNotFoundException", () =>
220 Effect.succeed([] as deadline.QueueSummary[]),
221 ),
222 );
223 yield* Effect.forEach(
224 queues,
225 (queue) =>
226 retryWhileConflict(
227 deadline.deleteQueue({ farmId, queueId: queue.queueId }),
228 ).pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void)),
229 { concurrency: 4, discard: true },
230 );
231
232 // 5. Fleets (deletion drains workers asynchronously).
233 const fleets = yield* deadline.listFleets.items({ farmId }).pipe(
234 EffectStream.runCollect,
235 Effect.map((chunk) => Array.from(chunk)),
236 Effect.catchTag("ResourceNotFoundException", () =>
237 Effect.succeed([] as deadline.FleetSummary[]),
238 ),
239 );
240 yield* Effect.forEach(
241 fleets,
242 (fleet) =>
243 retryWhileConflict(
244 deadline.deleteFleet({ farmId, fleetId: fleet.fleetId }),
245 ).pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void)),
246 { concurrency: 4, discard: true },
247 );
248
249 // 6. Limits (their queue-limit associations were deleted above).
250 const limits = yield* deadline.listLimits.items({ farmId }).pipe(
251 EffectStream.runCollect,
252 Effect.map((chunk) => Array.from(chunk)),
253 Effect.catchTag("ResourceNotFoundException", () =>
254 Effect.succeed([] as deadline.LimitSummary[]),
255 ),
256 );
257 yield* Effect.forEach(
258 limits,
259 // deleteLimit is idempotent server-side: neither ResourceNotFound nor
260 // Conflict is in its error union — deleting a missing limit succeeds.
261 (limit) => deadline.deleteLimit({ farmId, limitId: limit.limitId }),
262 { concurrency: 4, discard: true },
263 );
264
265 // 7. Storage profiles (after queues — a queue's allowedStorageProfileIds
266 // reference blocks profile deletion with ConflictException).
267 const storageProfiles = yield* deadline.listStorageProfiles
268 .items({ farmId })
269 .pipe(
270 EffectStream.runCollect,
271 Effect.map((chunk) => Array.from(chunk)),
272 Effect.catchTag("ResourceNotFoundException", () =>
273 Effect.succeed([] as deadline.StorageProfileSummary[]),
274 ),
275 );
276 yield* Effect.forEach(
277 storageProfiles,
278 (profile) =>
279 retryWhileConflict(
280 deadline.deleteStorageProfile({
281 farmId,
282 storageProfileId: profile.storageProfileId,
283 }),
284 ).pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void)),
285 { concurrency: 4, discard: true },
286 );
287});
288
289// Explicitly-typed retry wrappers — an inline `Effect.retry` in provider
290// lifecycle code leaks `Retry.Return`'s conditional type into declaration
291// emit and widens the provider layer to `unknown` for every consumer of
292// `AWS.providers()`.
293
294/**
295 * Retry through Deadline `ConflictException` (STATUS_CONFLICT while a
296 * sub-resource drains or a concurrent mutation settles). Bounded.
297 */
298export const retryWhileConflict = <A, E extends { readonly _tag: string }, R>(
299 self: Effect.Effect<A, E, R>,
300): Effect.Effect<A, E, R> =>
301 Effect.retry(self, {
302 while: (e) => e._tag === "ConflictException",
303 schedule: Schedule.max([Schedule.spaced("6 seconds"), Schedule.recurs(9)]),
304 });
305
306/**
307 * A freshly-created farm rejects sub-resource mutations for a short window:
308 * "Farm ... infrastructure setup in progress" surfaces as a 429
309 * `ThrottlingException` or a `ConflictException`, and the farm's internal
310 * components (e.g. its BudgetTracker) surface as `ResourceNotFoundException`
311 * before they are provisioned. Bounded retry through the settling window
312 * (~60s). Genuine throttling is also safely absorbed here.
313 */
314export const retryWhileFarmSettling = <
315 A,
316 E extends { readonly _tag: string },
317 R,
318>(
319 self: Effect.Effect<A, E, R>,
320): Effect.Effect<A, E, R> =>
321 Effect.retry(self, {
322 while: (e) =>
323 e._tag === "ConflictException" ||
324 e._tag === "InternalServerException" ||
325 e._tag === "ResourceNotFoundException" ||
326 e._tag === "ThrottlingException",
327 schedule: Schedule.max([Schedule.spaced("6 seconds"), Schedule.recurs(9)]),
328 });
329
330/**
331 * A freshly-created IAM role may not have propagated when Deadline validates
332 * it — createFleet/createMonitor surface this as AccessDeniedException.
333 * Bounded retry through the propagation window (~60s).
334 */
335export const retryThroughIamPropagation = <
336 A,
337 E extends { readonly _tag: string },
338 R,
339>(
340 self: Effect.Effect<A, E, R>,
341): Effect.Effect<A, E, R> =>
342 Effect.retry(self, {
343 while: (e) => e._tag === "AccessDeniedException",
344 schedule: Schedule.max([Schedule.spaced("6 seconds"), Schedule.recurs(9)]),
345 });