1import * as bqdt from "@distilled.cloud/gcp/bigquerydatatransfer_v1";
2import * as Data from "effect/Data";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
7import { Unowned } from "../../AdoptPolicy.ts";
8import { isResolved } from "../../Diff.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { GcpEnvironment } from "../Environment.ts";
12import {
13 alchemyLabelKeys,
14 createInternalLabels,
15 hasAlchemyLabels,
16} from "../Labels.ts";
17import type { Providers } from "../Providers.ts";
18
19const backoff = Schedule.min([
20 Schedule.exponential(Duration.millis(300), 1.5),
21 Schedule.spaced(Duration.seconds(2)),
22]);
23
24export type TransferParams = Record<string, unknown>;
25
26export type EmailPreferences = {
27 /**
28 * Send email to the config owner when a transfer run fails.
29 */
30 enableFailureEmail?: boolean;
31};
32
33export type ScheduleOptions = {
34 /**
35 * Disable automatic scheduling. Runs can still be started with
36 * StartManualTransferRuns. When true, `schedule` is ignored.
37 */
38 disableAutoScheduling?: boolean;
39 /**
40 * RFC3339 time to start scheduling runs. The first run is at or after
41 * this time according to `schedule`.
42 */
43 startTime?: string;
44 /**
45 * RFC3339 time to stop scheduling runs. A run is not scheduled at or
46 * after this time.
47 */
48 endTime?: string;
49};
50
51export type EventDrivenSchedule = {
52 /**
53 * Pub/Sub subscription that triggers runs. Only `google_cloud_storage`
54 * supports this. Format: `projects/{project}/subscriptions/{subscription}`.
55 */
56 pubsubSubscription?: string;
57};
58
59export type TimeBasedSchedule = {
60 /** RFC3339 time to start scheduling runs. */
61 startTime?: string;
62 /** RFC3339 time to stop scheduling runs. */
63 endTime?: string;
64 /**
65 * Recurrence in UTC (e.g. `"every 24 hours"`, `"first sunday of quarter
66 * 00:00"`). Empty uses the data source default.
67 */
68 schedule?: string;
69};
70
71export type ScheduleOptionsV2 = {
72 /**
73 * Event-driven schedule. Cannot be combined with `timeBasedSchedule` or
74 * `manualSchedule`.
75 */
76 eventDrivenSchedule?: EventDrivenSchedule;
77 /**
78 * Time-based schedule. Cannot be combined with `eventDrivenSchedule` or
79 * `manualSchedule`. Replaces top-level `schedule` / `scheduleOptions`.
80 */
81 timeBasedSchedule?: TimeBasedSchedule;
82 /**
83 * Manual schedule (equivalent to `disableAutoScheduling: true`). Empty
84 * object selects this mode.
85 */
86 manualSchedule?: Record<string, never>;
87};
88
89export type EncryptionConfiguration = {
90 /**
91 * Cloud KMS key used to encrypt transferred BigQuery data, as
92 * `projects/{project}/locations/{location}/keyRings/{keyRing}/cryptoKeys/{cryptoKey}`.
93 */
94 kmsKeyName?: string;
95};
96
97export type DataplexConfiguration = {
98 /**
99 * Dataplex Universal Catalog entry group, as
100 * `projects/{project}/locations/{region}/entryGroups/{entryGroup}`.
101 */
102 entryGroup?: string;
103};
104
105export type MetadataDestination = {
106 /** Dataplex Universal Catalog destination. */
107 dataplexConfiguration?: DataplexConfiguration;
108};
109
110export type ManagedTableType =
111 | "MANAGED_TABLE_TYPE_UNSPECIFIED"
112 | "NATIVE"
113 | "BIGLAKE"
114 | (string & {});
115
116export type TransferConfigProps = {
117 /**
118 * Data Transfer location (`us-central1`, `us-east1`, `us`, …). Must
119 * match the destination dataset location. Immutable — changing it
120 * replaces the config. `US-CENTRAL1` is accepted and normalized to
121 * `us-central1`.
122 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
123 */
124 location?: string;
125 /**
126 * Data source id (e.g. `"scheduled_query"`, `"google_cloud_storage"`).
127 * Immutable — changing it replaces the config. The full list is
128 * available from `projects.locations.dataSources.list`.
129 */
130 dataSourceId: string;
131 /**
132 * User-specified display name. Transfer configs have no labels field,
133 * so Alchemy ownership (`alchemy-stack` / `alchemy-stage` /
134 * `alchemy-id`) is stored in a `[alchemy …]` prefix for `read` /
135 * `list` / nuke.
136 */
137 displayName?: string;
138 /**
139 * Destination BigQuery dataset id. Required for most data sources
140 * (including scheduled queries and Cloud Storage). Must live in the
141 * same location as this config.
142 */
143 destinationDatasetId?: string;
144 /**
145 * Parameters specific to `dataSourceId`. Values are typically strings.
146 * Scheduled query: `query`, `destination_table_name_template`,
147 * `write_disposition`. Cloud Storage: `data_path_template`,
148 * `destination_table_name_template`, `file_format`.
149 */
150 params: TransferParams;
151 /**
152 * Recurrence in UTC (e.g. `"every 24 hours"`). Empty uses the data
153 * source default. Ignored when `scheduleOptions.disableAutoScheduling`
154 * is true. Do not set together with `scheduleOptionsV2`.
155 */
156 schedule?: string;
157 /**
158 * Time-window and auto-schedule options. Do not set together with
159 * `scheduleOptionsV2`.
160 */
161 scheduleOptions?: ScheduleOptions;
162 /**
163 * V2 schedule (time-based, event-driven, or manual). Replaces
164 * `schedule` and `scheduleOptions` — the two families cannot be
165 * combined.
166 */
167 scheduleOptionsV2?: ScheduleOptionsV2;
168 /**
169 * Days of data to refresh on each run. `0` uses the data source
170 * default. Only valid when the data source supports it.
171 */
172 dataRefreshWindowDays?: number;
173 /**
174 * When true, the config exists but no runs are scheduled.
175 * @default false
176 */
177 disabled?: boolean;
178 /**
179 * Pub/Sub topic notified when a run finishes, as
180 * `projects/{project}/topics/{topic}`.
181 */
182 notificationPubsubTopic?: string;
183 /**
184 * Email notification preferences for the config owner.
185 */
186 emailPreferences?: EmailPreferences;
187 /**
188 * Optional CMEK for transferred BigQuery data.
189 */
190 encryptionConfiguration?: EncryptionConfiguration;
191 /**
192 * Classification of the destination table (`NATIVE`, `BIGLAKE`).
193 */
194 managedTableType?: ManagedTableType;
195 /**
196 * Metadata destination (Dataplex) when the transfer writes catalog
197 * metadata rather than a dataset.
198 */
199 metadataDestination?: MetadataDestination;
200 /**
201 * Service account the transfer runs as. The caller needs
202 * `iam.serviceAccounts.actAs`. Not all data sources support this;
203 * omitted uses the caller's credentials. Not returned by get — passed
204 * on create and whenever it changes.
205 */
206 serviceAccountName?: string;
207 /**
208 * OAuth version info for data sources that need user credentials
209 * (e.g. `youtube_channel`). Do not set with `serviceAccountName`.
210 */
211 versionInfo?: string;
212 /**
213 * Deprecated OAuth authorization code. Prefer `versionInfo`.
214 */
215 authorizationCode?: string;
216};
217
218export type TransferConfig = Resource<
219 "GCP.BigQueryDataTransfer.TransferConfig",
220 TransferConfigProps,
221 {
222 /** Full resource name `projects/{project}/locations/{location}/transferConfigs/{id}`. */
223 name: string;
224 /** Server-assigned config id (last path segment, usually a UUID). */
225 transferConfigId: string;
226 /** Project id. */
227 project: string;
228 /** Location id (`us-central1`, …). */
229 location: string;
230 /** User display name with the Alchemy ownership prefix stripped. */
231 displayName: string | undefined;
232 /** Data source id. */
233 dataSourceId: string | undefined;
234 /** Destination dataset id. */
235 destinationDatasetId: string | undefined;
236 /** Data-source parameters currently stored on the config. */
237 params: TransferParams;
238 /** Recurrence string. */
239 schedule: string | undefined;
240 /** Time-window / auto-schedule options. */
241 scheduleOptions: ScheduleOptions | undefined;
242 /** V2 schedule options. */
243 scheduleOptionsV2: ScheduleOptionsV2 | undefined;
244 /** Refresh window in days. */
245 dataRefreshWindowDays: number | undefined;
246 /** Whether the config is disabled. */
247 disabled: boolean;
248 /** Run-completion Pub/Sub topic. */
249 notificationPubsubTopic: string | undefined;
250 /** Email notification preferences. */
251 emailPreferences: EmailPreferences | undefined;
252 /** CMEK configuration. */
253 encryptionConfiguration: EncryptionConfiguration | undefined;
254 /** Destination table classification. */
255 managedTableType: string | undefined;
256 /** Metadata destination. */
257 metadataDestination: MetadataDestination | undefined;
258 /** Region of the destination dataset. */
259 datasetRegion: string | undefined;
260 /** Next scheduled run (RFC3339). */
261 nextRunTime: string | undefined;
262 /** State of the most recently updated run. */
263 state: string | undefined;
264 /** Last modification time (RFC3339). */
265 updateTime: string | undefined;
266 /** Deprecated user id. */
267 userId: string | undefined;
268 /** Owner whose credentials the transfer uses. */
269 ownerInfo: bqdt.UserInfo | undefined;
270 /** Latest config-level error, if any. */
271 error: bqdt.Status | undefined;
272 },
273 never,
274 Providers
275>;
276
277/**
278 * A BigQuery Data Transfer Service transfer configuration.
279 *
280 * Transfer configs have no labels field — Alchemy stamps ownership into
281 * the display name (`[alchemy alchemy-stack=… alchemy-stage=…
282 * alchemy-id=…]`) so `read`, `list`, and `pnpm nuke:gcp` can find them.
283 * The physical id is a server-assigned UUID. `dataSourceId` and
284 * `location` are immutable; changing either replaces the config.
285 *
286 * ### Creating a Transfer Config
287 * **Example:** Scheduled query with auto-scheduling disabled
288 * ```typescript
289 * const dataset = yield* GCP.BigQuery.Dataset("Analytics", {
290 * location: "US-CENTRAL1",
291 * forceDestroy: true,
292 * });
293 * const nightly = yield* GCP.BigQueryDataTransfer.TransferConfig("Nightly", {
294 * dataSourceId: "scheduled_query",
295 * destinationDatasetId: dataset.datasetId,
296 * displayName: "nightly",
297 * schedule: "every 24 hours",
298 * scheduleOptions: { disableAutoScheduling: true },
299 * params: {
300 * query: "SELECT 1 AS n",
301 * destination_table_name_template: "nightly",
302 * write_disposition: "WRITE_TRUNCATE",
303 * },
304 * });
305 * ```
306 *
307 * **Example:** Cloud Storage load
308 * ```typescript
309 * const load = yield* GCP.BigQueryDataTransfer.TransferConfig("GcsLoad", {
310 * dataSourceId: "google_cloud_storage",
311 * destinationDatasetId: dataset.datasetId,
312 * schedule: "every 24 hours",
313 * scheduleOptions: { disableAutoScheduling: true },
314 * params: {
315 * data_path_template: "gs://my-bucket/*.csv",
316 * destination_table_name_template: "events",
317 * file_format: "CSV",
318 * skip_leading_rows: "1",
319 * max_bad_records: "0",
320 * },
321 * });
322 * ```
323 *
324 * @resource
325 * @category BigQueryDataTransfer
326 */
327export const TransferConfig = Resource<TransferConfig>(
328 "GCP.BigQueryDataTransfer.TransferConfig",
329);
330
331export class TransferConfigNotResolved extends Data.TaggedError(
332 "GCP.BigQueryDataTransfer.TransferConfigNotResolved",
333)<{
334 name: string;
335}> {}
336
337const lastSegment = (value: string) => {
338 const trimmed = value.replace(/\/+$/, "");
339 const parts = trimmed.split("/");
340 return parts[parts.length - 1] || trimmed;
341};
342
343const normalizeLocation = (location: string) =>
344 lastSegment(location).toLowerCase();
345
346const parentOf = (project: string, location: string) =>
347 `projects/${project}/locations/${location}`;
348
349const parseName = (name: string) => {
350 const parts = name.split("/").filter((part) => part.length > 0);
351 const configsAt = parts.lastIndexOf("transferConfigs");
352 const locationsAt = parts.lastIndexOf("locations");
353 const projectsAt = parts.lastIndexOf("projects");
354 return {
355 project:
356 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
357 location:
358 locationsAt >= 0 && parts[locationsAt + 1] ? parts[locationsAt + 1]! : "",
359 transferConfigId:
360 configsAt >= 0 && parts[configsAt + 1]
361 ? parts[configsAt + 1]!
362 : lastSegment(name),
363 };
364};
365
366const encodeDisplayName = (
367 labels: Record<string, string>,
368 displayName: string | undefined,
369): string => {
370 const marker = `[alchemy ${alchemyLabelKeys.stack}=${labels[alchemyLabelKeys.stack]} ${alchemyLabelKeys.stage}=${labels[alchemyLabelKeys.stage]} ${alchemyLabelKeys.id}=${labels[alchemyLabelKeys.id]}]`;
371 const trimmed = displayName?.replace(/[\r\n]+/g, " ").trim();
372 const combined =
373 trimmed && trimmed.length > 0 ? `${marker} ${trimmed}` : marker;
374 return combined.slice(0, 1024);
375};
376
377const parseDisplayName = (
378 displayName: string | undefined,
379): {
380 labels: Record<string, string>;
381 displayName: string | undefined;
382} => {
383 if (!displayName?.startsWith("[alchemy ")) {
384 return { labels: {}, displayName };
385 }
386 const end = displayName.indexOf("]");
387 if (end < 0) return { labels: {}, displayName };
388 const labels: Record<string, string> = {};
389 for (const part of displayName.slice("[alchemy ".length, end).split(/\s+/)) {
390 const eq = part.indexOf("=");
391 if (eq > 0) {
392 labels[part.slice(0, eq)] = part.slice(eq + 1);
393 }
394 }
395 const rest = displayName.slice(end + 1).replace(/^[\s\n]+/, "");
396 return { labels, displayName: rest.length > 0 ? rest : undefined };
397};
398
399const hasOwnershipMarker = (displayName: string | undefined): boolean =>
400 Object.keys(parseDisplayName(displayName).labels).some((key) =>
401 key.startsWith("alchemy-"),
402 );
403
404const compact = <T extends Record<string, unknown>>(value: T): T =>
405 Object.fromEntries(
406 Object.entries(value).filter(([, item]) => item !== undefined),
407 ) as T;
408
409const jsonEqual = (left: unknown, right: unknown) =>
410 JSON.stringify(left ?? null) === JSON.stringify(right ?? null);
411
412const canonParams = (params: TransferParams | undefined): TransferParams => {
413 const out: TransferParams = {};
414 for (const key of Object.keys(params ?? {}).sort()) {
415 const value = params![key];
416 if (value === undefined) continue;
417 out[key] = value;
418 }
419 return out;
420};
421
422const toScheduleOptions = (
423 options: bqdt.ScheduleOptions | undefined,
424): ScheduleOptions | undefined => {
425 if (options === undefined) return undefined;
426 if (
427 options.disableAutoScheduling === undefined &&
428 options.startTime === undefined &&
429 options.endTime === undefined
430 ) {
431 return undefined;
432 }
433 return compact({
434 disableAutoScheduling: options.disableAutoScheduling,
435 startTime: options.startTime,
436 endTime: options.endTime,
437 });
438};
439
440const toScheduleOptionsV2 = (
441 options: bqdt.ScheduleOptionsV2 | undefined,
442): ScheduleOptionsV2 | undefined => {
443 if (options === undefined) return undefined;
444 const timeBased =
445 options.timeBasedSchedule === undefined
446 ? undefined
447 : compact({
448 startTime: options.timeBasedSchedule.startTime,
449 endTime: options.timeBasedSchedule.endTime,
450 schedule: options.timeBasedSchedule.schedule,
451 });
452 const eventDriven =
453 options.eventDrivenSchedule === undefined
454 ? undefined
455 : compact({
456 pubsubSubscription: options.eventDrivenSchedule.pubsubSubscription,
457 });
458 const manual =
459 options.manualSchedule === undefined
460 ? undefined
461 : ({} as Record<string, never>);
462 if (
463 timeBased === undefined &&
464 eventDriven === undefined &&
465 manual === undefined
466 ) {
467 return undefined;
468 }
469 return compact({
470 timeBasedSchedule:
471 timeBased && Object.keys(timeBased).length > 0 ? timeBased : undefined,
472 eventDrivenSchedule:
473 eventDriven && Object.keys(eventDriven).length > 0
474 ? eventDriven
475 : undefined,
476 manualSchedule: manual,
477 });
478};
479
480const toEmailPreferences = (
481 preferences: bqdt.EmailPreferences | undefined,
482): EmailPreferences | undefined => {
483 if (preferences === undefined) return undefined;
484 if (preferences.enableFailureEmail === undefined) return undefined;
485 return { enableFailureEmail: preferences.enableFailureEmail };
486};
487
488const toEncryption = (
489 encryption: bqdt.EncryptionConfiguration | undefined,
490): EncryptionConfiguration | undefined => {
491 if (encryption?.kmsKeyName === undefined) return undefined;
492 return { kmsKeyName: encryption.kmsKeyName };
493};
494
495const toMetadataDestination = (
496 destination: bqdt.MetadataDestination | undefined,
497): MetadataDestination | undefined => {
498 const entryGroup = destination?.dataplexConfiguration?.entryGroup;
499 if (entryGroup === undefined) return undefined;
500 return { dataplexConfiguration: { entryGroup } };
501};
502
503const toAttrs = (
504 config: bqdt.TransferConfig,
505 project: string,
506 location: string,
507) => {
508 const name = config.name ?? "";
509 const parsed = parseName(name);
510 const { displayName } = parseDisplayName(config.displayName);
511 return {
512 name,
513 transferConfigId: parsed.transferConfigId,
514 project: parsed.project || project,
515 location: parsed.location || location,
516 displayName,
517 dataSourceId: config.dataSourceId,
518 destinationDatasetId: config.destinationDatasetId,
519 params: config.params ?? {},
520 schedule: config.schedule,
521 scheduleOptions: toScheduleOptions(config.scheduleOptions),
522 scheduleOptionsV2: toScheduleOptionsV2(config.scheduleOptionsV2),
523 dataRefreshWindowDays: config.dataRefreshWindowDays,
524 disabled: config.disabled === true,
525 notificationPubsubTopic: config.notificationPubsubTopic,
526 emailPreferences: toEmailPreferences(config.emailPreferences),
527 encryptionConfiguration: toEncryption(config.encryptionConfiguration),
528 managedTableType: config.managedTableType,
529 metadataDestination: toMetadataDestination(config.metadataDestination),
530 datasetRegion: config.datasetRegion,
531 nextRunTime: config.nextRunTime,
532 state: config.state,
533 updateTime: config.updateTime,
534 userId: config.userId,
535 ownerInfo: config.ownerInfo,
536 error: config.error,
537 };
538};
539
540const getByName = (name: string) =>
541 bqdt
542 .getProjectsLocationsTransferConfigs({ name })
543 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
544
545const listAt = (parent: string) =>
546 bqdt.listProjectsLocationsTransferConfigs
547 .pages({
548 parent,
549 pageSize: 1000,
550 })
551 .pipe(
552 Stream.flatMap((page) => Stream.fromIterable(page.transferConfigs ?? [])),
553 Stream.runCollect,
554 Effect.map((chunk) => Array.from(chunk)),
555 Effect.catchTag("NotFound", () =>
556 Effect.succeed([] as bqdt.TransferConfig[]),
557 ),
558 );
559
560const findOwned = Effect.fn(function* (
561 parent: string,
562 id: string,
563 dataSourceId: string | undefined,
564) {
565 const configs = yield* listAt(parent);
566 for (const config of configs) {
567 const { labels } = parseDisplayName(config.displayName);
568 if (!(yield* hasAlchemyLabels(id, labels))) continue;
569 if (
570 dataSourceId !== undefined &&
571 (config.dataSourceId ?? "") !== dataSourceId
572 ) {
573 continue;
574 }
575 if (config.name === undefined) continue;
576 // List can lag a delete; confirm the config still exists.
577 return yield* getByName(config.name);
578 }
579 return undefined;
580});
581
582const toBody = (
583 news: TransferConfigProps,
584 displayName: string,
585): bqdt.TransferConfig =>
586 compact({
587 displayName,
588 dataSourceId: news.dataSourceId,
589 destinationDatasetId: news.destinationDatasetId,
590 params: news.params,
591 schedule: news.schedule,
592 scheduleOptions: news.scheduleOptions,
593 scheduleOptionsV2: news.scheduleOptionsV2,
594 dataRefreshWindowDays: news.dataRefreshWindowDays,
595 disabled: news.disabled === true ? true : news.disabled,
596 notificationPubsubTopic: news.notificationPubsubTopic,
597 emailPreferences: news.emailPreferences,
598 encryptionConfiguration: news.encryptionConfiguration,
599 managedTableType: news.managedTableType,
600 metadataDestination: news.metadataDestination,
601 }) as bqdt.TransferConfig;
602
603const retryTransient = <A, E extends { _tag: string }, R>(
604 effect: Effect.Effect<A, E, R>,
605) =>
606 effect.pipe(
607 Effect.retry({
608 while: (error) => error._tag === "NotFound" || error._tag === "Conflict",
609 times: 8,
610 schedule: backoff,
611 }),
612 );
613
614const listLocationParents = (project: string, region: string) =>
615 bqdt.listProjectsLocations
616 .pages({
617 name: `projects/${project}`,
618 pageSize: 100,
619 })
620 .pipe(
621 Stream.flatMap((page) => Stream.fromIterable(page.locations ?? [])),
622 Stream.map((location) => location.name),
623 Stream.filter((name): name is string => !!name && name.length > 0),
624 Stream.runCollect,
625 Effect.map((chunk) => Array.from(chunk)),
626 Effect.catchTag("NotFound", () =>
627 Effect.succeed([parentOf(project, region)]),
628 ),
629 );
630
631export const TransferConfigProvider = () =>
632 Provider.succeed(TransferConfig, {
633 stables: [
634 "name",
635 "transferConfigId",
636 "project",
637 "location",
638 "dataSourceId",
639 ],
640
641 diff: Effect.fn(function* ({ news, olds, output }) {
642 if (!isResolved(news)) return;
643 const previousLocation = olds?.location ?? output?.location;
644 const nextLocation =
645 typeof news.location === "string" ? news.location : undefined;
646 if (
647 previousLocation !== undefined &&
648 nextLocation !== undefined &&
649 normalizeLocation(previousLocation) !== normalizeLocation(nextLocation)
650 ) {
651 return { action: "replace" as const, deleteFirst: false };
652 }
653
654 const previousSource = olds?.dataSourceId ?? output?.dataSourceId;
655 const nextSource =
656 typeof news.dataSourceId === "string" ? news.dataSourceId : undefined;
657 if (
658 previousSource !== undefined &&
659 nextSource !== undefined &&
660 nextSource !== previousSource
661 ) {
662 return { action: "replace" as const, deleteFirst: false };
663 }
664 return undefined;
665 }),
666
667 read: Effect.fn(function* ({ id, olds, output }) {
668 const env = yield* GcpEnvironment.current;
669 const location = normalizeLocation(
670 olds?.location ?? output?.location ?? env.region,
671 );
672 const parent = parentOf(env.project, location);
673 const existing =
674 output?.name !== undefined
675 ? yield* getByName(output.name)
676 : yield* findOwned(parent, id, olds?.dataSourceId);
677 if (existing === undefined) return undefined;
678 const attrs = toAttrs(existing, env.project, location);
679 return (yield* hasAlchemyLabels(
680 id,
681 parseDisplayName(existing.displayName).labels,
682 ))
683 ? attrs
684 : Unowned(attrs);
685 }),
686
687 list: () =>
688 Effect.gen(function* () {
689 const env = yield* GcpEnvironment.current;
690 const parents = yield* listLocationParents(env.project, env.region);
691 const unique =
692 parents.length > 0
693 ? [...new Set(parents)]
694 : [parentOf(env.project, env.region)];
695 const pages = yield* Effect.forEach(
696 unique,
697 (parent) => listAt(parent),
698 { concurrency: 4 },
699 );
700 return pages
701 .flat()
702 .filter((config) => hasOwnershipMarker(config.displayName))
703 .map((config) => toAttrs(config, env.project, env.region));
704 }),
705
706 reconcile: Effect.fn(function* ({ id, news, output, olds }) {
707 const env = yield* GcpEnvironment.current;
708 const location = normalizeLocation(
709 news.location ?? output?.location ?? env.region,
710 );
711 const parent = parentOf(env.project, location);
712 const ownership = yield* createInternalLabels(id);
713 const desiredDisplayName = encodeDisplayName(ownership, news.displayName);
714 const desiredDisabled = news.disabled === true;
715 const desiredBody = toBody(
716 { ...news, disabled: desiredDisabled },
717 desiredDisplayName,
718 );
719
720 let current =
721 output?.name !== undefined ? yield* getByName(output.name) : undefined;
722 if (current === undefined) {
723 current = yield* findOwned(parent, id, news.dataSourceId);
724 }
725
726 if (current === undefined) {
727 const created = yield* bqdt
728 .createProjectsLocationsTransferConfigs({
729 parent,
730 serviceAccountName: news.serviceAccountName,
731 versionInfo: news.versionInfo,
732 authorizationCode: news.authorizationCode,
733 body: desiredBody,
734 })
735 .pipe(
736 Effect.catchTag("Conflict", () =>
737 findOwned(parent, id, news.dataSourceId),
738 ),
739 );
740 current = created ?? undefined;
741 }
742
743 if (current === undefined) {
744 return yield* new TransferConfigNotResolved({
745 name: `${parent}/transferConfigs`,
746 });
747 }
748
749 const name = current.name ?? "";
750 const mask: string[] = [];
751 if ((current.displayName ?? "") !== desiredDisplayName) {
752 mask.push("displayName");
753 }
754 if (
755 (current.destinationDatasetId ?? "") !==
756 (news.destinationDatasetId ?? "")
757 ) {
758 mask.push("destinationDatasetId");
759 }
760 if (!jsonEqual(canonParams(current.params), canonParams(news.params))) {
761 mask.push("params");
762 }
763 if (
764 news.schedule !== undefined &&
765 (current.schedule ?? "") !== news.schedule
766 ) {
767 mask.push("schedule");
768 }
769 if (
770 news.scheduleOptions !== undefined &&
771 !jsonEqual(
772 toScheduleOptions(current.scheduleOptions) ?? null,
773 compact({ ...news.scheduleOptions }),
774 )
775 ) {
776 mask.push("scheduleOptions");
777 }
778 if (
779 news.scheduleOptionsV2 !== undefined &&
780 !jsonEqual(
781 toScheduleOptionsV2(current.scheduleOptionsV2) ?? null,
782 news.scheduleOptionsV2,
783 )
784 ) {
785 mask.push("scheduleOptionsV2");
786 }
787 if (
788 news.dataRefreshWindowDays !== undefined &&
789 (current.dataRefreshWindowDays ?? 0) !== news.dataRefreshWindowDays
790 ) {
791 mask.push("dataRefreshWindowDays");
792 }
793 if ((current.disabled === true) !== desiredDisabled) {
794 mask.push("disabled");
795 }
796 if (
797 news.notificationPubsubTopic !== undefined &&
798 (current.notificationPubsubTopic ?? "") !== news.notificationPubsubTopic
799 ) {
800 mask.push("notificationPubsubTopic");
801 }
802 if (
803 news.emailPreferences !== undefined &&
804 !jsonEqual(
805 toEmailPreferences(current.emailPreferences) ?? null,
806 compact({ ...news.emailPreferences }),
807 )
808 ) {
809 mask.push("emailPreferences");
810 }
811 if (
812 news.encryptionConfiguration !== undefined &&
813 (current.encryptionConfiguration?.kmsKeyName ?? "") !==
814 (news.encryptionConfiguration.kmsKeyName ?? "")
815 ) {
816 mask.push("encryptionConfiguration");
817 }
818 if (
819 news.managedTableType !== undefined &&
820 (current.managedTableType ?? "") !== news.managedTableType
821 ) {
822 mask.push("managedTableType");
823 }
824 if (
825 news.metadataDestination !== undefined &&
826 !jsonEqual(
827 toMetadataDestination(current.metadataDestination) ?? null,
828 news.metadataDestination,
829 )
830 ) {
831 mask.push("metadataDestination");
832 }
833
834 const serviceAccountChanged =
835 (olds?.serviceAccountName ?? "") !== (news.serviceAccountName ?? "");
836
837 if (mask.length > 0 || serviceAccountChanged) {
838 current = yield* retryTransient(
839 bqdt.patchProjectsLocationsTransferConfigs({
840 name,
841 updateMask: mask.length > 0 ? mask.join(",") : "displayName",
842 serviceAccountName: news.serviceAccountName,
843 versionInfo: news.versionInfo,
844 authorizationCode: news.authorizationCode,
845 body: { ...desiredBody, name },
846 }),
847 );
848 current = (yield* getByName(name)) ?? current;
849 }
850
851 return toAttrs(current, env.project, location);
852 }),
853
854 delete: Effect.fn(function* ({ output }) {
855 yield* bqdt
856 .deleteProjectsLocationsTransferConfigs({ name: output.name })
857 .pipe(
858 Effect.retry({
859 while: (error) => error._tag === "Conflict",
860 times: 8,
861 schedule: Schedule.spaced("2 seconds"),
862 }),
863 Effect.catchTag("NotFound", () => Effect.void),
864 );
865 }),
866 });