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_PARTITION_COUNT,
11 DEFAULT_PER_PARTITION_BYTES,
12 DEFAULT_PUBLISH_MIB,
13 DEFAULT_SUBSCRIBE_MIB,
14 DEFAULT_ZONE,
15 ResourceNotResolved,
16 countOf,
17 expandName,
18 fieldMask,
19 getTopic,
20 hasOwnershipMarker,
21 ignoreMissing,
22 listOwnedTopics,
23 normalizeLocation,
24 ownedByAlchemy,
25 parentOf,
26 parseName,
27 partitionBody,
28 partitionKey,
29 regionOf,
30 replaceOnIdentity,
31 resourceName,
32 retentionBody,
33 retentionKey,
34 retryInUse,
35 sameRef,
36 toResourceId,
37 waitUntilGone,
38} from "./internal.ts";
39
40export type TopicPartitionCapacity = {
41 /**
42 * Publish throughput per partition in MiB/s. Must be between 4 and 16.
43 * @default 4
44 */
45 publishMibPerSec?: number;
46 /**
47 * Subscribe throughput per partition in MiB/s. Must be between 4 and 32.
48 * @default 4
49 */
50 subscribeMibPerSec?: number;
51};
52
53export type TopicPartitionConfig = {
54 /**
55 * Number of partitions. Must be at least 1. After create, the count
56 * can increase but not decrease — shrinking replaces the topic.
57 * @default 1
58 */
59 count?: number | string;
60 /**
61 * Per-partition throughput. Required unless the topic uses a
62 * reservation that supplies capacity.
63 */
64 capacity?: TopicPartitionCapacity;
65};
66
67export type TopicRetentionConfig = {
68 /**
69 * Provisioned storage per partition, in bytes. The API minimum is
70 * 30 GiB (`32212254720`). Older messages are dropped when a partition
71 * exceeds this size, regardless of `period`.
72 * @default "32212254720"
73 */
74 perPartitionBytes?: string;
75 /**
76 * How long a published message is retained (e.g. `"86400s"`). If
77 * omitted, messages are retained until `perPartitionBytes` is
78 * exhausted.
79 */
80 period?: string;
81};
82
83export type TopicReservationConfig = {
84 /**
85 * Reservation that supplies this topic's throughput, as a full
86 * resource name or reservation id. Regional topics require a
87 * reservation in the same region.
88 */
89 throughputReservation?: string;
90};
91
92export type AdminTopicProps = {
93 /**
94 * Topic id (the `{topic}` segment of
95 * `projects/{project}/locations/{location}/topics/{topic}`). If
96 * omitted, a unique name is generated from the stack, stage, and
97 * logical id. Pub/Sub Lite has no labels field, so Alchemy stamps
98 * ownership into a `+alc.{stack}.{stage}.{id}` suffix. Immutable —
99 * changing it replaces the topic.
100 */
101 topicId?: string;
102 /**
103 * Zone (`us-central1-a`) or region (`us-central1`). Zonal topics can
104 * provision their own partition capacity. Regional topics must attach
105 * a reservation. Immutable — changing it replaces the topic.
106 * @default "us-central1-a"
107 */
108 location?: string;
109 /**
110 * Partition count and per-partition throughput.
111 */
112 partitionConfig?: TopicPartitionConfig;
113 /**
114 * Message retention. `perPartitionBytes` defaults to 30 GiB.
115 */
116 retentionConfig?: TopicRetentionConfig;
117 /**
118 * Optional reservation that this topic draws throughput from.
119 */
120 reservationConfig?: TopicReservationConfig;
121};
122
123export type AdminTopic = Resource<
124 "GCP.PubSubLite.AdminTopic",
125 AdminTopicProps,
126 {
127 /** Full resource name `projects/{project}/locations/{location}/topics/{topic}`. */
128 name: string;
129 /** Topic id (last path segment, including the Alchemy ownership suffix). */
130 topicId: string;
131 /** Project id. */
132 project: string;
133 /** Zone or region id. */
134 location: string;
135 /** Partition configuration currently applied. */
136 partitionConfig: pubsublite.PartitionConfig | undefined;
137 /** Retention configuration currently applied. */
138 retentionConfig: pubsublite.RetentionConfig | undefined;
139 /** Reservation configuration currently applied. */
140 reservationConfig: pubsublite.ReservationConfig | undefined;
141 },
142 never,
143 Providers
144>;
145
146/**
147 * A Pub/Sub Lite topic — a zonal or regional partitioned log.
148 *
149 * Pub/Sub Lite topics have no labels. Alchemy stamps
150 * `alchemy-stack` / `alchemy-stage` / `alchemy-id` into the topic id so
151 * `list` / `pnpm nuke:gcp` can find them. `topicId` and `location`
152 * replace the topic. Partition count can only increase in place;
153 * shrinking it replaces the topic. Capacity, retention, and reservation
154 * attachment update in place.
155 *
156 * ### Creating a Topic
157 * **Example:** Zonal topic with generated name
158 * ```typescript
159 * const topic = yield* GCP.PubSubLite.AdminTopic("Events", {});
160 * ```
161 *
162 * **Example:** Explicit capacity and retention
163 * ```typescript
164 * const topic = yield* GCP.PubSubLite.AdminTopic("Events", {
165 * location: "us-central1-a",
166 * partitionConfig: {
167 * count: 1,
168 * capacity: { publishMibPerSec: 4, subscribeMibPerSec: 8 },
169 * },
170 * retentionConfig: { perPartitionBytes: "32212254720", period: "86400s" },
171 * });
172 * ```
173 *
174 * ### Using a Reservation
175 * **Example:** Regional topic attached to a reservation
176 * ```typescript
177 * const reservation = yield* GCP.PubSubLite.AdminReservation("Capacity", {
178 * location: "us-central1",
179 * throughputCapacity: "4",
180 * });
181 * const topic = yield* GCP.PubSubLite.AdminTopic("Events", {
182 * location: "us-central1",
183 * reservationConfig: { throughputReservation: reservation.name },
184 * });
185 * ```
186 *
187 * @resource
188 * @category PubSubLite
189 */
190export const AdminTopic = Resource<AdminTopic>("GCP.PubSubLite.AdminTopic");
191
192const COLLECTION = "reservations";
193const TOPIC_COLLECTION = "topics";
194
195const desiredPartition = (news: AdminTopicProps): pubsublite.PartitionConfig =>
196 partitionBody({
197 count: news.partitionConfig?.count ?? DEFAULT_PARTITION_COUNT,
198 capacity: {
199 publishMibPerSec:
200 news.partitionConfig?.capacity?.publishMibPerSec ?? DEFAULT_PUBLISH_MIB,
201 subscribeMibPerSec:
202 news.partitionConfig?.capacity?.subscribeMibPerSec ??
203 DEFAULT_SUBSCRIBE_MIB,
204 },
205 });
206
207const desiredRetention = (news: AdminTopicProps): pubsublite.RetentionConfig =>
208 retentionBody({
209 perPartitionBytes:
210 news.retentionConfig?.perPartitionBytes ?? DEFAULT_PER_PARTITION_BYTES,
211 period: news.retentionConfig?.period,
212 });
213
214const desiredReservation = (
215 news: AdminTopicProps,
216 project: string,
217 location: string,
218): pubsublite.ReservationConfig | undefined => {
219 const reservation = news.reservationConfig?.throughputReservation;
220 if (reservation === undefined || reservation.length === 0) {
221 return news.reservationConfig === undefined
222 ? undefined
223 : { throughputReservation: undefined };
224 }
225 return {
226 throughputReservation: expandName(
227 reservation,
228 project,
229 regionOf(location),
230 COLLECTION,
231 ),
232 };
233};
234
235const toAttrs = (
236 topic: pubsublite.Topic,
237 project: string,
238): AdminTopic["Attributes"] => {
239 const name = topic.name ?? "";
240 const parsed = parseName(name, TOPIC_COLLECTION);
241 return {
242 name,
243 topicId: parsed.id,
244 project: parsed.project || project,
245 location: parsed.location || DEFAULT_ZONE,
246 partitionConfig: topic.partitionConfig,
247 retentionConfig: topic.retentionConfig,
248 reservationConfig: topic.reservationConfig,
249 };
250};
251
252export const AdminTopicProvider = () =>
253 Provider.succeed(AdminTopic, {
254 stables: ["name", "topicId", "project", "location"],
255
256 diff: Effect.fn(function* ({ news, olds, output }) {
257 if (!isResolved(news)) return undefined;
258 const previousCount = countOf(
259 olds?.partitionConfig?.count ?? output?.partitionConfig?.count,
260 );
261 const nextCount = countOf(
262 news.partitionConfig?.count ??
263 olds?.partitionConfig?.count ??
264 output?.partitionConfig?.count,
265 );
266 return replaceOnIdentity({
267 previousId: olds?.topicId ?? output?.topicId,
268 nextId: news.topicId ?? olds?.topicId ?? output?.topicId,
269 previousLocation: normalizeLocation(
270 olds?.location ?? output?.location,
271 DEFAULT_ZONE,
272 ),
273 nextLocation: normalizeLocation(
274 news.location ?? olds?.location ?? output?.location,
275 DEFAULT_ZONE,
276 ),
277 extra: nextCount < previousCount,
278 });
279 }),
280
281 read: Effect.fn(function* ({ id, olds, output }) {
282 const env = yield* GcpEnvironment.current;
283 const topicId = yield* toResourceId(id, olds?.topicId, output?.topicId);
284 const location = normalizeLocation(
285 olds?.location ?? output?.location,
286 DEFAULT_ZONE,
287 );
288 const name =
289 output?.name ??
290 resourceName(env.project, location, TOPIC_COLLECTION, topicId);
291 const existing = yield* getTopic(name);
292 if (existing === undefined) return undefined;
293 const attrs = toAttrs(existing, env.project);
294 const owned = yield* ownedByAlchemy(id, attrs.topicId);
295 if (owned) return attrs;
296 return hasOwnershipMarker(attrs.topicId) ? undefined : Unowned(attrs);
297 }),
298
299 list: () =>
300 Effect.gen(function* () {
301 const env = yield* GcpEnvironment.current;
302 const items = yield* listOwnedTopics();
303 return items.map((item) => toAttrs(item, env.project));
304 }),
305
306 reconcile: Effect.fn(function* ({ id, news, output }) {
307 const env = yield* GcpEnvironment.current;
308 const topicId = yield* toResourceId(id, news.topicId, output?.topicId);
309 const location = normalizeLocation(
310 news.location ?? output?.location,
311 DEFAULT_ZONE,
312 );
313 const name = resourceName(
314 env.project,
315 location,
316 TOPIC_COLLECTION,
317 topicId,
318 );
319 const partitionConfig = desiredPartition(news);
320 const retentionConfig = desiredRetention(news);
321 const reservationConfig = desiredReservation(news, env.project, location);
322
323 let current = yield* getTopic(output?.name ?? name);
324
325 if (current === undefined) {
326 const body: pubsublite.Topic = {
327 partitionConfig,
328 retentionConfig,
329 };
330 if (reservationConfig?.throughputReservation) {
331 body.reservationConfig = reservationConfig;
332 }
333 const created = yield* pubsublite
334 .createAdminProjectsLocationsTopics({
335 parent: parentOf(env.project, location),
336 topicId,
337 body,
338 })
339 .pipe(Effect.catchTag("Conflict", () => getTopic(name)));
340 current = created ?? undefined;
341 }
342
343 if (current === undefined) {
344 return yield* new ResourceNotResolved({ name });
345 }
346
347 const observedName = current.name ?? name;
348 const reservationChanged =
349 reservationConfig !== undefined &&
350 !sameRef(
351 current.reservationConfig?.throughputReservation,
352 reservationConfig.throughputReservation,
353 );
354 const mask = fieldMask([
355 partitionKey(current.partitionConfig) !==
356 partitionKey(partitionConfig) && "partitionConfig",
357 retentionKey(current.retentionConfig) !==
358 retentionKey(retentionConfig) && "retentionConfig",
359 reservationChanged && "reservationConfig",
360 ]);
361 if (mask.length > 0) {
362 const body: pubsublite.Topic = {
363 name: observedName,
364 partitionConfig,
365 retentionConfig,
366 };
367 if (reservationConfig !== undefined) {
368 body.reservationConfig = reservationConfig;
369 }
370 current = yield* pubsublite.patchAdminProjectsLocationsTopics({
371 name: observedName,
372 updateMask: mask,
373 body,
374 });
375 }
376
377 return toAttrs(current, env.project);
378 }),
379
380 delete: Effect.fn(function* ({ output }) {
381 yield* retryInUse(
382 ignoreMissing(
383 pubsublite.deleteAdminProjectsLocationsTopics({
384 name: output.name,
385 }),
386 ),
387 );
388 yield* waitUntilGone(getTopic(output.name));
389 }),
390 });