1import * as datapipelines from "@distilled.cloud/gcp/datapipelines_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 { GcpEnvironment } from "../Environment.ts";
9import { createInternalLabels } from "../Labels.ts";
10import type { Providers } from "../Providers.ts";
11import {
12 DEFAULT_STATE,
13 DEFAULT_TYPE,
14 findOwnedPipeline,
15 getPipeline,
16 ignoreMissing,
17 jsonEqual,
18 listOwnedPipelines,
19 locationParent,
20 normalizeLocation,
21 ownedByAlchemy,
22 ownershipSources,
23 parseName,
24 replaceOnIdentity,
25 resourceName,
26 sameText,
27 scheduleKey,
28 toDisplayName,
29 toPipelineId,
30 updateMaskOf,
31 userSources,
32} from "./internal.ts";
33
34export type PipelineType =
35 | datapipelines.GoogleCloudDatapipelinesV1PipelineTypeEnum
36 | (string & {});
37
38export type PipelineState =
39 | datapipelines.GoogleCloudDatapipelinesV1PipelineStateEnum
40 | (string & {});
41
42export type ScheduleSpec = {
43 /**
44 * Unix-cron schedule (for example `"0 0 * * *"`). When set, Data
45 * Pipelines creates an internal Cloud Scheduler job.
46 */
47 schedule?: string;
48 /**
49 * tz-database time zone used to interpret `schedule`. Empty uses UTC.
50 */
51 timeZone?: string;
52};
53
54export type Workload = datapipelines.GoogleCloudDatapipelinesV1Workload;
55
56export type PipelineProps = {
57 /**
58 * Pipeline id (the `{pipeline}` segment of
59 * `projects/{project}/locations/{location}/pipelines/{pipeline}`). If
60 * omitted, a unique RFC1035 name is generated from the stack, stage,
61 * and logical id. Immutable — changing it replaces the pipeline.
62 */
63 pipelineId?: string;
64 /**
65 * Location (`us-central1`, …). Immutable — changing it replaces the
66 * pipeline. `US-CENTRAL1` is accepted and normalized to `us-central1`.
67 * Data Pipelines is only available in App Engine regions.
68 * @default the stack's GCP region (`GCP.Region`, profile region, `us-central1`)
69 */
70 location?: string;
71 /**
72 * Display name. Letters, numbers, hyphens, and underscores only. If
73 * omitted, Alchemy uses the generated pipeline id.
74 */
75 displayName?: string;
76 /**
77 * Pipeline type (`PIPELINE_TYPE_BATCH` or `PIPELINE_TYPE_STREAMING`).
78 * Batch pipelines run on `scheduleInfo`; streaming pipelines launch a
79 * job at create time. Immutable — changing it replaces the pipeline.
80 * @default "PIPELINE_TYPE_BATCH"
81 */
82 type?: PipelineType;
83 /**
84 * Desired execution state (`STATE_ACTIVE`, `STATE_PAUSED`,
85 * `STATE_ARCHIVED`, …). Applied on create. Later archive requests call
86 * `pipelines.stop` (permanent). State cannot be patched.
87 * @default "STATE_ACTIVE"
88 */
89 state?: PipelineState;
90 /**
91 * Cron used to launch batch jobs. Omit to manage jobs with
92 * `pipelines.run` instead of Cloud Scheduler.
93 */
94 scheduleInfo?: ScheduleSpec;
95 /**
96 * Service account email for the internal Cloud Scheduler job. Defaults
97 * to the Compute Engine default service account.
98 */
99 schedulerServiceAccountEmail?: string;
100 /**
101 * Dataflow classic or Flex template used to launch jobs. Exactly one of
102 * `dataflowLaunchTemplateRequest` or `dataflowFlexTemplateRequest`.
103 */
104 workload?: Workload;
105 /**
106 * Source annotations (for example Dataplex). Immutable. Alchemy
107 * ownership (`alchemy-stack` / `alchemy-stage` / `alchemy-id`) is
108 * merged in automatically because pipelines have no labels field.
109 */
110 pipelineSources?: Record<string, string>;
111};
112
113export type Pipeline = Resource<
114 "GCP.DataPipelines.Pipeline",
115 PipelineProps,
116 {
117 /** Full resource name `projects/{project}/locations/{location}/pipelines/{pipeline}`. */
118 name: string;
119 /** Pipeline id (last path segment). */
120 pipelineId: string;
121 /** Location id. */
122 location: string;
123 /** Project id. */
124 project: string;
125 /** Parent `projects/{project}/locations/{location}`. */
126 parent: string;
127 /** Display name. */
128 displayName: string | undefined;
129 /** Pipeline type. */
130 type: string | undefined;
131 /** Execution state. */
132 state: string | undefined;
133 /** Cron schedule, if configured. */
134 scheduleInfo: ScheduleSpec | undefined;
135 /** Scheduler service account email, if set. */
136 schedulerServiceAccountEmail: string | undefined;
137 /** Dataflow template used to launch jobs. */
138 workload: Workload | undefined;
139 /** User source annotations (Alchemy ownership keys stripped). */
140 pipelineSources: Record<string, string>;
141 /** Number of launched jobs. */
142 jobCount: number | undefined;
143 /** RFC3339 creation timestamp. */
144 createTime: string | undefined;
145 /** RFC3339 last-update timestamp. */
146 lastUpdateTime: string | undefined;
147 },
148 never,
149 Providers
150>;
151
152/**
153 * A Data Pipelines pipeline that launches recurring Dataflow jobs.
154 *
155 * Pipelines have no labels field, so Alchemy stamps ownership into
156 * `pipelineSources` for `list` / nuke. Pipeline id, location, and type
157 * are identity — changing any of them replaces the pipeline. Display
158 * name, workload, schedule, and scheduler service account update in
159 * place. Archiving uses `pipelines.stop` and is permanent. Creating a
160 * pipeline requires the Data Pipelines API
161 * (`datapipelines.googleapis.com`).
162 *
163 * ### Creating a Pipeline
164 * **Example:** Generated name
165 * ```typescript
166 * const pipeline = yield* GCP.DataPipelines.Pipeline("Batch", {
167 * type: "PIPELINE_TYPE_BATCH",
168 * });
169 * ```
170 *
171 * **Example:** Classic Word Count template on a yearly cron
172 * ```typescript
173 * const pipeline = yield* GCP.DataPipelines.Pipeline("WordCount", {
174 * pipelineId: "word-count",
175 * type: "PIPELINE_TYPE_BATCH",
176 * displayName: "word-count",
177 * scheduleInfo: { schedule: "0 0 1 1 *", timeZone: "UTC" },
178 * workload: {
179 * dataflowLaunchTemplateRequest: {
180 * projectId: "my-project",
181 * location: "us-central1",
182 * gcsPath: "gs://dataflow-templates/latest/Word_Count",
183 * launchParameters: {
184 * jobName: "word-count",
185 * parameters: {
186 * inputFile: "gs://dataflow-samples/shakespeare/kinglear.txt",
187 * output: "gs://my-bucket/out",
188 * },
189 * environment: { tempLocation: "gs://my-bucket/tmp" },
190 * },
191 * },
192 * },
193 * });
194 * ```
195 *
196 * ### Updating a Pipeline
197 * Change props on the same logical id; the engine keeps the physical id.
198 *
199 * **Example:** Rename and change the schedule
200 * ```typescript
201 * const pipeline = yield* GCP.DataPipelines.Pipeline("WordCount", {
202 * type: "PIPELINE_TYPE_BATCH",
203 * displayName: "word-count-v2",
204 * scheduleInfo: { schedule: "0 0 1 2 *", timeZone: "UTC" },
205 * });
206 * ```
207 *
208 * @resource
209 * @category DataPipelines
210 */
211export const Pipeline = Resource<Pipeline>("GCP.DataPipelines.Pipeline");
212
213export class PipelineNotResolved extends Data.TaggedError(
214 "GCP.DataPipelines.PipelineNotResolved",
215)<{
216 name: string;
217}> {}
218
219const toAttrs = (
220 pipeline: datapipelines.GoogleCloudDatapipelinesV1Pipeline,
221 project: string,
222 region: string,
223) => {
224 const name = pipeline.name ?? "";
225 const parsed = parseName(name, region);
226 return {
227 name,
228 pipelineId: parsed.pipelineId,
229 location: parsed.location,
230 project: parsed.project || project,
231 parent: parsed.parent || locationParent(project, parsed.location),
232 displayName: pipeline.displayName,
233 type: pipeline.type,
234 state: pipeline.state,
235 scheduleInfo: pipeline.scheduleInfo
236 ? {
237 schedule: pipeline.scheduleInfo.schedule,
238 timeZone: pipeline.scheduleInfo.timeZone,
239 }
240 : undefined,
241 schedulerServiceAccountEmail: pipeline.schedulerServiceAccountEmail,
242 workload: pipeline.workload,
243 pipelineSources: userSources(pipeline.pipelineSources),
244 jobCount: pipeline.jobCount,
245 createTime: pipeline.createTime,
246 lastUpdateTime: pipeline.lastUpdateTime,
247 };
248};
249
250const fillWorkload = (
251 workload: Workload | undefined,
252 project: string,
253 location: string,
254): Workload | undefined => {
255 if (workload === undefined) return undefined;
256 const classic = workload.dataflowLaunchTemplateRequest;
257 const flex = workload.dataflowFlexTemplateRequest;
258 return {
259 dataflowLaunchTemplateRequest: classic
260 ? {
261 ...classic,
262 projectId: classic.projectId ?? project,
263 location: classic.location ?? location,
264 }
265 : undefined,
266 dataflowFlexTemplateRequest: flex
267 ? {
268 ...flex,
269 projectId: flex.projectId ?? project,
270 location: flex.location ?? location,
271 }
272 : undefined,
273 };
274};
275
276const desiredSchedule = (
277 info: ScheduleSpec | undefined,
278): datapipelines.GoogleCloudDatapipelinesV1ScheduleSpec | undefined => {
279 if (info === undefined) return undefined;
280 if ((info.schedule ?? "") === "" && (info.timeZone ?? "") === "") {
281 return undefined;
282 }
283 return {
284 schedule: info.schedule,
285 timeZone: info.timeZone,
286 };
287};
288
289export const PipelineProvider = () =>
290 Provider.succeed(Pipeline, {
291 stables: [
292 "name",
293 "pipelineId",
294 "location",
295 "project",
296 "parent",
297 "createTime",
298 ],
299
300 diff: Effect.fn(function* ({ news, olds, output }) {
301 if (!isResolved(news)) return undefined;
302 const env = yield* GcpEnvironment.current;
303 const previousLocation = normalizeLocation(
304 olds?.location ?? output?.location,
305 env.region,
306 );
307 const nextLocation = normalizeLocation(
308 news.location ?? olds?.location ?? output?.location,
309 env.region,
310 );
311 const previousType = olds?.type ?? output?.type;
312 const nextType = news.type ?? previousType;
313 return replaceOnIdentity({
314 previousId: olds?.pipelineId ?? output?.pipelineId,
315 nextId: news.pipelineId,
316 previousLocation,
317 nextLocation,
318 previousType,
319 nextType,
320 });
321 }),
322
323 read: Effect.fn(function* ({ id, olds, output }) {
324 const env = yield* GcpEnvironment.current;
325 const pipelineId = yield* toPipelineId(
326 id,
327 olds?.pipelineId,
328 output?.pipelineId,
329 );
330 const location = normalizeLocation(
331 olds?.location ?? output?.location,
332 env.region,
333 );
334 const name =
335 output?.name ?? resourceName(env.project, location, pipelineId);
336 const existing = yield* findOwnedPipeline(
337 id,
338 env.project,
339 env.region,
340 name,
341 );
342 if (existing === undefined) return undefined;
343 const attrs = toAttrs(existing, env.project, env.region);
344 return (yield* ownedByAlchemy(id, existing.pipelineSources))
345 ? attrs
346 : Unowned(attrs);
347 }),
348
349 list: () =>
350 Effect.gen(function* () {
351 const env = yield* GcpEnvironment.current;
352 const rows = yield* listOwnedPipelines(env.project, env.region);
353 return rows.map((row) => toAttrs(row, env.project, env.region));
354 }),
355
356 reconcile: Effect.fn(function* ({ id, news, output }) {
357 const env = yield* GcpEnvironment.current;
358 const pipelineId = yield* toPipelineId(
359 id,
360 news.pipelineId,
361 output?.pipelineId,
362 );
363 const location = normalizeLocation(
364 news.location ?? output?.location,
365 env.region,
366 );
367 const parent = locationParent(env.project, location);
368 const name = resourceName(env.project, location, pipelineId);
369 const displayName = toDisplayName(news.displayName, pipelineId);
370 const type = news.type ?? output?.type ?? DEFAULT_TYPE;
371 const state = news.state ?? output?.state ?? DEFAULT_STATE;
372 const ownership = yield* createInternalLabels(id);
373 const pipelineSources = ownershipSources(
374 ownership,
375 news.pipelineSources ?? output?.pipelineSources,
376 );
377 const scheduleInfo = desiredSchedule(news.scheduleInfo);
378 const workload = fillWorkload(news.workload, env.project, location);
379 const schedulerServiceAccountEmail = news.schedulerServiceAccountEmail;
380
381 let current = yield* findOwnedPipeline(
382 id,
383 env.project,
384 env.region,
385 output?.name ?? name,
386 );
387
388 if (current === undefined) {
389 const created = yield* datapipelines
390 .createProjectsLocationsPipelines({
391 parent,
392 body: {
393 name,
394 displayName,
395 type,
396 state,
397 scheduleInfo,
398 schedulerServiceAccountEmail,
399 workload,
400 pipelineSources,
401 },
402 })
403 .pipe(Effect.catchTag("Conflict", () => getPipeline(name)));
404 current = created ?? undefined;
405 }
406
407 if (current === undefined) {
408 return yield* new PipelineNotResolved({ name });
409 }
410
411 const currentName = current.name ?? name;
412 const displayChanged = !sameText(current.displayName, displayName);
413 const scheduleChanged =
414 news.scheduleInfo !== undefined &&
415 !jsonEqual(
416 scheduleKey(current.scheduleInfo),
417 scheduleKey(scheduleInfo),
418 );
419 const schedulerChanged =
420 news.schedulerServiceAccountEmail !== undefined &&
421 !sameText(
422 current.schedulerServiceAccountEmail,
423 schedulerServiceAccountEmail,
424 );
425 const workloadChanged =
426 news.workload !== undefined && !jsonEqual(current.workload, workload);
427 const updateMask = updateMaskOf(
428 displayChanged ? "displayName" : undefined,
429 scheduleChanged ? "scheduleInfo" : undefined,
430 schedulerChanged ? "schedulerServiceAccountEmail" : undefined,
431 workloadChanged ? "workload" : undefined,
432 );
433
434 if (updateMask.length > 0) {
435 current = yield* datapipelines.patchProjectsLocationsPipelines({
436 name: currentName,
437 updateMask,
438 body: {
439 displayName,
440 scheduleInfo,
441 schedulerServiceAccountEmail,
442 workload,
443 },
444 });
445 }
446
447 const wantsArchived = state === "STATE_ARCHIVED";
448 if (wantsArchived && current.state !== "STATE_ARCHIVED") {
449 current = yield* datapipelines
450 .stopProjectsLocationsPipelines({
451 name: currentName,
452 body: {},
453 })
454 .pipe(
455 Effect.catchTag("NotFound", () => getPipeline(currentName)),
456 Effect.flatMap((row) =>
457 row !== undefined
458 ? Effect.succeed(row)
459 : new PipelineNotResolved({ name: currentName }),
460 ),
461 );
462 }
463
464 return toAttrs(current, env.project, env.region);
465 }),
466
467 delete: Effect.fn(function* ({ output }) {
468 if (!output.name) return;
469 yield* ignoreMissing(
470 datapipelines.deleteProjectsLocationsPipelines({ name: output.name }),
471 );
472 }),
473 });