1import * as cloudtasks from "@distilled.cloud/gcp/cloudtasks_v2";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import type { Providers } from "../Providers.ts";
11
12const DEFAULT_STATE = "RUNNING" as const;
13const MAX_QUEUE_ID_LENGTH = 100;
14
15export type QueueRetryConfig = {
16 /** Attempts per task, including the first. `-1` means unlimited. */
17 maxAttempts?: number;
18 /** Time limit for retrying a failed task, e.g. `"3600s"`. `"0s"` is unlimited. */
19 maxRetryDuration?: string;
20 /** Minimum backoff after a failed attempt, e.g. `"0.1s"`. */
21 minBackoff?: string;
22 /** Maximum backoff after a failed attempt, e.g. `"3600s"`. */
23 maxBackoff?: string;
24 /** Times the retry interval doubles before increasing linearly. */
25 maxDoublings?: number;
26};
27
28export type QueueRateLimits = {
29 /** Maximum dispatch rate (tasks per second). Max `500`. */
30 maxDispatchesPerSecond?: number;
31 /** Maximum concurrent dispatches. Max `5000`. */
32 maxConcurrentDispatches?: number;
33};
34
35export type QueueStackdriverLoggingConfig = {
36 /** Fraction of operations to log, in `[0.0, 1.0]`. `0` logs nothing. */
37 samplingRatio?: number;
38};
39
40export type QueueHttpTarget = cloudtasks.HttpTarget;
41export type QueueAppEngineRouting = cloudtasks.AppEngineRouting;
42
43export type QueueState = "RUNNING" | "PAUSED";
44
45export type QueueProps = {
46 /**
47 * Queue id (the `{queue}` segment of
48 * `projects/{project}/locations/{location}/queues/{queue}`). If omitted,
49 * a unique name is generated from the stack, stage, and logical id.
50 * Immutable — changing it replaces the queue.
51 */
52 queueId?: string;
53 /**
54 * Location of the queue (e.g. `us-central1`). Immutable — changing it
55 * replaces the queue. `US-CENTRAL1` is accepted and normalized to
56 * `us-central1`.
57 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
58 */
59 location?: string;
60 /**
61 * Queue-level retry settings applied to Cloud Tasks–created tasks.
62 * Omitted fields keep the values Cloud Tasks currently has (including
63 * API defaults).
64 */
65 retryConfig?: QueueRetryConfig;
66 /**
67 * Dispatch rate limits. `maxBurstSize` is output-only and not settable.
68 * Omitted fields keep the current Cloud Tasks values.
69 */
70 rateLimits?: QueueRateLimits;
71 /**
72 * Cloud Logging sampling. If omitted, existing logging config is left
73 * unchanged.
74 */
75 stackdriverLoggingConfig?: QueueStackdriverLoggingConfig;
76 /**
77 * Queue-level HTTP target overrides for HTTP tasks. If omitted, the
78 * queue's current overrides are left unchanged.
79 */
80 httpTarget?: QueueHttpTarget;
81 /**
82 * App Engine routing override for App Engine tasks in this queue.
83 */
84 appEngineRoutingOverride?: QueueAppEngineRouting;
85 /**
86 * Desired queue state. `UpdateQueue` cannot change state — Alchemy
87 * calls PauseQueue / ResumeQueue. `DISABLED` is queue.yaml-only and
88 * is not settable here.
89 * @default "RUNNING"
90 */
91 state?: QueueState;
92};
93
94export type Queue = Resource<
95 "GCP.CloudTasks.Queue",
96 QueueProps,
97 {
98 /** Full resource name `projects/{project}/locations/{location}/queues/{queue}`. */
99 name: string;
100 /** Queue id (last path segment). */
101 queueId: string;
102 /** Project id. */
103 project: string;
104 /** Location id (e.g. `us-central1`). */
105 location: string;
106 /** Server-reported state (`RUNNING`, `PAUSED`, `DISABLED`). */
107 state: string | undefined;
108 /** Retry settings currently on the queue. */
109 retryConfig: QueueRetryConfig | undefined;
110 /** Dispatch rate limits currently on the queue. */
111 rateLimits: QueueRateLimits | undefined;
112 /** Logging config currently on the queue. */
113 stackdriverLoggingConfig: QueueStackdriverLoggingConfig | undefined;
114 /** Queue-level HTTP target overrides. */
115 httpTarget: QueueHttpTarget | undefined;
116 /** App Engine routing override, if any. */
117 appEngineRoutingOverride: QueueAppEngineRouting | undefined;
118 /** Last purge time, if the queue has ever been purged. */
119 purgeTime: string | undefined;
120 },
121 never,
122 Providers
123>;
124
125/**
126 * A Google Cloud Tasks queue.
127 *
128 * Cloud Tasks queues have no labels, so Alchemy identifies its queues by
129 * the generated `queueId`: `read` reports a queue with an explicit
130 * `queueId` as unowned (adopt it with `--adopt`), and nuke cannot
131 * discover queues.
132 *
133 * Changing `queueId` or `location` replaces the queue. Deleted queue
134 * names are tombstoned for up to 3 days — CreateQueue may appear to
135 * succeed; GetQueue is the source of truth.
136 *
137 * ### Creating a Queue
138 * **Example:** Generated name
139 * ```typescript
140 * const queue = yield* GCP.CloudTasks.Queue("jobs", {});
141 * ```
142 *
143 * **Example:** Rate limits, retries, and logging
144 * ```typescript
145 * const queue = yield* GCP.CloudTasks.Queue("jobs", {
146 * location: "us-central1",
147 * rateLimits: {
148 * maxDispatchesPerSecond: 10,
149 * maxConcurrentDispatches: 5,
150 * },
151 * retryConfig: {
152 * maxAttempts: 5,
153 * minBackoff: "1s",
154 * maxBackoff: "60s",
155 * },
156 * stackdriverLoggingConfig: { samplingRatio: 1 },
157 * });
158 * ```
159 *
160 * **Example:** Paused queue
161 * ```typescript
162 * const queue = yield* GCP.CloudTasks.Queue("jobs", {
163 * state: "PAUSED",
164 * });
165 * ```
166 *
167 * @resource
168 * @category CloudTasks
169 */
170export const Queue = Resource<Queue>("GCP.CloudTasks.Queue");
171
172export class QueueNotResolved extends Data.TaggedError(
173 "GCP.CloudTasks.QueueNotResolved",
174)<{
175 name: string;
176}> {}
177
178const lastSegment = (value: string) => {
179 const trimmed = value.replace(/\/+$/, "");
180 const parts = trimmed.split("/");
181 return parts[parts.length - 1] || trimmed;
182};
183
184const normalizeLocation = (location: string | undefined, fallback: string) =>
185 lastSegment(location ?? fallback).toLowerCase();
186
187const parseName = (name: string) => {
188 const parts = name.split("/");
189 return {
190 project: parts[1] ?? "",
191 location: parts[3] ?? "",
192 queueId: parts[5] ?? lastSegment(name),
193 };
194};
195
196const resourceName = (project: string, location: string, queueId: string) =>
197 `projects/${project}/locations/${location}/queues/${queueId}`;
198
199const locationParent = (project: string, location: string) =>
200 `projects/${project}/locations/${location}`;
201
202const toQueueId = (
203 id: string,
204 queueId: string | undefined,
205 existing?: string,
206) =>
207 Effect.gen(function* () {
208 if (queueId !== undefined) return queueId;
209 if (existing !== undefined) return existing;
210 const generated = yield* createPhysicalName({
211 id,
212 maxLength: MAX_QUEUE_ID_LENGTH,
213 lowercase: true,
214 });
215 const named = /^[a-z]/.test(generated) ? generated : `q${generated}`;
216 return named.replace(/-+$/g, "").slice(0, MAX_QUEUE_ID_LENGTH);
217 });
218
219const sortHeaderOverrides = (
220 headers: cloudtasks.HeaderOverrideList | undefined,
221): cloudtasks.HeaderOverrideList =>
222 [...(headers ?? [])]
223 .map((item) => ({
224 header: {
225 key: item.header?.key?.toLowerCase(),
226 value: item.header?.value,
227 },
228 }))
229 .sort((left, right) =>
230 (left.header?.key ?? "").localeCompare(right.header?.key ?? ""),
231 );
232
233const stable = (value: unknown): string =>
234 JSON.stringify(value, (_key, current) => {
235 if (current && typeof current === "object" && !Array.isArray(current)) {
236 return Object.fromEntries(
237 Object.entries(current as Record<string, unknown>)
238 .filter(([, item]) => item !== undefined)
239 .sort(([a], [b]) => a.localeCompare(b)),
240 );
241 }
242 return current;
243 });
244
245const normalizeDuration = (value: string | undefined) => {
246 if (value === undefined) return undefined;
247 const match = /^([0-9]+(?:\.[0-9]+)?)s$/.exec(value.trim());
248 if (!match) return value;
249 return `${Number(match[1])}s`;
250};
251
252const normalizeRetry = (config: QueueRetryConfig) => ({
253 maxAttempts: config.maxAttempts,
254 maxRetryDuration: normalizeDuration(config.maxRetryDuration),
255 minBackoff: normalizeDuration(config.minBackoff),
256 maxBackoff: normalizeDuration(config.maxBackoff),
257 maxDoublings: config.maxDoublings,
258});
259
260const overlayChanged = (
261 desired: object | undefined,
262 observed: object | undefined,
263 normalize: (value: object) => unknown = (value) => value,
264): boolean => {
265 if (desired === undefined) return false;
266 const picked: Record<string, unknown> = {};
267 const source = (observed ?? {}) as Record<string, unknown>;
268 for (const key of Object.keys(desired)) {
269 picked[key] = source[key];
270 }
271 return stable(normalize(desired)) !== stable(normalize(picked));
272};
273
274const toRateLimits = (
275 config: QueueRateLimits,
276 observed?: cloudtasks.RateLimits,
277): cloudtasks.RateLimits => {
278 const merged: cloudtasks.RateLimits = { ...observed, ...config };
279 delete merged.maxBurstSize;
280 return merged;
281};
282
283const toUserRateLimits = (
284 config: cloudtasks.RateLimits | undefined,
285): QueueRateLimits | undefined => {
286 if (config === undefined) return undefined;
287 const next: QueueRateLimits = {};
288 if (config.maxDispatchesPerSecond !== undefined) {
289 next.maxDispatchesPerSecond = config.maxDispatchesPerSecond;
290 }
291 if (config.maxConcurrentDispatches !== undefined) {
292 next.maxConcurrentDispatches = config.maxConcurrentDispatches;
293 }
294 return Object.keys(next).length > 0 ? next : undefined;
295};
296
297const toAttrs = (
298 queue: cloudtasks.Queue,
299 project: string,
300): Queue["Attributes"] => {
301 const name = queue.name ?? "";
302 const parsed = parseName(name);
303 return {
304 name,
305 queueId: parsed.queueId,
306 project: parsed.project || project,
307 location: parsed.location,
308 state: queue.state,
309 retryConfig: queue.retryConfig,
310 rateLimits: toUserRateLimits(queue.rateLimits),
311 stackdriverLoggingConfig: queue.stackdriverLoggingConfig,
312 httpTarget: queue.httpTarget,
313 appEngineRoutingOverride: queue.appEngineRoutingOverride,
314 purgeTime: queue.purgeTime,
315 };
316};
317
318const getByName = (name: string) =>
319 cloudtasks
320 .getProjectsLocationsQueues({ name })
321 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
322
323const sameHttpTarget = (
324 left: cloudtasks.HttpTarget,
325 right: cloudtasks.HttpTarget | undefined,
326) => {
327 const normalize = (target: cloudtasks.HttpTarget | undefined) => {
328 if (target === undefined) return undefined;
329 return {
330 ...target,
331 headerOverrides: sortHeaderOverrides(target.headerOverrides),
332 };
333 };
334 return stable(normalize(left)) === stable(normalize(right));
335};
336
337export const QueueProvider = () =>
338 Provider.succeed(Queue, {
339 stables: ["name", "queueId", "project", "location"],
340
341 diff: Effect.fn(function* ({ news, olds, output }) {
342 const env = yield* GcpEnvironment.current;
343 if (!isResolved(news)) return undefined;
344 const previousId = olds?.queueId ?? output?.queueId;
345 const nextId = news.queueId ?? previousId;
346 const idChanged =
347 previousId !== undefined &&
348 nextId !== undefined &&
349 nextId !== previousId;
350
351 const previousLocation = normalizeLocation(
352 olds?.location ?? output?.location,
353 env.region,
354 );
355 const nextLocation = normalizeLocation(
356 news.location ?? output?.location,
357 env.region,
358 );
359 if (idChanged || previousLocation !== nextLocation) {
360 return { action: "replace" as const };
361 }
362 return undefined;
363 }),
364
365 read: Effect.fn(function* ({ id, olds, output }) {
366 const env = yield* GcpEnvironment.current;
367 const queueId = yield* toQueueId(id, olds?.queueId, output?.queueId);
368 const location = normalizeLocation(
369 olds?.location ?? output?.location,
370 env.region,
371 );
372 const name = output?.name ?? resourceName(env.project, location, queueId);
373 const existing = yield* getByName(name);
374 if (existing === undefined) return undefined;
375 const attrs = toAttrs(existing, env.project);
376 const generated = yield* toQueueId(id, undefined);
377 return attrs.queueId === generated ? attrs : Unowned(attrs);
378 }),
379
380 reconcile: Effect.fn(function* ({ id, news, output }) {
381 const env = yield* GcpEnvironment.current;
382 const queueId = yield* toQueueId(id, news.queueId, output?.queueId);
383 const location = normalizeLocation(
384 news.location ?? output?.location,
385 env.region,
386 );
387 const name = resourceName(env.project, location, queueId);
388 const parent = locationParent(env.project, location);
389 const desiredHttpTarget = news.httpTarget;
390 const desiredState = news.state ?? DEFAULT_STATE;
391
392 let current = yield* getByName(name);
393
394 if (current === undefined) {
395 const body: cloudtasks.Queue = {
396 name,
397 httpTarget: desiredHttpTarget,
398 };
399 if (news.retryConfig !== undefined) {
400 body.retryConfig = news.retryConfig;
401 }
402 if (news.rateLimits !== undefined) {
403 body.rateLimits = toRateLimits(news.rateLimits);
404 }
405 if (news.stackdriverLoggingConfig !== undefined) {
406 body.stackdriverLoggingConfig = news.stackdriverLoggingConfig;
407 }
408 if (news.appEngineRoutingOverride !== undefined) {
409 body.appEngineRoutingOverride = news.appEngineRoutingOverride;
410 }
411 // CreateQueue can appear to succeed for a tombstoned name. GetQueue
412 // is the source of truth (up to ~3 days after delete).
413 yield* cloudtasks
414 .createProjectsLocationsQueues({
415 parent,
416 body,
417 })
418 .pipe(Effect.catchTag("Conflict", () => Effect.void));
419 current = yield* getByName(name);
420 }
421
422 if (current === undefined) {
423 return yield* new QueueNotResolved({ name });
424 }
425
426 const mask: string[] = [];
427 const patchBody: cloudtasks.Queue = { name };
428
429 if (
430 overlayChanged(news.retryConfig, current.retryConfig, (value) =>
431 normalizeRetry(value as QueueRetryConfig),
432 )
433 ) {
434 patchBody.retryConfig = {
435 ...current.retryConfig,
436 ...news.retryConfig,
437 };
438 mask.push("retryConfig");
439 }
440
441 const desiredRateLimits = news.rateLimits;
442 if (
443 desiredRateLimits !== undefined &&
444 overlayChanged(desiredRateLimits, toUserRateLimits(current.rateLimits))
445 ) {
446 patchBody.rateLimits = toRateLimits(
447 desiredRateLimits,
448 current.rateLimits,
449 );
450 mask.push("rateLimits");
451 }
452
453 if (
454 overlayChanged(
455 news.stackdriverLoggingConfig,
456 current.stackdriverLoggingConfig,
457 )
458 ) {
459 patchBody.stackdriverLoggingConfig = {
460 ...current.stackdriverLoggingConfig,
461 ...news.stackdriverLoggingConfig,
462 };
463 mask.push("stackdriverLoggingConfig");
464 }
465
466 if (
467 desiredHttpTarget !== undefined &&
468 !sameHttpTarget(desiredHttpTarget, current.httpTarget)
469 ) {
470 patchBody.httpTarget = desiredHttpTarget;
471 mask.push("httpTarget");
472 }
473
474 if (
475 overlayChanged(
476 news.appEngineRoutingOverride,
477 current.appEngineRoutingOverride,
478 )
479 ) {
480 patchBody.appEngineRoutingOverride = {
481 ...current.appEngineRoutingOverride,
482 ...news.appEngineRoutingOverride,
483 };
484 mask.push("appEngineRoutingOverride");
485 }
486
487 if (mask.length > 0) {
488 current = yield* cloudtasks.patchProjectsLocationsQueues({
489 name,
490 updateMask: mask.join(","),
491 body: patchBody,
492 });
493 }
494
495 if (desiredState === "PAUSED" && current.state === "RUNNING") {
496 current = yield* cloudtasks.pauseProjectsLocationsQueues({
497 name,
498 body: {},
499 });
500 } else if (
501 desiredState === "RUNNING" &&
502 (current.state === "PAUSED" || current.state === "DISABLED")
503 ) {
504 current = yield* cloudtasks.resumeProjectsLocationsQueues({
505 name,
506 body: {},
507 });
508 }
509
510 return toAttrs(current, env.project);
511 }),
512
513 delete: Effect.fn(function* ({ output }) {
514 yield* cloudtasks
515 .deleteProjectsLocationsQueues({ name: output.name })
516 .pipe(Effect.catchTag("NotFound", () => Effect.void));
517 }),
518 });