1import * as composer from "@distilled.cloud/gcp/composer_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
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 { tagRecord } from "../../Tags.ts";
12import { GcpEnvironment } from "../Environment.ts";
13import { waitForOperation as waitForGcpOperation } from "../Operation.ts";
14import {
15 createInternalLabels,
16 diffLabels,
17 hasAlchemyLabels,
18 stripInternalLabels,
19 toLabels,
20} from "../Labels.ts";
21import type { Providers } from "../Providers.ts";
22
23// Composer rejects 64-character ids (even on GET) despite documenting
24// "a letter followed by up to 63" characters.
25const MAX_NAME_LENGTH = 63;
26
27/**
28 * Composer rejects `locations/-` (`Unexpected location: -`). Nuke walks
29 * known Composer regions instead.
30 */
31const LIST_LOCATIONS = [
32 "africa-south1",
33 "asia-east1",
34 "asia-east2",
35 "asia-northeast1",
36 "asia-northeast2",
37 "asia-northeast3",
38 "asia-south1",
39 "asia-south2",
40 "asia-southeast1",
41 "asia-southeast2",
42 "australia-southeast1",
43 "australia-southeast2",
44 "europe-central2",
45 "europe-north1",
46 "europe-southwest1",
47 "europe-west1",
48 "europe-west2",
49 "europe-west3",
50 "europe-west4",
51 "europe-west6",
52 "europe-west8",
53 "europe-west9",
54 "europe-west10",
55 "europe-west12",
56 "me-central1",
57 "me-central2",
58 "me-west1",
59 "northamerica-northeast1",
60 "northamerica-northeast2",
61 "southamerica-east1",
62 "southamerica-west1",
63 "us-central1",
64 "us-east1",
65 "us-east4",
66 "us-east5",
67 "us-south1",
68 "us-west1",
69 "us-west2",
70 "us-west3",
71 "us-west4",
72] as const;
73
74export type EnvironmentConfig = composer.EnvironmentConfig;
75export type SoftwareConfig = composer.SoftwareConfig;
76export type NodeConfig = composer.NodeConfig;
77export type WorkloadsConfig = composer.WorkloadsConfig;
78export type PrivateEnvironmentConfig = composer.PrivateEnvironmentConfig;
79export type StorageConfig = composer.StorageConfig;
80export type MaintenanceWindow = composer.MaintenanceWindow;
81export type WebServerNetworkAccessControl =
82 composer.WebServerNetworkAccessControl;
83export type MasterAuthorizedNetworksConfig =
84 composer.MasterAuthorizedNetworksConfig;
85export type RecoveryConfig = composer.RecoveryConfig;
86export type DataRetentionConfig = composer.DataRetentionConfig;
87export type DatabaseConfig = composer.DatabaseConfig;
88export type WebServerConfig = composer.WebServerConfig;
89export type EncryptionConfig = composer.EncryptionConfig;
90
91export type EnvironmentSize =
92 | composer.EnvironmentConfigEnvironmentSizeEnum
93 | (string & {});
94export type ResilienceMode =
95 | composer.EnvironmentConfigResilienceModeEnum
96 | (string & {});
97export type EnvironmentState = composer.EnvironmentStateEnum | (string & {});
98
99export type EnvironmentProps = {
100 /**
101 * Environment id (the `{environment}` segment of
102 * `projects/{project}/locations/{location}/environments/{environment}`).
103 * If omitted, a unique RFC1035 name is generated from the stack, stage,
104 * and logical id. Must start with a lowercase letter, be 1-63 characters,
105 * and match `[a-z]([-a-z0-9]*[a-z0-9])?`. Immutable — changing it
106 * replaces the environment.
107 */
108 environmentId?: string;
109 /**
110 * Region (`us-central1`, `us-east1`, …). Immutable — changing it
111 * replaces the environment. `US-CENTRAL1` is accepted and normalized to
112 * `us-central1`.
113 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
114 */
115 location?: string;
116 /**
117 * User labels. Alchemy ownership labels are merged in automatically.
118 */
119 labels?: Record<string, string>;
120 /**
121 * Configuration for software, nodes, workloads, size, networking, and
122 * related settings. Node network/subnetwork, service account, encryption,
123 * private-IP, python version, and storage bucket are immutable.
124 *
125 * New environments require an explicit `nodeConfig.serviceAccount` —
126 * Composer rejects creates without one (`Composer environment service
127 * account is required to be explicitly specified`). The account needs
128 * `roles/composer.worker` (or broader) on the project. Changing it
129 * replaces the environment.
130 */
131 config?: EnvironmentConfig;
132 /**
133 * Cloud Storage bucket used by the environment (`bucket` has no `gs://`
134 * prefix). Immutable — changing it replaces the environment.
135 */
136 storageConfig?: StorageConfig;
137};
138
139export type Environment = Resource<
140 "GCP.Composer.Environment",
141 EnvironmentProps,
142 {
143 /** Full resource name `projects/{project}/locations/{location}/environments/{environment}`. */
144 name: string;
145 /** Environment id (last path segment). */
146 environmentId: string;
147 /** Project id. */
148 project: string;
149 /** Region id (`us-central1`, …). */
150 location: string;
151 /** Server-generated UUID. */
152 uuid: string | undefined;
153 /** Server-reported state (`RUNNING`, `CREATING`, …). */
154 state: string | undefined;
155 /** User labels (Alchemy ownership labels stripped). */
156 labels: Record<string, string>;
157 /** Configuration currently applied, including output-only URIs. */
158 config: EnvironmentConfig | undefined;
159 /** Storage configuration currently applied. */
160 storageConfig: StorageConfig | undefined;
161 /** Apache Airflow Web UI URI. */
162 airflowUri: string | undefined;
163 /** Airflow Web UI URI for workforce identity federation. */
164 airflowByoidUri: string | undefined;
165 /** GKE cluster that runs this environment. */
166 gkeCluster: string | undefined;
167 /** Cloud Storage prefix of the DAGs for this environment. */
168 dagGcsPrefix: string | undefined;
169 /** RFC3339 creation timestamp. */
170 createTime: string | undefined;
171 /** RFC3339 last-update timestamp. */
172 updateTime: string | undefined;
173 },
174 never,
175 Providers
176>;
177
178/**
179 * A Cloud Composer environment (Managed Apache Airflow).
180 *
181 * Changing `environmentId`, `location`, node network/subnetwork, node
182 * service account, encryption, private-IP settings, python version, or the
183 * storage bucket replaces the environment. Labels, PyPI packages, Airflow config overrides, env vars,
184 * image version, workloads, and environment size update in place — one
185 * update type per Composer patch, applied sequentially.
186 *
187 * Provisioning typically takes 20–45 minutes. Polls the LRO via
188 * `getProjectsLocationsOperations` (Composer has no wait long-poll).
189 *
190 * New environments must name their service account explicitly in
191 * `config.nodeConfig.serviceAccount`; Composer no longer falls back to the
192 * default Compute Engine account. The account needs `roles/composer.worker`
193 * (or broader) on the project. Changing it replaces the environment.
194 *
195 * ### Creating an Environment
196 * **Example:** Generated name, Composer 3 small
197 * ```typescript
198 * const airflow = yield* GCP.Composer.Environment("Airflow", {
199 * config: {
200 * environmentSize: "ENVIRONMENT_SIZE_SMALL",
201 * nodeConfig: {
202 * serviceAccount: "composer-env@my-project.iam.gserviceaccount.com",
203 * },
204 * softwareConfig: { imageVersion: "composer-3-airflow-2" },
205 * },
206 * });
207 * ```
208 *
209 * **Example:** Explicit id, labels, and PyPI packages
210 * ```typescript
211 * const airflow = yield* GCP.Composer.Environment("Airflow", {
212 * environmentId: "app-airflow",
213 * location: "us-central1",
214 * labels: { env: "prod" },
215 * config: {
216 * environmentSize: "ENVIRONMENT_SIZE_SMALL",
217 * nodeConfig: {
218 * serviceAccount: "composer-env@my-project.iam.gserviceaccount.com",
219 * },
220 * softwareConfig: {
221 * imageVersion: "composer-3-airflow-2",
222 * pypiPackages: { numpy: "==2.1.0" },
223 * },
224 * },
225 * });
226 * ```
227 *
228 * ### Updating software
229 * **Example:** Override Airflow config and env vars
230 * ```typescript
231 * const airflow = yield* GCP.Composer.Environment("Airflow", {
232 * environmentId: "app-airflow",
233 * config: {
234 * softwareConfig: {
235 * airflowConfigOverrides: { "core-dags_are_paused_at_creation": "True" },
236 * envVariables: { EXAMPLE_VAR: "test" },
237 * },
238 * },
239 * });
240 * ```
241 *
242 * @resource
243 * @category Composer
244 */
245export const Environment = Resource<Environment>("GCP.Composer.Environment");
246
247export class EnvironmentNotResolved extends Data.TaggedError(
248 "GCP.Composer.EnvironmentNotResolved",
249)<{
250 name: string;
251}> {}
252
253export class EnvironmentNotReady extends Data.TaggedError(
254 "GCP.Composer.EnvironmentNotReady",
255)<{
256 name: string;
257 state: string;
258}> {}
259
260export class EnvironmentFailed extends Data.TaggedError(
261 "GCP.Composer.EnvironmentFailed",
262)<{
263 name: string;
264 state: string | undefined;
265}> {}
266
267export class EnvironmentStillExists extends Data.TaggedError(
268 "GCP.Composer.EnvironmentStillExists",
269)<{
270 name: string;
271}> {}
272
273const lastSegment = (value: string) => {
274 const trimmed = value.replace(/\/+$/, "");
275 const parts = trimmed.split("/");
276 return parts[parts.length - 1] || trimmed;
277};
278
279const normalizeLocation = (
280 location: string | undefined,
281 defaultLocation: string,
282) => lastSegment(location ?? defaultLocation).toLowerCase();
283
284const rfc1035 = (name: string): string => {
285 let next = name
286 .toLowerCase()
287 .replace(/[^a-z0-9-]/g, "-")
288 .replace(/-+/g, "-")
289 .replace(/^-+|-+$/g, "");
290 if (!/^[a-z]/.test(next)) next = `c${next}`;
291 next = next.slice(0, MAX_NAME_LENGTH).replace(/-+$/g, "");
292 if (next.length === 0) return "environment";
293 if (!/[a-z0-9]$/.test(next)) next = `${next.slice(0, MAX_NAME_LENGTH - 1)}0`;
294 return next.slice(0, MAX_NAME_LENGTH);
295};
296
297const resourceName = (
298 project: string,
299 location: string,
300 environmentId: string,
301) => `projects/${project}/locations/${location}/environments/${environmentId}`;
302
303const parseName = (name: string, defaultLocation: string) => {
304 const parts = name.split("/").filter((part) => part.length > 0);
305 const environmentsAt = parts.lastIndexOf("environments");
306 const locationsAt = parts.lastIndexOf("locations");
307 const projectsAt = parts.lastIndexOf("projects");
308 return {
309 project:
310 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
311 location:
312 locationsAt >= 0 && parts[locationsAt + 1]
313 ? parts[locationsAt + 1]!
314 : defaultLocation,
315 environmentId:
316 environmentsAt >= 0 && parts[environmentsAt + 1]
317 ? parts[environmentsAt + 1]!
318 : lastSegment(name),
319 };
320};
321
322const userLabels = (
323 labels: Record<string, string | undefined> | null | undefined,
324): Record<string, string> => stripInternalLabels(tagRecord(labels));
325
326const toId = (
327 id: string,
328 environmentId: string | undefined,
329 existing?: string,
330) =>
331 Effect.gen(function* () {
332 return (
333 environmentId ??
334 existing ??
335 rfc1035(
336 yield* createPhysicalName({
337 id,
338 maxLength: MAX_NAME_LENGTH,
339 lowercase: true,
340 }),
341 )
342 );
343 });
344
345const comparable = (value: unknown): unknown => {
346 if (value === undefined) return undefined;
347 if (value === null) return null;
348 if (Array.isArray(value)) return value.map(comparable);
349 if (typeof value === "object") {
350 return Object.fromEntries(
351 Object.entries(value as Record<string, unknown>)
352 .filter(([, item]) => item !== undefined)
353 .sort(([a], [b]) => a.localeCompare(b))
354 .map(([key, item]) => [key, comparable(item)]),
355 );
356 }
357 return value;
358};
359
360const jsonKey = (value: unknown) => JSON.stringify(comparable(value) ?? null);
361
362const mapOf = (
363 map: Record<string, string | undefined> | null | undefined,
364): Record<string, string> =>
365 Object.fromEntries(
366 Object.entries(map ?? {}).filter(
367 (entry): entry is [string, string] => entry[1] !== undefined,
368 ),
369 );
370
371const mapKey = (map: Record<string, string | undefined> | null | undefined) =>
372 jsonKey(mapOf(map));
373
374const sizeOf = (size: string | undefined) => {
375 const value = (size ?? "").toUpperCase();
376 return value === "ENVIRONMENT_SIZE_UNSPECIFIED" ? "" : value;
377};
378
379const resilienceOf = (mode: string | undefined) => {
380 const value = (mode ?? "").toUpperCase();
381 return value === "RESILIENCE_MODE_UNSPECIFIED" ? "" : value;
382};
383
384const pluginsOf = (mode: string | undefined) => {
385 const value = (mode ?? "").toUpperCase();
386 return value === "WEB_SERVER_PLUGINS_MODE_UNSPECIFIED" ? "" : value;
387};
388
389const imageVersionMatches = (
390 desired: string | undefined,
391 observed: string | undefined,
392) => {
393 if (desired === undefined || desired.length === 0) return true;
394 if (observed === undefined) return false;
395 if (desired === observed) return true;
396 if (!observed.startsWith(desired)) return false;
397 const next = observed[desired.length];
398 return next === undefined || next === "-" || next === ".";
399};
400
401const networkId = (value: string | undefined) =>
402 value === undefined || value.length === 0 ? undefined : lastSegment(value);
403
404const fieldChanged = (next: unknown, prev: unknown) =>
405 next !== undefined && prev !== undefined && jsonKey(next) !== jsonKey(prev);
406
407const networkChanged = (next: string | undefined, prev: string | undefined) =>
408 next !== undefined &&
409 prev !== undefined &&
410 networkId(next) !== networkId(prev);
411
412const listChanged = (
413 next: ReadonlyArray<string> | undefined,
414 prev: ReadonlyArray<string> | undefined,
415) =>
416 next !== undefined &&
417 prev !== undefined &&
418 jsonKey([...next].map((item) => item.toLowerCase()).sort()) !==
419 jsonKey([...prev].map((item) => item.toLowerCase()).sort());
420
421const immutableChanged = (
422 news: EnvironmentProps,
423 olds: Partial<EnvironmentProps> | undefined,
424 output: Environment["Attributes"] | undefined,
425) => {
426 const prevConfig = olds?.config ?? output?.config;
427 const nextConfig = news.config;
428 const prevStorage = olds?.storageConfig ?? output?.storageConfig;
429 const nextStorage = news.storageConfig;
430 const prevNode = prevConfig?.nodeConfig;
431 const nextNode = nextConfig?.nodeConfig;
432 const prevPrivate = prevConfig?.privateEnvironmentConfig;
433 const nextPrivate = nextConfig?.privateEnvironmentConfig;
434
435 return (
436 (nextConfig !== undefined &&
437 prevConfig !== undefined &&
438 (networkChanged(nextNode?.network, prevNode?.network) ||
439 networkChanged(nextNode?.subnetwork, prevNode?.subnetwork) ||
440 fieldChanged(nextNode?.machineType, prevNode?.machineType) ||
441 fieldChanged(nextNode?.diskSizeGb, prevNode?.diskSizeGb) ||
442 networkChanged(nextNode?.location, prevNode?.location) ||
443 networkChanged(nextNode?.serviceAccount, prevNode?.serviceAccount) ||
444 listChanged(nextNode?.oauthScopes, prevNode?.oauthScopes) ||
445 listChanged(nextNode?.tags, prevNode?.tags) ||
446 fieldChanged(
447 nextNode?.ipAllocationPolicy,
448 prevNode?.ipAllocationPolicy,
449 ) ||
450 fieldChanged(
451 nextNode?.enableIpMasqAgent,
452 prevNode?.enableIpMasqAgent,
453 ) ||
454 fieldChanged(
455 nextNode?.composerInternalIpv4CidrBlock,
456 prevNode?.composerInternalIpv4CidrBlock,
457 ) ||
458 networkChanged(
459 nextNode?.composerNetworkAttachment,
460 prevNode?.composerNetworkAttachment,
461 ) ||
462 fieldChanged(
463 nextConfig.encryptionConfig?.kmsKeyName,
464 prevConfig.encryptionConfig?.kmsKeyName,
465 ) ||
466 fieldChanged(
467 nextConfig.softwareConfig?.pythonVersion,
468 prevConfig.softwareConfig?.pythonVersion,
469 ) ||
470 fieldChanged(
471 nextConfig.databaseConfig?.zone,
472 prevConfig.databaseConfig?.zone,
473 ) ||
474 fieldChanged(
475 nextPrivate?.enablePrivateEnvironment,
476 prevPrivate?.enablePrivateEnvironment,
477 ) ||
478 fieldChanged(
479 nextPrivate?.networkingType,
480 prevPrivate?.networkingType,
481 ) ||
482 networkChanged(
483 nextPrivate?.cloudComposerConnectionSubnetwork,
484 prevPrivate?.cloudComposerConnectionSubnetwork,
485 ) ||
486 fieldChanged(
487 nextPrivate?.cloudComposerNetworkIpv4CidrBlock,
488 prevPrivate?.cloudComposerNetworkIpv4CidrBlock,
489 ) ||
490 fieldChanged(
491 nextPrivate?.cloudSqlIpv4CidrBlock,
492 prevPrivate?.cloudSqlIpv4CidrBlock,
493 ) ||
494 fieldChanged(
495 nextPrivate?.webServerIpv4CidrBlock,
496 prevPrivate?.webServerIpv4CidrBlock,
497 ) ||
498 fieldChanged(
499 nextPrivate?.enablePrivatelyUsedPublicIps,
500 prevPrivate?.enablePrivatelyUsedPublicIps,
501 ) ||
502 fieldChanged(
503 nextPrivate?.privateClusterConfig,
504 prevPrivate?.privateClusterConfig,
505 ) ||
506 fieldChanged(
507 nextPrivate?.networkingConfig,
508 prevPrivate?.networkingConfig,
509 ) ||
510 fieldChanged(
511 nextPrivate?.enablePrivateBuildsOnly,
512 prevPrivate?.enablePrivateBuildsOnly,
513 ))) ||
514 (nextStorage?.bucket !== undefined &&
515 prevStorage?.bucket !== undefined &&
516 nextStorage.bucket !== prevStorage.bucket)
517 );
518};
519
520const toAttrs = (
521 environment: composer.Environment,
522 project: string,
523 region: string,
524) => {
525 const name = environment.name ?? "";
526 const parsed = parseName(name, region);
527 return {
528 name,
529 environmentId: parsed.environmentId,
530 project: parsed.project || project,
531 location: parsed.location,
532 uuid: environment.uuid,
533 state: environment.state,
534 labels: userLabels(environment.labels),
535 config: environment.config,
536 storageConfig: environment.storageConfig,
537 airflowUri: environment.config?.airflowUri,
538 airflowByoidUri: environment.config?.airflowByoidUri,
539 gkeCluster: environment.config?.gkeCluster,
540 dagGcsPrefix: environment.config?.dagGcsPrefix,
541 createTime: environment.createTime,
542 updateTime: environment.updateTime,
543 };
544};
545
546const isPlaceholder = (environment: composer.Environment) => {
547 const name = environment.name ?? "";
548 return name.endsWith("/environments/-") || name.endsWith("/environments/");
549};
550
551const getByName = (name: string) =>
552 composer
553 .getProjectsLocationsEnvironments({ name })
554 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
555
556// Environment creates run 20-30 minutes, deletes 10-20.
557const waitForOperation = (operation: composer.Operation) =>
558 waitForGcpOperation(
559 operation,
560 (name) => composer.getProjectsLocationsOperations({ name }),
561 { budget: "45 minutes", interval: "15 seconds" },
562 );
563
564const waitUntilRunning = (name: string) =>
565 getByName(name).pipe(
566 Effect.filterOrFail(
567 (environment): environment is composer.Environment =>
568 environment !== undefined,
569 () => new EnvironmentNotResolved({ name }),
570 ),
571 Effect.filterOrFail(
572 (environment) => environment.state !== "ERROR",
573 (environment) =>
574 new EnvironmentFailed({
575 name,
576 state: environment.state,
577 }),
578 ),
579 Effect.filterOrFail(
580 (environment) => environment.state === "RUNNING",
581 (environment) =>
582 new EnvironmentNotReady({
583 name,
584 state: environment.state ?? "STATE_UNSPECIFIED",
585 }),
586 ),
587 Effect.retry({
588 while: (error) =>
589 error._tag === "GCP.Composer.EnvironmentNotResolved" ||
590 error._tag === "GCP.Composer.EnvironmentNotReady",
591 times: 240,
592 schedule: Schedule.spaced("15 seconds"),
593 }),
594 );
595
596const waitUntilGone = (name: string) =>
597 getByName(name).pipe(
598 Effect.flatMap((environment) =>
599 environment === undefined
600 ? Effect.void
601 : Effect.fail(new EnvironmentStillExists({ name })),
602 ),
603 Effect.retry({
604 while: (error) => error._tag === "GCP.Composer.EnvironmentStillExists",
605 times: 120,
606 schedule: Schedule.spaced("15 seconds"),
607 }),
608 );
609
610const listOwnedAt = (project: string, location: string) =>
611 composer.listProjectsLocationsEnvironments
612 .pages({
613 parent: `projects/${project}/locations/${location}`,
614 pageSize: 1000,
615 })
616 .pipe(
617 Stream.flatMap((page) => Stream.fromIterable(page.environments ?? [])),
618 Stream.filter(
619 (environment) =>
620 !isPlaceholder(environment) &&
621 Object.keys(environment.labels ?? {}).some((key) =>
622 key.startsWith("alchemy-"),
623 ),
624 ),
625 Stream.map((environment) => toAttrs(environment, project, location)),
626 Stream.runCollect,
627 Effect.map((chunk) => Array.from(chunk)),
628 // A location Composer does not serve has no environments.
629 Effect.catchTag("NotFound", () => Effect.succeed([])),
630 );
631
632const toCreateConfig = (
633 config: composer.EnvironmentConfig | undefined,
634): composer.EnvironmentConfig | undefined => {
635 if (config === undefined) return undefined;
636 const {
637 airflowByoidUri: _airflowByoidUri,
638 airflowUri: _airflowUri,
639 gkeCluster: _gkeCluster,
640 dagGcsPrefix: _dagGcsPrefix,
641 ...rest
642 } = config;
643 return rest;
644};
645
646type EnvironmentPatch = {
647 mask: string;
648 body: composer.Environment;
649};
650
651const desiredPatches = (
652 news: EnvironmentProps,
653 current: composer.Environment,
654 desiredLabels: Record<string, string>,
655 labelsChanged: boolean,
656): EnvironmentPatch[] => {
657 const patches: EnvironmentPatch[] = [];
658 if (labelsChanged) {
659 patches.push({ mask: "labels", body: { labels: desiredLabels } });
660 }
661
662 const nextSoftware = news.config?.softwareConfig;
663 const currentSoftware = current.config?.softwareConfig;
664
665 if (
666 nextSoftware?.pypiPackages !== undefined &&
667 mapKey(nextSoftware.pypiPackages) !== mapKey(currentSoftware?.pypiPackages)
668 ) {
669 patches.push({
670 mask: "config.softwareConfig.pypiPackages",
671 body: {
672 config: { softwareConfig: { pypiPackages: nextSoftware.pypiPackages } },
673 },
674 });
675 }
676 if (
677 nextSoftware?.airflowConfigOverrides !== undefined &&
678 mapKey(nextSoftware.airflowConfigOverrides) !==
679 mapKey(currentSoftware?.airflowConfigOverrides)
680 ) {
681 patches.push({
682 mask: "config.softwareConfig.airflowConfigOverrides",
683 body: {
684 config: {
685 softwareConfig: {
686 airflowConfigOverrides: nextSoftware.airflowConfigOverrides,
687 },
688 },
689 },
690 });
691 }
692 if (
693 nextSoftware?.envVariables !== undefined &&
694 mapKey(nextSoftware.envVariables) !== mapKey(currentSoftware?.envVariables)
695 ) {
696 patches.push({
697 mask: "config.softwareConfig.envVariables",
698 body: {
699 config: { softwareConfig: { envVariables: nextSoftware.envVariables } },
700 },
701 });
702 }
703 if (
704 nextSoftware?.imageVersion !== undefined &&
705 !imageVersionMatches(
706 nextSoftware.imageVersion,
707 currentSoftware?.imageVersion,
708 )
709 ) {
710 patches.push({
711 mask: "config.softwareConfig.imageVersion",
712 body: {
713 config: { softwareConfig: { imageVersion: nextSoftware.imageVersion } },
714 },
715 });
716 }
717 if (
718 nextSoftware?.schedulerCount !== undefined &&
719 (currentSoftware?.schedulerCount ?? 0) !== nextSoftware.schedulerCount
720 ) {
721 patches.push({
722 mask: "config.softwareConfig.schedulerCount",
723 body: {
724 config: {
725 softwareConfig: { schedulerCount: nextSoftware.schedulerCount },
726 },
727 },
728 });
729 }
730 if (
731 nextSoftware?.cloudDataLineageIntegration !== undefined &&
732 jsonKey(nextSoftware.cloudDataLineageIntegration) !==
733 jsonKey(currentSoftware?.cloudDataLineageIntegration)
734 ) {
735 patches.push({
736 mask: "config.softwareConfig.cloudDataLineageIntegration",
737 body: {
738 config: {
739 softwareConfig: {
740 cloudDataLineageIntegration:
741 nextSoftware.cloudDataLineageIntegration,
742 },
743 },
744 },
745 });
746 }
747 if (
748 nextSoftware?.webServerPluginsMode !== undefined &&
749 pluginsOf(nextSoftware.webServerPluginsMode) !==
750 pluginsOf(currentSoftware?.webServerPluginsMode)
751 ) {
752 patches.push({
753 mask: "config.softwareConfig.webServerPluginsMode",
754 body: {
755 config: {
756 softwareConfig: {
757 webServerPluginsMode: nextSoftware.webServerPluginsMode,
758 },
759 },
760 },
761 });
762 }
763
764 const nextConfig = news.config;
765 if (
766 nextConfig?.nodeCount !== undefined &&
767 (current.config?.nodeCount ?? 0) !== nextConfig.nodeCount
768 ) {
769 patches.push({
770 mask: "config.nodeCount",
771 body: { config: { nodeCount: nextConfig.nodeCount } },
772 });
773 }
774 if (
775 nextConfig?.webServerNetworkAccessControl !== undefined &&
776 jsonKey(nextConfig.webServerNetworkAccessControl) !==
777 jsonKey(current.config?.webServerNetworkAccessControl)
778 ) {
779 patches.push({
780 mask: "config.webServerNetworkAccessControl",
781 body: {
782 config: {
783 webServerNetworkAccessControl:
784 nextConfig.webServerNetworkAccessControl,
785 },
786 },
787 });
788 }
789 if (
790 nextConfig?.databaseConfig?.machineType !== undefined &&
791 (current.config?.databaseConfig?.machineType ?? "") !==
792 nextConfig.databaseConfig.machineType
793 ) {
794 patches.push({
795 mask: "config.databaseConfig.machineType",
796 body: {
797 config: {
798 databaseConfig: {
799 machineType: nextConfig.databaseConfig.machineType,
800 },
801 },
802 },
803 });
804 }
805 if (
806 nextConfig?.webServerConfig?.machineType !== undefined &&
807 (current.config?.webServerConfig?.machineType ?? "") !==
808 nextConfig.webServerConfig.machineType
809 ) {
810 patches.push({
811 mask: "config.webServerConfig.machineType",
812 body: {
813 config: {
814 webServerConfig: {
815 machineType: nextConfig.webServerConfig.machineType,
816 },
817 },
818 },
819 });
820 }
821 if (
822 nextConfig?.maintenanceWindow !== undefined &&
823 jsonKey(nextConfig.maintenanceWindow) !==
824 jsonKey(current.config?.maintenanceWindow)
825 ) {
826 patches.push({
827 mask: "config.maintenanceWindow",
828 body: { config: { maintenanceWindow: nextConfig.maintenanceWindow } },
829 });
830 }
831 if (
832 nextConfig?.workloadsConfig !== undefined &&
833 jsonKey(nextConfig.workloadsConfig) !==
834 jsonKey(current.config?.workloadsConfig)
835 ) {
836 patches.push({
837 mask: "config.workloadsConfig",
838 body: { config: { workloadsConfig: nextConfig.workloadsConfig } },
839 });
840 }
841 if (
842 nextConfig?.environmentSize !== undefined &&
843 sizeOf(nextConfig.environmentSize) !==
844 sizeOf(current.config?.environmentSize)
845 ) {
846 patches.push({
847 mask: "config.environmentSize",
848 body: { config: { environmentSize: nextConfig.environmentSize } },
849 });
850 }
851 if (
852 nextConfig?.resilienceMode !== undefined &&
853 resilienceOf(nextConfig.resilienceMode) !==
854 resilienceOf(current.config?.resilienceMode)
855 ) {
856 patches.push({
857 mask: "config.resilienceMode",
858 body: { config: { resilienceMode: nextConfig.resilienceMode } },
859 });
860 }
861 if (
862 nextConfig?.masterAuthorizedNetworksConfig !== undefined &&
863 jsonKey(nextConfig.masterAuthorizedNetworksConfig) !==
864 jsonKey(current.config?.masterAuthorizedNetworksConfig)
865 ) {
866 patches.push({
867 mask: "config.masterAuthorizedNetworksConfig",
868 body: {
869 config: {
870 masterAuthorizedNetworksConfig:
871 nextConfig.masterAuthorizedNetworksConfig,
872 },
873 },
874 });
875 }
876 if (
877 nextConfig?.recoveryConfig !== undefined &&
878 jsonKey(nextConfig.recoveryConfig) !==
879 jsonKey(current.config?.recoveryConfig)
880 ) {
881 patches.push({
882 mask: "config.recoveryConfig",
883 body: { config: { recoveryConfig: nextConfig.recoveryConfig } },
884 });
885 }
886 if (
887 nextConfig?.dataRetentionConfig !== undefined &&
888 jsonKey(nextConfig.dataRetentionConfig) !==
889 jsonKey(current.config?.dataRetentionConfig)
890 ) {
891 patches.push({
892 mask: "config.dataRetentionConfig",
893 body: {
894 config: { dataRetentionConfig: nextConfig.dataRetentionConfig },
895 },
896 });
897 }
898
899 return patches;
900};
901
902export const EnvironmentProvider = () =>
903 Provider.succeed(Environment, {
904 stables: [
905 "name",
906 "environmentId",
907 "project",
908 "location",
909 "uuid",
910 "createTime",
911 ],
912
913 diff: Effect.fn(function* ({ news, olds, output }) {
914 if (!isResolved(news)) return undefined;
915 const env = yield* GcpEnvironment.current;
916
917 const previousId = olds?.environmentId ?? output?.environmentId;
918 const nextId = news.environmentId ?? previousId;
919 const previousLocation = normalizeLocation(
920 olds?.location ?? output?.location,
921 env.region,
922 );
923 const nextLocation = normalizeLocation(
924 news.location ?? output?.location,
925 env.region,
926 );
927 const replace =
928 (previousId !== undefined &&
929 nextId !== undefined &&
930 nextId !== previousId) ||
931 previousLocation !== nextLocation ||
932 immutableChanged(news, olds, output);
933
934 if (!replace) return undefined;
935 return {
936 action: "replace" as const,
937 deleteFirst:
938 previousLocation === nextLocation &&
939 previousId !== undefined &&
940 nextId === previousId,
941 };
942 }),
943
944 read: Effect.fn(function* ({ id, olds, output }) {
945 const env = yield* GcpEnvironment.current;
946 const environmentId = yield* toId(
947 id,
948 olds?.environmentId,
949 output?.environmentId,
950 );
951 const location = normalizeLocation(
952 olds?.location ?? output?.location,
953 env.region,
954 );
955 const name =
956 output?.name ?? resourceName(env.project, location, environmentId);
957 const existing = yield* getByName(name);
958 if (existing === undefined) return undefined;
959 const attrs = toAttrs(existing, env.project, env.region);
960 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
961 ? attrs
962 : Unowned(attrs);
963 }),
964
965 list: () =>
966 Effect.gen(function* () {
967 const env = yield* GcpEnvironment.current;
968 const pages = yield* Effect.forEach(
969 Array.from(new Set<string>([...LIST_LOCATIONS, env.region])),
970 (location) => listOwnedAt(env.project, location),
971 { concurrency: 8 },
972 );
973 return pages.flat();
974 }),
975
976 reconcile: Effect.fn(function* ({ id, news, output }) {
977 const env = yield* GcpEnvironment.current;
978 const environmentId = yield* toId(
979 id,
980 news.environmentId,
981 output?.environmentId,
982 );
983 const location = normalizeLocation(
984 news.location ?? output?.location,
985 env.region,
986 );
987 const name = resourceName(env.project, location, environmentId);
988 const desiredLabels = {
989 ...toLabels(news.labels),
990 ...(yield* createInternalLabels(id)),
991 };
992
993 let current = yield* getByName(output?.name ?? name);
994
995 if (current === undefined) {
996 const created = yield* composer
997 .createProjectsLocationsEnvironments({
998 parent: `projects/${env.project}/locations/${location}`,
999 body: {
1000 name,
1001 labels: desiredLabels,
1002 config: toCreateConfig(news.config),
1003 storageConfig: news.storageConfig,
1004 },
1005 })
1006 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
1007 if (created !== undefined) {
1008 // ALREADY_EXISTS (6): a concurrent create won the race.
1009 yield* waitForOperation(created).pipe(
1010 Effect.catchTag("GCP.OperationFailed", (error) =>
1011 error.code === 6 ? Effect.void : Effect.fail(error),
1012 ),
1013 );
1014 }
1015 current = yield* waitUntilRunning(name);
1016 }
1017
1018 if (current === undefined) {
1019 return yield* new EnvironmentNotResolved({ name });
1020 }
1021
1022 if (current.state === "ERROR") {
1023 return yield* new EnvironmentFailed({
1024 name,
1025 state: current.state,
1026 });
1027 }
1028
1029 if (current.state !== "RUNNING") {
1030 current = yield* waitUntilRunning(name);
1031 }
1032
1033 const observedLabels = tagRecord(current.labels);
1034 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
1035 const labelsChanged = upsert.length > 0 || removed.length > 0;
1036 const patches = desiredPatches(
1037 news,
1038 current,
1039 desiredLabels,
1040 labelsChanged,
1041 );
1042
1043 for (const patch of patches) {
1044 const operation = yield* composer
1045 .patchProjectsLocationsEnvironments({
1046 name: current.name ?? name,
1047 updateMask: patch.mask,
1048 body: patch.body,
1049 })
1050 .pipe(
1051 Effect.retry({
1052 while: (error) => error._tag === "Conflict",
1053 times: 8,
1054 schedule: Schedule.spaced("5 seconds"),
1055 }),
1056 );
1057 yield* waitForOperation(operation);
1058 current = yield* waitUntilRunning(current.name ?? name);
1059 }
1060
1061 return toAttrs(current, env.project, env.region);
1062 }),
1063
1064 delete: Effect.fn(function* ({ output }) {
1065 const operation = yield* composer
1066 .deleteProjectsLocationsEnvironments({ name: output.name })
1067 .pipe(
1068 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
1069 Effect.retry({
1070 while: (error) => error._tag === "Conflict",
1071 times: 8,
1072 schedule: Schedule.spaced("5 seconds"),
1073 }),
1074 );
1075 if (operation !== undefined) {
1076 // NOT_FOUND (5): the environment was already gone.
1077 yield* waitForOperation(operation).pipe(
1078 Effect.catchTag("GCP.OperationFailed", (error) =>
1079 error.code === 5 ? Effect.void : Effect.fail(error),
1080 ),
1081 );
1082 }
1083 yield* waitUntilGone(output.name);
1084 }),
1085 });