1import * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
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 { projectNumber } from "../Host.ts";
14import { updateIamMembership } from "../IamPolicy.ts";
15import {
16 createInternalLabels,
17 diffLabels,
18 hasAlchemyLabels,
19 stripInternalLabels,
20 toLabels,
21} from "../Labels.ts";
22import type { Providers } from "../Providers.ts";
23
24export type SubscriptionRetryPolicy = {
25 /**
26 * Minimum delay between consecutive deliveries of a given message
27 * (e.g. `"10s"`). Must be between 0 and 600 seconds.
28 */
29 minimumBackoff?: string;
30 /**
31 * Maximum delay between consecutive deliveries of a given message
32 * (e.g. `"600s"`). Must be between 0 and 600 seconds.
33 */
34 maximumBackoff?: string;
35};
36
37export type SubscriptionDeadLetterPolicy = {
38 /**
39 * Topic that receives dead-lettered messages, as
40 * `projects/{project}/topics/{topic}`.
41 */
42 deadLetterTopic?: string;
43 /**
44 * Maximum delivery attempts before dead-lettering. Must be between 5 and
45 * 100. `0` is treated as `5`.
46 */
47 maxDeliveryAttempts?: number;
48};
49
50export type SubscriptionExpirationPolicy = {
51 /**
52 * Idle time-to-live (e.g. `"2678400s"` for 31 days). Minimum is 1 day.
53 * Omit `ttl` to never expire. If the whole policy is omitted, GCP
54 * applies a 31-day default.
55 */
56 ttl?: string;
57};
58
59export type SubscriptionOidcToken = {
60 /** Service account email Pub/Sub uses to sign the JWT. */
61 serviceAccountEmail?: string;
62 /** JWT audience. Defaults to the push endpoint. */
63 audience?: string;
64};
65
66export type SubscriptionPushConfig = {
67 /**
68 * HTTPS endpoint that receives pushed messages
69 * (e.g. `"https://example.com/push"`).
70 */
71 pushEndpoint?: string;
72 /**
73 * Endpoint attributes. The only currently supported key is
74 * `x-goog-version` (`v1` / `v1beta1` / `v1beta2`).
75 */
76 attributes?: Record<string, string>;
77 /**
78 * Attach an OIDC JWT as `Authorization` on every push.
79 */
80 oidcToken?: SubscriptionOidcToken;
81};
82
83export type SubscriptionProps = {
84 /**
85 * Subscription id (the `{subscription}` segment of
86 * `projects/{project}/subscriptions/{subscription}`). If omitted, a
87 * unique name is generated from the stack, stage, and logical id.
88 * Changing it replaces the subscription.
89 */
90 subscriptionId?: string;
91 /**
92 * Topic this subscription receives from. Accepts a full resource name
93 * (`projects/{project}/topics/{topic}`) or a topic id. Immutable —
94 * changing it replaces the subscription.
95 */
96 topic: string;
97 /**
98 * User labels. Alchemy ownership labels are merged in automatically.
99 */
100 labels?: Record<string, string>;
101 /**
102 * Approximate time Pub/Sub waits for an ack before redelivering, in
103 * seconds. Minimum 10, maximum 600. `0` is treated as 10.
104 * @default 10
105 */
106 ackDeadlineSeconds?: number;
107 /**
108 * Retain acknowledged messages so they can be seeked until they fall
109 * out of `messageRetentionDuration`.
110 * @default false
111 */
112 retainAckedMessages?: boolean;
113 /**
114 * How long to retain unacknowledged messages (e.g. `"86400s"`). Cannot
115 * be more than 31 days or less than 10 minutes. Defaults to 7 days.
116 */
117 messageRetentionDuration?: string;
118 /**
119 * Deliver messages with the same ordering key in publish order.
120 * Immutable — changing it replaces the subscription.
121 * @default false
122 */
123 enableMessageOrdering?: boolean;
124 /**
125 * Guarantee that an acknowledged message is not redelivered (best
126 * effort; distinct publishes of the same payload still get distinct
127 * `message_id`s).
128 * @default false
129 */
130 enableExactlyOnceDelivery?: boolean;
131 /**
132 * Pub/Sub filter language expression. Only matching messages are
133 * delivered. Immutable — changing it replaces the subscription.
134 */
135 filter?: string;
136 /**
137 * When the subscription expires if idle. Omit for the 31-day default.
138 */
139 expirationPolicy?: SubscriptionExpirationPolicy;
140 /**
141 * Retry backoff after a NACK or ack-deadline exceed.
142 */
143 retryPolicy?: SubscriptionRetryPolicy;
144 /**
145 * Dead-letter topic and max delivery attempts. Omit to disable.
146 *
147 * Pub/Sub forwards dead letters as the project's Pub/Sub service agent
148 * (`service-{projectNumber}@gcp-sa-pubsub.iam.gserviceaccount.com`), so
149 * reconcile grants it `roles/pubsub.publisher` on the dead-letter topic
150 * and `roles/pubsub.subscriber` on this subscription. The topic grant is
151 * left in place on delete: the agent is shared by every subscription in
152 * the project that dead-letters to that topic.
153 */
154 deadLetterPolicy?: SubscriptionDeadLetterPolicy;
155 /**
156 * Push delivery endpoint. Omit (or leave empty) for a pull
157 * subscription.
158 */
159 pushConfig?: SubscriptionPushConfig;
160};
161
162export type Subscription = Resource<
163 "GCP.PubSub.Subscription",
164 SubscriptionProps,
165 {
166 /** Full resource name `projects/{project}/subscriptions/{subscription}`. */
167 name: string;
168 /** Subscription id (last path segment). */
169 subscriptionId: string;
170 /** Project id. */
171 project: string;
172 /** Topic resource name this subscription is attached to. */
173 topic: string;
174 /** User labels (Alchemy ownership labels stripped). */
175 labels: Record<string, string>;
176 /** Ack deadline in seconds. */
177 ackDeadlineSeconds: number;
178 /** Whether acknowledged messages are retained. */
179 retainAckedMessages: boolean;
180 /** Message retention duration, if set. */
181 messageRetentionDuration: string | undefined;
182 /** Whether message ordering is enabled. */
183 enableMessageOrdering: boolean;
184 /** Whether exactly-once delivery is enabled. */
185 enableExactlyOnceDelivery: boolean;
186 /** Filter expression, if set. */
187 filter: string | undefined;
188 /** Server-reported subscription state. */
189 state: string | undefined;
190 /** Idle expiration policy, if set. */
191 expirationPolicy: SubscriptionExpirationPolicy | undefined;
192 /** Retry policy, if set. */
193 retryPolicy: SubscriptionRetryPolicy | undefined;
194 /** Dead-letter policy, if set. */
195 deadLetterPolicy: SubscriptionDeadLetterPolicy | undefined;
196 /** Push config, if this is a push subscription. */
197 pushConfig: SubscriptionPushConfig | undefined;
198 },
199 never,
200 Providers
201>;
202
203/**
204 * A Google Cloud Pub/Sub subscription.
205 *
206 * ### Creating a Subscription
207 * **Example:** Pull subscription on a topic
208 * ```typescript
209 * const topic = yield* GCP.PubSub.Topic("events", {});
210 * const subscription = yield* GCP.PubSub.Subscription("orders", {
211 * topic: topic.name,
212 * });
213 * ```
214 *
215 * **Example:** Explicit id, labels, and ack deadline
216 * ```typescript
217 * const topic = yield* GCP.PubSub.Topic("events", {});
218 * const subscription = yield* GCP.PubSub.Subscription("orders", {
219 * subscriptionId: "order-events",
220 * topic: topic.name,
221 * ackDeadlineSeconds: 30,
222 * labels: { env: "prod" },
223 * });
224 * ```
225 *
226 * ### Pulling Messages
227 * **Example:** Pull and acknowledge
228 * ```typescript
229 * const pull = yield* GCP.PubSub.Pull(subscription);
230 * const { receivedMessages } = yield* pull({
231 * body: { maxMessages: 1 },
232 * });
233 * const ackIds = (receivedMessages ?? [])
234 * .map((message) => message.ackId)
235 * .filter((ackId): ackId is string => ackId !== undefined);
236 * if (ackIds.length > 0) {
237 * const acknowledge = yield* GCP.PubSub.Acknowledge(subscription);
238 * yield* acknowledge({ body: { ackIds } });
239 * }
240 * ```
241 *
242 * @resource
243 * @category PubSub
244 */
245export const Subscription = Resource<Subscription>("GCP.PubSub.Subscription");
246
247export class SubscriptionNotResolved extends Data.TaggedError(
248 "GCP.PubSub.SubscriptionNotResolved",
249)<{
250 name: string;
251}> {}
252
253const DEFAULT_ACK_DEADLINE_SECONDS = 10;
254
255const subscriptionIdOf = (name: string) => name.split("/").pop() ?? name;
256
257const resourceName = (project: string, subscriptionId: string) =>
258 `projects/${project}/subscriptions/${subscriptionId}`;
259
260const topicNameOf = (project: string, topic: string) =>
261 topic.startsWith("projects/") ? topic : `projects/${project}/topics/${topic}`;
262
263const topicKey = (topic: string | undefined) =>
264 topic === undefined ? undefined : subscriptionIdOf(topic);
265
266const userLabels = (
267 labels: Record<string, string | undefined> | null | undefined,
268): Record<string, string> => stripInternalLabels(tagRecord(labels));
269
270const toId = (
271 id: string,
272 subscriptionId: string | undefined,
273 existing?: string,
274) =>
275 Effect.gen(function* () {
276 return (
277 subscriptionId ??
278 existing ??
279 (yield* createPhysicalName({
280 id,
281 maxLength: 255,
282 lowercase: true,
283 forbiddenPrefixes: ["goog"],
284 }))
285 );
286 });
287
288const toPushConfig = (
289 config: pubsub.PushConfig | undefined,
290): SubscriptionPushConfig | undefined => {
291 if (config === undefined) return undefined;
292 const attributes = tagRecord(config.attributes);
293 const pushEndpoint = config.pushEndpoint;
294 if (pushEndpoint === undefined && Object.keys(attributes).length === 0) {
295 return undefined;
296 }
297 return {
298 pushEndpoint,
299 attributes: Object.keys(attributes).length > 0 ? attributes : undefined,
300 oidcToken: config.oidcToken,
301 };
302};
303
304const toAttrs = (subscription: pubsub.Subscription, project: string) => {
305 const name = subscription.name ?? "";
306 return {
307 name,
308 subscriptionId: subscriptionIdOf(name),
309 project,
310 topic: subscription.topic ?? "",
311 labels: userLabels(subscription.labels),
312 ackDeadlineSeconds:
313 subscription.ackDeadlineSeconds ?? DEFAULT_ACK_DEADLINE_SECONDS,
314 retainAckedMessages: subscription.retainAckedMessages === true,
315 messageRetentionDuration: subscription.messageRetentionDuration,
316 enableMessageOrdering: subscription.enableMessageOrdering === true,
317 enableExactlyOnceDelivery: subscription.enableExactlyOnceDelivery === true,
318 filter: subscription.filter,
319 state: subscription.state,
320 expirationPolicy: subscription.expirationPolicy,
321 retryPolicy: subscription.retryPolicy,
322 deadLetterPolicy: subscription.deadLetterPolicy,
323 pushConfig: toPushConfig(subscription.pushConfig),
324 };
325};
326
327const getByName = (name: string) =>
328 pubsub
329 .getProjectsSubscriptions({ subscription: name })
330 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
331
332const waitUntilGone = (name: string) =>
333 getByName(name).pipe(
334 Effect.filterOrFail(
335 (existing) => existing === undefined,
336 () => new SubscriptionStillExists({ name }),
337 ),
338 Effect.retry({
339 while: (error) => error._tag === "GCP.PubSub.SubscriptionStillExists",
340 schedule: Schedule.spaced("2 seconds"),
341 times: 30,
342 }),
343 Effect.asVoid,
344 );
345
346export class SubscriptionStillExists extends Data.TaggedError(
347 "GCP.PubSub.SubscriptionStillExists",
348)<{ name: string }> {}
349
350// Pub/Sub serves reads from replicas that lag writes: one successful GET
351// does not mean the next one (or a publish routed to it) sees the
352// resource. Require consecutive successful reads before reporting ready.
353const CONSISTENT_READS = 3;
354
355const waitUntilPresent = (name: string) =>
356 Effect.gen(function* () {
357 let found: pubsub.Subscription | undefined;
358 for (let read = 0; read < CONSISTENT_READS; read++) {
359 if (read > 0) yield* Effect.sleep("1 second");
360 found = yield* getByName(name);
361 if (found === undefined) {
362 return yield* new SubscriptionNotResolved({ name });
363 }
364 }
365 return found!;
366 }).pipe(
367 Effect.retry({
368 while: (error) => error._tag === "GCP.PubSub.SubscriptionNotResolved",
369 schedule: Schedule.spaced("1 second"),
370 times: 60,
371 }),
372 );
373
374/**
375 * Let the project's Pub/Sub service agent forward dead letters: publish to
376 * the dead-letter topic and ack the original on the subscription.
377 */
378const grantDeadLetterAgent = (
379 project: string,
380 subscription: string,
381 deadLetterTopic: string,
382) =>
383 Effect.gen(function* () {
384 const number = yield* projectNumber(project);
385 const member = `serviceAccount:service-${number}@gcp-sa-pubsub.iam.gserviceaccount.com`;
386 yield* updateIamMembership({
387 kind: "pubsub.topic",
388 name: topicNameOf(project, deadLetterTopic),
389 member,
390 add: ["roles/pubsub.publisher"],
391 });
392 yield* updateIamMembership({
393 kind: "pubsub.subscription",
394 name: subscription,
395 member,
396 add: ["roles/pubsub.subscriber"],
397 });
398 });
399
400const sameOptionalString = (
401 left: string | undefined,
402 right: string | undefined,
403) => (left ?? "") === (right ?? "");
404
405const retryChanged = (
406 desired: SubscriptionRetryPolicy | undefined,
407 observed: pubsub.RetryPolicy | undefined,
408) =>
409 desired !== undefined &&
410 !(
411 sameOptionalString(desired.minimumBackoff, observed?.minimumBackoff) &&
412 sameOptionalString(desired.maximumBackoff, observed?.maximumBackoff)
413 );
414
415const deadLetterChanged = (
416 desired: SubscriptionDeadLetterPolicy | undefined,
417 observed: pubsub.DeadLetterPolicy | undefined,
418) =>
419 desired !== undefined &&
420 !(
421 sameOptionalString(desired.deadLetterTopic, observed?.deadLetterTopic) &&
422 (desired.maxDeliveryAttempts ?? 0) === (observed?.maxDeliveryAttempts ?? 0)
423 );
424
425const expirationChanged = (
426 desired: SubscriptionExpirationPolicy | undefined,
427 observed: pubsub.ExpirationPolicy | undefined,
428) => desired !== undefined && !sameOptionalString(desired.ttl, observed?.ttl);
429
430const pushChanged = (
431 desired: SubscriptionPushConfig | undefined,
432 observed: pubsub.PushConfig | undefined,
433) => {
434 if (desired === undefined) return false;
435 const observedAttributes = tagRecord(observed?.attributes);
436 const desiredAttributes = tagRecord(desired.attributes);
437 return (
438 (desired.pushEndpoint ?? "") !== (observed?.pushEndpoint ?? "") ||
439 JSON.stringify(desiredAttributes) !== JSON.stringify(observedAttributes) ||
440 (desired.oidcToken?.serviceAccountEmail ?? "") !==
441 (observed?.oidcToken?.serviceAccountEmail ?? "") ||
442 (desired.oidcToken?.audience ?? "") !==
443 (observed?.oidcToken?.audience ?? "")
444 );
445};
446
447const toCreateBody = (
448 topicName: string,
449 news: SubscriptionProps,
450 labels: Record<string, string>,
451): pubsub.Subscription => {
452 const body: pubsub.Subscription = {
453 topic: topicName,
454 labels,
455 };
456 if (news.ackDeadlineSeconds !== undefined) {
457 body.ackDeadlineSeconds = news.ackDeadlineSeconds;
458 }
459 if (news.retainAckedMessages !== undefined) {
460 body.retainAckedMessages = news.retainAckedMessages;
461 }
462 if (news.messageRetentionDuration !== undefined) {
463 body.messageRetentionDuration = news.messageRetentionDuration;
464 }
465 if (news.enableMessageOrdering !== undefined) {
466 body.enableMessageOrdering = news.enableMessageOrdering;
467 }
468 if (news.enableExactlyOnceDelivery !== undefined) {
469 body.enableExactlyOnceDelivery = news.enableExactlyOnceDelivery;
470 }
471 if (news.filter !== undefined) {
472 body.filter = news.filter;
473 }
474 if (news.expirationPolicy !== undefined) {
475 body.expirationPolicy = news.expirationPolicy;
476 }
477 if (news.retryPolicy !== undefined) {
478 body.retryPolicy = news.retryPolicy;
479 }
480 if (news.deadLetterPolicy !== undefined) {
481 body.deadLetterPolicy = news.deadLetterPolicy;
482 }
483 if (news.pushConfig !== undefined) {
484 body.pushConfig = news.pushConfig;
485 }
486 return body;
487};
488
489export const SubscriptionProvider = () =>
490 Provider.succeed(Subscription, {
491 stables: ["name", "subscriptionId", "project", "topic"],
492
493 diff: Effect.fn(function* ({ news, olds, output }) {
494 if (!isResolved(news)) return undefined;
495 const previousId = olds?.subscriptionId ?? output?.subscriptionId;
496 const nextId = news.subscriptionId ?? previousId;
497 const nameChanged =
498 news.subscriptionId !== undefined &&
499 previousId !== undefined &&
500 news.subscriptionId !== previousId;
501 const previousTopic = olds?.topic ?? output?.topic;
502 const topicChanged =
503 previousTopic !== undefined &&
504 topicKey(news.topic) !== topicKey(previousTopic);
505 const previousFilter = olds?.filter ?? output?.filter ?? "";
506 const nextFilter = news.filter ?? "";
507 const previousOrdering =
508 olds?.enableMessageOrdering ?? output?.enableMessageOrdering ?? false;
509 const nextOrdering = news.enableMessageOrdering ?? false;
510 if (
511 !nameChanged &&
512 !topicChanged &&
513 previousFilter === nextFilter &&
514 previousOrdering === nextOrdering
515 ) {
516 return undefined;
517 }
518 return {
519 action: "replace" as const,
520 deleteFirst:
521 !nameChanged &&
522 nextId !== undefined &&
523 previousId !== undefined &&
524 nextId === previousId,
525 };
526 }),
527
528 read: Effect.fn(function* ({ id, olds, output }) {
529 const env = yield* GcpEnvironment.current;
530 const subscriptionId = yield* toId(
531 id,
532 olds?.subscriptionId,
533 output?.subscriptionId,
534 );
535 const name = output?.name ?? resourceName(env.project, subscriptionId);
536 const existing = yield* getByName(name);
537 if (existing === undefined) return undefined;
538 const attrs = toAttrs(existing, env.project);
539 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
540 ? attrs
541 : Unowned(attrs);
542 }),
543
544 list: () =>
545 Effect.gen(function* () {
546 const env = yield* GcpEnvironment.current;
547 return yield* pubsub.listProjectsSubscriptions
548 .pages({
549 project: `projects/${env.project}`,
550 pageSize: 1000,
551 })
552 .pipe(
553 Stream.flatMap((page) =>
554 Stream.fromIterable(page.subscriptions ?? []),
555 ),
556 Stream.filter((subscription) =>
557 Object.keys(subscription.labels ?? {}).some((key) =>
558 key.startsWith("alchemy-"),
559 ),
560 ),
561 Stream.map((subscription) => toAttrs(subscription, env.project)),
562 Stream.runCollect,
563 Effect.map((chunk) => Array.from(chunk)),
564 );
565 }),
566
567 reconcile: Effect.fn(function* ({ id, news, output }) {
568 const env = yield* GcpEnvironment.current;
569 const subscriptionId = yield* toId(
570 id,
571 news.subscriptionId,
572 output?.subscriptionId,
573 );
574 const name = resourceName(env.project, subscriptionId);
575 const topicName = topicNameOf(env.project, news.topic);
576 const desiredLabels = {
577 ...toLabels(news.labels),
578 ...(yield* createInternalLabels(id)),
579 };
580
581 let current =
582 output !== undefined
583 ? yield* waitUntilPresent(name).pipe(
584 Effect.catchTag("GCP.PubSub.SubscriptionNotResolved", () =>
585 Effect.succeed(undefined),
586 ),
587 )
588 : yield* getByName(name);
589
590 if (current === undefined) {
591 const created = yield* pubsub
592 .createProjectsSubscriptions({
593 name,
594 body: toCreateBody(topicName, news, desiredLabels),
595 })
596 .pipe(Effect.catchTag("Conflict", () => waitUntilPresent(name)));
597 // Block until Pub/Sub serves the subscription: the create response
598 // alone does not mean a GET (or a publish routed to it) sees it yet.
599 current =
600 created === undefined ? undefined : yield* waitUntilPresent(name);
601 }
602
603 if (current === undefined) {
604 return yield* new SubscriptionNotResolved({ name });
605 }
606
607 const observedLabels = tagRecord(current.labels);
608 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
609 const labelsChanged = upsert.length > 0 || removed.length > 0;
610 const desiredAck =
611 news.ackDeadlineSeconds ??
612 current.ackDeadlineSeconds ??
613 DEFAULT_ACK_DEADLINE_SECONDS;
614 const ackChanged =
615 desiredAck !==
616 (current.ackDeadlineSeconds ?? DEFAULT_ACK_DEADLINE_SECONDS);
617 const desiredRetain =
618 news.retainAckedMessages ?? current.retainAckedMessages ?? false;
619 const retainChanged =
620 desiredRetain !== (current.retainAckedMessages === true);
621 const desiredExactlyOnce =
622 news.enableExactlyOnceDelivery ??
623 current.enableExactlyOnceDelivery ??
624 false;
625 const exactlyOnceChanged =
626 desiredExactlyOnce !== (current.enableExactlyOnceDelivery === true);
627 const retentionChanged =
628 news.messageRetentionDuration !== undefined &&
629 !sameOptionalString(
630 news.messageRetentionDuration,
631 current.messageRetentionDuration,
632 );
633 const expChanged = expirationChanged(
634 news.expirationPolicy,
635 current.expirationPolicy,
636 );
637 const retriesChanged = retryChanged(
638 news.retryPolicy,
639 current.retryPolicy,
640 );
641 const deadLettersChanged = deadLetterChanged(
642 news.deadLetterPolicy,
643 current.deadLetterPolicy,
644 );
645 const pushConfigChanged = pushChanged(
646 news.pushConfig,
647 current.pushConfig,
648 );
649
650 if (
651 labelsChanged ||
652 ackChanged ||
653 retainChanged ||
654 exactlyOnceChanged ||
655 retentionChanged ||
656 expChanged ||
657 retriesChanged ||
658 deadLettersChanged ||
659 pushConfigChanged
660 ) {
661 current = yield* pubsub
662 .patchProjectsSubscriptions({
663 name,
664 body: {
665 subscription: {
666 name,
667 topic: topicName,
668 labels: desiredLabels,
669 ackDeadlineSeconds: desiredAck,
670 retainAckedMessages: desiredRetain,
671 enableExactlyOnceDelivery: desiredExactlyOnce,
672 messageRetentionDuration:
673 news.messageRetentionDuration ??
674 current.messageRetentionDuration,
675 expirationPolicy:
676 news.expirationPolicy ?? current.expirationPolicy,
677 retryPolicy: news.retryPolicy ?? current.retryPolicy,
678 deadLetterPolicy:
679 news.deadLetterPolicy ?? current.deadLetterPolicy,
680 pushConfig: news.pushConfig ?? current.pushConfig,
681 },
682 updateMask: [
683 labelsChanged ? "labels" : undefined,
684 ackChanged ? "ackDeadlineSeconds" : undefined,
685 retainChanged ? "retainAckedMessages" : undefined,
686 exactlyOnceChanged ? "enableExactlyOnceDelivery" : undefined,
687 retentionChanged ? "messageRetentionDuration" : undefined,
688 expChanged ? "expirationPolicy" : undefined,
689 retriesChanged ? "retryPolicy" : undefined,
690 deadLettersChanged ? "deadLetterPolicy" : undefined,
691 pushConfigChanged ? "pushConfig" : undefined,
692 ]
693 .filter((field): field is string => field !== undefined)
694 .join(","),
695 },
696 })
697 .pipe(
698 Effect.retry({
699 while: (error) => error._tag === "NotFound",
700 times: 8,
701 schedule: Schedule.spaced("1 second"),
702 }),
703 );
704 }
705
706 const deadLetterTopic = current.deadLetterPolicy?.deadLetterTopic;
707 if (deadLetterTopic !== undefined && deadLetterTopic !== "") {
708 yield* grantDeadLetterAgent(env.project, name, deadLetterTopic);
709 }
710
711 return toAttrs(current, env.project);
712 }),
713
714 delete: Effect.fn(function* ({ output }) {
715 yield* pubsub
716 .deleteProjectsSubscriptions({ subscription: output.name })
717 .pipe(Effect.catchTag("NotFound", () => Effect.void));
718 // Pub/Sub reads lag deletes; block until it stops serving it.
719 yield* waitUntilGone(output.name);
720 }),
721 });