1import * as bigquery from "@distilled.cloud/gcp/bigquery_v2";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { GcpEnvironment } from "../Environment.ts";
11import {
12 ALCHEMY_LABEL_PREFIX,
13 alchemyLabelKeys,
14 createInternalLabels,
15 hasAlchemyLabels,
16} from "../Labels.ts";
17import type { Providers } from "../Providers.ts";
18
19const MAX_ROUTINE_ID_LENGTH = 256;
20const DEFAULT_LANGUAGE = "SQL";
21const DEFAULT_ARGUMENT_KIND = "FIXED_TYPE";
22
23export type RoutineType = bigquery.RoutineRoutineTypeEnum | (string & {});
24export type RoutineLanguage = bigquery.RoutineLanguageEnum | (string & {});
25export type RoutineArgument = bigquery.Argument;
26export type RoutineDataType = bigquery.StandardSqlDataType;
27export type RoutineReturnTableType = bigquery.StandardSqlTableType;
28export type RoutineSparkOptions = bigquery.SparkOptions;
29export type RoutineRemoteFunctionOptions = bigquery.RemoteFunctionOptions;
30export type RoutinePythonOptions = bigquery.PythonOptions;
31export type RoutineExternalRuntimeOptions = bigquery.ExternalRuntimeOptions;
32export type RoutineDeterminismLevel =
33 | bigquery.RoutineDeterminismLevelEnum
34 | (string & {});
35export type RoutineSecurityMode =
36 | bigquery.RoutineSecurityModeEnum
37 | (string & {});
38export type RoutineDataGovernanceType =
39 | bigquery.RoutineDataGovernanceTypeEnum
40 | (string & {});
41
42export type RoutineProps = {
43 /**
44 * Dataset id (the `{dataset}` segment of
45 * `projects/{project}/datasets/{dataset}`) or a full dataset resource
46 * name. Immutable — changing it replaces the routine.
47 */
48 datasetId: string;
49 /**
50 * Routine id (the `{routine}` segment of
51 * `projects/{project}/datasets/{dataset}/routines/{routine}`). If omitted,
52 * a unique name is generated from the stack, stage, and logical id. Must
53 * contain only letters, numbers, and underscores; max 256 characters.
54 * Immutable — changing it replaces the routine.
55 */
56 routineId?: string;
57 /**
58 * Routine kind (`SCALAR_FUNCTION`, `PROCEDURE`, `TABLE_VALUED_FUNCTION`,
59 * `AGGREGATE_FUNCTION`). Immutable — changing it replaces the routine.
60 */
61 routineType: RoutineType;
62 /**
63 * Body of the routine. For SQL functions this is the expression inside
64 * the `AS (…)` clause, not a full `CREATE FUNCTION` statement. For
65 * JavaScript, the evaluated string in the `AS` clause.
66 */
67 definitionBody: string;
68 /**
69 * Implementation language (`SQL`, `JAVASCRIPT`, `PYTHON`, `JAVA`,
70 * `SCALA`). Defaults to `SQL` unless `remoteFunctionOptions` is set.
71 */
72 language?: RoutineLanguage;
73 /**
74 * Input/output arguments. `mode` is valid on procedures only.
75 */
76 arguments?: RoutineArgument[];
77 /**
78 * Return type for scalar and aggregate functions. Optional for SQL
79 * (inferred at query time); required for other languages.
80 */
81 returnType?: RoutineDataType;
82 /**
83 * Return table schema when `routineType` is `TABLE_VALUED_FUNCTION`.
84 * Optional for SQL (inferred at query time).
85 */
86 returnTableType?: RoutineReturnTableType;
87 /**
88 * User-friendly description of this routine. Alchemy ownership is stored
89 * in a `[alchemy …]` prefix for `list` / nuke.
90 */
91 description?: string;
92 /**
93 * GCS paths of imported JavaScript libraries when `language` is
94 * `JAVASCRIPT`.
95 */
96 importedLibraries?: string[];
97 /**
98 * Determinism of a JavaScript UDF (`DETERMINISTIC`,
99 * `NOT_DETERMINISTIC`).
100 */
101 determinismLevel?: RoutineDeterminismLevel;
102 /**
103 * Security mode (`DEFINER`, `INVOKER`). If omitted, BigQuery infers it
104 * from the routine configuration.
105 */
106 securityMode?: RoutineSecurityMode;
107 /**
108 * Set to `DATA_MASKING` to register a custom masking function.
109 */
110 dataGovernanceType?: RoutineDataGovernanceType;
111 /**
112 * Extra validation for procedure bodies. Default `true`. Recursive
113 * procedures should set this to `false`.
114 * @default true
115 */
116 strictMode?: boolean;
117 /**
118 * Spark procedure options (`JAVA` / `SCALA` / PySpark).
119 */
120 sparkOptions?: RoutineSparkOptions;
121 /**
122 * Remote function endpoint and connection.
123 */
124 remoteFunctionOptions?: RoutineRemoteFunctionOptions;
125 /**
126 * Python UDF entry point and packages.
127 */
128 pythonOptions?: RoutinePythonOptions;
129 /**
130 * Container limits for Python UDFs.
131 */
132 externalRuntimeOptions?: RoutineExternalRuntimeOptions;
133};
134
135export type Routine = Resource<
136 "GCP.BigQuery.Routine",
137 RoutineProps,
138 {
139 /**
140 * Resource path
141 * `projects/{project}/datasets/{dataset}/routines/{routine}`.
142 */
143 name: string;
144 /** Opaque id `project:dataset.routine`. */
145 id: string;
146 /** Routine id (last path segment). */
147 routineId: string;
148 /** Dataset id. */
149 datasetId: string;
150 /** Project id. */
151 project: string;
152 /** Routine kind. */
153 routineType: string;
154 /** Implementation language, if set. */
155 language: string | undefined;
156 /** Routine body. */
157 definitionBody: string;
158 /** Arguments, if set. */
159 arguments: RoutineArgument[] | undefined;
160 /** Scalar/aggregate return type, if set. */
161 returnType: RoutineDataType | undefined;
162 /** TVF return table type, if set. */
163 returnTableType: RoutineReturnTableType | undefined;
164 /** Description, if set. */
165 description: string | undefined;
166 /** Imported JavaScript libraries. */
167 importedLibraries: string[] | undefined;
168 /** JavaScript determinism, if set. */
169 determinismLevel: string | undefined;
170 /** Security mode, if set. */
171 securityMode: string | undefined;
172 /** Data-governance classification, if set. */
173 dataGovernanceType: string | undefined;
174 /** Procedure strict-mode flag. */
175 strictMode: boolean | undefined;
176 /** Spark options, if set. */
177 sparkOptions: RoutineSparkOptions | undefined;
178 /** Remote function options, if set. */
179 remoteFunctionOptions: RoutineRemoteFunctionOptions | undefined;
180 /** Python UDF options, if set. */
181 pythonOptions: RoutinePythonOptions | undefined;
182 /** Python runtime options, if set. */
183 externalRuntimeOptions: RoutineExternalRuntimeOptions | undefined;
184 /** Creation time in milliseconds since epoch. */
185 creationTime: string | undefined;
186 /** Last-modified time in milliseconds since epoch. */
187 lastModifiedTime: string | undefined;
188 /** Server etag. */
189 etag: string | undefined;
190 },
191 never,
192 Providers
193>;
194
195/**
196 * A Google BigQuery routine — a user-defined function, table-valued
197 * function, aggregate function, or stored procedure.
198 *
199 * Routines have no labels; Alchemy stamps ownership into the description
200 * (`[alchemy alchemy-stack=… alchemy-stage=… alchemy-id=…]`) so `list` /
201 * `pnpm nuke:gcp` can find them inside datasets that carry `alchemy-*`
202 * labels. `datasetId`, `routineId`, and `routineType` are immutable
203 * (changing them replaces the routine). Body, arguments, description,
204 * language, and language-specific options update in place via a
205 * full-resource replace.
206 *
207 * ### Creating a Routine
208 * **Example:** SQL scalar function
209 * ```typescript
210 * const dataset = yield* GCP.BigQuery.Dataset("Analytics", {
211 * forceDestroy: true,
212 * });
213 * const add = yield* GCP.BigQuery.Routine("Add", {
214 * datasetId: dataset.datasetId,
215 * routineType: "SCALAR_FUNCTION",
216 * language: "SQL",
217 * arguments: [
218 * { name: "x", dataType: { typeKind: "INT64" } },
219 * { name: "y", dataType: { typeKind: "INT64" } },
220 * ],
221 * returnType: { typeKind: "INT64" },
222 * definitionBody: "x + y",
223 * });
224 * ```
225 *
226 * **Example:** SQL stored procedure
227 * ```typescript
228 * const proc = yield* GCP.BigQuery.Routine("Seed", {
229 * datasetId: dataset.datasetId,
230 * routineType: "PROCEDURE",
231 * language: "SQL",
232 * definitionBody: "SELECT 1",
233 * });
234 * ```
235 *
236 * **Example:** JavaScript UDF
237 * ```typescript
238 * const multiply = yield* GCP.BigQuery.Routine("Multiply", {
239 * datasetId: dataset.datasetId,
240 * routineType: "SCALAR_FUNCTION",
241 * language: "JAVASCRIPT",
242 * determinismLevel: "DETERMINISTIC",
243 * arguments: [
244 * { name: "x", dataType: { typeKind: "FLOAT64" } },
245 * { name: "y", dataType: { typeKind: "FLOAT64" } },
246 * ],
247 * returnType: { typeKind: "FLOAT64" },
248 * definitionBody: "return x * y;",
249 * });
250 * ```
251 *
252 * ### Calling a Routine
253 * **Example:** Invoke a SQL function with Query
254 * ```typescript
255 * const query = yield* GCP.BigQuery.Query(dataset);
256 * const result = yield* query({ query: "SELECT Add(1, 2) AS n" });
257 * ```
258 *
259 * @resource
260 * @category BigQuery
261 */
262export const Routine = Resource<Routine>("GCP.BigQuery.Routine");
263
264export class RoutineNotResolved extends Data.TaggedError(
265 "GCP.BigQuery.RoutineNotResolved",
266)<{
267 name: string;
268}> {}
269
270const lastSegment = (value: string) => {
271 const trimmed = value.replace(/\/+$/, "");
272 const parts = trimmed.split("/");
273 return parts[parts.length - 1] || trimmed;
274};
275
276const datasetIdOf = (datasetId: string) =>
277 datasetId.includes("/") ? lastSegment(datasetId) : datasetId;
278
279const resourceName = (project: string, datasetId: string, routineId: string) =>
280 `projects/${project}/datasets/${datasetId}/routines/${routineId}`;
281
282const toId = (id: string, routineId: string | undefined, existing?: string) =>
283 Effect.gen(function* () {
284 if (routineId !== undefined) return routineId;
285 if (existing !== undefined) return existing;
286 const generated = yield* createPhysicalName({
287 id,
288 maxLength: MAX_ROUTINE_ID_LENGTH,
289 lowercase: true,
290 delimiter: "_",
291 });
292 return generated.replaceAll("-", "_");
293 });
294
295const stable = (value: unknown): unknown => {
296 if (Array.isArray(value)) return value.map(stable);
297 if (value !== null && typeof value === "object") {
298 return Object.fromEntries(
299 Object.entries(value as Record<string, unknown>)
300 .filter(([, nested]) => nested !== undefined)
301 .sort(([left], [right]) => left.localeCompare(right))
302 .map(([key, nested]) => [key, stable(nested)]),
303 );
304 }
305 return value;
306};
307
308const jsonEqual = (left: unknown, right: unknown) =>
309 JSON.stringify(stable(left ?? null)) ===
310 JSON.stringify(stable(right ?? null));
311
312const canonType = (
313 type: RoutineDataType | undefined,
314): RoutineDataType | undefined => {
315 if (type === undefined) return undefined;
316 return {
317 typeKind: type.typeKind,
318 arrayElementType: canonType(type.arrayElementType),
319 rangeElementType: canonType(type.rangeElementType),
320 structType: type.structType
321 ? {
322 fields: (type.structType.fields ?? []).map((field) => ({
323 name: field.name,
324 type: canonType(field.type),
325 })),
326 }
327 : undefined,
328 };
329};
330
331const canonArguments = (args: RoutineArgument[] | undefined) =>
332 (args ?? []).map((argument) => ({
333 name: argument.name ?? "",
334 argumentKind: argument.argumentKind ?? DEFAULT_ARGUMENT_KIND,
335 mode: argument.mode,
336 isAggregate: argument.isAggregate,
337 dataType: canonType(argument.dataType),
338 }));
339
340const encodeDescription = (
341 labels: Record<string, string>,
342 description: string | undefined,
343): string => {
344 const marker = `[alchemy ${alchemyLabelKeys.stack}=${labels[alchemyLabelKeys.stack]} ${alchemyLabelKeys.stage}=${labels[alchemyLabelKeys.stage]} ${alchemyLabelKeys.id}=${labels[alchemyLabelKeys.id]}]`;
345 return description ? `${marker}\n${description}` : marker;
346};
347
348const parseDescription = (
349 description: string | undefined,
350): {
351 labels: Record<string, string>;
352 description: string | undefined;
353} => {
354 if (!description?.startsWith("[alchemy ")) {
355 return { labels: {}, description };
356 }
357 const end = description.indexOf("]");
358 if (end < 0) return { labels: {}, description };
359 const labels: Record<string, string> = {};
360 for (const part of description.slice("[alchemy ".length, end).split(/\s+/)) {
361 const eq = part.indexOf("=");
362 if (eq > 0) {
363 labels[part.slice(0, eq)] = part.slice(eq + 1);
364 }
365 }
366 const rest = description.slice(end + 1).replace(/^\n/, "");
367 return { labels, description: rest.length > 0 ? rest : undefined };
368};
369
370const hasOwnershipMarker = (description: string | undefined) =>
371 Object.keys(parseDescription(description).labels).some((key) =>
372 key.startsWith(ALCHEMY_LABEL_PREFIX),
373 );
374
375const toAttrs = (routine: bigquery.Routine, project: string) => {
376 const ref = routine.routineReference;
377 const routineId = ref?.routineId ?? "";
378 const datasetId = ref?.datasetId ?? "";
379 const projectId = ref?.projectId ?? project;
380 return {
381 name: resourceName(projectId, datasetId, routineId),
382 id: `${projectId}:${datasetId}.${routineId}`,
383 routineId,
384 datasetId,
385 project: projectId,
386 routineType: routine.routineType ?? "",
387 language: routine.language,
388 definitionBody: routine.definitionBody ?? "",
389 arguments: routine.arguments,
390 returnType: routine.returnType,
391 returnTableType: routine.returnTableType,
392 description: parseDescription(routine.description).description,
393 importedLibraries: routine.importedLibraries,
394 determinismLevel: routine.determinismLevel,
395 securityMode: routine.securityMode,
396 dataGovernanceType: routine.dataGovernanceType,
397 strictMode: routine.strictMode,
398 sparkOptions: routine.sparkOptions,
399 remoteFunctionOptions: routine.remoteFunctionOptions,
400 pythonOptions: routine.pythonOptions,
401 externalRuntimeOptions: routine.externalRuntimeOptions,
402 creationTime: routine.creationTime,
403 lastModifiedTime: routine.lastModifiedTime,
404 etag: routine.etag,
405 };
406};
407
408const getByRef = (projectId: string, datasetId: string, routineId: string) =>
409 bigquery
410 .getRoutines({ projectId, datasetId, routineId })
411 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
412
413const defaultLanguage = (news: RoutineProps, observed?: bigquery.Routine) => {
414 if (news.language !== undefined) return news.language;
415 if (observed?.language !== undefined) return observed.language;
416 if (
417 news.remoteFunctionOptions !== undefined ||
418 observed?.remoteFunctionOptions !== undefined
419 ) {
420 return undefined;
421 }
422 return DEFAULT_LANGUAGE;
423};
424
425const pick = <T>(news: T | undefined, observed: T | undefined) =>
426 news !== undefined ? news : observed;
427
428const toRoutineBody = (
429 projectId: string,
430 datasetId: string,
431 routineId: string,
432 news: RoutineProps,
433 encodedDescription: string,
434 observed?: bigquery.Routine,
435): bigquery.Routine => {
436 const body: bigquery.Routine = {
437 routineReference: { projectId, datasetId, routineId },
438 routineType: news.routineType,
439 definitionBody: news.definitionBody,
440 description: encodedDescription,
441 };
442 const language = defaultLanguage(news, observed);
443 if (language !== undefined) body.language = language;
444 const args = pick(news.arguments, observed?.arguments);
445 if (args !== undefined) body.arguments = args;
446 const returnType = pick(news.returnType, observed?.returnType);
447 if (returnType !== undefined) body.returnType = returnType;
448 const returnTableType = pick(news.returnTableType, observed?.returnTableType);
449 if (returnTableType !== undefined) body.returnTableType = returnTableType;
450 const importedLibraries = pick(
451 news.importedLibraries,
452 observed?.importedLibraries,
453 );
454 if (importedLibraries !== undefined) {
455 body.importedLibraries = importedLibraries;
456 }
457 const determinismLevel = pick(
458 news.determinismLevel,
459 observed?.determinismLevel,
460 );
461 if (determinismLevel !== undefined) body.determinismLevel = determinismLevel;
462 const securityMode = pick(news.securityMode, observed?.securityMode);
463 if (securityMode !== undefined) body.securityMode = securityMode;
464 const dataGovernanceType = pick(
465 news.dataGovernanceType,
466 observed?.dataGovernanceType,
467 );
468 if (dataGovernanceType !== undefined) {
469 body.dataGovernanceType = dataGovernanceType;
470 }
471 const strictMode = pick(news.strictMode, observed?.strictMode);
472 if (strictMode !== undefined) body.strictMode = strictMode;
473 const sparkOptions = pick(news.sparkOptions, observed?.sparkOptions);
474 if (sparkOptions !== undefined) body.sparkOptions = sparkOptions;
475 const remoteFunctionOptions = pick(
476 news.remoteFunctionOptions,
477 observed?.remoteFunctionOptions,
478 );
479 if (remoteFunctionOptions !== undefined) {
480 body.remoteFunctionOptions = remoteFunctionOptions;
481 }
482 const pythonOptions = pick(news.pythonOptions, observed?.pythonOptions);
483 if (pythonOptions !== undefined) body.pythonOptions = pythonOptions;
484 const externalRuntimeOptions = pick(
485 news.externalRuntimeOptions,
486 observed?.externalRuntimeOptions,
487 );
488 if (externalRuntimeOptions !== undefined) {
489 body.externalRuntimeOptions = externalRuntimeOptions;
490 }
491 return body;
492};
493
494const inputSnapshot = (routine: bigquery.Routine) => ({
495 routineType: (routine.routineType ?? "").toUpperCase(),
496 language: routine.language,
497 definitionBody: routine.definitionBody ?? "",
498 arguments: canonArguments(routine.arguments),
499 returnType: canonType(routine.returnType),
500 returnTableType: routine.returnTableType
501 ? {
502 columns: (routine.returnTableType.columns ?? []).map((column) => ({
503 name: column.name,
504 type: canonType(column.type),
505 })),
506 }
507 : undefined,
508 description: routine.description ?? "",
509 importedLibraries: routine.importedLibraries ?? [],
510 determinismLevel: routine.determinismLevel,
511 securityMode: routine.securityMode,
512 dataGovernanceType: routine.dataGovernanceType,
513 strictMode: routine.strictMode,
514 sparkOptions: routine.sparkOptions,
515 remoteFunctionOptions: routine.remoteFunctionOptions,
516 pythonOptions: routine.pythonOptions,
517 externalRuntimeOptions: routine.externalRuntimeOptions,
518});
519
520const hasAlchemyDatasetLabels = (
521 labels: Record<string, string | undefined> | null | undefined,
522) =>
523 Object.keys(labels ?? {}).some((key) => key.startsWith(ALCHEMY_LABEL_PREFIX));
524
525export const RoutineProvider = () =>
526 Provider.succeed(Routine, {
527 stables: [
528 "name",
529 "id",
530 "routineId",
531 "datasetId",
532 "project",
533 "routineType",
534 "creationTime",
535 ],
536
537 diff: Effect.fn(function* ({ news, olds, output }) {
538 if (!isResolved(news)) return undefined;
539 const previousRoutineId = olds?.routineId ?? output?.routineId;
540 const nextRoutineId = news.routineId ?? previousRoutineId;
541 const routineIdChanged =
542 news.routineId !== undefined &&
543 previousRoutineId !== undefined &&
544 news.routineId !== previousRoutineId;
545
546 const previousDataset = olds?.datasetId ?? output?.datasetId;
547 const nextDataset = datasetIdOf(news.datasetId);
548 const datasetChanged =
549 previousDataset !== undefined &&
550 datasetIdOf(previousDataset) !== nextDataset;
551
552 const previousType = (
553 olds?.routineType ??
554 output?.routineType ??
555 ""
556 ).toUpperCase();
557 const typeChanged =
558 previousType.length > 0 &&
559 news.routineType.toUpperCase() !== previousType;
560
561 if (!routineIdChanged && !datasetChanged && !typeChanged) {
562 return undefined;
563 }
564 return {
565 action: "replace" as const,
566 deleteFirst:
567 !routineIdChanged &&
568 !datasetChanged &&
569 nextRoutineId !== undefined &&
570 previousRoutineId !== undefined &&
571 nextRoutineId === previousRoutineId,
572 };
573 }),
574
575 read: Effect.fn(function* ({ id, olds, output }) {
576 const env = yield* GcpEnvironment.current;
577 const routineId = yield* toId(id, olds?.routineId, output?.routineId);
578 const datasetId = datasetIdOf(olds?.datasetId ?? output?.datasetId ?? "");
579 if (datasetId.length === 0) return undefined;
580 const existing = yield* getByRef(env.project, datasetId, routineId);
581 if (existing === undefined) return undefined;
582 const attrs = toAttrs(existing, env.project);
583 return (yield* hasAlchemyLabels(
584 id,
585 parseDescription(existing.description).labels,
586 ))
587 ? attrs
588 : Unowned(attrs);
589 }),
590
591 list: () =>
592 Effect.gen(function* () {
593 const env = yield* GcpEnvironment.current;
594 const datasets = yield* bigquery.listDatasets
595 .pages({
596 projectId: env.project,
597 maxResults: 1000,
598 all: true,
599 })
600 .pipe(
601 Stream.flatMap((page) => Stream.fromIterable(page.datasets ?? [])),
602 Stream.filter((dataset) => hasAlchemyDatasetLabels(dataset.labels)),
603 Stream.runCollect,
604 Effect.map((chunk) => Array.from(chunk)),
605 Effect.catchTag("NotFound", () =>
606 Effect.succeed([] as bigquery.DatasetListDatasetsItem[]),
607 ),
608 );
609 const pages = yield* Effect.forEach(
610 datasets,
611 (dataset) => {
612 const datasetId = dataset.datasetReference?.datasetId;
613 if (datasetId === undefined) {
614 return Effect.succeed([] as ReturnType<typeof toAttrs>[]);
615 }
616 return bigquery.listRoutines
617 .pages({
618 projectId: env.project,
619 datasetId,
620 maxResults: 1000,
621 readMask: "description,routineType,language",
622 })
623 .pipe(
624 Stream.flatMap((page) =>
625 Stream.fromIterable(page.routines ?? []),
626 ),
627 Stream.filter((routine) =>
628 hasOwnershipMarker(routine.description),
629 ),
630 Stream.map((routine) => toAttrs(routine, env.project)),
631 Stream.runCollect,
632 Effect.map((chunk) => Array.from(chunk)),
633 Effect.catchTag("NotFound", () =>
634 Effect.succeed([] as ReturnType<typeof toAttrs>[]),
635 ),
636 );
637 },
638 { concurrency: 4 },
639 );
640 return pages.flat();
641 }),
642
643 reconcile: Effect.fn(function* ({ id, news, output }) {
644 const env = yield* GcpEnvironment.current;
645 const routineId = yield* toId(id, news.routineId, output?.routineId);
646 const datasetId = datasetIdOf(news.datasetId);
647 const name = resourceName(env.project, datasetId, routineId);
648 const internalLabels = yield* createInternalLabels(id);
649
650 let current = yield* getByRef(env.project, datasetId, routineId);
651 const userDescription =
652 news.description !== undefined
653 ? news.description
654 : parseDescription(current?.description).description;
655 const encodedDescription = encodeDescription(
656 internalLabels,
657 userDescription,
658 );
659 const bodyOf = (observed?: bigquery.Routine) =>
660 toRoutineBody(
661 env.project,
662 datasetId,
663 routineId,
664 news,
665 encodedDescription,
666 observed,
667 );
668
669 if (current === undefined) {
670 const created = yield* bigquery
671 .insertRoutines({
672 projectId: env.project,
673 datasetId,
674 body: bodyOf(),
675 })
676 .pipe(
677 Effect.catchTag("Conflict", () =>
678 getByRef(env.project, datasetId, routineId),
679 ),
680 );
681 current = created ?? undefined;
682 }
683
684 if (current === undefined) {
685 return yield* new RoutineNotResolved({ name });
686 }
687
688 const desired = bodyOf(current);
689 const metadataChanged = !jsonEqual(
690 inputSnapshot(desired),
691 inputSnapshot(current),
692 );
693
694 if (metadataChanged) {
695 current = yield* bigquery
696 .updateRoutines({
697 projectId: env.project,
698 datasetId,
699 routineId,
700 body: desired,
701 })
702 .pipe(
703 Effect.catchTag("NotFound", () =>
704 bigquery
705 .insertRoutines({
706 projectId: env.project,
707 datasetId,
708 body: bodyOf(),
709 })
710 .pipe(
711 Effect.catchTag("Conflict", () =>
712 getByRef(env.project, datasetId, routineId),
713 ),
714 ),
715 ),
716 );
717 if (current === undefined) {
718 return yield* new RoutineNotResolved({ name });
719 }
720 }
721
722 return toAttrs(current, env.project);
723 }),
724
725 delete: Effect.fn(function* ({ output }) {
726 yield* bigquery
727 .deleteRoutines({
728 projectId: output.project,
729 datasetId: output.datasetId,
730 routineId: output.routineId,
731 })
732 .pipe(Effect.catchTag("NotFound", () => Effect.void));
733 }),
734 });