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 listAlchemyTables,
14 MAX_SCHEMA_BUNDLE_ID_LENGTH,
15 parentOwned,
16 parseResourceName,
17 tableIdOf,
18 tableNameOf,
19 toPhysicalId,
20 waitForOperation,
21 collectPages,
22} from "./operations.ts";
23
24export type InstancesTablesSchemaBundleProps = {
25 /**
26 * Parent instance. Full name `projects/{project}/instances/{instance}`
27 * or the instance id. Immutable — changing it replaces the bundle.
28 */
29 instance: string;
30 /**
31 * Parent table. Full name
32 * `projects/{project}/instances/{instance}/tables/{table}` or the table
33 * id. Immutable — changing it replaces the bundle.
34 */
35 table: string;
36 /**
37 * Schema bundle id (the `{schema_bundle}` segment of
38 * `.../tables/{table}/schemaBundles/{schema_bundle}`). If omitted, a
39 * unique name is generated from the stack, stage, and logical id.
40 * Immutable — changing it replaces the bundle.
41 */
42 schemaBundleId?: string;
43 /**
44 * Base64-encoded protobuf `FileDescriptorSet` (`protoc
45 * --include_imports --descriptor_set_out`).
46 */
47 protoDescriptors: string;
48 /**
49 * Ignore backwards-compatibility checks on update.
50 * @default true
51 */
52 ignoreWarnings?: boolean;
53};
54
55export type InstancesTablesSchemaBundle = Resource<
56 "GCP.Bigtable.InstancesTablesSchemaBundle",
57 InstancesTablesSchemaBundleProps,
58 {
59 /** Full resource name `.../tables/{table}/schemaBundles/{schema_bundle}`. */
60 name: string;
61 /** Schema bundle id (last path segment). */
62 schemaBundleId: string;
63 /** Parent table resource name. */
64 table: string;
65 /** Parent table id. */
66 tableId: string;
67 /** Parent instance resource name. */
68 instance: string;
69 /** Parent instance id. */
70 instanceId: string;
71 /** Project id. */
72 project: string;
73 /** Base64-encoded FileDescriptorSet currently stored. */
74 protoDescriptors: string | undefined;
75 /** Server etag. */
76 etag: string | undefined;
77 },
78 never,
79 Providers
80>;
81
82/**
83 * A Cloud Bigtable schema bundle — a named FileDescriptorSet used to
84 * decode protobuf cell values.
85 *
86 * The parent instance and table must already exist. Changing
87 * `schemaBundleId`, `instance`, or `table` replaces the bundle.
88 * `protoDescriptors` update in place (must be backwards compatible
89 * unless `ignoreWarnings` is true).
90 *
91 * Schema bundles have no labels field. Alchemy treats a bundle as owned
92 * when its parent instance carries Alchemy labels, so `list` /
93 * `pnpm nuke:gcp` can find it.
94 *
95 * ### Creating a Schema Bundle
96 * **Example:** Bundle from a FileDescriptorSet
97 * ```typescript
98 * const bundle = yield* GCP.Bigtable.InstancesTablesSchemaBundle("Rows", {
99 * instance: instance.name,
100 * table: table.name,
101 * protoDescriptors: descriptors,
102 * });
103 * ```
104 *
105 * @resource
106 * @category Bigtable
107 */
108export const InstancesTablesSchemaBundle =
109 Resource<InstancesTablesSchemaBundle>(
110 "GCP.Bigtable.InstancesTablesSchemaBundle",
111 );
112
113export class SchemaBundleNotResolved extends Data.TaggedError(
114 "GCP.Bigtable.SchemaBundleNotResolved",
115)<{
116 name: string;
117}> {}
118
119export class SchemaBundleStillExists extends Data.TaggedError(
120 "GCP.Bigtable.SchemaBundleStillExists",
121)<{
122 name: string;
123}> {}
124
125const toId = (
126 id: string,
127 schemaBundleId: string | undefined,
128 existing?: string,
129) => toPhysicalId(id, schemaBundleId, existing, MAX_SCHEMA_BUNDLE_ID_LENGTH);
130
131const descriptorsOf = (
132 schema: bigtable.ProtoSchema | { protoDescriptors?: string } | undefined,
133) => schema?.protoDescriptors ?? "";
134
135const toAttrs = (bundle: bigtable.SchemaBundle, project: string) => {
136 const name = bundle.name ?? "";
137 const parsed = parseResourceName(name);
138 return {
139 name,
140 schemaBundleId: parsed.schemaBundleId,
141 table: parsed.table,
142 tableId: parsed.tableId,
143 instance: parsed.instance,
144 instanceId: parsed.instanceId,
145 project: parsed.project || project,
146 protoDescriptors: bundle.protoSchema?.protoDescriptors,
147 etag: bundle.etag,
148 };
149};
150
151const getByName = (name: string) =>
152 bigtable
153 .getProjectsInstancesTablesSchemaBundles({ name })
154 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
155
156const waitUntilExists = (name: string) =>
157 getByName(name).pipe(
158 Effect.flatMap((bundle) =>
159 bundle
160 ? Effect.succeed(bundle)
161 : Effect.fail(new SchemaBundleNotResolved({ name })),
162 ),
163 Effect.retry({
164 while: (error) => error._tag === "GCP.Bigtable.SchemaBundleNotResolved",
165 times: 8,
166 schedule: Schedule.spaced("1 second"),
167 }),
168 );
169
170const waitUntilGone = (name: string) =>
171 getByName(name).pipe(
172 Effect.flatMap((bundle) =>
173 bundle === undefined
174 ? Effect.void
175 : Effect.fail(new SchemaBundleStillExists({ name })),
176 ),
177 Effect.retry({
178 while: (error) => error._tag === "GCP.Bigtable.SchemaBundleStillExists",
179 times: 8,
180 schedule: Schedule.spaced("2 seconds"),
181 }),
182 );
183
184export const InstancesTablesSchemaBundleProvider = () =>
185 Provider.succeed(InstancesTablesSchemaBundle, {
186 stables: [
187 "name",
188 "schemaBundleId",
189 "table",
190 "tableId",
191 "instance",
192 "instanceId",
193 "project",
194 ],
195
196 diff: Effect.fn(function* ({ news, olds, output }) {
197 if (!isResolved(news)) return undefined;
198 const previousId = olds?.schemaBundleId ?? output?.schemaBundleId;
199 const nextId = news.schemaBundleId ?? previousId;
200 const previousInstance = instanceIdOf(
201 olds?.instance ?? output?.instance ?? output?.instanceId ?? "",
202 );
203 const nextInstance = instanceIdOf(news.instance);
204 const previousTable = tableIdOf(
205 olds?.table ?? output?.table ?? output?.tableId ?? "",
206 );
207 const nextTable = tableIdOf(news.table);
208 if (
209 (previousId !== undefined &&
210 nextId !== undefined &&
211 previousId !== nextId) ||
212 (previousInstance.length > 0 && previousInstance !== nextInstance) ||
213 (previousTable.length > 0 && previousTable !== nextTable)
214 ) {
215 return { action: "replace" as const };
216 }
217 return undefined;
218 }),
219
220 read: Effect.fn(function* ({ id, olds, output }) {
221 const env = yield* GcpEnvironment.current;
222 const schemaBundleId = yield* toId(
223 id,
224 olds?.schemaBundleId,
225 output?.schemaBundleId,
226 );
227 const instanceRef = olds?.instance ?? output?.instance;
228 const tableRef = olds?.table ?? output?.table;
229 const name =
230 output?.name ??
231 (instanceRef && tableRef
232 ? `${tableNameOf(env.project, instanceRef, tableRef)}/schemaBundles/${schemaBundleId}`
233 : undefined);
234 if (name === undefined) return undefined;
235 const existing = yield* getByName(name);
236 if (existing === undefined) return undefined;
237 const attrs = toAttrs(existing, env.project);
238 return (yield* parentOwned(attrs.instance)) ? attrs : Unowned(attrs);
239 }),
240
241 list: () =>
242 Effect.gen(function* () {
243 const env = yield* GcpEnvironment.current;
244 const tables = (yield* listAlchemyTables(env.project)).filter(
245 (table): table is bigtable.Table & { name: string } =>
246 typeof table.name === "string" && table.name.length > 0,
247 );
248 const pages = yield* Effect.forEach(
249 tables,
250 (table) =>
251 collectPages(
252 bigtable.listProjectsInstancesTablesSchemaBundles.pages({
253 parent: table.name,
254 pageSize: 1000,
255 }),
256 (page) => page.schemaBundles,
257 ).pipe(
258 Effect.catchTag("NotFound", () =>
259 Effect.succeed([] as bigtable.SchemaBundle[]),
260 ),
261 ),
262 { concurrency: 4 },
263 );
264 return pages.flat().map((bundle) => toAttrs(bundle, env.project));
265 }),
266
267 reconcile: Effect.fn(function* ({ id, news, output }) {
268 const env = yield* GcpEnvironment.current;
269 const schemaBundleId = yield* toId(
270 id,
271 news.schemaBundleId,
272 output?.schemaBundleId,
273 );
274 const parent = tableNameOf(env.project, news.instance, news.table);
275 const name = `${parent}/schemaBundles/${schemaBundleId}`;
276 const ignoreWarnings = news.ignoreWarnings ?? true;
277 const protoSchema = { protoDescriptors: news.protoDescriptors };
278
279 let current = yield* getByName(output?.name ?? name);
280
281 if (current === undefined) {
282 const created = yield* bigtable
283 .createProjectsInstancesTablesSchemaBundles({
284 parent,
285 schemaBundleId,
286 body: { protoSchema },
287 })
288 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
289 if (created !== undefined) {
290 yield* waitForOperation(created, { alreadyExistsOk: true });
291 }
292 current = yield* waitUntilExists(name);
293 }
294
295 if (descriptorsOf(current.protoSchema) !== news.protoDescriptors) {
296 const patched = yield* bigtable
297 .patchProjectsInstancesTablesSchemaBundles({
298 name,
299 updateMask: "proto_schema",
300 ignoreWarnings,
301 body: { protoSchema, etag: current.etag },
302 })
303 .pipe(
304 Effect.retry({
305 while: (error) => error._tag === "Conflict",
306 times: 8,
307 schedule: Schedule.spaced("1 second"),
308 }),
309 );
310 yield* waitForOperation(patched);
311 current = yield* getByName(name);
312 if (current === undefined) {
313 return yield* new SchemaBundleNotResolved({ name });
314 }
315 }
316
317 return toAttrs(current, env.project);
318 }),
319
320 delete: Effect.fn(function* ({ output }) {
321 yield* bigtable
322 .deleteProjectsInstancesTablesSchemaBundles({
323 name: output.name,
324 })
325 .pipe(
326 Effect.catchTag("NotFound", () => Effect.void),
327 Effect.retry({
328 while: (error) => error._tag === "Conflict",
329 times: 8,
330 schedule: Schedule.spaced("2 seconds"),
331 }),
332 );
333 yield* waitUntilGone(output.name);
334 }),
335 });