1import * as bigtable from "@distilled.cloud/gcp/bigtableadmin_v2";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import type { Providers } from "../Providers.ts";
11import {
12 instanceIdOf,
13 instanceNameOf,
14 listAlchemyInstances,
15 materializedViewName,
16 MAX_MATERIALIZED_VIEW_ID_LENGTH,
17 parentOwned,
18 parseResourceName,
19 toPhysicalId,
20 waitForOperation,
21 collectPages,
22} from "./operations.ts";
23
24export type InstancesMaterializedViewProps = {
25 /**
26 * Parent instance. Full name `projects/{project}/instances/{instance}`
27 * or the instance id. Immutable — changing it replaces the view.
28 */
29 instance: string;
30 /**
31 * Materialized view id (the `{materialized_view}` segment of
32 * `.../instances/{instance}/materializedViews/{materialized_view}`).
33 * If omitted, a unique name is generated from the stack, stage, and
34 * logical id. Must be unique among table ids and view ids in the
35 * instance. Immutable — changing it replaces the view.
36 */
37 materializedViewId?: string;
38 /**
39 * GoogleSQL SELECT that defines the continuous materialized view.
40 * Must include `GROUP BY` or `ORDER BY` and cannot use `SELECT *`.
41 * Immutable — changing it replaces the view.
42 */
43 query: string;
44 /**
45 * Protect the view from Admin API deletes. Disabled automatically on
46 * destroy.
47 * @default false
48 */
49 deletionProtection?: boolean;
50};
51
52export type InstancesMaterializedView = Resource<
53 "GCP.Bigtable.InstancesMaterializedView",
54 InstancesMaterializedViewProps,
55 {
56 /** Full resource name `.../instances/{instance}/materializedViews/{materialized_view}`. */
57 name: string;
58 /** Materialized view id (last path segment). */
59 materializedViewId: string;
60 /** Parent instance resource name. */
61 instance: string;
62 /** Parent instance id. */
63 instanceId: string;
64 /** Project id. */
65 project: string;
66 /** Defining SELECT query. */
67 query: string | undefined;
68 /** Whether Admin API deletes are blocked. */
69 deletionProtection: boolean;
70 /** Per-cluster replication state. */
71 clusterStates:
72 | bigtable.GoogleBigtableAdminV2MaterializedViewClusterStateMap
73 | undefined;
74 /** Server etag. */
75 etag: string | undefined;
76 },
77 never,
78 Providers
79>;
80
81/**
82 * A Cloud Bigtable continuous materialized view — a SQL SELECT whose
83 * results are precomputed and kept current.
84 *
85 * The parent instance must already exist. The query must use `GROUP BY`
86 * or `ORDER BY` and cannot use `SELECT *`. Changing `materializedViewId`,
87 * `instance`, or `query` replaces the view. `deletionProtection` updates
88 * in place.
89 *
90 * Materialized views have no labels field. Alchemy treats a view as
91 * owned when its parent instance carries Alchemy labels, so `list` /
92 * `pnpm nuke:gcp` can find it.
93 *
94 * ### Creating a Materialized View
95 * **Example:** Count of rows grouped by a constant key
96 * ```typescript
97 * const events = yield* GCP.Bigtable.Table("Events", {
98 * instance: instance.name,
99 * columnFamilies: { cf: {} },
100 * });
101 * const view = yield* GCP.Bigtable.InstancesMaterializedView("Counts", {
102 * instance: instance.name,
103 * query: "SELECT '*' AS _key, COUNT(*) AS row_count FROM events GROUP BY _key",
104 * });
105 * ```
106 *
107 * @resource
108 * @category Bigtable
109 */
110export const InstancesMaterializedView = Resource<InstancesMaterializedView>(
111 "GCP.Bigtable.InstancesMaterializedView",
112);
113
114export class MaterializedViewNotResolved extends Data.TaggedError(
115 "GCP.Bigtable.MaterializedViewNotResolved",
116)<{
117 name: string;
118}> {}
119
120export class MaterializedViewStillExists extends Data.TaggedError(
121 "GCP.Bigtable.MaterializedViewStillExists",
122)<{
123 name: string;
124}> {}
125
126const toId = (
127 id: string,
128 materializedViewId: string | undefined,
129 existing?: string,
130) =>
131 toPhysicalId(
132 id,
133 materializedViewId,
134 existing,
135 MAX_MATERIALIZED_VIEW_ID_LENGTH,
136 );
137
138const queryOf = (value: string | undefined) => (value ?? "").trim();
139
140const toAttrs = (view: bigtable.MaterializedView, project: string) => {
141 const name = view.name ?? "";
142 const parsed = parseResourceName(name);
143 return {
144 name,
145 materializedViewId: parsed.materializedViewId,
146 instance: parsed.instance,
147 instanceId: parsed.instanceId,
148 project: parsed.project || project,
149 query: view.query,
150 deletionProtection: view.deletionProtection === true,
151 clusterStates: view.clusterStates,
152 etag: view.etag,
153 };
154};
155
156const getByName = (name: string) =>
157 bigtable
158 .getProjectsInstancesMaterializedViews({ name, view: "FULL" })
159 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
160
161const waitUntilExists = (name: string) =>
162 getByName(name).pipe(
163 Effect.flatMap((view) =>
164 view
165 ? Effect.succeed(view)
166 : Effect.fail(new MaterializedViewNotResolved({ name })),
167 ),
168 Effect.retry({
169 while: (error) =>
170 error._tag === "GCP.Bigtable.MaterializedViewNotResolved",
171 times: 8,
172 schedule: Schedule.spaced("1 second"),
173 }),
174 );
175
176const waitUntilGone = (name: string) =>
177 getByName(name).pipe(
178 Effect.flatMap((view) =>
179 view === undefined
180 ? Effect.void
181 : Effect.fail(new MaterializedViewStillExists({ name })),
182 ),
183 Effect.retry({
184 while: (error) =>
185 error._tag === "GCP.Bigtable.MaterializedViewStillExists",
186 times: 8,
187 schedule: Schedule.spaced("2 seconds"),
188 }),
189 );
190
191const unprotect = (name: string, etag: string | undefined) =>
192 Effect.gen(function* () {
193 const patched = yield* bigtable.patchProjectsInstancesMaterializedViews({
194 name,
195 updateMask: "deletion_protection",
196 body: { deletionProtection: false, etag },
197 });
198 yield* waitForOperation(patched);
199 });
200
201export const InstancesMaterializedViewProvider = () =>
202 Provider.succeed(InstancesMaterializedView, {
203 stables: [
204 "name",
205 "materializedViewId",
206 "instance",
207 "instanceId",
208 "project",
209 ],
210
211 diff: Effect.fn(function* ({ news, olds, output }) {
212 if (!isResolved(news)) return undefined;
213 const previousId = olds?.materializedViewId ?? output?.materializedViewId;
214 const nextId = news.materializedViewId ?? previousId;
215 const previousInstance = instanceIdOf(
216 olds?.instance ?? output?.instance ?? output?.instanceId ?? "",
217 );
218 const nextInstance = instanceIdOf(news.instance);
219 const previousQuery = queryOf(olds?.query ?? output?.query);
220 const nextQuery = queryOf(news.query);
221 if (
222 (previousId !== undefined &&
223 nextId !== undefined &&
224 previousId !== nextId) ||
225 (previousInstance.length > 0 && previousInstance !== nextInstance) ||
226 (previousQuery.length > 0 && previousQuery !== nextQuery)
227 ) {
228 return { action: "replace" as const };
229 }
230 return undefined;
231 }),
232
233 read: Effect.fn(function* ({ id, olds, output }) {
234 const env = yield* GcpEnvironment.current;
235 const materializedViewId = yield* toId(
236 id,
237 olds?.materializedViewId,
238 output?.materializedViewId,
239 );
240 const instanceRef = olds?.instance ?? output?.instance;
241 const name =
242 output?.name ??
243 (instanceRef
244 ? materializedViewName(
245 env.project,
246 instanceIdOf(instanceRef),
247 materializedViewId,
248 )
249 : undefined);
250 if (name === undefined) return undefined;
251 const existing = yield* getByName(name);
252 if (existing === undefined) return undefined;
253 const attrs = toAttrs(existing, env.project);
254 return (yield* parentOwned(attrs.instance)) ? attrs : Unowned(attrs);
255 }),
256
257 list: () =>
258 Effect.gen(function* () {
259 const env = yield* GcpEnvironment.current;
260 const instances = (yield* listAlchemyInstances(env.project)).filter(
261 (instance): instance is bigtable.Instance & { name: string } =>
262 typeof instance.name === "string" && instance.name.length > 0,
263 );
264 const pages = yield* Effect.forEach(
265 instances,
266 (instance) =>
267 collectPages(
268 bigtable.listProjectsInstancesMaterializedViews.pages({
269 parent: instance.name,
270 pageSize: 1000,
271 }),
272 (page) => page.materializedViews,
273 ).pipe(
274 Effect.catchTag("NotFound", () =>
275 Effect.succeed([] as bigtable.MaterializedView[]),
276 ),
277 ),
278 { concurrency: 4 },
279 );
280 return pages.flat().map((view) => toAttrs(view, env.project));
281 }),
282
283 reconcile: Effect.fn(function* ({ id, news, output }) {
284 const env = yield* GcpEnvironment.current;
285 const materializedViewId = yield* toId(
286 id,
287 news.materializedViewId,
288 output?.materializedViewId,
289 );
290 const parent = instanceNameOf(env.project, news.instance);
291 const name = `${parent}/materializedViews/${materializedViewId}`;
292 const desiredProtection = news.deletionProtection === true;
293
294 let current = yield* getByName(output?.name ?? name);
295
296 if (current === undefined) {
297 const created = yield* bigtable
298 .createProjectsInstancesMaterializedViews({
299 parent,
300 materializedViewId,
301 body: {
302 query: news.query,
303 deletionProtection: desiredProtection,
304 },
305 })
306 .pipe(
307 // A just-created source table rejects views for a short while.
308 Effect.retry({
309 while: (error) => error._tag === "SourceTableNotReady",
310 times: 8,
311 schedule: Schedule.spaced("5 seconds"),
312 }),
313 Effect.catchTag("Conflict", () => Effect.succeed(undefined)),
314 );
315 if (created !== undefined) {
316 yield* waitForOperation(created, { alreadyExistsOk: true });
317 }
318 current = yield* waitUntilExists(name);
319 }
320
321 if ((current.deletionProtection === true) !== desiredProtection) {
322 const patched = yield* bigtable
323 .patchProjectsInstancesMaterializedViews({
324 name,
325 updateMask: "deletion_protection",
326 body: {
327 deletionProtection: desiredProtection,
328 etag: current.etag,
329 },
330 })
331 .pipe(
332 Effect.retry({
333 while: (error) => error._tag === "Conflict",
334 times: 8,
335 schedule: Schedule.spaced("1 second"),
336 }),
337 );
338 yield* waitForOperation(patched);
339 current = yield* getByName(name);
340 if (current === undefined) {
341 return yield* new MaterializedViewNotResolved({ name });
342 }
343 }
344
345 return toAttrs(current, env.project);
346 }),
347
348 delete: Effect.fn(function* ({ output }) {
349 const current = yield* getByName(output.name);
350 if (current === undefined) return;
351 if (current.deletionProtection === true) {
352 yield* unprotect(output.name, current.etag);
353 }
354 yield* bigtable
355 .deleteProjectsInstancesMaterializedViews({
356 name: output.name,
357 })
358 .pipe(
359 Effect.catchTag("NotFound", () => Effect.void),
360 Effect.retry({
361 while: (error) => error._tag === "Conflict",
362 times: 8,
363 schedule: Schedule.spaced("2 seconds"),
364 }),
365 );
366 yield* waitUntilGone(output.name);
367 }),
368 });