1import * as emr from "@distilled.cloud/aws/emr-serverless";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Option from "effect/Option";
5import * as Stream 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, diffTags, hasAlchemyTags } from "../../Tags.ts";
12import { toWireMinutes } from "../../Util/Duration.ts";
13import type { Providers } from "../Providers.ts";
14import {
15 awaitApplicationCreated,
16 awaitApplicationStopped,
17} from "./internal.ts";
18
19/**
20 * The engine an EMR Serverless application runs. Changing the type replaces
21 * the application.
22 */
23export type ApplicationType = "SPARK" | "HIVE";
24
25/**
26 * Auto-stop behavior for an EMR Serverless {@link Application}.
27 */
28export interface AutoStopConfiguration {
29 /**
30 * Whether the application stops automatically after being idle.
31 * @default true
32 */
33 enabled?: boolean;
34 /**
35 * Idle time after which the application stops automatically — e.g.
36 * `"5 minutes"` or `Duration.minutes(5)`. Sent to AWS as whole minutes.
37 * @default "15 minutes"
38 */
39 idleTimeout?: Duration.Input;
40}
41
42/** Convert the alchemy-facing auto-stop config to the wire shape (minutes). */
43const toWireAutoStop = (config: AutoStopConfiguration): emr.AutoStopConfig => ({
44 enabled: config.enabled,
45 idleTimeoutMinutes: toWireMinutes(config.idleTimeout),
46});
47
48export interface ApplicationProps {
49 /**
50 * Name of the application (1-64 characters: letters, digits, `.` `_` `/`
51 * `#` `-`). Changing the name replaces the application.
52 * @default a generated physical name
53 */
54 applicationName?: string;
55 /**
56 * The engine type. Changing the type replaces the application.
57 * @default "SPARK"
58 */
59 type?: ApplicationType;
60 /**
61 * The Amazon EMR release associated with the application, e.g.
62 * `emr-7.9.0`. Updatable in place (the application must be in a `CREATED`
63 * or `STOPPED` state; the provider stops a `STARTED` application before
64 * applying the update).
65 */
66 releaseLabel: string;
67 /**
68 * The CPU architecture of the application.
69 * @default "X86_64"
70 */
71 architecture?: "X86_64" | "ARM64";
72 /**
73 * Pre-initialized capacity kept warm per worker type (e.g. `Driver` and
74 * `Executor` for Spark; `HiveDriver` and `TezTask` for Hive). Warm workers
75 * bill while the application is started.
76 */
77 initialCapacity?: Record<string, emr.InitialCapacityConfig>;
78 /**
79 * The maximum aggregate vCPU, memory and disk the application may scale to.
80 */
81 maximumCapacity?: emr.MaximumAllowedResources;
82 /**
83 * Whether the application starts automatically on job submission.
84 * @default enabled
85 */
86 autoStartConfiguration?: emr.AutoStartConfig;
87 /**
88 * Whether (and after how long idle) the application stops
89 * automatically.
90 * @default enabled after 15 idle minutes
91 */
92 autoStopConfiguration?: AutoStopConfiguration;
93 /**
94 * VPC connectivity (subnet + security group IDs) for jobs that must reach
95 * resources in a VPC. Omit for the default non-VPC connectivity.
96 */
97 networkConfiguration?: emr.NetworkConfiguration;
98 /**
99 * Tags to apply to the application. Merged with the internal Alchemy tags.
100 */
101 tags?: Record<string, string>;
102}
103
104export interface Application extends Resource<
105 "AWS.EMRServerless.Application",
106 ApplicationProps,
107 {
108 /** The ID of the application. */
109 applicationId: string;
110 /** The name of the application. */
111 applicationName: string;
112 /** The ARN of the application. */
113 applicationArn: string;
114 /** The application type (`SPARK` or `HIVE`). */
115 type: string;
116 /** The EMR release label the application runs (e.g. `emr-7.5.0`). */
117 releaseLabel: string;
118 /** The application state (e.g. `CREATED`, `STARTED`, `STOPPED`). */
119 state?: string;
120 },
121 {},
122 Providers
123> {}
124
125/**
126 * An Amazon EMR Serverless application — a serverless Spark or Hive
127 * environment that automatically provisions and scales workers per job,
128 * with no cluster to manage. An application in the `CREATED` or `STOPPED`
129 * state costs nothing; billing only occurs for workers while the application
130 * is started (including any pre-initialized `initialCapacity`).
131 *
132 * ### Creating Applications
133 * **Example:** Spark Application
134 * ```typescript
135 * import * as AWS from "alchemy/AWS";
136 *
137 * const app = yield* AWS.EMRServerless.Application("Spark", {
138 * releaseLabel: "emr-7.9.0",
139 * });
140 * // app.applicationId is passed to StartJobRun
141 * ```
142 *
143 * **Example:** Hive Application with Auto-Stop Tuning
144 * ```typescript
145 * const app = yield* AWS.EMRServerless.Application("Hive", {
146 * type: "HIVE",
147 * releaseLabel: "emr-7.9.0",
148 * autoStartConfiguration: { enabled: true },
149 * autoStopConfiguration: { enabled: true, idleTimeout: "5 minutes" },
150 * });
151 * ```
152 *
153 * ### Capacity
154 * **Example:** Pre-Initialized Capacity for Low-Latency Jobs
155 * ```typescript
156 * const app = yield* AWS.EMRServerless.Application("Warm", {
157 * releaseLabel: "emr-7.9.0",
158 * initialCapacity: {
159 * Driver: {
160 * workerCount: 1,
161 * workerConfiguration: { cpu: "2 vCPU", memory: "4 GB" },
162 * },
163 * Executor: {
164 * workerCount: 2,
165 * workerConfiguration: { cpu: "2 vCPU", memory: "4 GB" },
166 * },
167 * },
168 * maximumCapacity: { cpu: "16 vCPU", memory: "64 GB" },
169 * });
170 * ```
171 *
172 * ### Networking
173 * **Example:** VPC-Connected Application
174 * ```typescript
175 * const app = yield* AWS.EMRServerless.Application("InVpc", {
176 * releaseLabel: "emr-7.9.0",
177 * networkConfiguration: {
178 * subnetIds: [subnet.subnetId],
179 * securityGroupIds: [securityGroup.securityGroupId],
180 * },
181 * });
182 * ```
183 *
184 * @resource
185 */
186export const Application = Resource<Application>(
187 "AWS.EMRServerless.Application",
188);
189
190/** Normalize a possibly-undefined structural prop for drift comparison. */
191const canonical = (value: unknown): string =>
192 value === undefined ? "" : JSON.stringify(value);
193
194/**
195 * Derive a deterministic create token from the resource's instance ID so a
196 * retried create after a crashed reconcile never double-provisions.
197 */
198const createToken = (instanceId: string): string =>
199 instanceId.replaceAll(/[^a-zA-Z0-9]/g, "").slice(0, 64) || "alchemy";
200
201export const ApplicationProvider = () =>
202 Provider.effect(
203 Application,
204 Effect.gen(function* () {
205 const createName = Effect.fn(function* (
206 id: string,
207 props: { applicationName?: string | undefined },
208 ) {
209 return (
210 props.applicationName ??
211 (yield* createPhysicalName({ id, maxLength: 64 }))
212 );
213 });
214
215 const toAttributes = (application: emr.Application) => ({
216 applicationId: application.applicationId,
217 applicationName: application.name ?? application.applicationId,
218 applicationArn: application.arn,
219 type: application.type,
220 releaseLabel: application.releaseLabel,
221 state: application.state,
222 });
223
224 const observeById = Effect.fn(function* (applicationId: string) {
225 return yield* emr.getApplication({ applicationId }).pipe(
226 Effect.map((response) => response.application),
227 Effect.catchTag("ResourceNotFoundException", () =>
228 Effect.succeed(undefined),
229 ),
230 );
231 });
232
233 // Applications can only be fetched by id; when the id is unknown (state
234 // persistence failed before the first reconcile completed), find the
235 // live (non-terminated) application carrying the derived physical name.
236 const observeByName = Effect.fn(function* (name: string) {
237 const summary = yield* emr.listApplications.items({}).pipe(
238 Stream.filter((s) => s.name === name && s.state !== "TERMINATED"),
239 Stream.runHead,
240 Effect.map(Option.getOrUndefined),
241 );
242 if (summary === undefined) {
243 return undefined;
244 }
245 return yield* observeById(summary.id);
246 });
247
248 const observe = Effect.fn(function* (
249 applicationId: string | undefined,
250 name: string,
251 ) {
252 const byId =
253 applicationId !== undefined
254 ? yield* observeById(applicationId)
255 : undefined;
256 if (byId !== undefined && byId.state !== "TERMINATED") {
257 return byId;
258 }
259 return yield* observeByName(name);
260 });
261
262 const syncTags = Effect.fn(function* (
263 application: emr.Application,
264 desired: Record<string, string>,
265 ) {
266 // Diff against OBSERVED cloud tags (adoption may bring foreign tags).
267 const observed = Object.fromEntries(
268 Object.entries(application.tags ?? {}).filter(
269 (entry): entry is [string, string] => entry[1] !== undefined,
270 ),
271 );
272 const { upsert, removed } = diffTags(observed, desired);
273 if (upsert.length > 0) {
274 yield* emr.tagResource({
275 resourceArn: application.arn,
276 tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
277 });
278 }
279 if (removed.length > 0) {
280 yield* emr.untagResource({
281 resourceArn: application.arn,
282 tagKeys: removed,
283 });
284 }
285 });
286
287 return Application.Provider.of({
288 stables: ["applicationId", "applicationName", "applicationArn", "type"],
289
290 list: () =>
291 Effect.gen(function* () {
292 const pages = yield* emr.listApplications
293 .pages({})
294 .pipe(Stream.runCollect);
295 return Array.from(pages)
296 .flatMap((page) => page.applications)
297 .filter((s) => s.state !== "TERMINATED")
298 .map((s) => ({
299 applicationId: s.id,
300 applicationName: s.name ?? s.id,
301 applicationArn: s.arn,
302 type: s.type,
303 releaseLabel: s.releaseLabel,
304 state: s.state,
305 }));
306 }),
307
308 read: Effect.fn(function* ({ id, olds, output }) {
309 const name =
310 output?.applicationName ?? (yield* createName(id, olds ?? {}));
311 const application = yield* observe(output?.applicationId, name);
312 if (application === undefined) {
313 return undefined;
314 }
315 const attrs = toAttributes(application);
316 return (yield* hasAlchemyTags(id, application.tags))
317 ? attrs
318 : Unowned(attrs);
319 }),
320
321 diff: Effect.fn(function* ({ id, news, olds }) {
322 if (!isResolved(news)) return undefined;
323 const oldName = yield* createName(id, olds);
324 const newName = yield* createName(id, news);
325 if (oldName !== newName) {
326 return { action: "replace" } as const;
327 }
328 if ((olds.type ?? "SPARK") !== (news.type ?? "SPARK")) {
329 return { action: "replace" } as const;
330 }
331 // releaseLabel/capacity/autoStart/autoStop/network/tags → update
332 }),
333
334 reconcile: Effect.fn(function* ({
335 id,
336 news,
337 output,
338 session,
339 instanceId,
340 }) {
341 const name = output?.applicationName ?? (yield* createName(id, news));
342 const type = news.type ?? "SPARK";
343 const internalTags = yield* createInternalTags(id);
344 const desiredTags = { ...news.tags, ...internalTags };
345 const desiredAutoStop =
346 news.autoStopConfiguration !== undefined
347 ? toWireAutoStop(news.autoStopConfiguration)
348 : undefined;
349
350 // 1. OBSERVE — cloud state is authoritative; output is an id cache
351 let application = yield* observe(output?.applicationId, name);
352
353 // 2. ENSURE — create if missing; ~1 min async until CREATED
354 if (application === undefined || application.state === "TERMINATED") {
355 const created = yield* emr
356 .createApplication({
357 clientToken: createToken(instanceId),
358 name,
359 type,
360 releaseLabel: news.releaseLabel,
361 architecture: news.architecture,
362 initialCapacity: news.initialCapacity,
363 maximumCapacity: news.maximumCapacity,
364 autoStartConfiguration: news.autoStartConfiguration,
365 autoStopConfiguration: desiredAutoStop,
366 networkConfiguration: news.networkConfiguration,
367 tags: desiredTags,
368 })
369 .pipe(
370 // a concurrent reconciler already created it — observe instead
371 Effect.catchTag("ConflictException", () =>
372 Effect.succeed(undefined),
373 ),
374 );
375 const applicationId =
376 created?.applicationId ??
377 (yield* observeByName(name))?.applicationId;
378 if (applicationId === undefined) {
379 return yield* Effect.fail(
380 new emr.ResourceNotFoundException({
381 message: `application ${name} not visible after create`,
382 }),
383 );
384 }
385 yield* session.note(`creating application ${name} (async)...`);
386 application = yield* awaitApplicationCreated(applicationId);
387 } else if (application.state === "CREATING") {
388 application = yield* awaitApplicationCreated(
389 application.applicationId,
390 );
391 }
392
393 // 3. SYNC — diff each mutable aspect (observed vs desired), apply a
394 // single updateApplication carrying only the drifted fields.
395 const drift = {
396 releaseLabel:
397 news.releaseLabel !== application.releaseLabel
398 ? news.releaseLabel
399 : undefined,
400 architecture:
401 news.architecture !== undefined &&
402 news.architecture !== application.architecture
403 ? news.architecture
404 : undefined,
405 initialCapacity:
406 news.initialCapacity !== undefined &&
407 canonical(news.initialCapacity) !==
408 canonical(application.initialCapacity)
409 ? news.initialCapacity
410 : undefined,
411 maximumCapacity:
412 news.maximumCapacity !== undefined &&
413 canonical(news.maximumCapacity) !==
414 canonical(application.maximumCapacity)
415 ? news.maximumCapacity
416 : undefined,
417 autoStartConfiguration:
418 news.autoStartConfiguration !== undefined &&
419 canonical(news.autoStartConfiguration) !==
420 canonical(application.autoStartConfiguration)
421 ? news.autoStartConfiguration
422 : undefined,
423 autoStopConfiguration:
424 desiredAutoStop !== undefined &&
425 canonical(desiredAutoStop) !==
426 canonical(application.autoStopConfiguration)
427 ? desiredAutoStop
428 : undefined,
429 networkConfiguration:
430 news.networkConfiguration !== undefined &&
431 canonical(news.networkConfiguration) !==
432 canonical(application.networkConfiguration)
433 ? news.networkConfiguration
434 : undefined,
435 };
436 if (Object.values(drift).some((v) => v !== undefined)) {
437 // Updates require a CREATED or STOPPED state — stop a started
438 // application first (autoStart re-starts it on job submission).
439 if (
440 application.state === "STARTED" ||
441 application.state === "STARTING"
442 ) {
443 yield* emr.stopApplication({
444 applicationId: application.applicationId,
445 });
446 }
447 yield* awaitApplicationStopped(application.applicationId);
448 // Fresh token per update: distinct desired states must never be
449 // deduplicated against each other.
450 const updateToken = yield* Effect.sync(() => crypto.randomUUID());
451 const updated = yield* emr.updateApplication({
452 applicationId: application.applicationId,
453 clientToken: updateToken,
454 ...drift,
455 });
456 application = updated.application;
457 }
458
459 // 3b. SYNC TAGS — diff against observed cloud tags
460 yield* syncTags(application, desiredTags);
461
462 yield* session.note(application.applicationId);
463 return toAttributes(application);
464 }),
465
466 delete: Effect.fn(function* ({ output }) {
467 const application = yield* emr
468 .getApplication({ applicationId: output.applicationId })
469 .pipe(
470 Effect.map((response) => response.application),
471 Effect.catchTag("ResourceNotFoundException", () =>
472 Effect.succeed(undefined),
473 ),
474 );
475 if (application === undefined || application.state === "TERMINATED") {
476 return;
477 }
478 // Deletion requires a CREATED or STOPPED state.
479 if (
480 application.state === "STARTED" ||
481 application.state === "STARTING"
482 ) {
483 yield* emr.stopApplication({
484 applicationId: application.applicationId,
485 });
486 }
487 yield* awaitApplicationStopped(application.applicationId).pipe(
488 Effect.catchTag("ResourceNotFoundException", () =>
489 Effect.succeed(undefined),
490 ),
491 );
492 yield* emr
493 .deleteApplication({ applicationId: output.applicationId })
494 .pipe(
495 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
496 );
497 }),
498 });
499 }),
500 );