1import * as workflows from "@distilled.cloud/gcp/workflows_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 {
14 createInternalLabels,
15 diffLabels,
16 hasAlchemyLabels,
17 stripInternalLabels,
18 toLabels,
19} from "../Labels.ts";
20import { waitForOperation as waitForGcpOperation } from "../Operation.ts";
21import type { Providers } from "../Providers.ts";
22
23const MAX_NAME_LENGTH = 64;
24
25export type CallLogLevel = workflows.WorkflowCallLogLevelEnum | (string & {});
26export type ExecutionHistoryLevel =
27 | workflows.WorkflowExecutionHistoryLevelEnum
28 | (string & {});
29
30export type WorkflowProps = {
31 /**
32 * Workflow id (the `{workflow}` segment of
33 * `projects/{project}/locations/{location}/workflows/{workflow}`). If
34 * omitted, a unique name is generated from the stack, stage, and logical
35 * id. Must match `[A-Za-z][A-Za-z0-9_-]{0,62}[A-Za-z0-9]` (1–64
36 * characters, start with a letter, end with a letter or number).
37 * Immutable — changing it replaces the workflow.
38 */
39 workflowId?: string;
40 /**
41 * Workflows location (`us-central1`, `us-east1`, …). Immutable —
42 * changing it replaces the workflow. `US-CENTRAL1` is accepted and
43 * normalized to `us-central1`.
44 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
45 */
46 location?: string;
47 /**
48 * Workflow YAML/JSON source. Max 128KB. Updating this creates a new
49 * revision used by subsequent executions.
50 */
51 sourceContents: string;
52 /**
53 * Human-readable description (max 1000 Unicode characters).
54 */
55 description?: string;
56 /**
57 * User labels. Alchemy ownership labels are merged in automatically.
58 */
59 labels?: Record<string, string>;
60 /**
61 * Service account the workflow runs as. Format
62 * `projects/{project}/serviceAccounts/{account}` or an email. Defaults
63 * to the project's Compute Engine default service account. Updating this
64 * creates a new revision.
65 */
66 serviceAccount?: string;
67 /**
68 * Platform logging for `call` steps and responses.
69 * `LOG_ALL_CALLS`, `LOG_ERRORS_ONLY`, or `LOG_NONE`.
70 */
71 callLogLevel?: CallLogLevel;
72 /**
73 * User-defined environment variables for this revision. At most 20
74 * entries; each value up to 4KiB. Keys cannot be empty or start with
75 * `GOOGLE` or `WORKFLOWS`. Updating this creates a new revision.
76 */
77 userEnvVars?: Record<string, string>;
78 /**
79 * Execution history recorded for runs of this workflow.
80 * `EXECUTION_HISTORY_BASIC` or `EXECUTION_HISTORY_DETAILED`.
81 */
82 executionHistoryLevel?: ExecutionHistoryLevel;
83 /**
84 * Cloud KMS key used to encrypt workflow and execution data, as
85 * `projects/{project}/locations/{location}/keyRings/{keyRing}/cryptoKeys/{cryptoKey}`.
86 */
87 cryptoKeyName?: string;
88};
89
90export type Workflow = Resource<
91 "GCP.Workflows.Workflow",
92 WorkflowProps,
93 {
94 /** Full resource name `projects/{project}/locations/{location}/workflows/{workflow}`. */
95 name: string;
96 /** Workflow id (last path segment). */
97 workflowId: string;
98 /** Location id (`us-central1`, …). */
99 location: string;
100 /** Project id. */
101 project: string;
102 /** Workflow YAML/JSON source of the latest revision. */
103 sourceContents: string | undefined;
104 /** User-provided description. */
105 description: string | undefined;
106 /** User labels (Alchemy ownership labels stripped). */
107 labels: Record<string, string>;
108 /** Service account the latest revision runs as. */
109 serviceAccount: string | undefined;
110 /** Call logging level. */
111 callLogLevel: string | undefined;
112 /** User environment variables on the latest revision. */
113 userEnvVars: Record<string, string>;
114 /** Execution history level. */
115 executionHistoryLevel: string | undefined;
116 /** CMEK key, if any. */
117 cryptoKeyName: string | undefined;
118 /** CMEK key version currently in use. */
119 cryptoKeyVersion: string | undefined;
120 /** Latest revision id (`000001-a4d`). */
121 revisionId: string | undefined;
122 /** RFC3339 timestamp of the latest revision. */
123 revisionCreateTime: string | undefined;
124 /** Server-reported deployment state (`ACTIVE`, `UNAVAILABLE`). */
125 state: string | undefined;
126 /** Error details when `state` is `UNAVAILABLE`. */
127 stateError: workflows.StateError | undefined;
128 /** RFC3339 creation timestamp. */
129 createTime: string | undefined;
130 /** RFC3339 last-update timestamp. */
131 updateTime: string | undefined;
132 },
133 never,
134 Providers
135>;
136
137/**
138 * A Google Cloud Workflows workflow — YAML or JSON orchestration that
139 * Workflows executes as a revisioned program.
140 *
141 * Changing `workflowId` or `location` replaces the workflow. Updating
142 * source, service account, or environment variables creates a new
143 * revision; already-running executions keep the old revision.
144 *
145 * ### Creating a Workflow
146 * **Example:** Generated name
147 * ```typescript
148 * const greet = yield* GCP.Workflows.Workflow("Greet", {
149 * sourceContents: `main:
150 * steps:
151 * - done:
152 * return: hello
153 * `,
154 * });
155 * ```
156 *
157 * **Example:** Named workflow with labels and env vars
158 * ```typescript
159 * const greet = yield* GCP.Workflows.Workflow("Greet", {
160 * workflowId: "order-greet",
161 * location: "us-central1",
162 * description: "say hello",
163 * labels: { env: "prod" },
164 * callLogLevel: "LOG_ERRORS_ONLY",
165 * userEnvVars: { greeting: "hello" },
166 * sourceContents: `main:
167 * steps:
168 * - done:
169 * return: hello
170 * `,
171 * });
172 * ```
173 *
174 * ### Executing a Workflow
175 * **Example:** Start an execution
176 * ```typescript
177 * const createExecution = yield* GCP.Workflows.CreateExecution(greet);
178 * const execution = yield* createExecution({
179 * body: { argument: JSON.stringify({ name: "world" }) },
180 * });
181 * ```
182 *
183 * @resource
184 * @category Workflows
185 */
186export const Workflow = Resource<Workflow>("GCP.Workflows.Workflow");
187
188export class WorkflowNotResolved extends Data.TaggedError(
189 "GCP.Workflows.WorkflowNotResolved",
190)<{
191 name: string;
192}> {}
193
194/** The deployed workflow is in state `UNAVAILABLE` (see `stateError`). */
195export class WorkflowUnavailable extends Data.TaggedError(
196 "GCP.Workflows.WorkflowUnavailable",
197)<{
198 name: string;
199 message: string;
200}> {}
201
202export class WorkflowStillExists extends Data.TaggedError(
203 "GCP.Workflows.WorkflowStillExists",
204)<{
205 name: string;
206}> {}
207
208const lastSegment = (value: string) => {
209 const trimmed = value.replace(/\/+$/, "");
210 const parts = trimmed.split("/");
211 return parts[parts.length - 1] || trimmed;
212};
213
214const normalizeLocation = (location: string | undefined, fallback: string) =>
215 lastSegment(location ?? fallback).toLowerCase();
216
217const resourceName = (project: string, location: string, workflowId: string) =>
218 `projects/${project}/locations/${location}/workflows/${workflowId}`;
219
220const parentOf = (project: string, location: string) =>
221 `projects/${project}/locations/${location}`;
222
223const parseName = (name: string) => {
224 const parts = name.split("/").filter((part) => part.length > 0);
225 const workflowsAt = parts.lastIndexOf("workflows");
226 const locationsAt = parts.lastIndexOf("locations");
227 const projectsAt = parts.lastIndexOf("projects");
228 return {
229 project:
230 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
231 location:
232 locationsAt >= 0 && parts[locationsAt + 1] ? parts[locationsAt + 1]! : "",
233 workflowId:
234 workflowsAt >= 0 && parts[workflowsAt + 1]
235 ? parts[workflowsAt + 1]!
236 : lastSegment(name),
237 };
238};
239
240const userLabels = (
241 labels: Record<string, string | undefined> | null | undefined,
242): Record<string, string> => stripInternalLabels(tagRecord(labels));
243
244const userMap = (
245 map: Record<string, string | undefined> | null | undefined,
246): Record<string, string> => tagRecord(map);
247
248const toId = (id: string, workflowId: string | undefined, existing?: string) =>
249 Effect.gen(function* () {
250 return (
251 workflowId ??
252 existing ??
253 (yield* createPhysicalName({
254 id,
255 maxLength: MAX_NAME_LENGTH,
256 lowercase: true,
257 }))
258 );
259 });
260
261const compact = <T extends Record<string, unknown>>(value: T): T =>
262 Object.fromEntries(
263 Object.entries(value).filter(([, item]) => item !== undefined),
264 ) as T;
265
266const mapJson = (map: Record<string, string | undefined> | null | undefined) =>
267 JSON.stringify(
268 Object.fromEntries(
269 Object.entries(map ?? {})
270 .filter(([, value]) => value !== undefined)
271 .sort(([left], [right]) => left.localeCompare(right)),
272 ),
273 );
274
275const logLevelOf = (value: string | undefined) =>
276 !value || value === "CALL_LOG_LEVEL_UNSPECIFIED" ? undefined : value;
277
278const historyLevelOf = (value: string | undefined) =>
279 !value || value === "EXECUTION_HISTORY_LEVEL_UNSPECIFIED" ? undefined : value;
280
281const toAttrs = (workflow: workflows.Workflow, project: string) => {
282 const name = workflow.name ?? "";
283 const parsed = parseName(name);
284 return {
285 name,
286 workflowId: parsed.workflowId,
287 location: parsed.location,
288 project: parsed.project || project,
289 sourceContents: workflow.sourceContents,
290 description: workflow.description,
291 labels: userLabels(workflow.labels),
292 serviceAccount: workflow.serviceAccount,
293 callLogLevel: workflow.callLogLevel,
294 userEnvVars: userMap(workflow.userEnvVars),
295 executionHistoryLevel: workflow.executionHistoryLevel,
296 cryptoKeyName: workflow.cryptoKeyName,
297 cryptoKeyVersion: workflow.cryptoKeyVersion,
298 revisionId: workflow.revisionId,
299 revisionCreateTime: workflow.revisionCreateTime,
300 state: workflow.state,
301 stateError: workflow.stateError,
302 createTime: workflow.createTime,
303 updateTime: workflow.updateTime,
304 };
305};
306
307const getByName = (name: string) =>
308 workflows
309 .getProjectsLocationsWorkflows({ name })
310 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
311
312/**
313 * Wait for a workflow operation. ALREADY_EXISTS (code 6) counts as success
314 * (create race); with `notFoundOk`, so does NOT_FOUND (code 5, delete race)
315 * and an operation that is already gone.
316 */
317const waitForOperation = (
318 operation: workflows.Operation,
319 options?: { notFoundOk?: boolean },
320) =>
321 waitForGcpOperation(
322 operation,
323 (name) => workflows.getProjectsLocationsOperations({ name }),
324 { budget: "5 minutes", interval: "2 seconds" },
325 ).pipe(
326 Effect.catchIf(
327 (error) =>
328 (error._tag === "GCP.OperationFailed" &&
329 (error.code === 6 ||
330 (options?.notFoundOk === true && error.code === 5))) ||
331 (options?.notFoundOk === true && error._tag === "NotFound"),
332 () => Effect.succeed(operation),
333 ),
334 );
335
336const waitUntilReady = (name: string) =>
337 getByName(name).pipe(
338 Effect.filterOrFail(
339 (workflow): workflow is workflows.Workflow => workflow !== undefined,
340 () => new WorkflowNotResolved({ name }),
341 ),
342 Effect.filterOrFail(
343 (workflow) => workflow.state !== "UNAVAILABLE",
344 (workflow) =>
345 new WorkflowUnavailable({
346 name,
347 message: workflow.stateError?.details ?? "workflow is UNAVAILABLE",
348 }),
349 ),
350 Effect.filterOrFail(
351 (workflow) => workflow.state === "ACTIVE" || workflow.state === undefined,
352 () => new WorkflowNotResolved({ name }),
353 ),
354 Effect.retry({
355 while: (error) => error._tag === "GCP.Workflows.WorkflowNotResolved",
356 times: 8,
357 schedule: Schedule.spaced("1 second"),
358 }),
359 );
360
361const waitUntilGone = (name: string) =>
362 getByName(name).pipe(
363 Effect.flatMap((workflow) =>
364 workflow === undefined
365 ? Effect.void
366 : Effect.fail(new WorkflowStillExists({ name })),
367 ),
368 Effect.retry({
369 while: (error) => error._tag === "GCP.Workflows.WorkflowStillExists",
370 times: 10,
371 schedule: Schedule.spaced("1 second"),
372 }),
373 );
374
375export const WorkflowProvider = () =>
376 Provider.succeed(Workflow, {
377 stables: ["name", "workflowId", "location", "project", "createTime"],
378
379 diff: Effect.fn(function* ({ news, olds, output }) {
380 const env = yield* GcpEnvironment.current;
381 if (!isResolved(news)) return undefined;
382 const previousId = olds?.workflowId ?? output?.workflowId;
383 const nextId = news.workflowId ?? previousId;
384 const idChanged =
385 previousId !== undefined &&
386 nextId !== undefined &&
387 nextId !== previousId;
388
389 const previousLocation = normalizeLocation(
390 olds?.location ?? output?.location,
391 env.region,
392 );
393 const nextLocation = normalizeLocation(
394 news.location ?? output?.location,
395 env.region,
396 );
397 const locationChanged = previousLocation !== nextLocation;
398
399 if (idChanged || locationChanged) {
400 return { action: "replace" as const, deleteFirst: false };
401 }
402 return undefined;
403 }),
404
405 read: Effect.fn(function* ({ id, olds, output }) {
406 const env = yield* GcpEnvironment.current;
407 const workflowId = yield* toId(id, olds?.workflowId, output?.workflowId);
408 const location = normalizeLocation(
409 olds?.location ?? output?.location,
410 env.region,
411 );
412 const name =
413 output?.name ?? resourceName(env.project, location, workflowId);
414 const existing = yield* getByName(name);
415 if (existing === undefined) return undefined;
416 const attrs = toAttrs(existing, env.project);
417 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
418 ? attrs
419 : Unowned(attrs);
420 }),
421
422 list: () =>
423 Effect.gen(function* () {
424 const env = yield* GcpEnvironment.current;
425 return yield* workflows.listProjectsLocationsWorkflows
426 .pages({
427 parent: `projects/${env.project}/locations/-`,
428 pageSize: 1000,
429 })
430 .pipe(
431 Stream.flatMap((page) => Stream.fromIterable(page.workflows ?? [])),
432 Stream.filter((workflow) =>
433 Object.keys(workflow.labels ?? {}).some((key) =>
434 key.startsWith("alchemy-"),
435 ),
436 ),
437 Stream.map((workflow) => toAttrs(workflow, env.project)),
438 Stream.runCollect,
439 Effect.map((chunk) => Array.from(chunk)),
440 Effect.catchTag("NotFound", () => Effect.succeed([])),
441 );
442 }),
443
444 reconcile: Effect.fn(function* ({ id, news, output }) {
445 const env = yield* GcpEnvironment.current;
446 const workflowId = yield* toId(id, news.workflowId, output?.workflowId);
447 const location = normalizeLocation(
448 news.location ?? output?.location,
449 env.region,
450 );
451 const name = resourceName(env.project, location, workflowId);
452 const parent = parentOf(env.project, location);
453 const desiredLabels = {
454 ...toLabels(news.labels),
455 ...(yield* createInternalLabels(id)),
456 };
457 const desiredBody = compact({
458 sourceContents: news.sourceContents,
459 description: news.description,
460 labels: desiredLabels,
461 serviceAccount: news.serviceAccount,
462 callLogLevel: news.callLogLevel,
463 userEnvVars: news.userEnvVars,
464 executionHistoryLevel: news.executionHistoryLevel,
465 cryptoKeyName: news.cryptoKeyName,
466 }) as workflows.Workflow;
467
468 let current = yield* getByName(name);
469
470 if (current === undefined) {
471 const created = yield* workflows
472 .createProjectsLocationsWorkflows({
473 parent,
474 workflowId,
475 body: desiredBody,
476 })
477 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
478 if (created !== undefined) {
479 yield* waitForOperation(created);
480 }
481 current = yield* waitUntilReady(name);
482 }
483
484 if (current === undefined) {
485 return yield* new WorkflowNotResolved({ name });
486 }
487
488 const observedLabels = tagRecord(current.labels);
489 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
490 const labelsChanged = upsert.length > 0 || removed.length > 0;
491 const sourceChanged =
492 (current.sourceContents ?? "") !== news.sourceContents;
493 const descriptionChanged =
494 (current.description ?? "") !== (news.description ?? "");
495 const serviceAccountChanged =
496 news.serviceAccount !== undefined &&
497 lastSegment(current.serviceAccount ?? "") !==
498 lastSegment(news.serviceAccount);
499 const callLogLevelChanged =
500 news.callLogLevel !== undefined &&
501 logLevelOf(current.callLogLevel) !== logLevelOf(news.callLogLevel);
502 const envVarsChanged =
503 news.userEnvVars !== undefined &&
504 mapJson(current.userEnvVars) !== mapJson(news.userEnvVars);
505 const historyChanged =
506 news.executionHistoryLevel !== undefined &&
507 historyLevelOf(current.executionHistoryLevel) !==
508 historyLevelOf(news.executionHistoryLevel);
509 const cryptoKeyChanged =
510 news.cryptoKeyName !== undefined &&
511 (current.cryptoKeyName ?? "") !== news.cryptoKeyName;
512
513 const updateMask = [
514 labelsChanged ? "labels" : undefined,
515 sourceChanged ? "sourceContents" : undefined,
516 descriptionChanged ? "description" : undefined,
517 serviceAccountChanged ? "serviceAccount" : undefined,
518 callLogLevelChanged ? "callLogLevel" : undefined,
519 envVarsChanged ? "userEnvVars" : undefined,
520 historyChanged ? "executionHistoryLevel" : undefined,
521 cryptoKeyChanged ? "cryptoKeyName" : undefined,
522 ].filter((field): field is string => field !== undefined);
523
524 if (updateMask.length > 0) {
525 const operation = yield* workflows
526 .patchProjectsLocationsWorkflows({
527 name,
528 updateMask: updateMask.join(","),
529 body: desiredBody,
530 })
531 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
532 if (operation === undefined) {
533 const created = yield* workflows
534 .createProjectsLocationsWorkflows({
535 parent,
536 workflowId,
537 body: desiredBody,
538 })
539 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
540 if (created !== undefined) {
541 yield* waitForOperation(created);
542 }
543 } else {
544 yield* waitForOperation(operation);
545 }
546 current = yield* waitUntilReady(name);
547 }
548
549 if (current === undefined) {
550 return yield* new WorkflowNotResolved({ name });
551 }
552
553 return toAttrs(current, env.project);
554 }),
555
556 delete: Effect.fn(function* ({ output }) {
557 const operation = yield* workflows
558 .deleteProjectsLocationsWorkflows({ name: output.name })
559 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
560 if (operation !== undefined) {
561 yield* waitForOperation(operation, { notFoundOk: true });
562 }
563 yield* waitUntilGone(output.name);
564 }),
565 });