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 instanceNameOf,
13 listAlchemyInstances,
14 MAX_TABLE_ID_LENGTH,
15 parentOwned,
16 parseResourceName,
17 toPhysicalId,
18 waitForOperation,
19 collectPages,
20} from "./operations.ts";
21
22export type GcRule = {
23 /** Keep only the most recent N versions of each cell. */
24 maxNumVersions?: number;
25 /** Delete cells older than this duration (e.g. `"2592000s"`). */
26 maxAge?: string;
27};
28
29export type ColumnFamily = {
30 /** Garbage-collection rule for this family. */
31 gcRule?: GcRule;
32};
33
34export type ChangeStreamConfig = {
35 /**
36 * How long change-stream data is retained (`1d`–`7d`, e.g. `"86400s"`).
37 */
38 retentionPeriod?: string;
39};
40
41export type AutomatedBackupPolicy = {
42 /** How long automated backups are retained (`3d`–`90d`). */
43 retentionPeriod?: string;
44 /** Backup frequency. Only `"24h"` is supported. */
45 frequency?: string;
46};
47
48export type TableProps = {
49 /**
50 * Parent instance. Full name `projects/{project}/instances/{instance}`
51 * or the instance id. Immutable — changing it replaces the table.
52 */
53 instance: string;
54 /**
55 * Table id (the `{table}` segment of
56 * `.../instances/{instance}/tables/{table}`). If omitted, a unique name
57 * is generated from the stack, stage, and logical id. Must be 1-50
58 * characters. Immutable — changing it replaces the table.
59 */
60 tableId?: string;
61 /**
62 * Column families keyed by family id. Omitted on update leaves families
63 * unchanged; a provided map is reconciled (create / update / drop).
64 */
65 columnFamilies?: Record<string, ColumnFamily>;
66 /**
67 * Protect the table (and its instance) from Admin API deletes.
68 * Disabled automatically on destroy.
69 * @default false
70 */
71 deletionProtection?: boolean;
72 /**
73 * Timestamp granularity. Immutable.
74 * @default "MILLIS"
75 */
76 granularity?: bigtable.TableGranularityEnum | (string & {});
77 /**
78 * Enable a change stream on this table.
79 */
80 changeStreamConfig?: ChangeStreamConfig;
81 /**
82 * Automated backup policy. Omit to disable.
83 */
84 automatedBackupPolicy?: AutomatedBackupPolicy;
85};
86
87export type Table = Resource<
88 "GCP.Bigtable.Table",
89 TableProps,
90 {
91 /** Full resource name `projects/{project}/instances/{instance}/tables/{table}`. */
92 name: string;
93 /** Table id (last path segment). */
94 tableId: string;
95 /** Parent instance resource name. */
96 instance: string;
97 /** Instance id. */
98 instanceId: string;
99 /** Project id. */
100 project: string;
101 /** Column families currently configured. */
102 columnFamilies: Record<string, ColumnFamily>;
103 /** Whether Admin API deletes are blocked. */
104 deletionProtection: boolean;
105 /** Timestamp granularity. */
106 granularity: string | undefined;
107 /** Change-stream config, if enabled. */
108 changeStreamConfig: ChangeStreamConfig | undefined;
109 /** Automated backup policy, if enabled. */
110 automatedBackupPolicy: AutomatedBackupPolicy | undefined;
111 /** Restore info, if this table was restored from a backup. */
112 restoreInfo: bigtable.RestoreInfo | undefined;
113 },
114 never,
115 Providers
116>;
117
118/**
119 * A Cloud Bigtable table — rows keyed by row key, grouped into column
120 * families.
121 *
122 * The parent instance must already exist. Changing `tableId`,
123 * `instance`, or `granularity` replaces the table. Column families,
124 * deletion protection, change streams, and automated backups update in
125 * place. Row read/write is a data-plane API; use {@link GetTable} for
126 * admin metadata.
127 *
128 * Tables have no labels field. Alchemy treats a table as owned when its
129 * parent instance carries Alchemy labels, so `list` / `pnpm nuke:gcp`
130 * can find it.
131 *
132 * ### Creating a Table
133 * **Example:** Table with one column family
134 * ```typescript
135 * const instance = yield* GCP.Bigtable.Instance("Data", {});
136 * const table = yield* GCP.Bigtable.Table("Users", {
137 * instance: instance.name,
138 * columnFamilies: {
139 * cf: { gcRule: { maxNumVersions: 3 } },
140 * },
141 * });
142 * ```
143 *
144 * ### Reading Table Metadata
145 * **Example:** Get the bound table
146 * ```typescript
147 * const getTable = yield* GCP.Bigtable.GetTable(table);
148 * const live = yield* getTable({ view: "SCHEMA_VIEW" });
149 * ```
150 *
151 * @resource
152 * @category Bigtable
153 */
154export const Table = Resource<Table>("GCP.Bigtable.Table");
155
156export class TableNotResolved extends Data.TaggedError(
157 "GCP.Bigtable.TableNotResolved",
158)<{
159 name: string;
160}> {}
161
162export class TableStillExists extends Data.TaggedError(
163 "GCP.Bigtable.TableStillExists",
164)<{
165 name: string;
166}> {}
167
168const DEFAULT_GRANULARITY = "MILLIS";
169const SCHEMA_VIEW = "SCHEMA_VIEW";
170
171const normalizeGranularity = (value: string | undefined) => {
172 const next = (value ?? DEFAULT_GRANULARITY).toUpperCase();
173 return next === "TIMESTAMP_GRANULARITY_UNSPECIFIED"
174 ? DEFAULT_GRANULARITY
175 : next;
176};
177
178const gcKey = (rule: GcRule | bigtable.GcRule | undefined) =>
179 JSON.stringify({
180 maxNumVersions: rule?.maxNumVersions ?? 0,
181 maxAge: rule?.maxAge ?? "",
182 });
183
184const familiesOf = (
185 families:
186 | Record<string, ColumnFamily | bigtable.ColumnFamily | undefined>
187 | null
188 | undefined,
189): Record<string, ColumnFamily> => {
190 const result: Record<string, ColumnFamily> = {};
191 for (const [id, family] of Object.entries(families ?? {})) {
192 if (family === undefined) continue;
193 result[id] = { gcRule: family.gcRule };
194 }
195 return result;
196};
197
198const toId = (id: string, tableId: string | undefined, existing?: string) =>
199 toPhysicalId(id, tableId, existing, MAX_TABLE_ID_LENGTH);
200
201const toAttrs = (table: bigtable.Table, project: string) => {
202 const name = table.name ?? "";
203 const parsed = parseResourceName(name);
204 return {
205 name,
206 tableId: parsed.tableId,
207 instance: parsed.instance,
208 instanceId: parsed.instanceId,
209 project: parsed.project || project,
210 columnFamilies: familiesOf(table.columnFamilies),
211 deletionProtection: table.deletionProtection === true,
212 granularity: table.granularity,
213 changeStreamConfig: table.changeStreamConfig,
214 automatedBackupPolicy: table.automatedBackupPolicy,
215 restoreInfo: table.restoreInfo,
216 };
217};
218
219const getByName = (name: string) =>
220 bigtable
221 .getProjectsInstancesTables({ name, view: SCHEMA_VIEW })
222 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
223
224const waitUntilGone = (name: string) =>
225 getByName(name).pipe(
226 Effect.flatMap((table) =>
227 table === undefined
228 ? Effect.void
229 : Effect.fail(new TableStillExists({ name })),
230 ),
231 Effect.retry({
232 while: (error) => error._tag === "GCP.Bigtable.TableStillExists",
233 times: 8,
234 schedule: Schedule.spaced("2 seconds"),
235 }),
236 );
237
238const toFamilyBody = (family: ColumnFamily): bigtable.ColumnFamily => ({
239 gcRule: family.gcRule,
240});
241
242const changeStreamKey = (config: ChangeStreamConfig | undefined) =>
243 JSON.stringify({ retention: config?.retentionPeriod ?? "" });
244
245const backupKey = (policy: AutomatedBackupPolicy | undefined) =>
246 JSON.stringify({
247 retention: policy?.retentionPeriod ?? "",
248 frequency: policy?.frequency ?? "",
249 });
250
251const familyMods = (
252 observed: Record<string, ColumnFamily>,
253 desired: Record<string, ColumnFamily>,
254): bigtable.Modification[] => {
255 const mods: bigtable.Modification[] = [];
256 for (const [id, family] of Object.entries(desired)) {
257 if (observed[id] === undefined) {
258 mods.push({ id, create: toFamilyBody(family) });
259 } else if (gcKey(observed[id]?.gcRule) !== gcKey(family.gcRule)) {
260 mods.push({
261 id,
262 update: toFamilyBody(family),
263 updateMask: "gc_rule",
264 });
265 }
266 }
267 for (const id of Object.keys(observed)) {
268 if (desired[id] === undefined) {
269 mods.push({ id, drop: true });
270 }
271 }
272 return mods;
273};
274
275export const TableProvider = () =>
276 Provider.succeed(Table, {
277 stables: ["name", "tableId", "instance", "instanceId", "project"],
278
279 diff: Effect.fn(function* ({ news, olds, output }) {
280 if (!isResolved(news)) return undefined;
281 const previousId = olds?.tableId ?? output?.tableId;
282 const nextId = news.tableId ?? previousId;
283 const previousInstance = olds?.instance ?? output?.instance;
284 const previousInstanceId = previousInstance
285 ? parseResourceName(
286 previousInstance.includes("/instances/")
287 ? previousInstance
288 : `projects/_/instances/${previousInstance}`,
289 ).instanceId
290 : output?.instanceId;
291 const nextInstanceId = parseResourceName(
292 news.instance.includes("/instances/")
293 ? news.instance
294 : `projects/_/instances/${news.instance}`,
295 ).instanceId;
296 const previousGranularity = normalizeGranularity(
297 olds?.granularity ?? output?.granularity,
298 );
299 const nextGranularity = normalizeGranularity(
300 news.granularity ?? output?.granularity,
301 );
302 if (
303 (previousId !== undefined &&
304 nextId !== undefined &&
305 previousId !== nextId) ||
306 (previousInstanceId !== undefined &&
307 previousInstanceId !== nextInstanceId) ||
308 (news.granularity !== undefined &&
309 previousGranularity !== nextGranularity)
310 ) {
311 return { action: "replace" as const };
312 }
313 return undefined;
314 }),
315
316 read: Effect.fn(function* ({ id, olds, output }) {
317 const env = yield* GcpEnvironment.current;
318 const tableId = yield* toId(id, olds?.tableId, output?.tableId);
319 const instanceRef = olds?.instance ?? output?.instance;
320 const name =
321 output?.name ??
322 (instanceRef
323 ? `${instanceNameOf(env.project, instanceRef)}/tables/${tableId}`
324 : undefined);
325 if (name === undefined) return undefined;
326 const existing = yield* getByName(name);
327 if (existing === undefined) return undefined;
328 const attrs = toAttrs(existing, env.project);
329 return (yield* parentOwned(attrs.instance)) ? attrs : Unowned(attrs);
330 }),
331
332 list: () =>
333 Effect.gen(function* () {
334 const env = yield* GcpEnvironment.current;
335 const instances = (yield* listAlchemyInstances(env.project)).filter(
336 (instance): instance is bigtable.Instance & { name: string } =>
337 !!instance.name,
338 );
339 const pages = yield* Effect.forEach(
340 instances,
341 (instance) =>
342 collectPages(
343 bigtable.listProjectsInstancesTables.pages({
344 parent: instance.name,
345 pageSize: 1000,
346 }),
347 (page) => page.tables,
348 ).pipe(
349 Effect.catchTag("NotFound", () =>
350 Effect.succeed([] as bigtable.Table[]),
351 ),
352 ),
353 { concurrency: 4 },
354 );
355 return pages.flat().map((table) => toAttrs(table, env.project));
356 }),
357
358 reconcile: Effect.fn(function* ({ id, news, output }) {
359 const env = yield* GcpEnvironment.current;
360 const tableId = yield* toId(id, news.tableId, output?.tableId);
361 const parent = instanceNameOf(env.project, news.instance);
362 const name = `${parent}/tables/${tableId}`;
363 const desiredProtection = news.deletionProtection === true;
364 const families = news.columnFamilies;
365
366 let current = yield* getByName(name);
367
368 if (current === undefined) {
369 const created = yield* bigtable
370 .createProjectsInstancesTables({
371 parent,
372 body: {
373 tableId,
374 table: {
375 granularity: news.granularity
376 ? normalizeGranularity(news.granularity)
377 : undefined,
378 columnFamilies:
379 families === undefined
380 ? undefined
381 : Object.fromEntries(
382 Object.entries(families).map(([familyId, family]) => [
383 familyId,
384 toFamilyBody(family),
385 ]),
386 ),
387 deletionProtection: desiredProtection,
388 changeStreamConfig: news.changeStreamConfig,
389 automatedBackupPolicy: news.automatedBackupPolicy,
390 },
391 },
392 })
393 .pipe(
394 Effect.catchTag("Conflict", (error) =>
395 getByName(name).pipe(
396 Effect.flatMap((existing) =>
397 existing ? Effect.succeed(existing) : Effect.fail(error),
398 ),
399 ),
400 ),
401 Effect.retry({
402 while: (error) => error._tag === "Conflict",
403 times: 8,
404 schedule: Schedule.spaced("2 seconds"),
405 }),
406 );
407 current = created ?? undefined;
408 }
409
410 if (current === undefined) {
411 return yield* new TableNotResolved({ name });
412 }
413
414 if (families !== undefined) {
415 const mods = familyMods(familiesOf(current.columnFamilies), families);
416 if (mods.length > 0) {
417 if (current.deletionProtection === true) {
418 const unlocked = yield* bigtable.patchProjectsInstancesTables({
419 name,
420 updateMask: "deletion_protection",
421 body: { deletionProtection: false },
422 });
423 yield* waitForOperation(unlocked);
424 current = (yield* getByName(name)) ?? current;
425 }
426 current = yield* bigtable.modifyColumnFamiliesProjectsInstancesTables(
427 {
428 name,
429 body: { modifications: mods },
430 },
431 );
432 }
433 }
434
435 const mask: string[] = [];
436 const patchBody: bigtable.Table = {};
437 if ((current.deletionProtection === true) !== desiredProtection) {
438 patchBody.deletionProtection = desiredProtection;
439 mask.push("deletion_protection");
440 }
441 if (
442 news.changeStreamConfig !== undefined &&
443 changeStreamKey(news.changeStreamConfig) !==
444 changeStreamKey(current.changeStreamConfig)
445 ) {
446 patchBody.changeStreamConfig = news.changeStreamConfig;
447 mask.push("change_stream_config");
448 }
449 if (
450 news.automatedBackupPolicy !== undefined &&
451 backupKey(news.automatedBackupPolicy) !==
452 backupKey(current.automatedBackupPolicy)
453 ) {
454 patchBody.automatedBackupPolicy = news.automatedBackupPolicy;
455 mask.push("automated_backup_policy");
456 }
457
458 if (mask.length > 0) {
459 const patched = yield* bigtable.patchProjectsInstancesTables({
460 name,
461 updateMask: mask.join(","),
462 body: patchBody,
463 });
464 yield* waitForOperation(patched);
465 current = yield* getByName(name);
466 if (current === undefined) {
467 return yield* new TableNotResolved({ name });
468 }
469 }
470
471 return toAttrs(current, env.project);
472 }),
473
474 delete: Effect.fn(function* ({ output }) {
475 const current = yield* getByName(output.name);
476 if (current === undefined) return;
477 if (current.deletionProtection === true) {
478 const patched = yield* bigtable.patchProjectsInstancesTables({
479 name: output.name,
480 updateMask: "deletion_protection",
481 body: { deletionProtection: false },
482 });
483 yield* waitForOperation(patched);
484 }
485 yield* bigtable.deleteProjectsInstancesTables({ name: output.name }).pipe(
486 Effect.catchTag("NotFound", () => Effect.void),
487 Effect.retry({
488 while: (error) => error._tag === "Conflict",
489 times: 8,
490 schedule: Schedule.spaced("2 seconds"),
491 }),
492 );
493 yield* waitUntilGone(output.name);
494 }),
495 });