1import * as mediaconvert from "@distilled.cloud/aws/mediaconvert";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
11import type { Providers } from "../Providers.ts";
12import { readMcTags, syncMcTags } from "./internal.ts";
13
14export interface QueueProps {
15 /**
16 * Name of the queue. Must be unique within the account/region and match
17 * `^[\w-]+$`. If omitted, a unique name is generated. Changing the name
18 * replaces the queue.
19 */
20 queueName?: string;
21 /**
22 * Optional description of the queue.
23 */
24 description?: string;
25 /**
26 * Pricing plan for the queue. `ON_DEMAND` queues bill per transcode minute;
27 * `RESERVED` queues purchase a reservation (a paid commitment). Changing the
28 * pricing plan replaces the queue.
29 * @default "ON_DEMAND"
30 */
31 pricingPlan?: mediaconvert.PricingPlan;
32 /**
33 * Reservation plan for a `RESERVED` queue (commitment, renewal type, and the
34 * number of reserved render units). Ignored for `ON_DEMAND` queues.
35 */
36 reservationPlanSettings?: mediaconvert.ReservationPlanSettings;
37 /**
38 * Initial/desired queue status. `ACTIVE` queues process jobs; `PAUSED`
39 * queues hold submitted jobs until resumed.
40 * @default "ACTIVE"
41 */
42 status?: mediaconvert.QueueStatus;
43 /**
44 * Maximum number of jobs the queue processes concurrently (ON_DEMAND).
45 */
46 concurrentJobs?: number;
47 /**
48 * User-defined tags for the queue.
49 */
50 tags?: Record<string, string>;
51}
52
53export interface Queue extends Resource<
54 "AWS.MediaConvert.Queue",
55 QueueProps,
56 {
57 queueName: string;
58 queueArn: string;
59 type: string | undefined;
60 status: string | undefined;
61 pricingPlan: string | undefined;
62 },
63 never,
64 Providers
65> {}
66
67/**
68 * An AWS Elemental MediaConvert queue — the pool that submitted transcode jobs
69 * are scheduled against. Every account has a system `Default` on-demand queue;
70 * create additional queues to isolate workloads or to purchase reserved
71 * render capacity.
72 *
73 * ### Creating a Queue
74 * **Example:** On-Demand Queue
75 * ```typescript
76 * const queue = yield* MediaConvert.Queue("Transcode", {
77 * description: "Marketing video transcodes",
78 * });
79 * ```
80 *
81 * **Example:** Paused Queue
82 * ```typescript
83 * const queue = yield* MediaConvert.Queue("Transcode", {
84 * status: "PAUSED",
85 * tags: { team: "media" },
86 * });
87 * ```
88 *
89 * ### Reserved Capacity
90 * **Example:** Reserved Queue with a One-Year Commitment
91 * ```typescript
92 * const queue = yield* MediaConvert.Queue("Reserved", {
93 * pricingPlan: "RESERVED",
94 * reservationPlanSettings: {
95 * Commitment: "ONE_YEAR",
96 * RenewalType: "EXPIRE",
97 * ReservedSlots: 1,
98 * },
99 * });
100 * ```
101 *
102 * @resource
103 */
104export const Queue = Resource<Queue>("AWS.MediaConvert.Queue");
105
106export const QueueProvider = () =>
107 Provider.effect(
108 Queue,
109 Effect.gen(function* () {
110 const createName = Effect.fn(function* (id: string, props: QueueProps) {
111 return (
112 props.queueName ?? (yield* createPhysicalName({ id, maxLength: 128 }))
113 );
114 });
115
116 const toAttrs = (queue: mediaconvert.Queue & { Name: string }) => ({
117 queueName: queue.Name,
118 queueArn: queue.Arn!,
119 type: queue.Type,
120 status: queue.Status,
121 pricingPlan: queue.PricingPlan,
122 });
123
124 /** Get a queue by name; typed not-found → undefined. */
125 const getQueue = Effect.fn(function* (name: string) {
126 const response = yield* mediaconvert
127 .getQueue({ Name: name })
128 .pipe(
129 Effect.catchTag("NotFoundException", () =>
130 Effect.succeed(undefined),
131 ),
132 );
133 return response?.Queue;
134 });
135
136 return {
137 stables: ["queueName", "queueArn"],
138
139 diff: Effect.fn(function* ({ id, olds, news }) {
140 if (!isResolved(news)) return undefined;
141 const oldName = yield* createName(id, olds);
142 const newName = yield* createName(id, news);
143 // The name is the queue's identity; a change means a new queue.
144 if (oldName !== newName) return { action: "replace" } as const;
145 // Pricing plan is immutable once the queue exists.
146 if (
147 (olds.pricingPlan ?? "ON_DEMAND") !==
148 (news.pricingPlan ?? "ON_DEMAND")
149 ) {
150 return { action: "replace" } as const;
151 }
152 }),
153
154 read: Effect.fn(function* ({ id, olds, output }) {
155 const name = output?.queueName ?? (yield* createName(id, olds ?? {}));
156 const queue = yield* getQueue(name);
157 if (queue === undefined) return undefined;
158 const attrs = toAttrs(queue);
159 const tags = yield* readMcTags(queue.Arn!);
160 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
161 }),
162
163 reconcile: Effect.fn(function* ({ id, news, output, session }) {
164 const internalTags = yield* createInternalTags(id);
165 const desiredTags = { ...internalTags, ...news.tags };
166 const name = output?.queueName ?? (yield* createName(id, news));
167
168 // 1. Observe — cloud state is authoritative; output is an id cache.
169 let queue = yield* getQueue(name);
170
171 // 2. Ensure — create if missing.
172 if (queue === undefined) {
173 const created = yield* mediaconvert.createQueue({
174 Name: name,
175 Description: news.description,
176 PricingPlan: news.pricingPlan,
177 ReservationPlanSettings: news.reservationPlanSettings,
178 Status: news.status,
179 ConcurrentJobs: news.concurrentJobs,
180 Tags: desiredTags,
181 });
182 queue = created.Queue!;
183 } else {
184 // 3. Sync — MediaConvert's UpdateQueue is a full in-place update of
185 // the mutable fields. Apply whenever any of them drift.
186 const statusDrift =
187 (news.status ?? "ACTIVE") !== (queue.Status ?? "ACTIVE");
188 const descDrift =
189 (news.description ?? undefined) !==
190 (queue.Description ?? undefined);
191 const concurrencyDrift =
192 news.concurrentJobs !== undefined &&
193 news.concurrentJobs !== queue.ConcurrentJobs;
194 if (statusDrift || descDrift || concurrencyDrift) {
195 const updated = yield* mediaconvert.updateQueue({
196 Name: name,
197 Description: news.description,
198 Status: news.status,
199 ConcurrentJobs: news.concurrentJobs,
200 ReservationPlanSettings: news.reservationPlanSettings,
201 });
202 queue = updated.Queue!;
203 }
204 }
205
206 // 3b. Sync tags — diff against OBSERVED cloud tags.
207 yield* syncMcTags(queue.Arn!, desiredTags);
208
209 yield* session.note(name);
210 return toAttrs(queue as mediaconvert.Queue & { Name: string });
211 }),
212
213 delete: Effect.fn(function* ({ output }) {
214 yield* mediaconvert.deleteQueue({ Name: output.queueName }).pipe(
215 Effect.catchTag("NotFoundException", () => Effect.void),
216 // A queue draining in-flight jobs can transiently reject deletion.
217 Effect.retry({
218 while: (e) => e._tag === "ConflictException",
219 schedule: Schedule.max([
220 Schedule.fixed("3 seconds"),
221 Schedule.recurs(10),
222 ]),
223 }),
224 );
225 }),
226
227 list: () =>
228 mediaconvert.listQueues.pages({}).pipe(
229 Stream.runCollect,
230 Effect.map((chunk) =>
231 Array.from(chunk).flatMap((page) => page.Queues ?? []),
232 ),
233 Effect.map((queues) =>
234 queues
235 // The built-in `Default` queue is an AWS SYSTEM queue that
236 // always exists and can never be deleted — keep it out of
237 // enumeration for account-wide teardown (nuke).
238 .filter((q) => q.Type !== "SYSTEM")
239 .map(toAttrs),
240 ),
241 ),
242 };
243 }),
244 );