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 { createPhysicalName } from "../../PhysicalName.ts";
8import { isResolved } from "../../Diff.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { tagRecord } from "../../Tags.ts";
12import { GcpEnvironment } from "../Environment.ts";
13import {
14 createInternalLabels,
15 diffLabels,
16 hasAlchemyLabels,
17 stripInternalLabels,
18 toLabels,
19} from "../Labels.ts";
20import type { Providers } from "../Providers.ts";
21
22export type TopicProps = {
23 /**
24 * Topic id (the `{topic}` segment of `projects/{project}/topics/{topic}`).
25 * If omitted, a unique name is generated from the stack, stage, and
26 * logical id.
27 */
28 topicId?: string;
29 /**
30 * User labels. Alchemy ownership labels are merged in automatically.
31 */
32 labels?: Record<string, string>;
33 /**
34 * Cloud KMS key used to encrypt messages, as
35 * `projects/{project}/locations/{location}/keyRings/{keyRing}/cryptoKeys/{cryptoKey}`.
36 */
37 kmsKeyName?: string;
38 /**
39 * Minimum duration to retain published messages (e.g. `"86400s"`).
40 */
41 messageRetentionDuration?: string;
42};
43
44export type Topic = Resource<
45 "GCP.PubSub.Topic",
46 TopicProps,
47 {
48 /** Full resource name `projects/{project}/topics/{topic}`. */
49 name: string;
50 /** Topic id (last path segment). */
51 topicId: string;
52 /** Project id. */
53 project: string;
54 /** User labels (Alchemy ownership labels stripped). */
55 labels: Record<string, string>;
56 /** KMS key used for encryption, if any. */
57 kmsKeyName: string | undefined;
58 /** Message retention duration, if set. */
59 messageRetentionDuration: string | undefined;
60 /** Server-reported topic state. */
61 state: string | undefined;
62 },
63 never,
64 Providers
65>;
66
67/**
68 * A Google Cloud Pub/Sub topic.
69 *
70 * ### Creating a Topic
71 * **Example:** Generated name
72 * ```typescript
73 * const topic = yield* GCP.PubSub.Topic("events", {});
74 * ```
75 *
76 * **Example:** Explicit id and labels
77 * ```typescript
78 * const topic = yield* GCP.PubSub.Topic("events", {
79 * topicId: "order-events",
80 * labels: { env: "prod" },
81 * });
82 * ```
83 *
84 * ### Updating a Topic
85 * **Example:** Change labels
86 * ```typescript
87 * const topic = yield* GCP.PubSub.Topic("events", {
88 * labels: { env: "prod" },
89 * });
90 * ```
91 *
92 * ### Binding from a Function
93 * **Example:** Publish from Cloud Run
94 * ```typescript
95 * export class Api extends GCP.Function<Api>()(
96 * "Api",
97 * { main: import.meta.url },
98 * Effect.gen(function* () {
99 * const topic = yield* GCP.PubSub.Topic("events", {});
100 * const publish = yield* GCP.PubSub.Publish(topic);
101 * return {
102 * fetch: Effect.gen(function* () {
103 * yield* publish({
104 * body: { messages: [{ data: btoa("hello") }] },
105 * }).pipe(Effect.orDie);
106 * return HttpServerResponse.text("ok");
107 * }),
108 * };
109 * }).pipe(Effect.provide([GCP.PubSub.PublishHttp])),
110 * ) {}
111 * ```
112 *
113 * ### Destroying a Topic
114 * **Example:** `alchemy destroy` deletes the topic after its subscriptions.
115 *
116 * @resource
117 * @category PubSub
118 */
119export const Topic = Resource<Topic>("GCP.PubSub.Topic");
120
121export class TopicStillExists extends Data.TaggedError(
122 "GCP.PubSub.TopicStillExists",
123)<{ name: string }> {}
124
125export class TopicNotResolved extends Data.TaggedError(
126 "GCP.PubSub.TopicNotResolved",
127)<{
128 name: string;
129}> {}
130
131const topicIdOf = (name: string) => name.split("/").pop() ?? name;
132
133const resourceName = (project: string, topicId: string) =>
134 `projects/${project}/topics/${topicId}`;
135
136const userLabels = (
137 labels: Record<string, string | undefined> | null | undefined,
138): Record<string, string> => stripInternalLabels(tagRecord(labels));
139
140const toId = (id: string, topicId: string | undefined, existing?: string) =>
141 Effect.gen(function* () {
142 return (
143 topicId ??
144 existing ??
145 (yield* createPhysicalName({ id, maxLength: 255, lowercase: true }))
146 );
147 });
148
149const toAttrs = (topic: pubsub.Topic, project: string) => {
150 const name = topic.name ?? "";
151 return {
152 name,
153 topicId: topicIdOf(name),
154 project,
155 labels: userLabels(topic.labels),
156 kmsKeyName: topic.kmsKeyName,
157 messageRetentionDuration: topic.messageRetentionDuration,
158 state: topic.state,
159 };
160};
161
162const getByName = (name: string) =>
163 pubsub
164 .getProjectsTopics({ topic: name })
165 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
166
167// Pub/Sub serves reads from replicas that lag writes: one successful GET
168// does not mean the next one (or a publish routed to it) sees the
169// resource. Require consecutive successful reads before reporting ready.
170const CONSISTENT_READS = 3;
171
172const waitUntilPresent = (name: string) =>
173 Effect.gen(function* () {
174 let found: pubsub.Topic | undefined;
175 for (let read = 0; read < CONSISTENT_READS; read++) {
176 if (read > 0) yield* Effect.sleep("1 second");
177 found = yield* getByName(name);
178 if (found === undefined) {
179 return yield* new TopicNotResolved({ name });
180 }
181 }
182 return found!;
183 }).pipe(
184 Effect.retry({
185 while: (error) => error._tag === "GCP.PubSub.TopicNotResolved",
186 schedule: Schedule.spaced("1 second"),
187 times: 60,
188 }),
189 );
190
191export const TopicProvider = () =>
192 Provider.succeed(Topic, {
193 stables: ["name", "topicId", "project"],
194
195 // topicId is the physical identity: a new id is a new topic, and the
196 // old one must be deleted (not left behind by an in-place "update").
197 diff: Effect.fn(function* ({ id, news, olds, output }) {
198 if (!isResolved(news)) return undefined;
199 const previous = output?.topicId ?? olds?.topicId;
200 if (previous === undefined) return undefined;
201 const next = yield* toId(id, news.topicId, output?.topicId);
202 return next !== previous
203 ? { action: "replace" as const, deleteFirst: false }
204 : undefined;
205 }),
206
207 read: Effect.fn(function* ({ id, olds, output }) {
208 const env = yield* GcpEnvironment.current;
209 const topicId = yield* toId(id, olds?.topicId, output?.topicId);
210 const name = output?.name ?? resourceName(env.project, topicId);
211 const existing = yield* getByName(name);
212 if (existing === undefined) return undefined;
213 const attrs = toAttrs(existing, env.project);
214 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
215 ? attrs
216 : Unowned(attrs);
217 }),
218
219 list: () =>
220 Effect.gen(function* () {
221 const env = yield* GcpEnvironment.current;
222 return yield* pubsub.listProjectsTopics
223 .pages({
224 project: `projects/${env.project}`,
225 pageSize: 1000,
226 })
227 .pipe(
228 Stream.flatMap((page) => Stream.fromIterable(page.topics ?? [])),
229 Stream.filter((topic) =>
230 Object.keys(topic.labels ?? {}).some((key) =>
231 key.startsWith("alchemy-"),
232 ),
233 ),
234 Stream.map((topic) => toAttrs(topic, env.project)),
235 Stream.runCollect,
236 Effect.map((chunk) => Array.from(chunk)),
237 );
238 }),
239
240 reconcile: Effect.fn(function* ({ id, news, output }) {
241 const env = yield* GcpEnvironment.current;
242 const topicId = yield* toId(id, news.topicId, output?.topicId);
243 const name = resourceName(env.project, topicId);
244 const desiredLabels = {
245 ...toLabels(news.labels),
246 ...(yield* createInternalLabels(id)),
247 };
248
249 let current = yield* getByName(name);
250
251 if (current === undefined) {
252 yield* pubsub
253 .createProjectsTopics({
254 name,
255 body: {
256 labels: desiredLabels,
257 kmsKeyName: news.kmsKeyName,
258 messageRetentionDuration: news.messageRetentionDuration,
259 },
260 })
261 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
262 // Block until Pub/Sub serves the topic, not just the create response.
263 current = yield* waitUntilPresent(name);
264 }
265
266 if (current === undefined) {
267 return yield* new TopicNotResolved({ name });
268 }
269
270 const observedLabels = tagRecord(current.labels);
271 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
272 const labelsChanged = upsert.length > 0 || removed.length > 0;
273 const kmsChanged = (current.kmsKeyName ?? "") !== (news.kmsKeyName ?? "");
274 const retentionChanged =
275 (current.messageRetentionDuration ?? "") !==
276 (news.messageRetentionDuration ?? "");
277
278 if (labelsChanged || kmsChanged || retentionChanged) {
279 current = yield* pubsub.patchProjectsTopics({
280 name,
281 body: {
282 topic: {
283 name,
284 labels: desiredLabels,
285 kmsKeyName: news.kmsKeyName,
286 messageRetentionDuration: news.messageRetentionDuration,
287 },
288 updateMask: [
289 labelsChanged ? "labels" : undefined,
290 kmsChanged ? "kmsKeyName" : undefined,
291 retentionChanged ? "messageRetentionDuration" : undefined,
292 ]
293 .filter((field): field is string => field !== undefined)
294 .join(","),
295 },
296 });
297 }
298
299 return toAttrs(current, env.project);
300 }),
301
302 delete: Effect.fn(function* ({ output }) {
303 yield* pubsub
304 .deleteProjectsTopics({ topic: output.name })
305 .pipe(Effect.catchTag("NotFound", () => Effect.void));
306 // Pub/Sub reads lag deletes; block until it stops serving it.
307 yield* getByName(output.name).pipe(
308 Effect.filterOrFail(
309 (existing) => existing === undefined,
310 () => new TopicStillExists({ name: output.name }),
311 ),
312 Effect.retry({
313 while: (error) => error._tag === "GCP.PubSub.TopicStillExists",
314 schedule: Schedule.spaced("2 seconds"),
315 times: 30,
316 }),
317 );
318 }),
319 });