1import * as batch from "@distilled.cloud/aws/batch";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
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, diffTags } from "../../Tags.ts";
10import { AWSEnvironment, type AccountID } from "../Environment.ts";
11import type { Providers } from "../Providers.ts";
12import type { RegionID } from "../Region.ts";
13import { pollBatch, retryBatch } from "./internal.ts";
14
15export type JobQueueName = string;
16export type JobQueueArn =
17 `arn:aws:batch:${RegionID}:${AccountID}:job-queue/${JobQueueName}`;
18
19/**
20 * Raised when a job queue's asynchronous deletion does not complete within
21 * the provider's poll budget. Reporting success here would let the engine
22 * delete the queue's compute environments while the association still exists,
23 * wedging the whole chain.
24 */
25export class JobQueueDeleteTimeoutError extends Data.TaggedError(
26 "JobQueueDeleteTimeoutError",
27)<{
28 readonly jobQueueName: string;
29 readonly status: string | undefined;
30 readonly statusReason: string | undefined;
31 readonly message: string;
32}> {}
33
34export interface JobQueueProps {
35 /**
36 * Name of the job queue. If omitted, a unique name is generated.
37 * Up to 128 characters (letters, numbers, hyphens, underscores).
38 */
39 jobQueueName?: string;
40 /**
41 * Compute environment ARNs the queue schedules onto, in preference order
42 * (index 0 is tried first).
43 */
44 computeEnvironments: string[];
45 /**
46 * Queue priority — queues with a higher value are evaluated first when
47 * they share compute environments.
48 * @default 1
49 */
50 priority?: number;
51 /**
52 * Whether the queue accepts new job submissions.
53 * @default "ENABLED"
54 */
55 state?: "ENABLED" | "DISABLED";
56 /**
57 * User-defined tags to apply to the job queue.
58 */
59 tags?: Record<string, string>;
60}
61
62export interface JobQueue extends Resource<
63 "AWS.Batch.JobQueue",
64 JobQueueProps,
65 {
66 jobQueueName: JobQueueName;
67 jobQueueArn: JobQueueArn;
68 state: "ENABLED" | "DISABLED";
69 status: string;
70 priority: number;
71 computeEnvironments: string[];
72 tags: Record<string, string>;
73 },
74 never,
75 Providers
76> {}
77
78/**
79 * An AWS Batch job queue. Jobs submitted to the queue are scheduled onto its
80 * associated compute environments in preference order.
81 *
82 * ### Creating Job Queues
83 * **Example:** Queue on a Fargate Compute Environment
84 * ```typescript
85 * const ce = yield* Batch.ComputeEnvironment("JobsCE", {});
86 * const queue = yield* Batch.JobQueue("JobsQueue", {
87 * computeEnvironments: [ce.computeEnvironmentArn],
88 * });
89 * ```
90 *
91 * **Example:** Prioritized queue
92 * ```typescript
93 * const critical = yield* Batch.JobQueue("CriticalQueue", {
94 * priority: 10,
95 * computeEnvironments: [ce.computeEnvironmentArn],
96 * });
97 * ```
98 *
99 * @resource
100 */
101export const JobQueue = Resource<JobQueue>("AWS.Batch.JobQueue");
102
103const toAttributes = (
104 q: batch.JobQueueDetail,
105 tags: Record<string, string>,
106) => ({
107 jobQueueName: q.jobQueueName!,
108 jobQueueArn: q.jobQueueArn as JobQueueArn,
109 state: (q.state ?? "ENABLED") as "ENABLED" | "DISABLED",
110 status: q.status ?? "VALID",
111 priority: q.priority ?? 1,
112 computeEnvironments: [...(q.computeEnvironmentOrder ?? [])]
113 .sort((a, b) => (a.order ?? 0) - (b.order ?? 0))
114 .flatMap((o) => (o.computeEnvironment ? [o.computeEnvironment] : [])),
115 tags,
116});
117
118const observedTagsOf = (q: { tags?: { [key: string]: string | undefined } }) =>
119 Object.fromEntries(
120 Object.entries(q.tags ?? {}).filter(
121 (e): e is [string, string] => typeof e[1] === "string",
122 ),
123 );
124
125const toOrder = (
126 computeEnvironments: string[],
127): batch.ComputeEnvironmentOrder[] =>
128 computeEnvironments.map((computeEnvironment, index) => ({
129 order: index + 1,
130 computeEnvironment,
131 }));
132
133export const JobQueueProvider = () =>
134 Provider.effect(
135 JobQueue,
136 Effect.gen(function* () {
137 const toName = (id: string, props: { jobQueueName?: string } = {}) =>
138 props.jobQueueName
139 ? Effect.succeed(props.jobQueueName)
140 : createPhysicalName({ id, maxLength: 128 });
141
142 // A deleted queue lingers as a DELETED record for a while and the name
143 // is immediately reusable, so a describe-by-name can briefly return
144 // both the tombstone and the live queue — prefer the live one.
145 const describeOne = (name: string) =>
146 batch
147 .describeJobQueues({ jobQueues: [name] })
148 .pipe(
149 Effect.map(
150 (res) =>
151 res.jobQueues?.find((q) => q.status !== "DELETED") ??
152 res.jobQueues?.[0],
153 ),
154 );
155
156 /** Wait until any CREATING/UPDATING/DELETING transition settles. */
157 const awaitSettled = (name: string) =>
158 pollBatch(
159 describeOne(name),
160 (q) =>
161 q === undefined ||
162 (q.status !== "CREATING" &&
163 q.status !== "UPDATING" &&
164 q.status !== "DELETING"),
165 );
166
167 const awaitDisabled = (name: string) =>
168 pollBatch(
169 describeOne(name),
170 (queue) =>
171 queue === undefined ||
172 queue.status === "DELETED" ||
173 queue.status === "DELETING" ||
174 (queue.state === "DISABLED" &&
175 queue.status !== "CREATING" &&
176 queue.status !== "UPDATING"),
177 );
178
179 return {
180 stables: ["jobQueueName", "jobQueueArn"],
181 diff: Effect.fn(function* ({ id, olds, news }) {
182 if (!isResolved(news)) return;
183 if (
184 (yield* toName(id, olds ?? {})) !== (yield* toName(id, news ?? {}))
185 ) {
186 return { action: "replace" } as const;
187 }
188 }),
189 read: Effect.fn(function* ({ id, olds, output }) {
190 const name = output?.jobQueueName ?? (yield* toName(id, olds ?? {}));
191 const q = yield* describeOne(name);
192 if (!q?.jobQueueArn || q.status === "DELETED") {
193 return undefined;
194 }
195 return toAttributes(q, observedTagsOf(q));
196 }),
197 list: () =>
198 Effect.gen(function* () {
199 const pages = yield* batch.describeJobQueues
200 .pages({})
201 .pipe(Stream.runCollect);
202 return Array.from(pages)
203 .flatMap((page) => page.jobQueues ?? [])
204 .flatMap((q) =>
205 q.jobQueueArn && q.status !== "DELETED"
206 ? [toAttributes(q, observedTagsOf(q))]
207 : [],
208 );
209 }),
210 reconcile: Effect.fn(function* ({ id, news, output, session }) {
211 const { accountId, region } = yield* AWSEnvironment.current;
212 const name = output?.jobQueueName ?? (yield* toName(id, news));
213 const arn =
214 `arn:aws:batch:${region}:${accountId}:job-queue/${name}` as JobQueueArn;
215 const internalTags = yield* createInternalTags(id);
216 const desiredTags = { ...internalTags, ...news.tags };
217 const desiredState = news.state ?? "ENABLED";
218 const desiredPriority = news.priority ?? 1;
219 const desiredOrder = toOrder(news.computeEnvironments);
220
221 // Observe — cloud state is authoritative.
222 let queue = yield* describeOne(name);
223 if (queue?.status === "DELETED") queue = undefined;
224
225 // Ensure — create if missing. The referenced compute environment may
226 // still be settling to VALID (its own reconcile waits, but adoption
227 // or manual drift can race) — retry through that window, and treat
228 // a concurrent create as a race to observe.
229 const create = retryBatch(
230 batch.createJobQueue({
231 jobQueueName: name,
232 state: desiredState,
233 priority: desiredPriority,
234 computeEnvironmentOrder: desiredOrder,
235 tags: desiredTags,
236 }),
237 (e) => e._tag === "ComputeEnvironmentNotValid",
238 ).pipe(Effect.catchTag("JobQueueAlreadyExists", () => Effect.void));
239
240 if (queue?.jobQueueArn && queue.status === "DELETING") {
241 // An interrupted destroy left the queue mid-deletion — let it
242 // finish, then fall through to a fresh create below.
243 queue = yield* awaitSettled(name);
244 if (queue?.status === "DELETED") queue = undefined;
245 }
246 if (!queue?.jobQueueArn) {
247 yield* create;
248 queue = yield* awaitSettled(name);
249 } else if (queue.status !== "VALID" && queue.status !== "INVALID") {
250 queue = yield* awaitSettled(name);
251 }
252 if (queue?.status === "DELETED") {
253 // The settle above can still land on a tombstone (deletion that
254 // completed between observe and settle) — one more create pass.
255 yield* create;
256 queue = yield* awaitSettled(name);
257 }
258
259 if (!queue?.jobQueueArn) {
260 return yield* Effect.die(
261 new Error(`JobQueue ${name} did not settle`),
262 );
263 }
264
265 // Sync — diff observed against desired, apply only the delta.
266 const observedOrder = [...(queue.computeEnvironmentOrder ?? [])]
267 .sort((a, b) => (a.order ?? 0) - (b.order ?? 0))
268 .map((o) => o.computeEnvironment)
269 .join(",");
270 const update: batch.UpdateJobQueueRequest = { jobQueue: name };
271 let dirty = false;
272 if ((queue.state ?? "ENABLED") !== desiredState) {
273 update.state = desiredState;
274 dirty = true;
275 }
276 if ((queue.priority ?? 1) !== desiredPriority) {
277 update.priority = desiredPriority;
278 dirty = true;
279 }
280 if (observedOrder !== news.computeEnvironments.join(",")) {
281 update.computeEnvironmentOrder = desiredOrder;
282 dirty = true;
283 }
284 if (dirty) {
285 yield* retryBatch(
286 batch.updateJobQueue(update),
287 (e) => e._tag === "JobQueueBeingModified",
288 );
289 queue = (yield* awaitSettled(name)) ?? queue;
290 }
291
292 // Sync tags — diff against OBSERVED cloud tags.
293 const { upsert, removed } = diffTags(
294 observedTagsOf(queue),
295 desiredTags,
296 );
297 if (upsert.length > 0) {
298 yield* batch.tagResource({
299 resourceArn: arn,
300 tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
301 });
302 }
303 if (removed.length > 0) {
304 yield* batch.untagResource({ resourceArn: arn, tagKeys: removed });
305 }
306
307 yield* session.note(arn);
308 return toAttributes(
309 { ...queue, jobQueueName: name, jobQueueArn: arn },
310 desiredTags,
311 );
312 }),
313 delete: Effect.fn(function* ({ output }) {
314 const name = output.jobQueueName;
315 const existing = yield* describeOne(name);
316 if (!existing || existing.status === "DELETED") return;
317
318 // Must be DISABLED before deletion.
319 if (
320 (existing.state ?? "ENABLED") !== "DISABLED" &&
321 existing.status !== "DELETING"
322 ) {
323 yield* retryBatch(
324 batch.updateJobQueue({ jobQueue: name, state: "DISABLED" }),
325 (e) => e._tag === "JobQueueBeingModified",
326 ).pipe(Effect.catchTag("JobQueueNotFound", () => Effect.void));
327 }
328 const disabled = yield* awaitDisabled(name);
329 if (!disabled || disabled.status === "DELETED") return;
330
331 // A resumed destroy can observe DELETING. Do not issue a redundant
332 // delete; just wait for the relationship to disappear. This keeps
333 // the downstream compute-environment delete strictly ordered.
334 if (disabled.status !== "DELETING") {
335 yield* retryBatch(
336 batch.deleteJobQueue({ jobQueue: name }),
337 (e) => e._tag === "JobQueueBeingModified",
338 ).pipe(Effect.catchTag("JobQueueNotFound", () => Effect.void));
339 }
340
341 // Wait until fully gone so downstream compute-environment deletion
342 // doesn't trip over the lingering association (queue deletion is
343 // asynchronous and takes up to ~1-2 minutes). An expired poll MUST
344 // NOT report success — the engine would then delete the queue's
345 // compute environments mid-teardown and wedge the chain.
346 const final = yield* pollBatch(
347 describeOne(name),
348 (q) => q === undefined || q.status === "DELETED",
349 );
350 if (final && final.status !== "DELETED") {
351 return yield* Effect.fail(
352 new JobQueueDeleteTimeoutError({
353 jobQueueName: name,
354 status: final.status,
355 statusReason: final.statusReason,
356 message: `Job queue ${name} was still ${final.status ?? "present"} after the delete poll budget${final.statusReason ? `: ${final.statusReason}` : ""}`,
357 }),
358 );
359 }
360 }),
361 };
362 }),
363 );