1import * as pubsublite from "@distilled.cloud/gcp/pubsublite_v1";
2import * as Effect from "effect/Effect";
3import { Unowned } from "../../AdoptPolicy.ts";
4import { isResolved } from "../../Diff.ts";
5import * as Provider from "../../Provider.ts";
6import { Resource } from "../../Resource.ts";
7import { GcpEnvironment } from "../Environment.ts";
8import type { Providers } from "../Providers.ts";
9import {
10 DEFAULT_ZONE,
11 ResourceNotResolved,
12 expandName,
13 fieldMask,
14 getSubscription,
15 hasOwnershipMarker,
16 ignoreMissing,
17 jsonEqual,
18 lastSegment,
19 listOwnedSubscriptions,
20 normalizeLocation,
21 ownedByAlchemy,
22 parentOf,
23 parseName,
24 replaceOnIdentity,
25 resourceName,
26 retryInUse,
27 sameRef,
28 sameText,
29 toResourceId,
30 waitUntilGone,
31} from "./internal.ts";
32
33export type SubscriptionDeliveryConfig = {
34 /**
35 * When messages become available to subscribers.
36 * `DELIVER_IMMEDIATELY` or `DELIVER_AFTER_STORED`.
37 */
38 deliveryRequirement?:
39 | pubsublite.DeliveryConfigDeliveryRequirementEnum
40 | (string & {});
41};
42
43export type SubscriptionPubSubConfig = {
44 /**
45 * Destination Pub/Sub topic
46 * (`projects/{project}/topics/{topic}`).
47 */
48 topic?: string;
49};
50
51export type SubscriptionExportConfig = {
52 /**
53 * Desired export state. Only `ACTIVE` and `PAUSED` are settable.
54 */
55 desiredState?: pubsublite.ExportConfigDesiredStateEnum | (string & {});
56 /**
57 * Optional Pub/Sub Lite topic that receives messages that cannot be
58 * exported. Must be in the same project and location.
59 */
60 deadLetterTopic?: string;
61 /**
62 * Export messages to a Pub/Sub topic. User subscriber clients must
63 * not connect to an export subscription.
64 */
65 pubsubConfig?: SubscriptionPubSubConfig;
66};
67
68export type AdminSubscriptionProps = {
69 /**
70 * Subscription id (the `{subscription}` segment of
71 * `projects/{project}/locations/{location}/subscriptions/{subscription}`).
72 * If omitted, a unique name is generated from the stack, stage, and
73 * logical id. Pub/Sub Lite has no labels field, so Alchemy stamps
74 * ownership into a `+alc.{stack}.{stage}.{id}` suffix. Immutable —
75 * changing it replaces the subscription.
76 */
77 subscriptionId?: string;
78 /**
79 * Zone or region. Must match the topic. If omitted, inferred from
80 * `topic` when it is a full resource name, otherwise `us-central1-a`.
81 * Immutable — changing it replaces the subscription.
82 */
83 location?: string;
84 /**
85 * Topic this subscription reads from. Accepts a full resource name or
86 * a topic id. Immutable — changing it replaces the subscription.
87 */
88 topic: string;
89 /**
90 * Message delivery settings.
91 */
92 deliveryConfig?: SubscriptionDeliveryConfig;
93 /**
94 * Optional export that writes messages to a Pub/Sub topic. User
95 * subscribers must not attach to an export subscription.
96 */
97 exportConfig?: SubscriptionExportConfig;
98 /**
99 * When true, the new subscription only receives messages published
100 * after it is created. Create-only.
101 * @default false
102 */
103 skipBacklog?: boolean;
104};
105
106export type AdminSubscription = Resource<
107 "GCP.PubSubLite.AdminSubscription",
108 AdminSubscriptionProps,
109 {
110 /** Full resource name `projects/{project}/locations/{location}/subscriptions/{subscription}`. */
111 name: string;
112 /** Subscription id (last path segment, including the Alchemy ownership suffix). */
113 subscriptionId: string;
114 /** Project id. */
115 project: string;
116 /** Zone or region id. */
117 location: string;
118 /** Topic this subscription is attached to. */
119 topic: string | undefined;
120 /** Delivery settings currently applied. */
121 deliveryConfig: pubsublite.DeliveryConfig | undefined;
122 /** Export configuration currently applied. */
123 exportConfig: pubsublite.ExportConfig | undefined;
124 },
125 never,
126 Providers
127>;
128
129/**
130 * A Pub/Sub Lite subscription — a named consumer of a Lite topic.
131 *
132 * Pub/Sub Lite subscriptions have no labels. Alchemy stamps
133 * `alchemy-stack` / `alchemy-stage` / `alchemy-id` into the
134 * subscription id so `list` / `pnpm nuke:gcp` can find them.
135 * `subscriptionId`, `location`, and `topic` replace the subscription.
136 * Delivery and export settings update in place. `skipBacklog` is
137 * create-only.
138 *
139 * ### Creating a Subscription
140 * **Example:** Pull subscription on a topic
141 * ```typescript
142 * const topic = yield* GCP.PubSubLite.AdminTopic("Events", {});
143 * const subscription = yield* GCP.PubSubLite.AdminSubscription("Inbox", {
144 * topic: topic.name,
145 * });
146 * ```
147 *
148 * **Example:** Deliver after stored
149 * ```typescript
150 * const subscription = yield* GCP.PubSubLite.AdminSubscription("Inbox", {
151 * topic: topic.name,
152 * deliveryConfig: { deliveryRequirement: "DELIVER_AFTER_STORED" },
153 * });
154 * ```
155 *
156 * ### Exporting to Pub/Sub
157 * **Example:** Export subscription
158 * ```typescript
159 * const lite = yield* GCP.PubSubLite.AdminTopic("Events", {});
160 * const classic = yield* GCP.PubSub.Topic("Mirror", {});
161 * const exported = yield* GCP.PubSubLite.AdminSubscription("Export", {
162 * topic: lite.name,
163 * exportConfig: {
164 * desiredState: "ACTIVE",
165 * pubsubConfig: { topic: classic.name },
166 * },
167 * });
168 * ```
169 *
170 * @resource
171 * @category PubSubLite
172 */
173export const AdminSubscription = Resource<AdminSubscription>(
174 "GCP.PubSubLite.AdminSubscription",
175);
176
177const COLLECTION = "subscriptions";
178const TOPIC_COLLECTION = "topics";
179
180const locationOfTopic = (topic: string, fallback: string) => {
181 if (!topic.includes("/locations/")) return fallback;
182 return parseName(topic, TOPIC_COLLECTION).location || fallback;
183};
184
185const desiredDelivery = (
186 news: AdminSubscriptionProps,
187): pubsublite.DeliveryConfig | undefined => {
188 if (news.deliveryConfig === undefined) return undefined;
189 return {
190 deliveryRequirement: news.deliveryConfig.deliveryRequirement,
191 };
192};
193
194const desiredExport = (
195 news: AdminSubscriptionProps,
196 project: string,
197 location: string,
198): pubsublite.ExportConfig | undefined => {
199 if (news.exportConfig === undefined) return undefined;
200 const deadLetter =
201 news.exportConfig.deadLetterTopic !== undefined &&
202 news.exportConfig.deadLetterTopic.length > 0
203 ? expandName(
204 news.exportConfig.deadLetterTopic,
205 project,
206 location,
207 TOPIC_COLLECTION,
208 )
209 : news.exportConfig.deadLetterTopic;
210 const pubsubTopic = news.exportConfig.pubsubConfig?.topic;
211 return {
212 desiredState: news.exportConfig.desiredState,
213 deadLetterTopic: deadLetter,
214 pubsubConfig:
215 pubsubTopic === undefined
216 ? news.exportConfig.pubsubConfig
217 : {
218 topic: pubsubTopic.startsWith("projects/")
219 ? pubsubTopic
220 : `projects/${project}/topics/${lastSegment(pubsubTopic)}`,
221 },
222 };
223};
224
225const toAttrs = (
226 subscription: pubsublite.Subscription,
227 project: string,
228): AdminSubscription["Attributes"] => {
229 const name = subscription.name ?? "";
230 const parsed = parseName(name, COLLECTION);
231 return {
232 name,
233 subscriptionId: parsed.id,
234 project: parsed.project || project,
235 location: parsed.location || DEFAULT_ZONE,
236 topic: subscription.topic,
237 deliveryConfig: subscription.deliveryConfig,
238 exportConfig: subscription.exportConfig,
239 };
240};
241
242export const AdminSubscriptionProvider = () =>
243 Provider.succeed(AdminSubscription, {
244 stables: ["name", "subscriptionId", "project", "location", "topic"],
245
246 diff: Effect.fn(function* ({ news, olds, output }) {
247 if (!isResolved(news)) return undefined;
248 const previousTopic = olds?.topic ?? output?.topic;
249 const topicChanged =
250 previousTopic !== undefined && !sameRef(previousTopic, news.topic);
251 const inferred = locationOfTopic(
252 news.topic,
253 normalizeLocation(
254 news.location ?? olds?.location ?? output?.location,
255 DEFAULT_ZONE,
256 ),
257 );
258 return replaceOnIdentity({
259 previousId: olds?.subscriptionId ?? output?.subscriptionId,
260 nextId:
261 news.subscriptionId ?? olds?.subscriptionId ?? output?.subscriptionId,
262 previousLocation: normalizeLocation(
263 olds?.location ?? output?.location,
264 DEFAULT_ZONE,
265 ),
266 nextLocation: normalizeLocation(
267 news.location ?? inferred,
268 DEFAULT_ZONE,
269 ),
270 extra: topicChanged,
271 });
272 }),
273
274 read: Effect.fn(function* ({ id, olds, output }) {
275 const env = yield* GcpEnvironment.current;
276 const subscriptionId = yield* toResourceId(
277 id,
278 olds?.subscriptionId,
279 output?.subscriptionId,
280 );
281 const location = normalizeLocation(
282 olds?.location ??
283 output?.location ??
284 (olds?.topic ? locationOfTopic(olds.topic, DEFAULT_ZONE) : undefined),
285 DEFAULT_ZONE,
286 );
287 const name =
288 output?.name ??
289 resourceName(env.project, location, COLLECTION, subscriptionId);
290 const existing = yield* getSubscription(name);
291 if (existing === undefined) return undefined;
292 const attrs = toAttrs(existing, env.project);
293 const owned = yield* ownedByAlchemy(id, attrs.subscriptionId);
294 if (owned) return attrs;
295 return hasOwnershipMarker(attrs.subscriptionId)
296 ? undefined
297 : Unowned(attrs);
298 }),
299
300 list: () =>
301 Effect.gen(function* () {
302 const env = yield* GcpEnvironment.current;
303 const items = yield* listOwnedSubscriptions();
304 return items.map((item) => toAttrs(item, env.project));
305 }),
306
307 reconcile: Effect.fn(function* ({ id, news, output }) {
308 const env = yield* GcpEnvironment.current;
309 const subscriptionId = yield* toResourceId(
310 id,
311 news.subscriptionId,
312 output?.subscriptionId,
313 );
314 const topicName = expandName(
315 news.topic,
316 env.project,
317 normalizeLocation(
318 news.location ??
319 output?.location ??
320 locationOfTopic(news.topic, DEFAULT_ZONE),
321 DEFAULT_ZONE,
322 ),
323 TOPIC_COLLECTION,
324 );
325 const location = normalizeLocation(
326 news.location ??
327 output?.location ??
328 locationOfTopic(topicName, DEFAULT_ZONE),
329 DEFAULT_ZONE,
330 );
331 const name = resourceName(
332 env.project,
333 location,
334 COLLECTION,
335 subscriptionId,
336 );
337 const deliveryConfig = desiredDelivery(news);
338 const exportConfig = desiredExport(news, env.project, location);
339
340 let current = yield* getSubscription(output?.name ?? name);
341
342 if (current === undefined) {
343 const body: pubsublite.Subscription = {
344 topic: topicName,
345 };
346 if (deliveryConfig !== undefined) body.deliveryConfig = deliveryConfig;
347 if (exportConfig !== undefined) body.exportConfig = exportConfig;
348 const created = yield* pubsublite
349 .createAdminProjectsLocationsSubscriptions({
350 parent: parentOf(env.project, location),
351 subscriptionId,
352 skipBacklog: news.skipBacklog,
353 body,
354 })
355 .pipe(Effect.catchTag("Conflict", () => getSubscription(name)));
356 current = created ?? undefined;
357 }
358
359 if (current === undefined) {
360 return yield* new ResourceNotResolved({ name });
361 }
362
363 const observedName = current.name ?? name;
364 const deliveryChanged =
365 deliveryConfig !== undefined &&
366 !sameText(
367 current.deliveryConfig?.deliveryRequirement,
368 deliveryConfig.deliveryRequirement,
369 );
370 const exportChanged =
371 exportConfig !== undefined &&
372 !jsonEqual(
373 {
374 desiredState: exportConfig.desiredState ?? "",
375 deadLetterTopic: exportConfig.deadLetterTopic ?? "",
376 pubsubTopic: exportConfig.pubsubConfig?.topic ?? "",
377 },
378 {
379 desiredState: current.exportConfig?.desiredState ?? "",
380 deadLetterTopic: current.exportConfig?.deadLetterTopic ?? "",
381 pubsubTopic: current.exportConfig?.pubsubConfig?.topic ?? "",
382 },
383 );
384 const mask = fieldMask([
385 deliveryChanged && "deliveryConfig",
386 exportChanged && "exportConfig",
387 ]);
388 if (mask.length > 0) {
389 const body: pubsublite.Subscription = { name: observedName };
390 if (deliveryConfig !== undefined) body.deliveryConfig = deliveryConfig;
391 if (exportConfig !== undefined) body.exportConfig = exportConfig;
392 current = yield* pubsublite.patchAdminProjectsLocationsSubscriptions({
393 name: observedName,
394 updateMask: mask,
395 body,
396 });
397 }
398
399 return toAttrs(current, env.project);
400 }),
401
402 delete: Effect.fn(function* ({ output }) {
403 yield* retryInUse(
404 ignoreMissing(
405 pubsublite.deleteAdminProjectsLocationsSubscriptions({
406 name: output.name,
407 }),
408 ),
409 );
410 yield* waitUntilGone(getSubscription(output.name));
411 }),
412 });