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 listWorkflowInvocations,
15 normalizeLocation,
16 parseResourceName,
17 replaceOnIdentity,
18 retryTransient,
19 sameJson,
20 waitUntilGone,
21} from "./internal.ts";
22import type { InvocationConfig } from "./RepositoriesWorkflowConfig.ts";
23
24export type RepositoriesWorkflowInvocationProps = {
25 /**
26 * Parent repository. Full name
27 * `projects/{project}/locations/{location}/repositories/{repository}`
28 * or the repository id (combined with `location`). Immutable —
29 * changing it replaces the invocation.
30 */
31 repository: string;
32 /**
33 * Region used when `repository` is a bare id.
34 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
35 */
36 location?: string;
37 /**
38 * Compilation result to invoke
39 * (`.../repositories/{repository}/compilationResults/{result}`).
40 * Mutually exclusive with `workflowConfig`. Immutable.
41 */
42 compilationResult?: string;
43 /**
44 * Workflow config to invoke
45 * (`.../repositories/{repository}/workflowConfigs/{workflowConfig}`).
46 * Mutually exclusive with `compilationResult`. Immutable.
47 */
48 workflowConfig?: string;
49 /**
50 * Invocation options. Immutable — changing them replaces the
51 * invocation.
52 */
53 invocationConfig?: InvocationConfig;
54};
55
56export type RepositoriesWorkflowInvocation = Resource<
57 "GCP.Dataform.RepositoriesWorkflowInvocation",
58 RepositoriesWorkflowInvocationProps,
59 {
60 /** Full resource name `.../repositories/{repository}/workflowInvocations/{invocation}`. */
61 name: string;
62 /** Workflow invocation id (last path segment). */
63 workflowInvocationId: string;
64 /** Parent repository resource name. */
65 repository: string;
66 /** Project id. */
67 project: string;
68 /** Location id. */
69 location: string;
70 /** Compilation result used, if any. */
71 compilationResult: string | undefined;
72 /** Workflow config used, if any. */
73 workflowConfig: string | undefined;
74 /** Resolved compilation result. */
75 resolvedCompilationResult: string | undefined;
76 /** Current state (`RUNNING`, `SUCCEEDED`, `FAILED`, …). */
77 state: string | undefined;
78 /** Inclusive start of the invocation. */
79 startTime: string | undefined;
80 /** Exclusive end of the invocation. */
81 endTime: string | undefined;
82 },
83 never,
84 Providers
85>;
86
87/**
88 * A Dataform workflow invocation — a single run of a compilation
89 * result (directly or via a workflow config).
90 *
91 * Invocations have no update API and no labels. Identity is
92 * server-assigned; changing `repository`, `compilationResult`,
93 * `workflowConfig`, or `invocationConfig` replaces the invocation.
94 * `list` / nuke finds invocations under Alchemy-labeled repositories.
95 * A running invocation is cancelled before delete.
96 *
97 * ### Creating a Workflow Invocation
98 * **Example:** Invoke a compilation result
99 * ```typescript
100 * const run = yield* GCP.Dataform.RepositoriesWorkflowInvocation("Run", {
101 * repository: repo.name,
102 * compilationResult: compilation.name,
103 * });
104 * ```
105 *
106 * **Example:** Invoke a workflow config
107 * ```typescript
108 * const run = yield* GCP.Dataform.RepositoriesWorkflowInvocation("Run", {
109 * repository: repo.name,
110 * workflowConfig: workflow.name,
111 * });
112 * ```
113 *
114 * @resource
115 * @category Dataform
116 */
117export const RepositoriesWorkflowInvocation =
118 Resource<RepositoriesWorkflowInvocation>(
119 "GCP.Dataform.RepositoriesWorkflowInvocation",
120 );
121
122export class RepositoriesWorkflowInvocationNotResolved extends Data.TaggedError(
123 "GCP.Dataform.RepositoriesWorkflowInvocationNotResolved",
124)<{
125 name: string;
126}> {}
127
128const toAttrs = (
129 invocation: dataform.WorkflowInvocation,
130 project: string,
131 region: string,
132) => {
133 const name = invocation.name ?? "";
134 const parsed = parseResourceName(name, "workflowInvocations", region);
135 return {
136 name,
137 workflowInvocationId: parsed.id,
138 repository: parsed.parent,
139 project: parsed.project || project,
140 location: parsed.location,
141 compilationResult: invocation.compilationResult,
142 workflowConfig: invocation.workflowConfig,
143 resolvedCompilationResult: invocation.resolvedCompilationResult,
144 state: invocation.state,
145 startTime: invocation.invocationTiming?.startTime,
146 endTime: invocation.invocationTiming?.endTime,
147 };
148};
149
150const getByName = (name: string) =>
151 name.length === 0 || name.includes("//")
152 ? Effect.succeed(undefined)
153 : dataform
154 .getProjectsLocationsRepositoriesWorkflowInvocations({ name })
155 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
156
157const invocationBody = (
158 news: RepositoriesWorkflowInvocationProps,
159): dataform.WorkflowInvocation => ({
160 compilationResult: news.compilationResult,
161 workflowConfig: news.workflowConfig,
162 invocationConfig: news.invocationConfig
163 ? {
164 transitiveDependentsIncluded:
165 news.invocationConfig.transitiveDependentsIncluded,
166 serviceAccount: news.invocationConfig.serviceAccount,
167 includedTags: news.invocationConfig.includedTags,
168 fullyRefreshIncrementalTablesEnabled:
169 news.invocationConfig.fullyRefreshIncrementalTablesEnabled,
170 queryPriority: news.invocationConfig.queryPriority,
171 transitiveDependenciesIncluded:
172 news.invocationConfig.transitiveDependenciesIncluded,
173 includedTargets: news.invocationConfig.includedTargets,
174 }
175 : undefined,
176});
177
178export const RepositoriesWorkflowInvocationProvider = () =>
179 Provider.succeed(RepositoriesWorkflowInvocation, {
180 stables: [
181 "name",
182 "workflowInvocationId",
183 "repository",
184 "project",
185 "location",
186 ],
187
188 diff: Effect.fn(function* ({ news, olds, output }) {
189 if (!isResolved(news)) return undefined;
190 const env = yield* GcpEnvironment.current;
191 const location = normalizeLocation(
192 news.location ?? olds?.location ?? output?.location,
193 env.region,
194 );
195 const compilationChanged =
196 (olds?.compilationResult ?? output?.compilationResult) !== undefined &&
197 news.compilationResult !== undefined &&
198 news.compilationResult !==
199 (olds?.compilationResult ?? output?.compilationResult);
200 const workflowChanged =
201 (olds?.workflowConfig ?? output?.workflowConfig) !== undefined &&
202 news.workflowConfig !== undefined &&
203 news.workflowConfig !==
204 (olds?.workflowConfig ?? output?.workflowConfig);
205 const configChanged =
206 news.invocationConfig !== undefined &&
207 olds?.invocationConfig !== undefined &&
208 !sameJson(olds.invocationConfig, news.invocationConfig);
209 return replaceOnIdentity({
210 previousLocation: olds?.location ?? output?.location,
211 nextLocation: news.location ?? olds?.location ?? output?.location,
212 previousParent: olds?.repository ?? output?.repository,
213 nextParent: expandRepository(news.repository, env.project, location),
214 extra: compilationChanged || workflowChanged || configChanged,
215 });
216 }),
217
218 read: Effect.fn(function* ({ olds, output }) {
219 const env = yield* GcpEnvironment.current;
220 const existing = yield* getByName(output?.name ?? "");
221 if (existing === undefined) return undefined;
222 const attrs = toAttrs(existing, env.project, env.region);
223 const parent = yield* dataform
224 .getProjectsLocationsRepositories({ name: attrs.repository })
225 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
226 if (parent === undefined) return Unowned(attrs);
227 return hasAlchemyLabelMap(parent.labels) ? attrs : Unowned(attrs);
228 }),
229
230 list: () =>
231 Effect.gen(function* () {
232 const env = yield* GcpEnvironment.current;
233 const invocations = yield* forEachOwnedRepository(
234 env.project,
235 env.region,
236 (repo) => listWorkflowInvocations(repo.name ?? ""),
237 );
238 return invocations.map((item) =>
239 toAttrs(item, env.project, env.region),
240 );
241 }),
242
243 reconcile: Effect.fn(function* ({ news, output }) {
244 const env = yield* GcpEnvironment.current;
245 const location = normalizeLocation(
246 news.location ?? output?.location,
247 env.region,
248 );
249 const repository = expandRepository(
250 news.repository,
251 env.project,
252 location,
253 );
254
255 let current = yield* getByName(output?.name ?? "");
256
257 if (current === undefined) {
258 current = yield* retryTransient(
259 dataform.createProjectsLocationsRepositoriesWorkflowInvocations({
260 parent: repository,
261 body: invocationBody(news),
262 }),
263 );
264 }
265
266 if (current === undefined) {
267 return yield* new RepositoriesWorkflowInvocationNotResolved({
268 name: output?.name ?? `${repository}/workflowInvocations`,
269 });
270 }
271
272 return toAttrs(current, env.project, env.region);
273 }),
274
275 delete: Effect.fn(function* ({ output }) {
276 if (!output.name) return;
277 const current = yield* getByName(output.name);
278 if (
279 current?.state === "RUNNING" ||
280 current?.state === "CANCELING" ||
281 current?.state === "STATE_UNSPECIFIED"
282 ) {
283 yield* retryTransient(
284 dataform.cancelProjectsLocationsRepositoriesWorkflowInvocations({
285 name: output.name,
286 body: {},
287 }),
288 ).pipe(
289 Effect.catchTag("NotFound", () => Effect.void),
290 Effect.catchTag("BadRequest", () => Effect.void),
291 );
292 }
293 yield* retryTransient(
294 dataform.deleteProjectsLocationsRepositoriesWorkflowInvocations({
295 name: output.name,
296 }),
297 ).pipe(Effect.catchTag("NotFound", () => Effect.void));
298 yield* waitUntilGone(getByName(output.name));
299 }),
300 });