1import { Credentials } from "@distilled.cloud/gcp/Credentials";
2import * as cloudrun from "@distilled.cloud/gcp/run_v2";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
7import { Unowned } from "../../AdoptPolicy.ts";
8import type * as Bundle from "../../Bundle/Bundle.ts";
9import { isResolved } from "../../Diff.ts";
10import { createPhysicalName } from "../../PhysicalName.ts";
11import {
12 Platform,
13 type Main,
14 type PlatformProps,
15 type PlatformServices,
16} from "../../Platform.ts";
17import * as Provider from "../../Provider.ts";
18import { Resource, type ResourceBinding } from "../../Resource.ts";
19import type { RuntimeContext } from "../../RuntimeContext.ts";
20import {
21 createContainerRuntimeContext,
22 type HostRuntimeContext,
23 type ServerHost,
24} from "../../Server/Process.ts";
25import type { Scope } from "effect/Scope";
26import { tagRecord } from "../../Tags.ts";
27import {
28 destroyHostImageRepository,
29 makeImageSource,
30} from "../ArtifactRegistry/ImageSource.ts";
31import { GcpEnvironment } from "../Environment.ts";
32import {
33 type LongRunningOperation,
34 waitForOperation as waitForLongRunningOperation,
35} from "../Operation.ts";
36import {
37 retryActAs,
38 type AppliedIamGrant,
39 type GcpHostBinding,
40} from "../Host.ts";
41import {
42 alchemyRuntimeEnv,
43 isManagedServiceAccount,
44 makeGcpBootstrap,
45 mergeContainerEnv,
46 releaseHostIdentity,
47 resolveHostIdentity,
48} from "../HostRuntime.ts";
49import {
50 createInternalLabels,
51 diffLabels,
52 hasAlchemyLabels,
53 stripInternalLabels,
54 toLabels,
55} from "../Labels.ts";
56import type { Providers } from "../Providers.ts";
57
58const MAX_NAME_LENGTH = 49;
59
60export type JobEnvVar = {
61 /** Environment variable name. */
62 name: string;
63 /** Literal value. Mutually exclusive with `valueSource`. */
64 value?: string;
65 /** Value sourced from Secret Manager. */
66 valueSource?: {
67 secretKeyRef?: {
68 /** Secret id or `projects/{project}/secrets/{secret}` name. */
69 secret: string;
70 /** Version (`latest`, integer, or alias). */
71 version?: string;
72 };
73 };
74};
75
76export type JobContainer = {
77 /** DNS_LABEL container name. */
78 name?: string;
79 /**
80 * Container image in Artifact Registry, Container Registry, or Docker Hub.
81 */
82 image: string;
83 /** Entrypoint. Image ENTRYPOINT is used when omitted. */
84 command?: string[];
85 /** Arguments to the entrypoint. Image CMD is used when omitted. */
86 args?: string[];
87 /** Environment variables. */
88 env?: JobEnvVar[];
89 /** Working directory inside the container. */
90 workingDir?: string;
91 /** CPU/memory limits (`cpu`, `memory`, `nvidia.com/gpu`). */
92 resources?: {
93 limits?: Record<string, string>;
94 cpuIdle?: boolean;
95 startupCpuBoost?: boolean;
96 };
97 /** Volume mounts. Names must match `volumes`. */
98 volumeMounts?: Array<{
99 name: string;
100 mountPath: string;
101 subPath?: string;
102 }>;
103};
104
105export type JobVpcAccess = {
106 /** Traffic VPC egress (`ALL_TRAFFIC`, `PRIVATE_RANGES_ONLY`). */
107 egress?: string;
108 /**
109 * Serverless VPC Access connector
110 * (`projects/{project}/locations/{location}/connectors/{connector}`).
111 */
112 connector?: string;
113 /** Direct VPC egress network interfaces. */
114 networkInterfaces?: Array<{
115 network?: string;
116 subnetwork?: string;
117 tags?: string[];
118 }>;
119};
120
121export type JobBinaryAuthorization = {
122 /** Use the project's default Binary Authorization policy. */
123 useDefault?: boolean;
124 /** Breakglass justification. Requires `useDefault`. */
125 breakglassJustification?: string;
126 /** Policy path `projects/{project}/platforms/cloudRun/{policy}`. */
127 policy?: string;
128};
129
130export type JobProps = PlatformProps & {
131 /**
132 * Job id (the `{job}` segment of
133 * `projects/{project}/locations/{location}/jobs/{job}`). If omitted, a
134 * unique RFC1035 name is generated. 1-49 characters, lowercase letter
135 * first. Immutable — changing it replaces the job.
136 */
137 jobId?: string;
138 /**
139 * Region (`us-central1`, `europe-west1`, …). Immutable — changing it
140 * replaces the job. `US-CENTRAL1` is accepted and normalized to
141 * `us-central1`.
142 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
143 */
144 location?: string;
145 /**
146 * User labels. Alchemy ownership labels are merged in automatically.
147 */
148 labels?: Record<string, string>;
149 /**
150 * User annotations. Cloud Run rejects `run.googleapis.com`,
151 * `cloud.googleapis.com`, `serving.knative.dev`, and
152 * `autoscaling.knative.dev` namespaces.
153 */
154 annotations?: Record<string, string>;
155 /**
156 * Launch stage (`GA`, `BETA`, `ALPHA`). Defaults to GA.
157 */
158 launchStage?: string;
159 /**
160 * Binary Authorization settings.
161 */
162 binaryAuthorization?: JobBinaryAuthorization;
163 /**
164 * Number of tasks the execution should run.
165 * @default 1
166 */
167 taskCount?: number;
168 /**
169 * Maximum number of tasks to run in parallel. `0` uses the maximum.
170 */
171 parallelism?: number;
172 /**
173 * Labels applied to each execution created from this job.
174 */
175 executionLabels?: Record<string, string>;
176 /**
177 * Annotations applied to each execution created from this job.
178 */
179 executionAnnotations?: Record<string, string>;
180 /**
181 * Containers that run as each task. Required unless `main` is set.
182 */
183 containers?: JobContainer[];
184 /**
185 * Module entrypoint for an Effect-native job (typically
186 * `import.meta.url`). Alchemy bundles the program, builds a container,
187 * and runs it to completion. Bindings grant IAM onto the runtime SA.
188 */
189 main?: string;
190 /**
191 * Named export to load from `main`.
192 * @default "default"
193 */
194 handler?: string;
195 /**
196 * Additional environment variables for the Effect-native container.
197 */
198 env?: Record<string, any>;
199 /**
200 * Bundler configuration for `main`.
201 */
202 build?: Bundle.BundleConfig;
203 /**
204 * Max time a task attempt may run (e.g. `"600s"`).
205 * @default "600s"
206 */
207 timeout?: string;
208 /**
209 * Retries per task before the task is marked failed.
210 * @default 3
211 */
212 maxRetries?: number;
213 /**
214 * Service account email used by the task. Defaults to the project's
215 * Compute Engine default.
216 */
217 serviceAccount?: string;
218 /**
219 * Execution environment (`EXECUTION_ENVIRONMENT_GEN1` or
220 * `EXECUTION_ENVIRONMENT_GEN2`).
221 */
222 executionEnvironment?: string;
223 /**
224 * Customer-managed encryption key
225 * (`projects/{project}/locations/{location}/keyRings/{keyRing}/cryptoKeys/{cryptoKey}`).
226 */
227 encryptionKey?: string;
228 /**
229 * VPC Access configuration for the task.
230 */
231 vpcAccess?: JobVpcAccess;
232};
233
234export type Job = Resource<
235 "GCP.Run.Job",
236 JobProps,
237 {
238 /** Full resource name `projects/{project}/locations/{location}/jobs/{job}`. */
239 name: string;
240 /** Job id (last path segment). */
241 jobId: string;
242 /** Project id. */
243 project: string;
244 /** Location id (`us-central1`, …). */
245 location: string;
246 /** User labels (Alchemy ownership labels stripped). */
247 labels: Record<string, string>;
248 /** User annotations. */
249 annotations: Record<string, string>;
250 /** Server-assigned UUID. */
251 uid: string | undefined;
252 /** Monotonic generation. */
253 generation: string | undefined;
254 /** Whether Cloud Run is still reconciling the job. */
255 reconciling: boolean;
256 /** RFC3339 creation timestamp. */
257 createTime: string | undefined;
258 /** RFC3339 last-update timestamp. */
259 updateTime: string | undefined;
260 /** Number of executions created for this job. */
261 executionCount: number | undefined;
262 /** Terminal condition state (`CONDITION_SUCCEEDED`, …). */
263 terminalConditionState: string | undefined;
264 /** Latest created execution name, if any. */
265 latestCreatedExecutionName: string | undefined;
266 /** Image of the first container. */
267 image: string | undefined;
268 /** Retries per task. */
269 maxRetries: number | undefined;
270 /** Task timeout. */
271 timeout: string | undefined;
272 /** Desired task count. */
273 taskCount: number | undefined;
274 /** Max parallelism. */
275 parallelism: number | undefined;
276 /** Task service account email. */
277 serviceAccount: string | undefined;
278 /** True when Alchemy minted the per-host runtime service account. */
279 managedServiceAccount: boolean;
280 /** IAM roles bindings granted to the runtime service account. */
281 iamGrants: AppliedIamGrant[];
282 /** Hash of the bundled `main` program and its bootstrap (Effect-native only). */
283 codeHash: string | undefined;
284 },
285 GcpHostBinding,
286 Providers
287>;
288
289export type JobRuntimeContext = HostRuntimeContext;
290export type JobServices = Credentials | GcpEnvironment | ServerHost;
291/**
292 * Effect-native Job shape: a `run` entry that executes to completion
293 * when the container starts, and/or a `fetch` HTTP handler.
294 */
295export type JobShape =
296 | void
297 | (Exclude<Main<JobServices>, void> & {
298 run?: Effect.Effect<
299 void,
300 never,
301 JobServices | PlatformServices | RuntimeContext | Scope
302 >;
303 });
304
305/**
306 * A Cloud Run Job — a container that runs to completion.
307 *
308 * Changing `jobId` or `location` replaces the job.
309 *
310 * ### Creating a Job
311 * **Example:** Generated name
312 * ```typescript
313 * const job = yield* GCP.Run.Job("Migrate", {
314 * containers: [
315 * { image: "us-docker.pkg.dev/cloudrun/container/job:latest" },
316 * ],
317 * });
318 * ```
319 *
320 * **Example:** Explicit id, labels, and retries
321 * ```typescript
322 * const job = yield* GCP.Run.Job("Migrate", {
323 * jobId: "order-migrate",
324 * location: "us-central1",
325 * labels: { env: "prod" },
326 * maxRetries: 1,
327 * timeout: "120s",
328 * containers: [
329 * {
330 * image: "us-docker.pkg.dev/cloudrun/container/job:latest",
331 * env: [{ name: "STAGE", value: "prod" }],
332 * },
333 * ],
334 * });
335 * ```
336 *
337 * ### Effect-native Job with bindings
338 * **Example:** Drain a cache then exit
339 * ```typescript
340 * export class Warm extends GCP.Run.Job<Warm>()(
341 * "Warm",
342 * { main: import.meta.url },
343 * Effect.gen(function* () {
344 * const redis = yield* GCP.Redis.ReadWriteRedis(cache);
345 * return {
346 * run: redis.set("warmed", "1"),
347 * };
348 * }).pipe(Effect.provide(GCP.Redis.ReadWriteRedisHttp)),
349 * ) {}
350 * ```
351 *
352 * ### Running a Job
353 * **Example:** Trigger an execution
354 * ```typescript
355 * const runJob = yield* GCP.Run.RunJob(job);
356 * yield* runJob();
357 * ```
358 *
359 * @resource
360 * @category Run
361 */
362export const Job: Platform<Job, JobServices, JobShape, JobRuntimeContext> =
363 Platform("GCP.Run.Job", {
364 createRuntimeContext: createContainerRuntimeContext("GCP.Run.Job") as (
365 id: string,
366 ) => JobRuntimeContext,
367 });
368
369export class JobNotResolved extends Data.TaggedError("GCP.Run.JobNotResolved")<{
370 name: string;
371}> {}
372
373export class JobNotReady extends Data.TaggedError("GCP.Run.JobNotReady")<{
374 name: string;
375 state: string;
376 message: string;
377}> {}
378
379export class JobReconciling extends Data.TaggedError("GCP.Run.JobReconciling")<{
380 name: string;
381 state: string;
382}> {}
383
384export class JobStillExists extends Data.TaggedError("GCP.Run.JobStillExists")<{
385 name: string;
386}> {}
387
388const lastSegment = (value: string) => {
389 const trimmed = value.replace(/\/+$/, "");
390 const parts = trimmed.split("/");
391 return parts[parts.length - 1] || trimmed;
392};
393
394const normalizeLocation = (
395 location: string | undefined,
396 defaultLocation: string,
397) => lastSegment(location ?? defaultLocation).toLowerCase();
398
399const rfc1035 = (name: string): string => {
400 let next = name
401 .toLowerCase()
402 .replace(/[^a-z0-9-]/g, "-")
403 .replace(/-+/g, "-")
404 .replace(/^-+|-+$/g, "");
405 if (!/^[a-z]/.test(next)) next = `j${next}`;
406 next = next.slice(0, MAX_NAME_LENGTH).replace(/-+$/g, "");
407 return next.length > 0 ? next : "job";
408};
409
410const resourceName = (project: string, location: string, jobId: string) =>
411 `projects/${project}/locations/${location}/jobs/${jobId}`;
412
413const parseName = (name: string, defaultLocation: string) => {
414 const parts = name.split("/").filter((part) => part.length > 0);
415 const jobsAt = parts.lastIndexOf("jobs");
416 const locationsAt = parts.lastIndexOf("locations");
417 const projectsAt = parts.lastIndexOf("projects");
418 return {
419 project:
420 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
421 location:
422 locationsAt >= 0 && parts[locationsAt + 1]
423 ? parts[locationsAt + 1]!
424 : defaultLocation,
425 jobId:
426 jobsAt >= 0 && parts[jobsAt + 1] ? parts[jobsAt + 1]! : lastSegment(name),
427 };
428};
429
430const userLabels = (
431 labels: Record<string, string | undefined> | null | undefined,
432): Record<string, string> => stripInternalLabels(tagRecord(labels));
433
434const userAnnotations = (
435 annotations: Record<string, string | undefined> | null | undefined,
436): Record<string, string> => tagRecord(annotations);
437
438const toId = (id: string, jobId: string | undefined, existing?: string) =>
439 Effect.gen(function* () {
440 if (jobId !== undefined) return jobId;
441 if (existing !== undefined) return existing;
442 return rfc1035(
443 yield* createPhysicalName({
444 id,
445 maxLength: MAX_NAME_LENGTH,
446 lowercase: true,
447 }),
448 );
449 });
450
451const HOST_TYPE = "GCP.Run.Job";
452
453const bootstrapFor = (news: JobProps) =>
454 makeGcpBootstrap("CloudRunJob", news.handler ?? "default");
455
456const toAttrs = (
457 job: cloudrun.GoogleCloudRunV2Job,
458 project: string,
459 region: string,
460 extras: { iamGrants?: AppliedIamGrant[]; codeHash?: string } = {},
461): Job["Attributes"] => {
462 const name = job.name ?? "";
463 const parsed = parseName(name, region);
464 const task = job.template?.template;
465 const container = task?.containers?.[0];
466 return {
467 name,
468 jobId: parsed.jobId,
469 project: parsed.project || project,
470 location: parsed.location,
471 labels: userLabels(job.labels),
472 annotations: userAnnotations(job.annotations),
473 uid: job.uid,
474 generation: job.generation,
475 reconciling: job.reconciling === true,
476 createTime: job.createTime,
477 updateTime: job.updateTime,
478 executionCount: job.executionCount,
479 terminalConditionState: job.terminalCondition?.state,
480 latestCreatedExecutionName: job.latestCreatedExecution?.name,
481 image: container?.image,
482 maxRetries: task?.maxRetries,
483 timeout: task?.timeout,
484 taskCount: job.template?.taskCount,
485 parallelism: job.template?.parallelism,
486 serviceAccount: task?.serviceAccount,
487 managedServiceAccount: isManagedServiceAccount({
488 project: parsed.project || project,
489 hostType: HOST_TYPE,
490 resourceName: name,
491 serviceAccount: task?.serviceAccount,
492 }),
493 iamGrants: extras.iamGrants ?? [],
494 codeHash: extras.codeHash,
495 };
496};
497
498const getByName = (name: string) =>
499 cloudrun
500 .getProjectsLocationsJobs({ name })
501 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
502
503const toApiContainers = (
504 containers: JobContainer[] | undefined,
505): cloudrun.GoogleCloudRunV2ContainerList =>
506 (containers ?? []).map((container) => ({
507 name: container.name,
508 image: container.image,
509 command: container.command,
510 args: container.args,
511 env: container.env?.map((item) => ({
512 name: item.name,
513 value: item.value,
514 valueSource: item.valueSource
515 ? {
516 secretKeyRef: item.valueSource.secretKeyRef
517 ? {
518 secret: item.valueSource.secretKeyRef.secret,
519 version: item.valueSource.secretKeyRef.version,
520 }
521 : undefined,
522 }
523 : undefined,
524 })),
525 workingDir: container.workingDir,
526 resources: container.resources
527 ? {
528 limits: container.resources.limits,
529 cpuIdle: container.resources.cpuIdle,
530 startupCpuBoost: container.resources.startupCpuBoost,
531 }
532 : undefined,
533 volumeMounts: container.volumeMounts,
534 }));
535
536const stable = (value: unknown): string =>
537 JSON.stringify(value, (_key, current) => {
538 if (current && typeof current === "object" && !Array.isArray(current)) {
539 return Object.fromEntries(
540 Object.entries(current as Record<string, unknown>)
541 .filter(([, item]) => item !== undefined)
542 .sort(([a], [b]) => a.localeCompare(b)),
543 );
544 }
545 return current;
546 });
547
548const normalizeDuration = (value: string | undefined) => {
549 if (value === undefined) return undefined;
550 const match = /^([0-9]+(?:\.[0-9]+)?)s$/.exec(value.trim());
551 if (!match) return value;
552 return `${Number(match[1])}s`;
553};
554
555const comparableEnv = (env: cloudrun.GoogleCloudRunV2EnvVarList | undefined) =>
556 [...(env ?? [])]
557 .map((item) => ({
558 name: item.name ?? "",
559 value: item.value ?? "",
560 secret: item.valueSource?.secretKeyRef?.secret ?? "",
561 version: item.valueSource?.secretKeyRef?.version ?? "",
562 }))
563 .sort((left, right) => left.name.localeCompare(right.name));
564
565const comparableContainers = (containers: JobContainer[] | undefined) =>
566 (containers ?? []).map((container) => ({
567 name: container.name ?? "",
568 image: container.image,
569 command: container.command ?? [],
570 args: container.args ?? [],
571 env: comparableEnv(container.env),
572 workingDir: container.workingDir ?? "",
573 limits: container.resources?.limits
574 ? tagRecord(container.resources.limits)
575 : {},
576 }));
577
578const comparableObservedContainers = (
579 containers: cloudrun.GoogleCloudRunV2ContainerList | undefined,
580 desired: JobContainer[] | undefined,
581) =>
582 (containers ?? []).map((container, index) => {
583 const user = desired?.[index];
584 return {
585 name: user?.name !== undefined ? (container.name ?? "") : "",
586 image: container.image ?? "",
587 command: container.command ?? [],
588 args: container.args ?? [],
589 env: comparableEnv(container.env),
590 workingDir: container.workingDir ?? "",
591 limits: user?.resources?.limits
592 ? tagRecord(container.resources?.limits)
593 : {},
594 };
595 });
596
597const desiredTemplate = (
598 news: JobProps,
599 current: cloudrun.GoogleCloudRunV2Job | undefined,
600): cloudrun.GoogleCloudRunV2ExecutionTemplate => {
601 const exec = current?.template;
602 const task = exec?.template;
603 return {
604 taskCount: news.taskCount ?? exec?.taskCount,
605 parallelism: news.parallelism ?? exec?.parallelism,
606 labels: news.executionLabels
607 ? toLabels(news.executionLabels)
608 : exec?.labels,
609 annotations: news.executionAnnotations ?? exec?.annotations,
610 template: {
611 containers: toApiContainers(news.containers),
612 timeout: news.timeout ?? task?.timeout,
613 maxRetries: news.maxRetries ?? task?.maxRetries,
614 serviceAccount: news.serviceAccount ?? task?.serviceAccount,
615 executionEnvironment:
616 news.executionEnvironment ?? task?.executionEnvironment,
617 encryptionKey: news.encryptionKey ?? task?.encryptionKey,
618 vpcAccess: news.vpcAccess ?? task?.vpcAccess,
619 volumes: task?.volumes,
620 gpuZonalRedundancyDisabled: task?.gpuZonalRedundancyDisabled,
621 nodeSelector: task?.nodeSelector,
622 },
623 };
624};
625
626/** Waits on a Cloud Run long-running operation (revisions roll out in minutes). */
627const waitForOperation = (
628 operation: cloudrun.GoogleLongrunningOperation,
629 options?: { notFoundOk?: boolean },
630) =>
631 waitForLongRunningOperation(
632 operation,
633 (name) => {
634 const get = cloudrun.getProjectsLocationsOperations({ name });
635 return options?.notFoundOk === true
636 ? get.pipe(
637 Effect.catchTag("NotFound", () =>
638 Effect.succeed<LongRunningOperation>({ name, done: true }),
639 ),
640 )
641 : get.pipe(
642 // A just-returned operation can briefly 404 on read.
643 Effect.retry({
644 while: (error) => error._tag === "NotFound",
645 times: 5,
646 schedule: Schedule.exponential("250 millis"),
647 }),
648 );
649 },
650 { budget: "10 minutes" },
651 ).pipe(
652 // google.rpc.Code NOT_FOUND: the resource was already gone.
653 Effect.catchTag("GCP.OperationFailed", (error) =>
654 options?.notFoundOk === true && error.code === 5
655 ? Effect.succeed<LongRunningOperation>(operation)
656 : Effect.fail(error),
657 ),
658 );
659
660const isPendingJob = (job: cloudrun.GoogleCloudRunV2Job) => {
661 const state = job.terminalCondition?.state ?? "";
662 return (
663 job.reconciling === true ||
664 state === "CONDITION_PENDING" ||
665 state === "CONDITION_RECONCILING" ||
666 state === "" ||
667 state === "STATE_UNSPECIFIED"
668 );
669};
670
671const waitUntilReady = (name: string) =>
672 getByName(name).pipe(
673 Effect.filterOrFail(
674 (job): job is cloudrun.GoogleCloudRunV2Job =>
675 job !== undefined && job.deleteTime === undefined,
676 () => new JobNotResolved({ name }),
677 ),
678 Effect.filterOrFail(
679 (job) => (job.terminalCondition?.state ?? "") !== "CONDITION_FAILED",
680 (job) =>
681 new JobNotReady({
682 name,
683 state: job.terminalCondition?.state ?? "",
684 message: job.terminalCondition?.message ?? "job failed",
685 }),
686 ),
687 Effect.filterOrFail(
688 (job) => !isPendingJob(job),
689 (job) =>
690 new JobReconciling({
691 name,
692 state: job.terminalCondition?.state || "reconciling",
693 }),
694 ),
695 Effect.retry({
696 while: (error) =>
697 error._tag === "GCP.Run.JobReconciling" ||
698 error._tag === "GCP.Run.JobNotResolved",
699 times: 10,
700 schedule: Schedule.spaced("4 seconds"),
701 }),
702 );
703
704const waitUntilGone = (name: string) =>
705 getByName(name).pipe(
706 Effect.flatMap((job) =>
707 job === undefined
708 ? Effect.void
709 : Effect.fail(new JobStillExists({ name })),
710 ),
711 Effect.retry({
712 while: (error) => error._tag === "GCP.Run.JobStillExists",
713 times: 10,
714 schedule: Schedule.spaced("2 seconds"),
715 }),
716 );
717
718export const JobProvider = () =>
719 Provider.succeed(Job, {
720 stables: ["name", "jobId", "project", "location", "uid", "createTime"],
721
722 diff: Effect.fn(function* ({ news, olds, output }) {
723 if (!isResolved(news)) return undefined;
724 const env = yield* GcpEnvironment.current;
725
726 const previousId = olds?.jobId ?? output?.jobId;
727 const nextId = news.jobId ?? previousId;
728 const previousLocation = normalizeLocation(
729 olds?.location ?? output?.location,
730 env.region,
731 );
732 const nextLocation = normalizeLocation(
733 news.location ?? output?.location,
734 env.region,
735 );
736
737 const idChanged =
738 previousId !== undefined &&
739 nextId !== undefined &&
740 nextId !== previousId;
741 const locationChanged = previousLocation !== nextLocation;
742 if (idChanged || locationChanged) {
743 return { action: "replace" as const, deleteFirst: false };
744 }
745 // A code-only change leaves the props untouched; hash the bundled
746 // program (and bootstrap) so it still surfaces as an update.
747 if (output !== undefined && news.main !== undefined) {
748 const images = yield* makeImageSource;
749 const hash = yield* images.hash({
750 source: { main: news.main, handler: news.handler, build: news.build },
751 isExternal: news.isExternal,
752 bootstrap: bootstrapFor(news),
753 });
754 if (hash !== undefined && hash !== output.codeHash) {
755 return { action: "update" as const };
756 }
757 }
758 return undefined;
759 }),
760
761 read: Effect.fn(function* ({ id, olds, output }) {
762 const env = yield* GcpEnvironment.current;
763 const jobId = yield* toId(id, olds?.jobId, output?.jobId);
764 const location = normalizeLocation(
765 olds?.location ?? output?.location,
766 env.region,
767 );
768 const name = output?.name ?? resourceName(env.project, location, jobId);
769 const existing = yield* getByName(name);
770 if (existing === undefined || existing.deleteTime !== undefined) {
771 return undefined;
772 }
773 const attrs = toAttrs(existing, env.project, env.region, {
774 iamGrants: output?.iamGrants,
775 codeHash: output?.codeHash,
776 });
777 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
778 ? attrs
779 : Unowned(attrs);
780 }),
781
782 list: () =>
783 Effect.gen(function* () {
784 const env = yield* GcpEnvironment.current;
785 return yield* cloudrun.listProjectsLocationsJobs
786 .pages({
787 parent: `projects/${env.project}/locations/-`,
788 pageSize: 1000,
789 })
790 .pipe(
791 Stream.flatMap((page) => Stream.fromIterable(page.jobs ?? [])),
792 Stream.filter(
793 (job) =>
794 job.deleteTime === undefined &&
795 Object.keys(job.labels ?? {}).some((key) =>
796 key.startsWith("alchemy-"),
797 ),
798 ),
799 Stream.map((job) => toAttrs(job, env.project, env.region)),
800 Stream.runCollect,
801 Effect.map((chunk) => Array.from(chunk)),
802 );
803 }),
804
805 reconcile: Effect.fn(function* ({ id, news, output, bindings, session }) {
806 const env = yield* GcpEnvironment.current;
807 const jobId = yield* toId(id, news.jobId, output?.jobId);
808 const location = normalizeLocation(
809 news.location ?? output?.location,
810 env.region,
811 );
812 const name = resourceName(env.project, location, jobId);
813 const desiredLabels = {
814 ...toLabels(news.labels),
815 ...(yield* createInternalLabels(id)),
816 };
817 const desiredAnnotations = userAnnotations(news.annotations);
818 const identity = yield* resolveHostIdentity({
819 project: env.project,
820 hostType: HOST_TYPE,
821 resourceName: name,
822 userServiceAccount: news.serviceAccount,
823 effectNative: news.main !== undefined,
824 bindings: bindings as ResourceBinding<GcpHostBinding>[],
825 output,
826 });
827 const serviceAccount = identity.serviceAccount;
828 let containers = news.containers;
829 let codeHash: string | undefined;
830 if (news.main !== undefined) {
831 const images = yield* makeImageSource;
832 const image = yield* images
833 .resolve({
834 id,
835 source: {
836 main: news.main,
837 handler: news.handler ?? "default",
838 build: news.build,
839 },
840 repositoryName: rfc1035(`${jobId}-src`),
841 location,
842 isExternal: news.isExternal,
843 bootstrap: bootstrapFor(news),
844 session,
845 })
846 .pipe(Effect.onError(() => identity.cleanup));
847 codeHash = image.codeHash;
848 const container = news.containers?.[0];
849 containers = [
850 {
851 ...container,
852 image: image.imageUri,
853 env: mergeContainerEnv(
854 container?.env,
855 identity.env,
856 yield* alchemyRuntimeEnv,
857 news.env,
858 ),
859 },
860 ];
861 } else {
862 const runtimeEnv = { ...identity.env, ...news.env };
863 if (Object.keys(runtimeEnv).length > 0 && containers) {
864 containers = containers.map((container, index) =>
865 index === 0
866 ? {
867 ...container,
868 env: mergeContainerEnv(container.env, runtimeEnv),
869 }
870 : container,
871 );
872 }
873 }
874 const effectiveNews: JobProps = { ...news, containers, serviceAccount };
875
876 let current = yield* getByName(name);
877 if (current?.deleteTime !== undefined) {
878 yield* waitUntilGone(name);
879 current = undefined;
880 }
881
882 if (current === undefined) {
883 const created = yield* cloudrun
884 .createProjectsLocationsJobs({
885 parent: `projects/${env.project}/locations/${location}`,
886 jobId,
887 body: {
888 labels: desiredLabels,
889 annotations:
890 Object.keys(desiredAnnotations).length > 0
891 ? desiredAnnotations
892 : undefined,
893 launchStage: news.launchStage,
894 binaryAuthorization: news.binaryAuthorization,
895 template: desiredTemplate(effectiveNews, undefined),
896 },
897 })
898 .pipe(
899 retryActAs,
900 Effect.catchTag("Conflict", () => Effect.succeed(undefined)),
901 Effect.onError(() => identity.cleanup),
902 );
903 if (created !== undefined) {
904 yield* waitForOperation(created);
905 }
906 current = yield* waitUntilReady(name);
907 }
908
909 if (current === undefined) {
910 return yield* new JobNotResolved({ name });
911 }
912
913 const observedLabels = tagRecord(current.labels);
914 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
915 const labelsChanged = upsert.length > 0 || removed.length > 0;
916 const annotationsChanged =
917 news.annotations !== undefined &&
918 stable(desiredAnnotations) !==
919 stable(userAnnotations(current.annotations));
920 const containersChanged =
921 stable(comparableContainers(effectiveNews.containers)) !==
922 stable(
923 comparableObservedContainers(
924 current.template?.template?.containers,
925 effectiveNews.containers,
926 ),
927 );
928 const timeoutChanged =
929 news.timeout !== undefined &&
930 normalizeDuration(news.timeout) !==
931 normalizeDuration(current.template?.template?.timeout);
932 const maxRetriesChanged =
933 news.maxRetries !== undefined &&
934 (current.template?.template?.maxRetries ?? 3) !== news.maxRetries;
935 const taskCountChanged =
936 news.taskCount !== undefined &&
937 (current.template?.taskCount ?? 1) !== news.taskCount;
938 const parallelismChanged =
939 news.parallelism !== undefined &&
940 (current.template?.parallelism ?? 0) !== news.parallelism;
941 const serviceAccountChanged =
942 (current.template?.template?.serviceAccount ?? "") !== serviceAccount;
943 const executionEnvironmentChanged =
944 news.executionEnvironment !== undefined &&
945 (current.template?.template?.executionEnvironment ?? "") !==
946 news.executionEnvironment;
947 const encryptionKeyChanged =
948 news.encryptionKey !== undefined &&
949 (current.template?.template?.encryptionKey ?? "") !==
950 news.encryptionKey;
951 const vpcChanged =
952 news.vpcAccess !== undefined &&
953 stable(news.vpcAccess) !==
954 stable(current.template?.template?.vpcAccess);
955 const launchStageChanged =
956 news.launchStage !== undefined &&
957 (current.launchStage ?? "GA") !== news.launchStage;
958 const binaryAuthorizationChanged =
959 news.binaryAuthorization !== undefined &&
960 stable(news.binaryAuthorization) !==
961 stable(current.binaryAuthorization);
962 const executionLabelsChanged =
963 news.executionLabels !== undefined &&
964 stable(toLabels(news.executionLabels)) !==
965 stable(tagRecord(current.template?.labels));
966 const executionAnnotationsChanged =
967 news.executionAnnotations !== undefined &&
968 stable(tagRecord(news.executionAnnotations)) !==
969 stable(tagRecord(current.template?.annotations));
970
971 if (
972 labelsChanged ||
973 annotationsChanged ||
974 containersChanged ||
975 timeoutChanged ||
976 maxRetriesChanged ||
977 taskCountChanged ||
978 parallelismChanged ||
979 serviceAccountChanged ||
980 executionEnvironmentChanged ||
981 encryptionKeyChanged ||
982 vpcChanged ||
983 launchStageChanged ||
984 binaryAuthorizationChanged ||
985 executionLabelsChanged ||
986 executionAnnotationsChanged
987 ) {
988 const operation = yield* cloudrun.patchProjectsLocationsJobs({
989 name,
990 body: {
991 name,
992 labels: desiredLabels,
993 annotations:
994 news.annotations !== undefined
995 ? desiredAnnotations
996 : current.annotations,
997 launchStage: news.launchStage ?? current.launchStage,
998 binaryAuthorization:
999 news.binaryAuthorization ?? current.binaryAuthorization,
1000 template: desiredTemplate(effectiveNews, current),
1001 },
1002 });
1003 yield* waitForOperation(operation);
1004 current = yield* waitUntilReady(name);
1005 }
1006
1007 if (current === undefined) {
1008 return yield* new JobNotResolved({ name });
1009 }
1010
1011 return toAttrs(current, env.project, env.region, {
1012 iamGrants: identity.grants,
1013 codeHash,
1014 });
1015 }),
1016
1017 delete: Effect.fn(function* ({ id, output }) {
1018 const operation = yield* cloudrun
1019 .deleteProjectsLocationsJobs({ name: output.name })
1020 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
1021 if (operation !== undefined) {
1022 yield* waitForOperation(operation, { notFoundOk: true });
1023 }
1024 yield* waitUntilGone(output.name);
1025 // Effect-native hosts build into a per-host repository on reconcile.
1026 yield* destroyHostImageRepository(
1027 id,
1028 output,
1029 rfc1035(`${output.jobId}-src`),
1030 );
1031 yield* releaseHostIdentity(output);
1032 }),
1033 });