1import * as dataproc from "@distilled.cloud/gcp/dataproc_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { tagRecord } from "../../Tags.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import {
11 createInternalLabels,
12 diffLabels,
13 hasAlchemyLabels,
14 toLabels,
15} from "../Labels.ts";
16import type { Providers } from "../Providers.ts";
17import type { WorkflowTemplateProps } from "./WorkflowTemplate.ts";
18import {
19 collectPages,
20 LIST_LOCATIONS,
21 MAX_MANAGED_CLUSTER_PREFIX_LENGTH,
22 MAX_POLICY_ID_LENGTH,
23 defaultWorkflowJobs,
24 defaultWorkflowPlacement,
25 emptyOnMissing,
26 fingerprint,
27 hasAlchemyLabelMap,
28 normalizeLocation,
29 parseResourceName,
30 regionParent,
31 rfc1035,
32 toPhysicalId,
33 userLabels,
34 waitUntilGone,
35} from "./internal.ts";
36
37export type RegionsWorkflowTemplateProps = Omit<
38 WorkflowTemplateProps,
39 "location"
40> & {
41 /**
42 * Region (`us-central1`, …). Immutable — changing it replaces the
43 * template. `US-CENTRAL1` is accepted and normalized to `us-central1`.
44 * @default the stack's GCP region (`GCP.Region`, profile region, `us-central1`)
45 */
46 region?: string;
47};
48
49export type RegionsWorkflowTemplate = Resource<
50 "GCP.Dataproc.RegionsWorkflowTemplate",
51 RegionsWorkflowTemplateProps,
52 {
53 /** Full resource name `projects/{project}/regions/{region}/workflowTemplates/{template}`. */
54 name: string;
55 /** Template id (last path segment). */
56 templateId: string;
57 /** Project id. */
58 project: string;
59 /** Region id. */
60 region: string;
61 /** User labels (Alchemy ownership labels stripped). */
62 labels: Record<string, string>;
63 /** Server version used for optimistic updates. */
64 version: number | undefined;
65 /** DAG timeout. */
66 dagTimeout: string | undefined;
67 /** RFC3339 creation timestamp. */
68 createTime: string | undefined;
69 /** RFC3339 last-update timestamp. */
70 updateTime: string | undefined;
71 },
72 never,
73 Providers
74>;
75
76/**
77 * A Dataproc workflow template (regions API).
78 *
79 * Same resource as {@link WorkflowTemplate} addressed via
80 * `projects/{project}/regions/{region}/workflowTemplates/{template}`.
81 *
82 * ### Creating a Template
83 * **Example:** Generated name
84 * ```typescript
85 * const template = yield* GCP.Dataproc.RegionsWorkflowTemplate("Nightly", {});
86 * ```
87 *
88 * **Example:** Explicit id and labels
89 * ```typescript
90 * const template = yield* GCP.Dataproc.RegionsWorkflowTemplate("Nightly", {
91 * templateId: "nightly-etl-reg",
92 * region: "us-central1",
93 * labels: { env: "prod" },
94 * });
95 * ```
96 *
97 * @resource
98 * @category Dataproc
99 */
100export const RegionsWorkflowTemplate = Resource<RegionsWorkflowTemplate>(
101 "GCP.Dataproc.RegionsWorkflowTemplate",
102);
103
104export class RegionsWorkflowTemplateNotResolved extends Data.TaggedError(
105 "GCP.Dataproc.RegionsWorkflowTemplateNotResolved",
106)<{
107 name: string;
108}> {}
109
110const resourceName = (project: string, region: string, templateId: string) =>
111 `${regionParent(project, region)}/workflowTemplates/${templateId}`;
112
113const toAttrs = (
114 template: dataproc.WorkflowTemplate,
115 project: string,
116 region: string,
117) => {
118 const name = template.name ?? "";
119 const parsed = parseResourceName(name, "workflowTemplates", region);
120 return {
121 name,
122 templateId: template.id ?? parsed.id,
123 project: parsed.project || project,
124 region: parsed.location || region,
125 labels: userLabels(template.labels),
126 version: template.version,
127 dagTimeout: template.dagTimeout,
128 createTime: template.createTime,
129 updateTime: template.updateTime,
130 };
131};
132
133const desiredBody = (
134 news: RegionsWorkflowTemplateProps,
135 templateId: string,
136 name: string,
137 desiredLabels: Record<string, string>,
138 version?: number,
139): dataproc.WorkflowTemplate => ({
140 id: templateId,
141 name,
142 labels: desiredLabels,
143 version,
144 jobs: news.jobs ?? defaultWorkflowJobs(),
145 placement:
146 news.placement ??
147 defaultWorkflowPlacement(
148 rfc1035(`${templateId}-c`, MAX_MANAGED_CLUSTER_PREFIX_LENGTH, "cluster"),
149 ),
150 dagTimeout: news.dagTimeout,
151 parameters: news.parameters,
152 encryptionConfig: news.kmsKey ? { kmsKey: news.kmsKey } : undefined,
153});
154
155const getByName = (name: string) =>
156 dataproc
157 .getProjectsRegionsWorkflowTemplates({ name })
158 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
159
160const listRegion = (project: string, region: string) =>
161 emptyOnMissing(
162 collectPages(
163 dataproc.listProjectsRegionsWorkflowTemplates.pages({
164 parent: regionParent(project, region),
165 pageSize: 1000,
166 }),
167 (page) => page.templates,
168 ).pipe(
169 Effect.map((items) =>
170 items
171 .filter((template) => hasAlchemyLabelMap(template.labels))
172 .map((template) => toAttrs(template, project, region)),
173 ),
174 ),
175 );
176
177const templateChanged = (
178 current: dataproc.WorkflowTemplate,
179 desired: dataproc.WorkflowTemplate,
180) =>
181 fingerprint(current.jobs) !== fingerprint(desired.jobs) ||
182 fingerprint(current.placement) !== fingerprint(desired.placement) ||
183 (current.dagTimeout ?? "") !== (desired.dagTimeout ?? "") ||
184 fingerprint(current.parameters) !== fingerprint(desired.parameters) ||
185 (current.encryptionConfig?.kmsKey ?? "") !==
186 (desired.encryptionConfig?.kmsKey ?? "");
187
188export const RegionsWorkflowTemplateProvider = () =>
189 Provider.succeed(RegionsWorkflowTemplate, {
190 stables: ["name", "templateId", "project", "region", "createTime"],
191
192 diff: Effect.fn(function* ({ news, olds, output }) {
193 if (!isResolved(news)) return undefined;
194 const env = yield* GcpEnvironment.current;
195 const previousId = olds?.templateId ?? output?.templateId;
196 const nextId = news.templateId ?? previousId;
197 const previousRegion = normalizeLocation(
198 olds?.region ?? output?.region,
199 env.region,
200 );
201 const nextRegion = normalizeLocation(
202 news.region ?? olds?.region ?? output?.region,
203 env.region,
204 );
205 if (
206 (previousId !== undefined &&
207 nextId !== undefined &&
208 previousId !== nextId) ||
209 (output !== undefined && previousRegion !== nextRegion)
210 ) {
211 return {
212 action: "replace" as const,
213 deleteFirst:
214 previousRegion === nextRegion &&
215 previousId !== undefined &&
216 nextId === previousId,
217 };
218 }
219 return undefined;
220 }),
221
222 read: Effect.fn(function* ({ id, olds, output }) {
223 const env = yield* GcpEnvironment.current;
224 const region = normalizeLocation(
225 olds?.region ?? output?.region,
226 env.region,
227 );
228 const templateId = yield* toPhysicalId(
229 id,
230 olds?.templateId,
231 output?.templateId,
232 MAX_POLICY_ID_LENGTH,
233 "template",
234 );
235 const name =
236 output?.name ?? resourceName(env.project, region, templateId);
237 const existing = yield* getByName(name);
238 if (existing === undefined) return undefined;
239 const attrs = toAttrs(existing, env.project, region);
240 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
241 ? attrs
242 : Unowned(attrs);
243 }),
244
245 list: () =>
246 Effect.gen(function* () {
247 const env = yield* GcpEnvironment.current;
248 const pages = yield* Effect.forEach(
249 [...new Set([env.region, ...LIST_LOCATIONS])],
250 (region) => listRegion(env.project, region),
251 { concurrency: 4 },
252 );
253 return pages.flat();
254 }),
255
256 reconcile: Effect.fn(function* ({ id, news, output }) {
257 const env = yield* GcpEnvironment.current;
258 const region = normalizeLocation(
259 news.region ?? output?.region,
260 env.region,
261 );
262 const templateId = yield* toPhysicalId(
263 id,
264 news.templateId,
265 output?.templateId,
266 MAX_POLICY_ID_LENGTH,
267 "template",
268 );
269 const name = resourceName(env.project, region, templateId);
270 const desiredLabels = {
271 ...toLabels(news.labels),
272 ...(yield* createInternalLabels(id)),
273 };
274
275 let current = yield* getByName(name);
276
277 if (current === undefined) {
278 const created = yield* dataproc
279 .createProjectsRegionsWorkflowTemplates({
280 parent: regionParent(env.project, region),
281 body: desiredBody(news, templateId, name, desiredLabels),
282 })
283 .pipe(Effect.catchTag("Conflict", () => getByName(name)));
284 current = created ?? undefined;
285 }
286
287 if (current === undefined) {
288 return yield* new RegionsWorkflowTemplateNotResolved({ name });
289 }
290
291 const desired = desiredBody(
292 news,
293 templateId,
294 current.name ?? name,
295 desiredLabels,
296 current.version,
297 );
298 const observedLabels = tagRecord(current.labels);
299 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
300 const labelsChanged = upsert.length > 0 || removed.length > 0;
301
302 if (labelsChanged || templateChanged(current, desired)) {
303 current = yield* dataproc.updateProjectsRegionsWorkflowTemplates({
304 name: current.name ?? name,
305 body: desired,
306 });
307 }
308
309 return toAttrs(current, env.project, region);
310 }),
311
312 delete: Effect.fn(function* ({ output }) {
313 if (!output.name) return;
314 yield* dataproc
315 .deleteProjectsRegionsWorkflowTemplates({ name: output.name })
316 .pipe(Effect.catchTag("NotFound", () => Effect.void));
317 yield* waitUntilGone(getByName(output.name), output.name);
318 }),
319 });