1import * as deadline from "@distilled.cloud/aws/deadline";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as EffectStream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { createInternalTags, hasAlchemyTags, type Tags } from "../../Tags.ts";
12import type { Providers } from "../Providers.ts";
13import {
14 asPlain,
15 deadlineArnOf,
16 fetchDeadlineTags,
17 reapDeadlineLogGroups,
18 retryWhileConflict,
19 retryWhileFarmSettling,
20 syncDeadlineTags,
21} from "./internal.ts";
22
23export type QueueStatus = deadline.QueueStatus;
24export type DefaultQueueBudgetAction = deadline.DefaultQueueBudgetAction;
25
26export interface QueueProps {
27 /**
28 * The identifier of the farm the queue belongs to. Changing it replaces
29 * the queue.
30 */
31 farmId: string;
32 /**
33 * Display name of the queue.
34 * @default ${app}-${stage}-${id}
35 */
36 displayName?: string;
37 /**
38 * A description of the queue.
39 */
40 description?: string;
41 /**
42 * Action taken on jobs in the queue when its associated budget is
43 * exhausted (`NONE`, `STOP_SCHEDULING_AND_COMPLETE_TASKS`,
44 * `STOP_SCHEDULING_AND_CANCEL_TASKS`).
45 * @default "NONE"
46 */
47 defaultBudgetAction?: DefaultQueueBudgetAction;
48 /**
49 * S3 bucket and root prefix used for job attachments.
50 */
51 jobAttachmentSettings?: deadline.JobAttachmentSettings;
52 /**
53 * ARN of the IAM role workers assume while running this queue's jobs.
54 */
55 roleArn?: string;
56 /**
57 * The OS user/group jobs run as on workers.
58 */
59 jobRunAsUser?: deadline.JobRunAsUser;
60 /**
61 * Names of file system locations (from storage profiles) that must be
62 * present on a worker for it to pick up this queue's jobs.
63 */
64 requiredFileSystemLocationNames?: string[];
65 /**
66 * Storage profile IDs allowed to be used with the queue.
67 */
68 allowedStorageProfileIds?: string[];
69 /**
70 * How the queue orders and schedules its jobs.
71 */
72 schedulingConfiguration?: deadline.SchedulingConfiguration;
73 /**
74 * Tags to associate with the queue.
75 */
76 tags?: Record<string, string>;
77}
78
79export interface Queue extends Resource<
80 "AWS.Deadline.Queue",
81 QueueProps,
82 {
83 /**
84 * The identifier of the farm the queue belongs to.
85 */
86 farmId: string;
87 /**
88 * Service-assigned unique identifier of the queue (`queue-...`).
89 */
90 queueId: string;
91 /**
92 * ARN of the queue.
93 */
94 queueArn: string;
95 /**
96 * The queue's display name.
97 */
98 displayName: string;
99 /**
100 * Current scheduling status of the queue.
101 */
102 status: QueueStatus;
103 /**
104 * The configured default budget action.
105 */
106 defaultBudgetAction: DefaultQueueBudgetAction;
107 /**
108 * ARN of the queue's job role, when configured.
109 */
110 roleArn: string | undefined;
111 /**
112 * Current tags reported for the queue.
113 */
114 tags: Record<string, string>;
115 },
116 never,
117 Providers
118> {}
119
120/**
121 * An AWS Deadline Cloud queue — accepts render jobs within a farm and
122 * schedules them onto associated fleets.
123 *
124 * ### Creating Queues
125 * **Example:** Basic Queue
126 * ```typescript
127 * import * as AWS from "alchemy/AWS";
128 *
129 * const farm = yield* AWS.Deadline.Farm("RenderFarm", {});
130 * const queue = yield* AWS.Deadline.Queue("RenderQueue", {
131 * farmId: farm.farmId,
132 * });
133 * ```
134 *
135 * **Example:** Queue with Job Attachments and Role
136 * ```typescript
137 * const queue = yield* AWS.Deadline.Queue("RenderQueue", {
138 * farmId: farm.farmId,
139 * roleArn: queueRole.roleArn,
140 * defaultBudgetAction: "STOP_SCHEDULING_AND_COMPLETE_TASKS",
141 * jobAttachmentSettings: {
142 * s3BucketName: bucket.bucketName,
143 * rootPrefix: "attachments/",
144 * },
145 * });
146 * ```
147 *
148 * @resource
149 */
150export const Queue = Resource<Queue>("AWS.Deadline.Queue");
151
152const createQueueName = (
153 id: string,
154 props: { displayName?: string | undefined },
155) =>
156 props.displayName
157 ? Effect.succeed(props.displayName)
158 : createPhysicalName({ id, maxLength: 100 });
159
160interface QueueState {
161 attrs: Queue["Attributes"];
162 described: deadline.GetQueueResponse;
163}
164
165const readQueueById = Effect.fn(function* (
166 farmId: string,
167 queueId: string,
168 arnOf: (path: string) => string,
169) {
170 const described = yield* deadline
171 .getQueue({ farmId, queueId })
172 .pipe(
173 Effect.catchTag("ResourceNotFoundException", () =>
174 Effect.succeed(undefined),
175 ),
176 );
177 if (!described) return undefined;
178 const queueArn = arnOf(`farm/${described.farmId}/queue/${described.queueId}`);
179 const state: QueueState = {
180 described,
181 attrs: {
182 farmId: described.farmId,
183 queueId: described.queueId,
184 queueArn,
185 displayName: described.displayName,
186 status: described.status,
187 defaultBudgetAction: described.defaultBudgetAction,
188 roleArn: described.roleArn,
189 tags: yield* fetchDeadlineTags(queueArn),
190 },
191 };
192 return state;
193});
194
195const findQueueByDisplayName = Effect.fn(function* (
196 farmId: string,
197 displayName: string,
198 arnOf: (path: string) => string,
199) {
200 const summaries = yield* deadline.listQueues.items({ farmId }).pipe(
201 EffectStream.runCollect,
202 Effect.map((chunk) => Array.from(chunk)),
203 // The parent farm may itself be gone.
204 Effect.catchTag("ResourceNotFoundException", () =>
205 Effect.succeed([] as deadline.QueueSummary[]),
206 ),
207 );
208 const match = summaries.find(
209 (summary) => summary.displayName === displayName,
210 );
211 if (!match) return undefined;
212 return yield* readQueueById(farmId, match.queueId, arnOf);
213});
214
215/**
216 * A queue still present after delete was initiated — retried by the bounded
217 * wait-until-gone schedule.
218 */
219class QueueStillExists extends Data.TaggedError("QueueStillExists")<{
220 readonly queueId: string;
221}> {}
222
223const retryWhileStillExists = <A, E extends { readonly _tag: string }, R>(
224 self: Effect.Effect<A, E, R>,
225): Effect.Effect<A, E, R> =>
226 Effect.retry(self, {
227 while: (e) => e._tag === "QueueStillExists",
228 schedule: Schedule.max([Schedule.spaced("6 seconds"), Schedule.recurs(9)]),
229 });
230
231const stringSetDelta = (
232 observed: readonly string[],
233 desired: readonly string[],
234) => ({
235 toAdd: desired.filter((value) => !observed.includes(value)),
236 toRemove: observed.filter((value) => !desired.includes(value)),
237});
238
239export const QueueProvider = () =>
240 Provider.effect(
241 Queue,
242 Effect.gen(function* () {
243 return {
244 stables: ["farmId", "queueId", "queueArn"],
245 // Keyed by a parent farm; cannot be enumerated account-wide without
246 // iterating every farm — treated as a sub-resource per the factory
247 // list() convention.
248 list: () => Effect.succeed([]),
249 read: Effect.fn(function* ({ id, olds, output }) {
250 const arnOf = yield* deadlineArnOf;
251 const farmId = output?.farmId ?? olds?.farmId;
252 if (farmId === undefined) return undefined;
253 const state = output?.queueId
254 ? yield* readQueueById(farmId, output.queueId, arnOf)
255 : yield* findQueueByDisplayName(
256 farmId,
257 yield* createQueueName(id, olds ?? {}),
258 arnOf,
259 );
260 if (!state) return undefined;
261 return (yield* hasAlchemyTags(id, state.attrs.tags as Tags))
262 ? state.attrs
263 : Unowned(state.attrs);
264 }),
265 diff: Effect.fn(function* ({ news, olds }) {
266 if (!isResolved(news)) return;
267 if (olds === undefined) return;
268 // The parent farm is fixed at creation.
269 if (olds.farmId !== news.farmId) {
270 return { action: "replace" } as const;
271 }
272 }),
273 reconcile: Effect.fn(function* ({ id, news, output, session }) {
274 if (news === undefined) {
275 return yield* Effect.fail(
276 new Error("AWS.Deadline.Queue requires props"),
277 );
278 }
279 const arnOf = yield* deadlineArnOf;
280 const farmId = news.farmId;
281 const displayName = yield* createQueueName(id, news);
282 const internalTags = yield* createInternalTags(id);
283 const desiredTags = { ...internalTags, ...news.tags };
284
285 // Observe.
286 let state = output?.queueId
287 ? yield* readQueueById(farmId, output.queueId, arnOf)
288 : yield* findQueueByDisplayName(farmId, displayName, arnOf);
289
290 // Ensure.
291 if (state === undefined) {
292 const created = yield* retryWhileFarmSettling(
293 deadline.createQueue({
294 farmId,
295 displayName,
296 description: news.description,
297 defaultBudgetAction: news.defaultBudgetAction,
298 jobAttachmentSettings: news.jobAttachmentSettings,
299 roleArn: news.roleArn,
300 jobRunAsUser: news.jobRunAsUser,
301 requiredFileSystemLocationNames:
302 news.requiredFileSystemLocationNames,
303 allowedStorageProfileIds: news.allowedStorageProfileIds,
304 schedulingConfiguration: news.schedulingConfiguration,
305 tags: desiredTags,
306 }),
307 );
308 yield* session.note(
309 `Created queue ${displayName} (${created.queueId})`,
310 );
311 state = yield* readQueueById(farmId, created.queueId, arnOf);
312 if (state === undefined) {
313 return yield* Effect.fail(
314 new Error(`failed to read created queue ${displayName}`),
315 );
316 }
317 }
318
319 // Sync mutable settings — compute the delta from OBSERVED state.
320 const described = state.described;
321 const fslDelta = stringSetDelta(
322 described.requiredFileSystemLocationNames ?? [],
323 news.requiredFileSystemLocationNames ?? [],
324 );
325 const spDelta = stringSetDelta(
326 described.allowedStorageProfileIds ?? [],
327 news.allowedStorageProfileIds ?? [],
328 );
329 const needsUpdate =
330 displayName !== described.displayName ||
331 (news.description !== undefined &&
332 news.description !== (asPlain(described.description) ?? "")) ||
333 (news.defaultBudgetAction !== undefined &&
334 news.defaultBudgetAction !== described.defaultBudgetAction) ||
335 (news.roleArn !== undefined &&
336 news.roleArn !== described.roleArn) ||
337 news.jobAttachmentSettings !== undefined ||
338 news.jobRunAsUser !== undefined ||
339 news.schedulingConfiguration !== undefined ||
340 fslDelta.toAdd.length > 0 ||
341 fslDelta.toRemove.length > 0 ||
342 spDelta.toAdd.length > 0 ||
343 spDelta.toRemove.length > 0;
344 if (needsUpdate) {
345 yield* retryWhileFarmSettling(
346 deadline.updateQueue({
347 farmId,
348 queueId: state.attrs.queueId,
349 displayName,
350 description: news.description,
351 defaultBudgetAction: news.defaultBudgetAction,
352 jobAttachmentSettings: news.jobAttachmentSettings,
353 roleArn: news.roleArn,
354 jobRunAsUser: news.jobRunAsUser,
355 requiredFileSystemLocationNamesToAdd:
356 fslDelta.toAdd.length > 0 ? fslDelta.toAdd : undefined,
357 requiredFileSystemLocationNamesToRemove:
358 fslDelta.toRemove.length > 0 ? fslDelta.toRemove : undefined,
359 allowedStorageProfileIdsToAdd:
360 spDelta.toAdd.length > 0 ? spDelta.toAdd : undefined,
361 allowedStorageProfileIdsToRemove:
362 spDelta.toRemove.length > 0 ? spDelta.toRemove : undefined,
363 schedulingConfiguration: news.schedulingConfiguration,
364 }),
365 );
366 yield* session.note(`Updated queue ${displayName}`);
367 }
368
369 // Sync tags — diff against observed cloud tags.
370 yield* syncDeadlineTags(state.attrs.queueArn, desiredTags);
371
372 yield* session.note(state.attrs.queueArn);
373 const final = yield* readQueueById(
374 farmId,
375 state.attrs.queueId,
376 arnOf,
377 );
378 if (!final) {
379 return yield* Effect.fail(
380 new Error(`failed to read reconciled queue ${displayName}`),
381 );
382 }
383 return final.attrs;
384 }),
385 delete: Effect.fn(function* ({ output }) {
386 yield* retryWhileConflict(
387 deadline.deleteQueue({
388 farmId: output.farmId,
389 queueId: output.queueId,
390 }),
391 ).pipe(
392 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
393 );
394 // Queue deletion is asynchronous; wait until it is gone so the
395 // parent farm's deletion does not hit a dependency conflict.
396 yield* retryWhileStillExists(
397 Effect.gen(function* () {
398 const described = yield* deadline
399 .getQueue({ farmId: output.farmId, queueId: output.queueId })
400 .pipe(
401 Effect.catchTag("ResourceNotFoundException", () =>
402 Effect.succeed(undefined),
403 ),
404 );
405 if (described !== undefined) {
406 return yield* Effect.fail(
407 new QueueStillExists({ queueId: output.queueId }),
408 );
409 }
410 }),
411 ).pipe(
412 // Exhausted retries: deletion is already converging server-side.
413 Effect.catchTag("QueueStillExists", () => Effect.void),
414 );
415 // Deadline auto-creates /aws/deadline/{farmId}/{queueId} for the
416 // queue's job/session logs and deleteQueue does NOT remove it —
417 // reap it so a deleted queue leaves no orphaned log group.
418 yield* reapDeadlineLogGroups(
419 `/aws/deadline/${output.farmId}/${output.queueId}`,
420 );
421 }),
422 };
423 }),
424 );