1import * as scheduler from "@distilled.cloud/gcp/cloudscheduler_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 { 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 { GcpEnvironment } from "../Environment.ts";
12import type { Providers } from "../Providers.ts";
13
14const DEFAULT_TIME_ZONE = "UTC";
15const DEFAULT_HTTP_METHOD = "POST";
16const MAX_NAME_LENGTH = 63;
17
18const backoff = Schedule.min([
19 Schedule.exponential(Duration.millis(300), 1.5),
20 Schedule.spaced(Duration.seconds(2)),
21]);
22
23export type HttpMethod =
24 | "POST"
25 | "GET"
26 | "HEAD"
27 | "PUT"
28 | "DELETE"
29 | "PATCH"
30 | "OPTIONS";
31
32export type OidcToken = {
33 /**
34 * Service account email used to mint the OIDC token. Must be in the same
35 * project as the job. The caller needs `iam.serviceAccounts.actAs`.
36 */
37 serviceAccountEmail?: string;
38 /**
39 * Audience claimed in the token. Defaults to the HTTP target URI.
40 */
41 audience?: string;
42};
43
44export type OAuthToken = {
45 /**
46 * Service account email used to mint the OAuth token. Must be in the same
47 * project as the job. The caller needs `iam.serviceAccounts.actAs`.
48 */
49 serviceAccountEmail?: string;
50 /**
51 * OAuth scope. Defaults to `https://www.googleapis.com/auth/cloud-platform`.
52 */
53 scope?: string;
54};
55
56export type HttpTarget = {
57 /**
58 * Full URI the job POSTs (or GETs, …) to. Must start with `http://` or
59 * `https://`.
60 */
61 uri: string;
62 /**
63 * HTTP method.
64 * @default "POST"
65 */
66 httpMethod?: HttpMethod;
67 /**
68 * Request headers. Cloud Scheduler overwrites Host, Content-Length,
69 * User-Agent, and `X-Google-*` / `X-CloudScheduler*` headers.
70 */
71 headers?: Record<string, string>;
72 /**
73 * UTF-8 request body. Allowed for POST, PUT, and PATCH. Alchemy
74 * base64-encodes this for the API.
75 */
76 body?: string;
77 /**
78 * Attach an OIDC token as `Authorization`.
79 */
80 oidcToken?: OidcToken;
81 /**
82 * Attach an OAuth token as `Authorization` (typically for
83 * `*.googleapis.com` targets).
84 */
85 oauthToken?: OAuthToken;
86};
87
88export type PubsubTarget = {
89 /**
90 * Pub/Sub topic to publish to, as
91 * `projects/{project}/topics/{topic}`. Must be in the same project.
92 */
93 topicName: string;
94 /**
95 * UTF-8 message payload. Alchemy base64-encodes this for the API.
96 * Either `data` or `attributes` is required.
97 */
98 data?: string;
99 /**
100 * Message attributes. Either `data` or `attributes` is required.
101 */
102 attributes?: Record<string, string>;
103};
104
105export type AppEngineRouting = {
106 /** App Engine service. Empty uses the default service. */
107 service?: string;
108 /** App Engine version. Empty uses the default version. */
109 version?: string;
110 /** App Engine instance. Empty uses any available instance. */
111 instance?: string;
112};
113
114export type AppEngineHttpTarget = {
115 /**
116 * Relative URI beginning with `/`. Empty uses `/`.
117 */
118 relativeUri?: string;
119 /**
120 * HTTP method. PATCH and OPTIONS are not permitted.
121 * @default "POST"
122 */
123 httpMethod?: HttpMethod;
124 /**
125 * Request headers.
126 */
127 headers?: Record<string, string>;
128 /**
129 * UTF-8 request body. Allowed for POST and PUT. Alchemy base64-encodes
130 * this for the API.
131 */
132 body?: string;
133 /**
134 * App Engine service / version / instance routing.
135 */
136 appEngineRouting?: AppEngineRouting;
137};
138
139export type RetryConfig = {
140 /**
141 * Retry attempts after a failed execution (0–5).
142 * @default 0
143 */
144 retryCount?: number;
145 /**
146 * Time limit for retrying a failed job (e.g. `"3600s"`). Zero means
147 * unlimited, unless `retryCount` is also 0.
148 */
149 maxRetryDuration?: string;
150 /**
151 * Times the backoff doubles before increasing linearly.
152 * @default 5
153 */
154 maxDoublings?: number;
155 /**
156 * Maximum backoff between retries.
157 * @default "3600s"
158 */
159 maxBackoffDuration?: string;
160 /**
161 * Minimum backoff between retries.
162 * @default "5s"
163 */
164 minBackoffDuration?: string;
165};
166
167export type JobProps = {
168 /**
169 * Job id (the `{job}` segment of
170 * `projects/{project}/locations/{location}/jobs/{job}`). If omitted, a
171 * unique name is generated from the stack, stage, and logical id. Must
172 * match `[A-Za-z0-9_-]{1,500}`. Immutable — changing it replaces the job.
173 */
174 jobId?: string;
175 /**
176 * Cloud Scheduler location (`us-central1`, `us-east1`, …). Immutable —
177 * changing it replaces the job. `US-CENTRAL1` is accepted and normalized
178 * to `us-central1`.
179 * @default the stack's GCP region (`GCP.Region`, profile region, `us-central1`)
180 */
181 location?: string;
182 /**
183 * Cron or english-like schedule (e.g. `"0 0 * * *"` or
184 * `"every 24 hours"`). Required on create.
185 */
186 schedule: string;
187 /**
188 * tz-database time zone used to interpret `schedule`.
189 * @default "UTC"
190 */
191 timeZone?: string;
192 /**
193 * Human-readable description (single line, max 499 characters).
194 */
195 description?: string;
196 /**
197 * Deadline for a single attempt (e.g. `"180s"`). HTTP targets default to
198 * 3 minutes and must be in `[15s, 30m]`.
199 */
200 attemptDeadline?: string;
201 /**
202 * Retry behavior after a failed attempt.
203 */
204 retryConfig?: RetryConfig;
205 /**
206 * HTTP/HTTPS target. Exactly one of `httpTarget`, `pubsubTarget`, or
207 * `appEngineHttpTarget` must be set.
208 */
209 httpTarget?: HttpTarget;
210 /**
211 * Pub/Sub target. Exactly one of `httpTarget`, `pubsubTarget`, or
212 * `appEngineHttpTarget` must be set.
213 */
214 pubsubTarget?: PubsubTarget;
215 /**
216 * App Engine HTTP target. Exactly one of `httpTarget`, `pubsubTarget`,
217 * or `appEngineHttpTarget` must be set.
218 */
219 appEngineHttpTarget?: AppEngineHttpTarget;
220 /**
221 * When true, the job exists but is not scheduled (`Job.State.PAUSED`).
222 * Synced via `jobs.pause` / `jobs.resume`.
223 * @default false
224 */
225 paused?: boolean;
226};
227
228export type Job = Resource<
229 "GCP.CloudScheduler.Job",
230 JobProps,
231 {
232 /** Full resource name `projects/{project}/locations/{location}/jobs/{job}`. */
233 name: string;
234 /** Job id (last path segment). */
235 jobId: string;
236 /** Location id (`us-central1`, …). */
237 location: string;
238 /** Project id. */
239 project: string;
240 /** Job description. */
241 description: string | undefined;
242 /** Cron or english-like schedule. */
243 schedule: string | undefined;
244 /** Time zone used to interpret the schedule. */
245 timeZone: string | undefined;
246 /** Server-reported job state. */
247 state: string | undefined;
248 /** Whether the job is paused. */
249 paused: boolean;
250 /** Attempt deadline. */
251 attemptDeadline: string | undefined;
252 /** Retry configuration. */
253 retryConfig: RetryConfig | undefined;
254 /** HTTP target, if set. Body is the API's base64 payload. */
255 httpTarget: scheduler.HttpTarget | undefined;
256 /** Pub/Sub target, if set. */
257 pubsubTarget: scheduler.PubsubTarget | undefined;
258 /** App Engine HTTP target, if set. */
259 appEngineHttpTarget: scheduler.AppEngineHttpTarget | undefined;
260 /** Next scheduled attempt (RFC3339). */
261 scheduleTime: string | undefined;
262 /** Last attempt start (RFC3339). */
263 lastAttemptTime: string | undefined;
264 /** Last user update (RFC3339). */
265 userUpdateTime: string | undefined;
266 /** Last attempt status. */
267 status: scheduler.Status | undefined;
268 /** Whether the job satisfies physical zone separation. */
269 satisfiesPzs: boolean | undefined;
270 },
271 never,
272 Providers
273>;
274
275/**
276 * A Cloud Scheduler job that invokes an HTTP, Pub/Sub, or App Engine
277 * target on a cron schedule.
278 *
279 * Jobs have no labels field, so ownership rests on the deterministic job
280 * name: `read` reports a job it finds without prior state as unowned
281 * (adopt it with `--adopt`). Name and location are immutable; changing
282 * either replaces the job.
283 *
284 * ### Creating a Job
285 * **Example:** Generated name, yearly HTTP GET
286 * ```typescript
287 * const ping = yield* GCP.CloudScheduler.Job("Ping", {
288 * schedule: "0 0 1 1 *",
289 * paused: true,
290 * httpTarget: {
291 * uri: "https://example.com/",
292 * httpMethod: "GET",
293 * },
294 * });
295 * ```
296 *
297 * **Example:** Named job with retries and a POST body
298 * ```typescript
299 * const notify = yield* GCP.CloudScheduler.Job("Notify", {
300 * jobId: "order-notify",
301 * location: "us-central1",
302 * schedule: "every 15 minutes",
303 * timeZone: "America/Chicago",
304 * description: "fan out order events",
305 * retryConfig: { retryCount: 3, minBackoffDuration: "10s" },
306 * httpTarget: {
307 * uri: "https://example.com/hooks/orders",
308 * httpMethod: "POST",
309 * headers: { "Content-Type": "application/json" },
310 * body: JSON.stringify({ source: "scheduler" }),
311 * },
312 * });
313 * ```
314 *
315 * **Example:** Pub/Sub target
316 * ```typescript
317 * const tick = yield* GCP.CloudScheduler.Job("Tick", {
318 * schedule: "0 * * * *",
319 * pubsubTarget: {
320 * topicName: topic.name,
321 * data: "tick",
322 * },
323 * });
324 * ```
325 *
326 * @resource
327 * @category CloudScheduler
328 */
329export const Job = Resource<Job>("GCP.CloudScheduler.Job");
330
331export class JobNotResolved extends Data.TaggedError(
332 "GCP.CloudScheduler.JobNotResolved",
333)<{
334 name: string;
335}> {}
336
337export class SchedulerJobNotReady extends Data.TaggedError(
338 "GCP.CloudScheduler.JobNotReady",
339)<{
340 name: string;
341 state: string | undefined;
342}> {}
343
344export class JobStateFailed extends Data.TaggedError(
345 "GCP.CloudScheduler.JobStateFailed",
346)<{
347 name: string;
348 state: string;
349 message: string;
350}> {}
351
352export class JobTargetMissing extends Data.TaggedError(
353 "GCP.CloudScheduler.JobTargetMissing",
354)<{
355 name: string;
356 message: string;
357}> {}
358
359const lastSegment = (value: string) => {
360 const trimmed = value.replace(/\/+$/, "");
361 const parts = trimmed.split("/");
362 return parts[parts.length - 1] || trimmed;
363};
364
365const normalizeLocation = (location: string | undefined, fallback: string) =>
366 lastSegment(location ?? fallback).toLowerCase();
367
368const resourceName = (project: string, location: string, jobId: string) =>
369 `projects/${project}/locations/${location}/jobs/${jobId}`;
370
371const parentOf = (project: string, location: string) =>
372 `projects/${project}/locations/${location}`;
373
374const parseName = (name: string, fallbackLocation: string) => {
375 const parts = name.split("/").filter((part) => part.length > 0);
376 const jobsAt = parts.lastIndexOf("jobs");
377 const locationsAt = parts.lastIndexOf("locations");
378 const projectsAt = parts.lastIndexOf("projects");
379 return {
380 project:
381 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
382 location:
383 locationsAt >= 0 && parts[locationsAt + 1]
384 ? parts[locationsAt + 1]!
385 : fallbackLocation,
386 jobId:
387 jobsAt >= 0 && parts[jobsAt + 1] ? parts[jobsAt + 1]! : lastSegment(name),
388 };
389};
390
391const toId = (id: string, jobId: string | undefined, existing?: string) =>
392 Effect.gen(function* () {
393 return (
394 jobId ??
395 existing ??
396 (yield* createPhysicalName({
397 id,
398 maxLength: MAX_NAME_LENGTH,
399 lowercase: true,
400 }))
401 );
402 });
403
404const encodeUtf8 = (value: string | undefined) =>
405 value === undefined
406 ? Effect.succeed(undefined as string | undefined)
407 : Effect.sync(() => Buffer.from(value, "utf8").toString("base64"));
408
409const methodOf = (value: string | undefined) =>
410 value && value !== "HTTP_METHOD_UNSPECIFIED" ? value : DEFAULT_HTTP_METHOD;
411
412const timeZoneOf = (value: string | undefined) =>
413 value && value.length > 0 ? value : DEFAULT_TIME_ZONE;
414
415const compact = <T extends Record<string, unknown>>(value: T): T =>
416 Object.fromEntries(
417 Object.entries(value).filter(([, item]) => item !== undefined),
418 ) as T;
419
420const toAttrs = (job: scheduler.Job, project: string, region: string) => {
421 const name = job.name ?? "";
422 const parsed = parseName(name, region);
423 return {
424 name,
425 jobId: parsed.jobId,
426 location: parsed.location,
427 project: parsed.project || project,
428 description: job.description,
429 schedule: job.schedule,
430 timeZone: job.timeZone,
431 state: job.state,
432 paused: job.state === "PAUSED",
433 attemptDeadline: job.attemptDeadline,
434 retryConfig: job.retryConfig,
435 httpTarget: job.httpTarget,
436 pubsubTarget: job.pubsubTarget,
437 appEngineHttpTarget: job.appEngineHttpTarget,
438 scheduleTime: job.scheduleTime,
439 lastAttemptTime: job.lastAttemptTime,
440 userUpdateTime: job.userUpdateTime,
441 status: job.status,
442 satisfiesPzs: job.satisfiesPzs,
443 };
444};
445
446const getByName = (name: string) =>
447 scheduler
448 .getProjectsLocationsJobs({ name })
449 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
450
451/** Poll until Cloud Scheduler reports the job in the desired ENABLED/PAUSED state. */
452const waitForJobState = (name: string, paused: boolean) =>
453 getByName(name).pipe(
454 Effect.flatMap(
455 (
456 job,
457 ): Effect.Effect<
458 scheduler.Job,
459 JobStateFailed | SchedulerJobNotReady
460 > => {
461 const desired = paused ? "PAUSED" : "ENABLED";
462 if (job?.state === desired) return Effect.succeed(job);
463 if (job?.state === "UPDATE_FAILED" || job?.state === "DISABLED") {
464 return Effect.fail(
465 new JobStateFailed({
466 name,
467 state: job.state,
468 message: job.status?.message ?? `job is ${job.state}`,
469 }),
470 );
471 }
472 return Effect.fail(
473 new SchedulerJobNotReady({ name, state: job?.state }),
474 );
475 },
476 ),
477 Effect.retry({
478 while: (error) => error._tag === "GCP.CloudScheduler.JobNotReady",
479 times: 30,
480 schedule: Schedule.spaced("2 seconds"),
481 }),
482 );
483
484const headersMatch = (
485 observed: scheduler.StringMap | undefined,
486 desired: Record<string, string> | undefined,
487) => {
488 if (desired === undefined) return true;
489 const current = observed ?? {};
490 for (const [key, value] of Object.entries(desired)) {
491 const found = Object.entries(current).find(
492 ([header]) => header.toLowerCase() === key.toLowerCase(),
493 );
494 if ((found?.[1] ?? "") !== value) return false;
495 }
496 return true;
497};
498
499const tokenEqual = (
500 observed:
501 | { serviceAccountEmail?: string; audience?: string; scope?: string }
502 | undefined,
503 desired:
504 | { serviceAccountEmail?: string; audience?: string; scope?: string }
505 | undefined,
506) => {
507 if (desired === undefined) return true;
508 return (
509 (observed?.serviceAccountEmail ?? "") ===
510 (desired.serviceAccountEmail ?? "") &&
511 (observed?.audience ?? "") === (desired.audience ?? "") &&
512 (observed?.scope ?? "") === (desired.scope ?? "")
513 );
514};
515
516const subsetEqual = (
517 observed: Record<string, unknown> | undefined,
518 desired: Record<string, unknown> | undefined,
519) => {
520 if (desired === undefined) return true;
521 const current = observed ?? {};
522 for (const [key, value] of Object.entries(desired)) {
523 if (value === undefined) continue;
524 if (current[key] !== value) return false;
525 }
526 return true;
527};
528
529const toApiHttpTarget = Effect.fn(function* (target: HttpTarget) {
530 return compact({
531 uri: target.uri,
532 httpMethod: target.httpMethod,
533 headers: target.headers,
534 body: yield* encodeUtf8(target.body),
535 oidcToken: target.oidcToken,
536 oauthToken: target.oauthToken,
537 }) as scheduler.HttpTarget;
538});
539
540const toApiPubsubTarget = Effect.fn(function* (target: PubsubTarget) {
541 return compact({
542 topicName: target.topicName,
543 data: yield* encodeUtf8(target.data),
544 attributes: target.attributes,
545 }) as scheduler.PubsubTarget;
546});
547
548const toApiAppEngineHttpTarget = Effect.fn(function* (
549 target: AppEngineHttpTarget,
550) {
551 return compact({
552 relativeUri: target.relativeUri,
553 httpMethod: target.httpMethod,
554 headers: target.headers,
555 body: yield* encodeUtf8(target.body),
556 appEngineRouting: target.appEngineRouting,
557 }) as scheduler.AppEngineHttpTarget;
558});
559
560const httpTargetDrift = (
561 observed: scheduler.HttpTarget | undefined,
562 desired: scheduler.HttpTarget | undefined,
563 news: HttpTarget | undefined,
564) => {
565 if (news === undefined || desired === undefined) return false;
566 if ((observed?.uri ?? "") !== (desired.uri ?? "")) return true;
567 if (methodOf(observed?.httpMethod) !== methodOf(desired.httpMethod)) {
568 return true;
569 }
570 if ((observed?.body ?? "") !== (desired.body ?? "")) return true;
571 if (!headersMatch(observed?.headers, news.headers)) return true;
572 if (!tokenEqual(observed?.oidcToken, news.oidcToken)) return true;
573 if (!tokenEqual(observed?.oauthToken, news.oauthToken)) return true;
574 return false;
575};
576
577const pubsubTargetDrift = (
578 observed: scheduler.PubsubTarget | undefined,
579 desired: scheduler.PubsubTarget | undefined,
580) => {
581 if (desired === undefined) return false;
582 if ((observed?.topicName ?? "") !== (desired.topicName ?? "")) return true;
583 if ((observed?.data ?? "") !== (desired.data ?? "")) return true;
584 return !subsetEqual(observed?.attributes, desired.attributes);
585};
586
587const appEngineTargetDrift = (
588 observed: scheduler.AppEngineHttpTarget | undefined,
589 desired: scheduler.AppEngineHttpTarget | undefined,
590 news: AppEngineHttpTarget | undefined,
591) => {
592 if (news === undefined || desired === undefined) return false;
593 if ((observed?.relativeUri ?? "") !== (desired.relativeUri ?? "")) {
594 return true;
595 }
596 if (methodOf(observed?.httpMethod) !== methodOf(desired.httpMethod)) {
597 return true;
598 }
599 if ((observed?.body ?? "") !== (desired.body ?? "")) return true;
600 if (!headersMatch(observed?.headers, news.headers)) return true;
601 return !subsetEqual(
602 observed?.appEngineRouting as Record<string, unknown> | undefined,
603 news.appEngineRouting as Record<string, unknown> | undefined,
604 );
605};
606
607const hasTarget = (news: JobProps) =>
608 news.httpTarget !== undefined ||
609 news.pubsubTarget !== undefined ||
610 news.appEngineHttpTarget !== undefined;
611
612const retryTransient = <A, E extends { _tag: string }, R>(
613 effect: Effect.Effect<A, E, R>,
614) =>
615 effect.pipe(
616 Effect.retry({
617 while: (error) => error._tag === "NotFound" || error._tag === "Conflict",
618 times: 8,
619 schedule: backoff,
620 }),
621 );
622
623const syncPaused = Effect.fn(function* (
624 name: string,
625 current: scheduler.Job,
626 paused: boolean,
627) {
628 if (paused && current.state === "ENABLED") {
629 const pausedJob = yield* retryTransient(
630 scheduler.pauseProjectsLocationsJobs({ name, body: {} }),
631 ).pipe(
632 Effect.catchTag(["BadRequest", "Conflict", "NotFound"], () =>
633 getByName(name).pipe(
634 Effect.flatMap((job) =>
635 job !== undefined && job.state === "PAUSED"
636 ? Effect.succeed(job)
637 : new JobNotResolved({ name }),
638 ),
639 ),
640 ),
641 );
642 return pausedJob ?? current;
643 }
644 if (!paused && current.state === "PAUSED") {
645 const resumed = yield* retryTransient(
646 scheduler.resumeProjectsLocationsJobs({ name, body: {} }),
647 ).pipe(
648 Effect.catchTag(["BadRequest", "Conflict", "NotFound"], () =>
649 getByName(name).pipe(
650 Effect.flatMap((job) =>
651 job !== undefined && job.state === "ENABLED"
652 ? Effect.succeed(job)
653 : new JobNotResolved({ name }),
654 ),
655 ),
656 ),
657 );
658 return resumed ?? current;
659 }
660 return current;
661});
662
663export const JobProvider = () =>
664 Provider.succeed(Job, {
665 stables: ["name", "jobId", "location", "project"],
666
667 diff: Effect.fn(function* ({ news, olds, output }) {
668 if (!isResolved(news)) return undefined;
669 const env = yield* GcpEnvironment.current;
670 const previousId = olds?.jobId ?? output?.jobId;
671 const nextId = news.jobId ?? previousId;
672 const idChanged =
673 previousId !== undefined &&
674 nextId !== undefined &&
675 nextId !== previousId;
676
677 const previousLocation = normalizeLocation(
678 olds?.location ?? output?.location,
679 env.region,
680 );
681 const nextLocation = normalizeLocation(
682 news.location ?? output?.location,
683 env.region,
684 );
685 const locationChanged = previousLocation !== nextLocation;
686
687 if (idChanged || locationChanged) {
688 return { action: "replace" as const, deleteFirst: false };
689 }
690 return undefined;
691 }),
692
693 read: Effect.fn(function* ({ id, olds, output }) {
694 const env = yield* GcpEnvironment.current;
695 const jobId = yield* toId(id, olds?.jobId, output?.jobId);
696 const location = normalizeLocation(
697 olds?.location ?? output?.location,
698 env.region,
699 );
700 const name = output?.name ?? resourceName(env.project, location, jobId);
701 const existing = yield* getByName(name);
702 if (existing === undefined) return undefined;
703 const attrs = toAttrs(existing, env.project, env.region);
704 // No labels field: without prior state the job may not be ours.
705 return output !== undefined ? attrs : Unowned(attrs);
706 }),
707
708 reconcile: Effect.fn(function* ({ id, news, output }) {
709 const env = yield* GcpEnvironment.current;
710 const jobId = yield* toId(id, news.jobId, output?.jobId);
711 const location = normalizeLocation(
712 news.location ?? output?.location,
713 env.region,
714 );
715 const name = resourceName(env.project, location, jobId);
716 const parent = parentOf(env.project, location);
717 if (!hasTarget(news)) {
718 return yield* new JobTargetMissing({
719 name,
720 message:
721 "Cloud Scheduler jobs require httpTarget, pubsubTarget, or appEngineHttpTarget",
722 });
723 }
724
725 // Descriptions are a single RE2 line (`^.{1,499}$`).
726 const desiredDescription =
727 news.description
728 ?.replace(/[\r\n]+/g, " ")
729 .trim()
730 .slice(0, 499) || undefined;
731 const desiredTimeZone = timeZoneOf(news.timeZone);
732 const desiredPaused = news.paused === true;
733 const httpTarget =
734 news.httpTarget !== undefined
735 ? yield* toApiHttpTarget(news.httpTarget)
736 : undefined;
737 const pubsubTarget =
738 news.pubsubTarget !== undefined
739 ? yield* toApiPubsubTarget(news.pubsubTarget)
740 : undefined;
741 const appEngineHttpTarget =
742 news.appEngineHttpTarget !== undefined
743 ? yield* toApiAppEngineHttpTarget(news.appEngineHttpTarget)
744 : undefined;
745
746 const desiredBody: scheduler.Job = compact({
747 name,
748 description: desiredDescription,
749 schedule: news.schedule,
750 timeZone: desiredTimeZone,
751 attemptDeadline: news.attemptDeadline,
752 retryConfig: news.retryConfig,
753 httpTarget,
754 pubsubTarget,
755 appEngineHttpTarget,
756 });
757
758 let current = yield* getByName(name);
759
760 if (current === undefined) {
761 const created = yield* scheduler
762 .createProjectsLocationsJobs({
763 parent,
764 body: desiredBody,
765 })
766 .pipe(Effect.catchTag("Conflict", () => getByName(name)));
767 current = created ?? undefined;
768 }
769
770 if (current === undefined) {
771 return yield* new JobNotResolved({ name });
772 }
773
774 const mask: string[] = [];
775 if ((current.description ?? "") !== (desiredDescription ?? "")) {
776 mask.push("description");
777 }
778 if ((current.schedule ?? "") !== news.schedule) {
779 mask.push("schedule");
780 }
781 if (timeZoneOf(current.timeZone) !== desiredTimeZone) {
782 mask.push("timeZone");
783 }
784 if (
785 news.attemptDeadline !== undefined &&
786 (current.attemptDeadline ?? "") !== news.attemptDeadline
787 ) {
788 mask.push("attemptDeadline");
789 }
790 if (
791 news.retryConfig !== undefined &&
792 !subsetEqual(
793 current.retryConfig as Record<string, unknown> | undefined,
794 news.retryConfig as Record<string, unknown>,
795 )
796 ) {
797 mask.push("retryConfig");
798 }
799 if (httpTargetDrift(current.httpTarget, httpTarget, news.httpTarget)) {
800 mask.push("httpTarget");
801 }
802 if (pubsubTargetDrift(current.pubsubTarget, pubsubTarget)) {
803 mask.push("pubsubTarget");
804 }
805 if (
806 appEngineTargetDrift(
807 current.appEngineHttpTarget,
808 appEngineHttpTarget,
809 news.appEngineHttpTarget,
810 )
811 ) {
812 mask.push("appEngineHttpTarget");
813 }
814
815 if (mask.length > 0) {
816 current = yield* retryTransient(
817 scheduler.patchProjectsLocationsJobs({
818 name,
819 updateMask: mask.join(","),
820 body: desiredBody,
821 }),
822 );
823 current = (yield* getByName(name)) ?? current;
824 }
825
826 current = yield* syncPaused(name, current, desiredPaused);
827 current = yield* waitForJobState(name, desiredPaused);
828 return toAttrs(current, env.project, env.region);
829 }),
830
831 delete: Effect.fn(function* ({ output }) {
832 yield* scheduler.deleteProjectsLocationsJobs({ name: output.name }).pipe(
833 Effect.retry({
834 while: (error) => error._tag === "Conflict",
835 times: 10,
836 schedule: Schedule.spaced("3 seconds"),
837 }),
838 Effect.catchTag("NotFound", () => Effect.void),
839 );
840 }),
841 });