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 {
14 createInternalLabels,
15 diffLabels,
16 hasAlchemyLabels,
17 stripInternalLabels,
18 toLabels,
19} from "../Labels.ts";
20import type { Providers } from "../Providers.ts";
21
22export type SnapshotProps = {
23 /**
24 * Snapshot id (the `{snapshot}` segment of
25 * `projects/{project}/snapshots/{snapshot}`). If omitted, a unique name
26 * is generated from the stack, stage, and logical id. Changing it
27 * replaces the snapshot.
28 */
29 snapshotId?: string;
30 /**
31 * Subscription whose backlog this snapshot retains. Accepts a full
32 * resource name (`projects/{project}/subscriptions/{subscription}`) or
33 * a subscription id. Immutable — changing it replaces the snapshot.
34 */
35 subscription: string;
36 /**
37 * User labels. Alchemy ownership labels are merged in automatically.
38 */
39 labels?: Record<string, string>;
40};
41
42export type Snapshot = Resource<
43 "GCP.PubSub.Snapshot",
44 SnapshotProps,
45 {
46 /** Full resource name `projects/{project}/snapshots/{snapshot}`. */
47 name: string;
48 /** Snapshot id (last path segment). */
49 snapshotId: string;
50 /** Project id. */
51 project: string;
52 /** Topic this snapshot is retaining messages from. */
53 topic: string | undefined;
54 /** RFC3339 time after which the snapshot is no longer guaranteed. */
55 expireTime: string | undefined;
56 /** User labels (Alchemy ownership labels stripped). */
57 labels: Record<string, string>;
58 },
59 never,
60 Providers
61>;
62
63/**
64 * A Google Cloud Pub/Sub snapshot of a subscription's message backlog.
65 *
66 * Snapshots capture unacked messages at creation time (and subsequent
67 * publishes to the topic) so a subscription can later `Seek` back to that
68 * point. The source `subscription` is immutable — changing it replaces the
69 * snapshot. Labels can be updated in place.
70 *
71 * ### Creating a Snapshot
72 * **Example:** Snapshot of a pull subscription
73 * ```typescript
74 * const topic = yield* GCP.PubSub.Topic("events", {});
75 * const subscription = yield* GCP.PubSub.Subscription("orders", {
76 * topic: topic.name,
77 * });
78 * const snapshot = yield* GCP.PubSub.Snapshot("checkpoint", {
79 * subscription: subscription.name,
80 * });
81 * ```
82 *
83 * **Example:** Explicit id and labels
84 * ```typescript
85 * const topic = yield* GCP.PubSub.Topic("events", {});
86 * const subscription = yield* GCP.PubSub.Subscription("orders", {
87 * topic: topic.name,
88 * });
89 * const snapshot = yield* GCP.PubSub.Snapshot("checkpoint", {
90 * snapshotId: "order-checkpoint",
91 * subscription: subscription.name,
92 * labels: { env: "prod" },
93 * });
94 * ```
95 *
96 * @resource
97 * @category PubSub
98 */
99export const Snapshot = Resource<Snapshot>("GCP.PubSub.Snapshot");
100
101export class SnapshotNotResolved extends Data.TaggedError(
102 "GCP.PubSub.SnapshotNotResolved",
103)<{
104 name: string;
105}> {}
106
107export class SnapshotStillExists extends Data.TaggedError(
108 "GCP.PubSub.SnapshotStillExists",
109)<{
110 name: string;
111}> {}
112
113const snapshotIdOf = (name: string) => name.split("/").pop() ?? name;
114
115const resourceName = (project: string, snapshotId: string) =>
116 `projects/${project}/snapshots/${snapshotId}`;
117
118const subscriptionNameOf = (project: string, subscription: string) =>
119 subscription.startsWith("projects/")
120 ? subscription
121 : `projects/${project}/subscriptions/${subscription}`;
122
123const resourceKey = (value: string | undefined) =>
124 value === undefined ? undefined : snapshotIdOf(value);
125
126const userLabels = (
127 labels: Record<string, string | undefined> | null | undefined,
128): Record<string, string> => stripInternalLabels(tagRecord(labels));
129
130const toId = (id: string, snapshotId: string | undefined, existing?: string) =>
131 Effect.gen(function* () {
132 if (snapshotId !== undefined) return snapshotId;
133 if (existing !== undefined) return existing;
134 const generated = yield* createPhysicalName({
135 id,
136 maxLength: 255,
137 lowercase: true,
138 forbiddenPrefixes: ["goog"],
139 });
140 return /^[a-z]/.test(generated) ? generated : `s${generated}`.slice(0, 255);
141 });
142
143const toAttrs = (snapshot: pubsub.Snapshot, project: string) => {
144 const name = snapshot.name ?? "";
145 return {
146 name,
147 snapshotId: snapshotIdOf(name),
148 project,
149 topic: snapshot.topic,
150 expireTime: snapshot.expireTime,
151 labels: userLabels(snapshot.labels),
152 };
153};
154
155const getByName = (name: string) =>
156 pubsub
157 .getProjectsSnapshots({ snapshot: name })
158 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
159
160const waitUntilGone = (name: string) =>
161 getByName(name).pipe(
162 Effect.filterOrFail(
163 (existing): existing is undefined => existing === undefined,
164 () => new SnapshotStillExists({ name }),
165 ),
166 Effect.asVoid,
167 Effect.retry({
168 while: (error) => error._tag === "GCP.PubSub.SnapshotStillExists",
169 // Pub/Sub reads lag deletes; allow up to a minute.
170 schedule: Schedule.spaced("2 seconds"),
171 times: 30,
172 }),
173 );
174
175const CONSISTENT_READS = 3;
176
177const waitUntilPresent = (name: string) =>
178 Effect.gen(function* () {
179 let found: pubsub.Snapshot | undefined;
180 for (let read = 0; read < CONSISTENT_READS; read++) {
181 if (read > 0) yield* Effect.sleep("1 second");
182 found = yield* getByName(name);
183 if (found === undefined) return yield* new SnapshotNotResolved({ name });
184 }
185 return found!;
186 }).pipe(
187 Effect.retry({
188 while: (error) => error._tag === "GCP.PubSub.SnapshotNotResolved",
189 schedule: Schedule.spaced("1 second"),
190 times: 60,
191 }),
192 );
193
194export const SnapshotProvider = () =>
195 Provider.succeed(Snapshot, {
196 stables: ["name", "snapshotId", "project", "topic"],
197
198 diff: Effect.fn(function* ({ news, olds, output }) {
199 if (!isResolved(news)) return undefined;
200 const previousId = olds?.snapshotId ?? output?.snapshotId;
201 const nameChanged =
202 news.snapshotId !== undefined &&
203 previousId !== undefined &&
204 news.snapshotId !== previousId;
205 const previousSubscription = olds?.subscription;
206 const subscriptionChanged =
207 previousSubscription !== undefined &&
208 resourceKey(news.subscription) !== resourceKey(previousSubscription);
209 if (!nameChanged && !subscriptionChanged) {
210 return undefined;
211 }
212 return {
213 action: "replace" as const,
214 deleteFirst:
215 !nameChanged &&
216 previousId !== undefined &&
217 (news.snapshotId ?? previousId) === previousId,
218 };
219 }),
220
221 read: Effect.fn(function* ({ id, olds, output }) {
222 const env = yield* GcpEnvironment.current;
223 const snapshotId = yield* toId(id, olds?.snapshotId, output?.snapshotId);
224 const name = output?.name ?? resourceName(env.project, snapshotId);
225 const existing = yield* getByName(name);
226 if (existing === undefined) return undefined;
227 const attrs = toAttrs(existing, env.project);
228 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
229 ? attrs
230 : Unowned(attrs);
231 }),
232
233 list: () =>
234 Effect.gen(function* () {
235 const env = yield* GcpEnvironment.current;
236 return yield* pubsub.listProjectsSnapshots
237 .pages({
238 project: `projects/${env.project}`,
239 pageSize: 1000,
240 })
241 .pipe(
242 Stream.flatMap((page) => Stream.fromIterable(page.snapshots ?? [])),
243 Stream.filter((snapshot) =>
244 Object.keys(snapshot.labels ?? {}).some((key) =>
245 key.startsWith("alchemy-"),
246 ),
247 ),
248 Stream.map((snapshot) => toAttrs(snapshot, env.project)),
249 Stream.runCollect,
250 Effect.map((chunk) => Array.from(chunk)),
251 );
252 }),
253
254 reconcile: Effect.fn(function* ({ id, news, olds, output }) {
255 const env = yield* GcpEnvironment.current;
256 const snapshotId = yield* toId(id, news.snapshotId, output?.snapshotId);
257 const name = resourceName(env.project, snapshotId);
258 const subscriptionName = subscriptionNameOf(
259 env.project,
260 news.subscription,
261 );
262 const desiredLabels = {
263 ...toLabels(news.labels),
264 ...(yield* createInternalLabels(id)),
265 };
266 // Snapshot GET does not return the source subscription. Recreate when
267 // persisted props show it changed — the field is immutable.
268 const subscriptionChanged =
269 olds?.subscription !== undefined &&
270 resourceKey(news.subscription) !== resourceKey(olds.subscription);
271
272 let current = yield* getByName(name);
273 if (current !== undefined && subscriptionChanged) {
274 yield* pubsub
275 .deleteProjectsSnapshots({ snapshot: name })
276 .pipe(Effect.catchTag("NotFound", () => Effect.void));
277 yield* waitUntilGone(name);
278 current = undefined;
279 }
280
281 if (current === undefined) {
282 yield* pubsub
283 .createProjectsSnapshots({
284 name,
285 body: {
286 subscription: subscriptionName,
287 labels: desiredLabels,
288 },
289 })
290 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
291 // Block until Pub/Sub consistently serves the snapshot; the next
292 // call (label patch) can otherwise hit a lagging replica.
293 current = yield* waitUntilPresent(name);
294 }
295
296 if (current === undefined) {
297 return yield* new SnapshotNotResolved({ name });
298 }
299
300 const observedLabels = tagRecord(current.labels);
301 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
302 if (upsert.length > 0 || removed.length > 0) {
303 current = yield* pubsub.patchProjectsSnapshots({
304 name,
305 body: {
306 snapshot: {
307 name,
308 labels: desiredLabels,
309 },
310 updateMask: "labels",
311 },
312 });
313 }
314
315 return toAttrs(current, env.project);
316 }),
317
318 delete: Effect.fn(function* ({ output }) {
319 yield* pubsub
320 .deleteProjectsSnapshots({ snapshot: output.name })
321 .pipe(Effect.catchTag("NotFound", () => Effect.void));
322 yield* waitUntilGone(output.name);
323 }),
324 });