1import * as bigquery from "@distilled.cloud/gcp/bigquery_v2";
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 hasAlchemyLabels,
16 stripInternalLabels,
17 toLabels,
18} from "../Labels.ts";
19import type { Providers } from "../Providers.ts";
20
21const MAX_JOB_ID_LENGTH = 1024;
22
23export type JobQuery = bigquery.JobConfigurationQuery;
24export type JobLoad = bigquery.JobConfigurationLoad;
25export type JobCopy = bigquery.JobConfigurationTableCopy;
26export type JobExtract = bigquery.JobConfigurationExtract;
27export type JobError = bigquery.ErrorProto;
28
29export type JobProps = {
30 /**
31 * Job id (the `{jobId}` segment of `projects/{project}/jobs/{jobId}`).
32 * If omitted, a unique name is generated from the stack, stage, and
33 * logical id. Letters, numbers, underscores, and dashes; max 1,024
34 * characters. Immutable — changing it replaces the job.
35 */
36 jobId?: string;
37 /**
38 * Geographic location (`US`, `EU`, `US-CENTRAL1`, `us-central1`, …).
39 * Immutable — changing it replaces the job.
40 * @default the stack's GCP region (`GCP.Region`, profile region, `us-central1`)
41 */
42 location?: string;
43 /**
44 * User labels. Alchemy ownership labels are merged in automatically.
45 * Immutable after create — changing labels replaces the job.
46 */
47 labels?: Record<string, string>;
48 /**
49 * Query job configuration. Mutually exclusive with `load`, `copy`, and
50 * `extract`. GoogleSQL is the default (`useLegacySql: false`).
51 */
52 query?: JobQuery;
53 /**
54 * Load job configuration. Mutually exclusive with `query`, `copy`, and
55 * `extract`.
56 */
57 load?: JobLoad;
58 /**
59 * Copy job configuration. Mutually exclusive with `query`, `load`, and
60 * `extract`.
61 */
62 copy?: JobCopy;
63 /**
64 * Extract job configuration. Mutually exclusive with `query`, `load`,
65 * and `copy`.
66 */
67 extract?: JobExtract;
68 /**
69 * Job timeout in milliseconds relative to creation time.
70 */
71 jobTimeoutMs?: string;
72 /**
73 * Reservation the job should run on, as
74 * `projects/{project}/locations/{location}/reservations/{reservation}`.
75 */
76 reservation?: string;
77};
78
79export type Job = Resource<
80 "GCP.BigQuery.Job",
81 JobProps,
82 {
83 /** Resource path `projects/{project}/jobs/{jobId}`. */
84 name: string;
85 /** Opaque id `project:jobId`. */
86 id: string;
87 /** Job id. */
88 jobId: string;
89 /** Project id. */
90 project: string;
91 /** Geographic location. */
92 location: string;
93 /** User labels (Alchemy ownership labels stripped). */
94 labels: Record<string, string>;
95 /** Job kind (`QUERY`, `LOAD`, `EXTRACT`, `COPY`, …). */
96 jobType: string | undefined;
97 /** Running state (`PENDING`, `RUNNING`, `DONE`). */
98 state: string | undefined;
99 /** Final error when the job completed unsuccessfully. */
100 errorResult: JobError | undefined;
101 /** Query text when this is a query job. */
102 query: string | undefined;
103 /** Self-link URL. */
104 selfLink: string | undefined;
105 /** Creator email. */
106 userEmail: string | undefined;
107 /** Creation time in milliseconds since epoch. */
108 creationTime: string | undefined;
109 /** Start time in milliseconds since epoch. */
110 startTime: string | undefined;
111 /** End time in milliseconds since epoch. */
112 endTime: string | undefined;
113 },
114 never,
115 Providers
116>;
117
118/**
119 * A Google BigQuery job — a one-shot query, load, copy, or extract.
120 *
121 * Jobs are immutable after insert. Changing `jobId`, `location`, labels,
122 * or the job configuration replaces the job. BigQuery job ids cannot be
123 * reused even after metadata delete, so replacement always creates a new
124 * id then deletes the previous generation. Destroy cancels a running job,
125 * then deletes its metadata.
126 *
127 * ### Creating a Job
128 * **Example:** GoogleSQL query
129 * ```typescript
130 * const job = yield* GCP.BigQuery.Job("Count", {
131 * query: { query: "SELECT 1 AS n", useLegacySql: false },
132 * labels: { env: "prod" },
133 * });
134 * ```
135 *
136 * **Example:** Query into a destination table
137 * ```typescript
138 * const dataset = yield* GCP.BigQuery.Dataset("Analytics", {
139 * forceDestroy: true,
140 * });
141 * const table = yield* GCP.BigQuery.Table("Results", {
142 * datasetId: dataset.datasetId,
143 * schema: [{ name: "n", type: "INTEGER" }],
144 * });
145 * const load = yield* GCP.BigQuery.Job("Seed", {
146 * query: {
147 * query: "SELECT 1 AS n",
148 * useLegacySql: false,
149 * destinationTable: {
150 * projectId: table.project,
151 * datasetId: table.datasetId,
152 * tableId: table.tableId,
153 * },
154 * writeDisposition: "WRITE_TRUNCATE",
155 * },
156 * });
157 * ```
158 *
159 * @resource
160 * @category BigQuery
161 */
162export const Job = Resource<Job>("GCP.BigQuery.Job");
163
164export class JobNotResolved extends Data.TaggedError(
165 "GCP.BigQuery.JobNotResolved",
166)<{
167 name: string;
168}> {}
169
170export class JobNotDone extends Data.TaggedError("GCP.BigQuery.JobNotDone")<{
171 jobId: string;
172 state: string | undefined;
173}> {}
174
175const userLabels = (
176 labels: Record<string, string | undefined> | null | undefined,
177): Record<string, string> => stripInternalLabels(tagRecord(labels));
178
179const resourceName = (project: string, jobId: string) =>
180 `projects/${project}/jobs/${jobId}`;
181
182const normalizeLocation = (location: string | undefined, fallback: string) =>
183 (location ?? fallback).toUpperCase();
184
185const toId = (id: string, jobId: string | undefined, existing?: string) =>
186 Effect.gen(function* () {
187 if (jobId !== undefined) return jobId;
188 if (existing !== undefined) return existing;
189 return yield* createPhysicalName({
190 id,
191 maxLength: MAX_JOB_ID_LENGTH,
192 lowercase: true,
193 });
194 });
195
196const jobIdOf = (
197 job: bigquery.Job | bigquery.JobListJobsItem,
198 fallback: string,
199) => job.jobReference?.jobId ?? fallback;
200
201const projectOf = (
202 job: bigquery.Job | bigquery.JobListJobsItem,
203 fallback: string,
204) => job.jobReference?.projectId ?? fallback;
205
206const locationOf = (
207 job: bigquery.Job | bigquery.JobListJobsItem,
208 fallback: string,
209) => job.jobReference?.location ?? fallback;
210
211const stateOf = (job: bigquery.Job | bigquery.JobListJobsItem) => {
212 const full = job as bigquery.Job;
213 return full.status?.state ?? (job as bigquery.JobListJobsItem).state;
214};
215
216const toAttrs = (
217 job: bigquery.Job | bigquery.JobListJobsItem,
218 fallbackProject: string,
219 region: string,
220) => {
221 const jobId = jobIdOf(job, "");
222 const project = projectOf(job, fallbackProject);
223 const full = job as bigquery.Job;
224 return {
225 name: resourceName(project, jobId),
226 id: full.id ?? `${project}:${jobId}`,
227 jobId,
228 project,
229 location: locationOf(job, region),
230 labels: userLabels(job.configuration?.labels),
231 jobType: job.configuration?.jobType,
232 state: stateOf(job),
233 errorResult:
234 full.status?.errorResult ?? (job as bigquery.JobListJobsItem).errorResult,
235 query: job.configuration?.query?.query,
236 selfLink: full.selfLink,
237 userEmail: full.user_email ?? (job as bigquery.JobListJobsItem).user_email,
238 creationTime: full.statistics?.creationTime ?? job.statistics?.creationTime,
239 startTime: full.statistics?.startTime ?? job.statistics?.startTime,
240 endTime: full.statistics?.endTime ?? job.statistics?.endTime,
241 };
242};
243
244const getByRef = (projectId: string, jobId: string, location: string) =>
245 bigquery
246 .getJobs({ projectId, jobId, location })
247 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
248
249const waitUntilDone = (projectId: string, jobId: string, location: string) =>
250 getByRef(projectId, jobId, location).pipe(
251 Effect.flatMap((job) => {
252 if (job === undefined) return Effect.succeed(undefined);
253 const state = job.status?.state ?? "";
254 if (state === "DONE") return Effect.succeed(job);
255 return Effect.fail(new JobNotDone({ jobId, state }));
256 }),
257 Effect.retry({
258 while: (error) => error._tag === "GCP.BigQuery.JobNotDone",
259 times: 10,
260 schedule: Schedule.spaced("2 seconds"),
261 }),
262 );
263
264const queryBody = (query: JobQuery): JobQuery => ({
265 ...query,
266 useLegacySql: query.useLegacySql ?? false,
267});
268
269const toJobBody = (
270 projectId: string,
271 jobId: string,
272 location: string,
273 news: JobProps,
274 labels: Record<string, string>,
275): bigquery.Job => ({
276 jobReference: { projectId, jobId, location },
277 configuration: {
278 labels,
279 query: news.query !== undefined ? queryBody(news.query) : undefined,
280 load: news.load,
281 copy: news.copy,
282 extract: news.extract,
283 jobTimeoutMs: news.jobTimeoutMs,
284 reservation: news.reservation,
285 },
286});
287
288const stable = (value: unknown): unknown => {
289 if (Array.isArray(value)) return value.map(stable);
290 if (value !== null && typeof value === "object") {
291 return Object.fromEntries(
292 Object.entries(value as Record<string, unknown>)
293 .filter(([, nested]) => nested !== undefined)
294 .sort(([left], [right]) => left.localeCompare(right))
295 .map(([key, nested]) => [key, stable(nested)]),
296 );
297 }
298 return value;
299};
300
301const jsonEqual = (left: unknown, right: unknown) =>
302 JSON.stringify(stable(left ?? null)) ===
303 JSON.stringify(stable(right ?? null));
304
305const pickDefined = (
306 news: Record<string, unknown>,
307 observed: Record<string, unknown>,
308) => {
309 const next: Record<string, unknown> = {};
310 const previous: Record<string, unknown> = {};
311 for (const [key, value] of Object.entries(news)) {
312 if (value !== undefined) {
313 next[key] = value;
314 previous[key] = observed[key];
315 }
316 }
317 return { news: next, observed: previous };
318};
319
320const nestedChanged = (
321 news: Record<string, unknown> | undefined,
322 olds: Record<string, unknown> | undefined,
323 hadOlds: boolean,
324) => {
325 if (news === undefined) return false;
326 if (olds === undefined) return hadOlds;
327 const picked = pickDefined(news, olds);
328 return !jsonEqual(picked.news, picked.observed);
329};
330
331export const JobProvider = () =>
332 Provider.succeed(Job, {
333 stables: ["name", "id", "jobId", "project", "location", "creationTime"],
334
335 diff: Effect.fn(function* ({ news, olds, output }) {
336 if (!isResolved(news)) return undefined;
337 const env = yield* GcpEnvironment.current;
338 const previousId = olds?.jobId ?? output?.jobId;
339 const jobIdChanged =
340 news.jobId !== undefined &&
341 previousId !== undefined &&
342 news.jobId !== previousId;
343
344 const previousLocation = normalizeLocation(
345 olds?.location ?? output?.location,
346 env.region,
347 );
348 const nextLocation = normalizeLocation(
349 news.location ?? olds?.location ?? output?.location,
350 env.region,
351 );
352 const locationChanged =
353 output !== undefined && previousLocation !== nextLocation;
354
355 const labelsChanged =
356 news.labels !== undefined &&
357 output !== undefined &&
358 !jsonEqual(toLabels(news.labels), output.labels);
359
360 const timeoutChanged =
361 news.jobTimeoutMs !== undefined &&
362 (olds?.jobTimeoutMs ?? "") !== news.jobTimeoutMs;
363 const reservationChanged =
364 news.reservation !== undefined &&
365 (olds?.reservation ?? "") !== news.reservation;
366
367 const replaced =
368 jobIdChanged ||
369 locationChanged ||
370 labelsChanged ||
371 timeoutChanged ||
372 reservationChanged ||
373 nestedChanged(
374 news.query as Record<string, unknown> | undefined,
375 olds?.query as Record<string, unknown> | undefined,
376 olds !== undefined,
377 ) ||
378 nestedChanged(
379 news.load as Record<string, unknown> | undefined,
380 olds?.load as Record<string, unknown> | undefined,
381 olds !== undefined,
382 ) ||
383 nestedChanged(
384 news.copy as Record<string, unknown> | undefined,
385 olds?.copy as Record<string, unknown> | undefined,
386 olds !== undefined,
387 ) ||
388 nestedChanged(
389 news.extract as Record<string, unknown> | undefined,
390 olds?.extract as Record<string, unknown> | undefined,
391 olds !== undefined,
392 );
393
394 if (!replaced) return undefined;
395 // Job ids cannot be reused after delete (the API returns NotFound on
396 // insert of a retired id), so never delete-first.
397 return { action: "replace" as const, deleteFirst: false };
398 }),
399
400 read: Effect.fn(function* ({ id, olds, output }) {
401 const env = yield* GcpEnvironment.current;
402 const jobId = yield* toId(id, olds?.jobId, output?.jobId);
403 const location = olds?.location ?? output?.location ?? env.region;
404 const project = output?.project ?? env.project;
405 const existing = yield* getByRef(project, jobId, location);
406 if (existing === undefined) return undefined;
407 const attrs = toAttrs(existing, project, env.region);
408 return (yield* hasAlchemyLabels(
409 id,
410 tagRecord(existing.configuration?.labels),
411 ))
412 ? attrs
413 : Unowned(attrs);
414 }),
415
416 list: () =>
417 Effect.gen(function* () {
418 const env = yield* GcpEnvironment.current;
419 return yield* bigquery.listJobs
420 .pages({
421 projectId: env.project,
422 maxResults: 1000,
423 projection: "full",
424 allUsers: true,
425 })
426 .pipe(
427 Stream.flatMap((page) => Stream.fromIterable(page.jobs ?? [])),
428 Stream.filter((job) =>
429 Object.keys(job.configuration?.labels ?? {}).some((key) =>
430 key.startsWith("alchemy-"),
431 ),
432 ),
433 Stream.map((job) => toAttrs(job, env.project, env.region)),
434 Stream.runCollect,
435 Effect.map((chunk) => Array.from(chunk)),
436 Effect.catchTag("Forbidden", () =>
437 bigquery.listJobs
438 .pages({
439 projectId: env.project,
440 maxResults: 1000,
441 projection: "full",
442 })
443 .pipe(
444 Stream.flatMap((page) =>
445 Stream.fromIterable(page.jobs ?? []),
446 ),
447 Stream.filter((job) =>
448 Object.keys(job.configuration?.labels ?? {}).some((key) =>
449 key.startsWith("alchemy-"),
450 ),
451 ),
452 Stream.map((job) => toAttrs(job, env.project, env.region)),
453 Stream.runCollect,
454 Effect.map((chunk) => Array.from(chunk)),
455 ),
456 ),
457 Effect.catchTag("NotFound", () =>
458 Effect.succeed([] as ReturnType<typeof toAttrs>[]),
459 ),
460 );
461 }),
462
463 reconcile: Effect.fn(function* ({ id, news, output }) {
464 const env = yield* GcpEnvironment.current;
465 const jobId = yield* toId(id, news.jobId, output?.jobId);
466 const location = news.location ?? output?.location ?? env.region;
467 const project = output?.project ?? env.project;
468 const name = resourceName(project, jobId);
469 const desiredLabels = {
470 ...toLabels(news.labels),
471 ...(yield* createInternalLabels(id)),
472 };
473
474 let current = yield* getByRef(project, jobId, location);
475
476 if (current === undefined) {
477 const created = yield* bigquery
478 .insertJobs({
479 projectId: project,
480 body: toJobBody(project, jobId, location, news, desiredLabels),
481 })
482 .pipe(
483 Effect.catchTag("Conflict", () =>
484 getByRef(project, jobId, location),
485 ),
486 );
487 current = created ?? undefined;
488 }
489
490 if (current === undefined) {
491 return yield* new JobNotResolved({ name });
492 }
493
494 const state = current.status?.state ?? "";
495 if (state !== "DONE") {
496 current = (yield* waitUntilDone(project, jobId, location)) ?? current;
497 }
498
499 return toAttrs(current, project, env.region);
500 }),
501
502 delete: Effect.fn(function* ({ output }) {
503 const state = output.state ?? "";
504 if (state === "PENDING" || state === "RUNNING") {
505 yield* bigquery
506 .cancelJobs({
507 projectId: output.project,
508 jobId: output.jobId,
509 location: output.location,
510 })
511 .pipe(
512 Effect.catchTag("NotFound", () => Effect.void),
513 Effect.catchTag("BadRequest", () => Effect.void),
514 );
515 yield* waitUntilDone(
516 output.project,
517 output.jobId,
518 output.location,
519 ).pipe(Effect.catchTag("GCP.BigQuery.JobNotDone", () => Effect.void));
520 }
521 yield* bigquery
522 .deleteJobs({
523 projectId: output.project,
524 jobId: output.jobId,
525 location: output.location,
526 })
527 .pipe(
528 Effect.retry({
529 while: (error) =>
530 error._tag === "Conflict" || error._tag === "BadRequest",
531 times: 8,
532 schedule: Schedule.spaced("2 seconds"),
533 }),
534 Effect.catchTag("NotFound", () => Effect.void),
535 );
536 }),
537 });