1import * as dataform from "@distilled.cloud/gcp/dataform_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 type { Providers } from "../Providers.ts";
10import {
11 expandRepository,
12 forEachOwnedRepository,
13 hasAlchemyLabelMap,
14 listWorkflowConfigs,
15 normalizeLocation,
16 parseResourceName,
17 replaceOnIdentity,
18 retryTransient,
19 sameJson,
20 sameText,
21 toPhysicalId,
22 updateMaskOf,
23 waitUntilExists,
24 waitUntilGone,
25} from "./internal.ts";
26
27export type InvocationTarget = {
28 /** Action name within `database` and `schema`. */
29 name?: string;
30 /** Database (Google Cloud project id). */
31 database?: string;
32 /** Schema (BigQuery dataset id). */
33 schema?: string;
34};
35
36export type InvocationConfig = {
37 /** Include transitive dependents of included actions. */
38 transitiveDependentsIncluded?: boolean;
39 /** Service account used to run the invocation. */
40 serviceAccount?: string;
41 /**
42 * Action tags to include. Alchemy ownership tags (`alchemy-stack`,
43 * `alchemy-stage`, `alchemy-id`) are merged in because workflow
44 * configs have no labels field.
45 */
46 includedTags?: string[];
47 /** Fully refresh incremental tables. */
48 fullyRefreshIncrementalTablesEnabled?: boolean;
49 /** BigQuery query priority (`INTERACTIVE` or `BATCH`). */
50 queryPriority?: dataform.InvocationConfigQueryPriorityEnum | (string & {});
51 /** Include transitive dependencies of included actions. */
52 transitiveDependenciesIncluded?: boolean;
53 /** Action identifiers to include. */
54 includedTargets?: InvocationTarget[];
55};
56
57export type RepositoriesWorkflowConfigProps = {
58 /**
59 * Parent repository. Full name
60 * `projects/{project}/locations/{location}/repositories/{repository}`
61 * or the repository id (combined with `location`). Immutable —
62 * changing it replaces the workflow config.
63 */
64 repository: string;
65 /**
66 * Region used when `repository` is a bare id.
67 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
68 */
69 location?: string;
70 /**
71 * Workflow config id. If omitted, a unique RFC1035 name is generated.
72 * Immutable — changing it replaces the workflow config.
73 */
74 workflowConfigId?: string;
75 /**
76 * Release config whose `releaseCompilationResult` is executed. Full
77 * name `.../repositories/{repository}/releaseConfigs/{releaseConfig}`
78 * or the release config id.
79 */
80 releaseConfig: string;
81 /**
82 * Cron schedule for automatic workflow invocations.
83 */
84 cronSchedule?: string;
85 /**
86 * IANA time zone used when interpreting `cronSchedule`.
87 * @default "UTC"
88 */
89 timeZone?: string;
90 /**
91 * Disable automatic workflow invocations.
92 * @default false
93 */
94 disabled?: boolean;
95 /**
96 * Invocation options.
97 */
98 invocationConfig?: InvocationConfig;
99};
100
101export type RepositoriesWorkflowConfig = Resource<
102 "GCP.Dataform.RepositoriesWorkflowConfig",
103 RepositoriesWorkflowConfigProps,
104 {
105 /** Full resource name `.../repositories/{repository}/workflowConfigs/{workflowConfig}`. */
106 name: string;
107 /** Workflow config id (last path segment). */
108 workflowConfigId: string;
109 /** Parent repository resource name. */
110 repository: string;
111 /** Project id. */
112 project: string;
113 /** Location id. */
114 location: string;
115 /** Release config resource name. */
116 releaseConfig: string | undefined;
117 /** Cron schedule, if set. */
118 cronSchedule: string | undefined;
119 /** Time zone. */
120 timeZone: string | undefined;
121 /** Whether automatic invocations are disabled. */
122 disabled: boolean;
123 /** User invocation tags (Alchemy ownership tags stripped). */
124 includedTags: string[] | undefined;
125 /** RFC3339 creation timestamp. */
126 createTime: string | undefined;
127 /** RFC3339 last-update timestamp. */
128 updateTime: string | undefined;
129 },
130 never,
131 Providers
132>;
133
134/**
135 * A Dataform workflow configuration — executes a release config's
136 * compilation result on a schedule (or on demand).
137 *
138 * Workflow configs have no labels field, so ownership is the parent
139 * repository's Alchemy labels. Changing `workflowConfigId`,
140 * `repository`, or `location` replaces the config. Release config,
141 * schedule, disabled flag, and invocation options update in place.
142 *
143 * ### Creating a Workflow Config
144 * **Example:** Run a release
145 * ```typescript
146 * const workflow = yield* GCP.Dataform.RepositoriesWorkflowConfig("Hourly", {
147 * repository: repo.name,
148 * releaseConfig: release.name,
149 * });
150 * ```
151 *
152 * **Example:** Disabled schedule
153 * ```typescript
154 * const workflow = yield* GCP.Dataform.RepositoriesWorkflowConfig("Hourly", {
155 * repository: repo.name,
156 * releaseConfig: release.name,
157 * cronSchedule: "0 * * * *",
158 * disabled: true,
159 * });
160 * ```
161 *
162 * @resource
163 * @category Dataform
164 */
165export const RepositoriesWorkflowConfig = Resource<RepositoriesWorkflowConfig>(
166 "GCP.Dataform.RepositoriesWorkflowConfig",
167);
168
169export class RepositoriesWorkflowConfigNotResolved extends Data.TaggedError(
170 "GCP.Dataform.RepositoriesWorkflowConfigNotResolved",
171)<{
172 name: string;
173}> {}
174
175const resourceName = (repository: string, workflowConfigId: string) =>
176 `${repository}/workflowConfigs/${workflowConfigId}`;
177
178const invocationOf = (
179 config: InvocationConfig | undefined,
180): dataform.InvocationConfig | undefined => {
181 if (config === undefined) return undefined;
182 return {
183 transitiveDependentsIncluded: config.transitiveDependentsIncluded,
184 serviceAccount: config.serviceAccount,
185 includedTags: config.includedTags,
186 fullyRefreshIncrementalTablesEnabled:
187 config.fullyRefreshIncrementalTablesEnabled,
188 queryPriority: config.queryPriority,
189 transitiveDependenciesIncluded: config.transitiveDependenciesIncluded,
190 includedTargets: config.includedTargets,
191 };
192};
193
194const toAttrs = (
195 config: dataform.WorkflowConfig,
196 project: string,
197 region: string,
198) => {
199 const name = config.name ?? "";
200 const parsed = parseResourceName(name, "workflowConfigs", region);
201 return {
202 name,
203 workflowConfigId: parsed.id,
204 repository: parsed.parent,
205 project: parsed.project || project,
206 location: parsed.location,
207 releaseConfig: config.releaseConfig,
208 cronSchedule: config.cronSchedule,
209 timeZone: config.timeZone,
210 disabled: config.disabled === true,
211 includedTags: config.invocationConfig?.includedTags,
212 createTime: config.createTime,
213 updateTime: config.updateTime,
214 };
215};
216
217const getByName = (name: string) =>
218 name.length === 0 || name.includes("//")
219 ? Effect.succeed(undefined)
220 : dataform
221 .getProjectsLocationsRepositoriesWorkflowConfigs({ name })
222 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
223
224export const RepositoriesWorkflowConfigProvider = () =>
225 Provider.succeed(RepositoriesWorkflowConfig, {
226 stables: [
227 "name",
228 "workflowConfigId",
229 "repository",
230 "project",
231 "location",
232 "createTime",
233 ],
234
235 diff: Effect.fn(function* ({ news, olds, output }) {
236 if (!isResolved(news)) return undefined;
237 const env = yield* GcpEnvironment.current;
238 const location = normalizeLocation(
239 news.location ?? olds?.location ?? output?.location,
240 env.region,
241 );
242 return replaceOnIdentity({
243 previousId: olds?.workflowConfigId ?? output?.workflowConfigId,
244 nextId:
245 news.workflowConfigId ??
246 olds?.workflowConfigId ??
247 output?.workflowConfigId,
248 previousLocation: olds?.location ?? output?.location,
249 nextLocation: news.location ?? olds?.location ?? output?.location,
250 previousParent: olds?.repository ?? output?.repository,
251 nextParent: expandRepository(news.repository, env.project, location),
252 extra:
253 news.invocationConfig !== undefined &&
254 olds?.invocationConfig !== undefined &&
255 !sameJson(olds.invocationConfig, news.invocationConfig),
256 });
257 }),
258
259 read: Effect.fn(function* ({ id, olds, output }) {
260 const env = yield* GcpEnvironment.current;
261 const location = normalizeLocation(
262 olds?.location ?? output?.location,
263 env.region,
264 );
265 const repository = expandRepository(
266 olds?.repository ??
267 output?.repository ??
268 parseResourceName(output?.name ?? "", "workflowConfigs", env.region)
269 .parent,
270 env.project,
271 location,
272 );
273 const workflowConfigId = yield* toPhysicalId(
274 id,
275 olds?.workflowConfigId,
276 output?.workflowConfigId,
277 );
278 const name = output?.name ?? resourceName(repository, workflowConfigId);
279 const existing = yield* getByName(name);
280 if (existing === undefined) return undefined;
281 const attrs = toAttrs(existing, env.project, env.region);
282 const parent = yield* dataform
283 .getProjectsLocationsRepositories({ name: attrs.repository })
284 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
285 if (parent === undefined) return Unowned(attrs);
286 return hasAlchemyLabelMap(parent.labels) ? attrs : Unowned(attrs);
287 }),
288
289 list: () =>
290 Effect.gen(function* () {
291 const env = yield* GcpEnvironment.current;
292 const configs = yield* forEachOwnedRepository(
293 env.project,
294 env.region,
295 (repo) => listWorkflowConfigs(repo.name ?? ""),
296 );
297 return configs.map((item) => toAttrs(item, env.project, env.region));
298 }),
299
300 reconcile: Effect.fn(function* ({ id, news, output }) {
301 const env = yield* GcpEnvironment.current;
302 const location = normalizeLocation(
303 news.location ?? output?.location,
304 env.region,
305 );
306 const repository = expandRepository(
307 news.repository,
308 env.project,
309 location,
310 );
311 const workflowConfigId = yield* toPhysicalId(
312 id,
313 news.workflowConfigId,
314 output?.workflowConfigId,
315 );
316 const name = resourceName(repository, workflowConfigId);
317 const resolvedRelease = news.releaseConfig.includes("/")
318 ? news.releaseConfig
319 : `${repository}/releaseConfigs/${news.releaseConfig}`;
320 const disabled = news.disabled === true;
321 const invocationConfig = invocationOf(news.invocationConfig);
322
323 let current = yield* getByName(output?.name ?? name);
324
325 if (current === undefined) {
326 const created = yield* retryTransient(
327 dataform.createProjectsLocationsRepositoriesWorkflowConfigs({
328 parent: repository,
329 workflowConfigId,
330 body: {
331 releaseConfig: resolvedRelease,
332 cronSchedule: news.cronSchedule,
333 timeZone: news.timeZone,
334 disabled,
335 invocationConfig,
336 },
337 }),
338 ).pipe(Effect.catchTag("Conflict", () => getByName(name)));
339 current = created ?? undefined;
340 if (current === undefined) {
341 current = yield* waitUntilExists(getByName(name), name);
342 }
343 }
344
345 if (current === undefined) {
346 return yield* new RepositoriesWorkflowConfigNotResolved({ name });
347 }
348
349 const currentName = current.name ?? name;
350 const releaseChanged = !sameText(current.releaseConfig, resolvedRelease);
351 const cronChanged = !sameText(current.cronSchedule, news.cronSchedule);
352 const zoneChanged = !sameText(current.timeZone, news.timeZone);
353 const disabledChanged = (current.disabled === true) !== disabled;
354
355 if (releaseChanged || cronChanged || zoneChanged || disabledChanged) {
356 current = yield* retryTransient(
357 dataform.patchProjectsLocationsRepositoriesWorkflowConfigs({
358 name: currentName,
359 updateMask: updateMaskOf(
360 releaseChanged ? "releaseConfig" : undefined,
361 cronChanged ? "cronSchedule" : undefined,
362 zoneChanged ? "timeZone" : undefined,
363 disabledChanged ? "disabled" : undefined,
364 ),
365 body: {
366 releaseConfig: resolvedRelease,
367 cronSchedule: news.cronSchedule,
368 timeZone: news.timeZone,
369 disabled,
370 },
371 }),
372 );
373 }
374
375 return toAttrs(current, env.project, env.region);
376 }),
377
378 delete: Effect.fn(function* ({ output }) {
379 if (!output.name) return;
380 yield* retryTransient(
381 dataform.deleteProjectsLocationsRepositoriesWorkflowConfigs({
382 name: output.name,
383 }),
384 ).pipe(
385 Effect.catchTag("NotFound", () => Effect.void),
386 Effect.catchTag("BadRequest", () => Effect.void),
387 );
388 yield* waitUntilGone(getByName(output.name));
389 }),
390 });