1import * as spanner from "@distilled.cloud/gcp/spanner_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 { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { GcpEnvironment } from "../Environment.ts";
11import { ALCHEMY_LABEL_PREFIX } from "../Labels.ts";
12import type { Providers } from "../Providers.ts";
13import { waitForOperation } from "./operations.ts";
14
15const DEFAULT_DIALECT = "GOOGLE_STANDARD_SQL";
16const MAX_DATABASE_ID_LENGTH = 30;
17
18export type DatabaseDialect =
19 | spanner.DatabaseDatabaseDialectEnum
20 | (string & {});
21
22export type EncryptionConfig = {
23 /**
24 * Cloud KMS key
25 * (`projects/{project}/locations/{location}/keyRings/{keyRing}/cryptoKeys/{cryptoKey}`).
26 * Immutable — changing it replaces the database.
27 */
28 kmsKeyName?: string;
29 /**
30 * KMS keys covering every region of the instance config. Immutable —
31 * changing them replaces the database.
32 */
33 kmsKeyNames?: string[];
34};
35
36export type DatabaseProps = {
37 /**
38 * Parent instance id or full name
39 * (`projects/{project}/instances/{instance}`). Immutable — changing it
40 * replaces the database.
41 */
42 instance: string;
43 /**
44 * Database id (the `{database}` segment of
45 * `projects/{project}/instances/{instance}/databases/{database}`). If
46 * omitted, a unique name is generated from the stack, stage, and
47 * logical id. Must match `^[a-z][-a-z0-9]*[a-z0-9]$` (2–30
48 * characters). Immutable — changing it replaces the database.
49 */
50 databaseId?: string;
51 /**
52 * SQL dialect. Immutable — changing it replaces the database.
53 * @default "GOOGLE_STANDARD_SQL"
54 */
55 databaseDialect?: DatabaseDialect;
56 /**
57 * Extra DDL statements run atomically with create. Immutable —
58 * changing them replaces the database.
59 */
60 extraStatements?: string[];
61 /**
62 * Customer-managed encryption. Immutable — changing it replaces the
63 * database.
64 */
65 encryptionConfig?: EncryptionConfig;
66 /**
67 * Drop protection. Alchemy defaults to disabled so `destroy` can drop
68 * the database. Delete disables protection first when it is on.
69 * @default false
70 */
71 enableDropProtection?: boolean;
72};
73
74export type Database = Resource<
75 "GCP.Spanner.Database",
76 DatabaseProps,
77 {
78 /** Full resource name `projects/{project}/instances/{instance}/databases/{database}`. */
79 name: string;
80 /** Database id (last path segment). */
81 databaseId: string;
82 /** Parent instance id. */
83 instanceId: string;
84 /** Project id. */
85 project: string;
86 /** SQL dialect. */
87 databaseDialect: string | undefined;
88 /** Whether drop protection is enabled. */
89 enableDropProtection: boolean;
90 /** Current database state (`CREATING`, `READY`, `READY_OPTIMIZING`). */
91 state: string | undefined;
92 /** Version retention period (e.g. `"1h"`). */
93 versionRetentionPeriod: string | undefined;
94 /** Earliest PITR timestamp, if available. */
95 earliestVersionTime: string | undefined;
96 /** Default leader region, if set. */
97 defaultLeader: string | undefined;
98 /** CMEK config, if any. */
99 encryptionConfig: EncryptionConfig | undefined;
100 /** Whether an update is in flight. */
101 reconciling: boolean | undefined;
102 /** RFC3339 creation timestamp. */
103 createTime: string | undefined;
104 },
105 never,
106 Providers
107>;
108
109/**
110 * A Cloud Spanner database inside an instance.
111 *
112 * Spanner databases have no labels field. `list` enumerates databases on
113 * alchemy-labeled instances so `pnpm nuke:gcp` can find leaked rows.
114 * Changing `instance`, `databaseId`, dialect, CMEK, or extra DDL
115 * replaces the database. Drop protection updates in place.
116 *
117 * ### Creating a Database
118 * **Example:** Generated name on a Spanner instance
119 * ```typescript
120 * const instance = yield* GCP.Spanner.Instance("App", {});
121 * const database = yield* GCP.Spanner.Database("AppDb", {
122 * instance: instance.instanceId,
123 * });
124 * ```
125 *
126 * **Example:** Explicit id, dialect, and extra DDL
127 * ```typescript
128 * const database = yield* GCP.Spanner.Database("AppDb", {
129 * instance: instance.instanceId,
130 * databaseId: "appdb",
131 * extraStatements: [
132 * "CREATE TABLE Users (UserId INT64 NOT NULL) PRIMARY KEY (UserId)",
133 * ],
134 * });
135 * ```
136 *
137 * ### Querying
138 * **Example:** Run SQL
139 * ```typescript
140 * const executeSql = yield* GCP.Spanner.ExecuteSql(database);
141 * const result = yield* executeSql({ sql: "SELECT 1 AS n" });
142 * ```
143 *
144 * ### Reading Schema
145 * **Example:** Fetch live DDL
146 * ```typescript
147 * const getDdl = yield* GCP.Spanner.GetDdl(database);
148 * const { statements } = yield* getDdl();
149 * ```
150 *
151 * @resource
152 * @category Spanner
153 */
154export const Database = Resource<Database>("GCP.Spanner.Database");
155
156export class DatabaseNotResolved extends Data.TaggedError(
157 "GCP.Spanner.DatabaseNotResolved",
158)<{
159 name: string;
160}> {}
161
162export class DatabaseNotReady extends Data.TaggedError(
163 "GCP.Spanner.DatabaseNotReady",
164)<{
165 name: string;
166 state: string;
167}> {}
168
169export class DatabaseStillExists extends Data.TaggedError(
170 "GCP.Spanner.DatabaseStillExists",
171)<{
172 name: string;
173}> {}
174
175const lastSegment = (value: string) => {
176 const trimmed = value.replace(/\/+$/, "");
177 const parts = trimmed.split("/");
178 return parts[parts.length - 1] || trimmed;
179};
180
181const instanceIdOf = (value: string) => lastSegment(value);
182
183const normalizeDialect = (value: string | undefined) => {
184 const next = (value ?? DEFAULT_DIALECT).toUpperCase();
185 return next.endsWith("_UNSPECIFIED") ? DEFAULT_DIALECT : next;
186};
187
188const resourceName = (
189 project: string,
190 instanceId: string,
191 databaseId: string,
192) => `projects/${project}/instances/${instanceId}/databases/${databaseId}`;
193
194const instanceName = (project: string, instanceId: string) =>
195 `projects/${project}/instances/${instanceId}`;
196
197const parseName = (name: string) => {
198 const parts = name.split("/").filter((part) => part.length > 0);
199 const databasesAt = parts.lastIndexOf("databases");
200 const instancesAt = parts.lastIndexOf("instances");
201 const projectsAt = parts.lastIndexOf("projects");
202 return {
203 project:
204 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
205 instanceId:
206 instancesAt >= 0 && parts[instancesAt + 1] ? parts[instancesAt + 1]! : "",
207 databaseId:
208 databasesAt >= 0 && parts[databasesAt + 1]
209 ? parts[databasesAt + 1]!
210 : lastSegment(name),
211 };
212};
213
214const toSpannerId = (name: string) => {
215 let next = name
216 .toLowerCase()
217 .replace(/[^a-z0-9-]/g, "-")
218 .replace(/-+/g, "-");
219 next = next.replace(/^-+/, "").replace(/-+$/g, "");
220 if (!/^[a-z]/.test(next)) next = `d${next}`;
221 next = next.slice(0, MAX_DATABASE_ID_LENGTH).replace(/-+$/g, "");
222 if (next.length < 2) next = `${next}xx`.slice(0, MAX_DATABASE_ID_LENGTH);
223 return next;
224};
225
226const toDatabaseId = (
227 id: string,
228 databaseId: string | undefined,
229 existing?: string,
230) =>
231 Effect.gen(function* () {
232 if (databaseId !== undefined) return databaseId;
233 if (existing !== undefined) return existing;
234 return toSpannerId(
235 yield* createPhysicalName({
236 id,
237 maxLength: MAX_DATABASE_ID_LENGTH,
238 lowercase: true,
239 }),
240 );
241 });
242
243const createStatementOf = (databaseId: string, dialect: string) =>
244 dialect === "POSTGRESQL"
245 ? `CREATE DATABASE "${databaseId}"`
246 : `CREATE DATABASE \`${databaseId}\``;
247
248const encryptionOf = (
249 config: spanner.EncryptionConfig | undefined,
250): EncryptionConfig | undefined => {
251 if (config === undefined) return undefined;
252 if (
253 (config.kmsKeyName === undefined || config.kmsKeyName.length === 0) &&
254 (config.kmsKeyNames === undefined || config.kmsKeyNames.length === 0)
255 ) {
256 return undefined;
257 }
258 return {
259 kmsKeyName: config.kmsKeyName,
260 kmsKeyNames: config.kmsKeyNames,
261 };
262};
263
264const encryptionKey = (config: EncryptionConfig | undefined) =>
265 JSON.stringify({
266 kmsKeyName: config?.kmsKeyName ?? "",
267 kmsKeyNames: [...(config?.kmsKeyNames ?? [])].sort(),
268 });
269
270const extraKey = (statements: string[] | undefined) =>
271 JSON.stringify(statements ?? []);
272
273const toAttrs = (
274 database: spanner.Database,
275 project: string,
276): Database["Attributes"] => {
277 const name = database.name ?? "";
278 const parsed = parseName(name);
279 return {
280 name,
281 databaseId: parsed.databaseId,
282 instanceId: parsed.instanceId,
283 project: parsed.project || project,
284 databaseDialect: database.databaseDialect,
285 enableDropProtection: database.enableDropProtection === true,
286 state: database.state,
287 versionRetentionPeriod: database.versionRetentionPeriod,
288 earliestVersionTime: database.earliestVersionTime,
289 defaultLeader: database.defaultLeader,
290 encryptionConfig: encryptionOf(database.encryptionConfig),
291 reconciling: database.reconciling,
292 createTime: database.createTime,
293 };
294};
295
296const getByName = (name: string) =>
297 spanner
298 .getProjectsInstancesDatabases({ name })
299 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
300
301const waitUntilExists = (name: string) =>
302 getByName(name).pipe(
303 Effect.flatMap((database) =>
304 database
305 ? Effect.succeed(database)
306 : Effect.fail(new DatabaseNotResolved({ name })),
307 ),
308 Effect.retry({
309 while: (error) => error._tag === "GCP.Spanner.DatabaseNotResolved",
310 times: 8,
311 schedule: Schedule.spaced("2 seconds"),
312 }),
313 );
314
315const isReadyState = (state: string | undefined) =>
316 state === "READY" || state === "READY_OPTIMIZING";
317
318const waitUntilReady = (name: string) =>
319 getByName(name).pipe(
320 Effect.filterOrFail(
321 (database): database is spanner.Database => database !== undefined,
322 () => new DatabaseNotResolved({ name }),
323 ),
324 Effect.filterOrFail(
325 (database) => isReadyState(database.state),
326 (database) =>
327 new DatabaseNotReady({
328 name,
329 state: database.state ?? "STATE_UNSPECIFIED",
330 }),
331 ),
332 Effect.retry({
333 while: (error) =>
334 error._tag === "GCP.Spanner.DatabaseNotReady" ||
335 error._tag === "GCP.Spanner.DatabaseNotResolved",
336 times: 10,
337 schedule: Schedule.spaced("8 seconds"),
338 }),
339 );
340
341const waitUntilGone = (name: string) =>
342 getByName(name).pipe(
343 Effect.flatMap((database) =>
344 database === undefined
345 ? Effect.void
346 : Effect.fail(new DatabaseStillExists({ name })),
347 ),
348 Effect.retry({
349 while: (error) => error._tag === "GCP.Spanner.DatabaseStillExists",
350 times: 10,
351 schedule: Schedule.spaced("8 seconds"),
352 }),
353 );
354
355const retryConcurrentChanges = <A, E extends { readonly _tag: string }, R>(
356 effect: Effect.Effect<A, E, R>,
357) =>
358 effect.pipe(
359 Effect.retry({
360 while: (error) => error._tag === "Conflict",
361 times: 8,
362 schedule: Schedule.spaced("5 seconds"),
363 }),
364 );
365
366const listAlchemyInstances = (project: string) =>
367 spanner.listProjectsInstances
368 .pages({
369 parent: `projects/${project}`,
370 pageSize: 1000,
371 })
372 .pipe(
373 Stream.flatMap((page) => Stream.fromIterable(page.instances ?? [])),
374 Stream.filter((instance) =>
375 Object.keys(instance.labels ?? {}).some((key) =>
376 key.startsWith(ALCHEMY_LABEL_PREFIX),
377 ),
378 ),
379 Stream.runCollect,
380 Effect.map((chunk) => Array.from(chunk)),
381 Effect.catchTag("NotFound", () =>
382 Effect.succeed([] as spanner.Instance[]),
383 ),
384 );
385
386export const DatabaseProvider = () =>
387 Provider.succeed(Database, {
388 stables: ["name", "databaseId", "instanceId", "project", "createTime"],
389
390 diff: Effect.fn(function* ({ news, olds, output }) {
391 if (!isResolved(news)) return undefined;
392
393 const previousInstance = olds?.instance ?? output?.instanceId;
394 const nextInstance = news.instance;
395 const previousId = olds?.databaseId ?? output?.databaseId;
396 const nextId = news.databaseId ?? previousId;
397 const previousDialect = normalizeDialect(
398 olds?.databaseDialect ?? output?.databaseDialect,
399 );
400 const nextDialect = normalizeDialect(
401 news.databaseDialect ?? output?.databaseDialect,
402 );
403 const previousKms = encryptionKey(
404 olds?.encryptionConfig ?? output?.encryptionConfig,
405 );
406 const nextKms = encryptionKey(
407 news.encryptionConfig ?? output?.encryptionConfig,
408 );
409 const extraChanged =
410 olds?.extraStatements !== undefined &&
411 news.extraStatements !== undefined &&
412 extraKey(olds.extraStatements) !== extraKey(news.extraStatements);
413
414 const instanceChanged =
415 previousInstance !== undefined &&
416 instanceIdOf(previousInstance) !== instanceIdOf(nextInstance);
417 const idChanged =
418 previousId !== undefined &&
419 nextId !== undefined &&
420 previousId !== nextId;
421
422 const replace =
423 instanceChanged ||
424 idChanged ||
425 previousDialect !== nextDialect ||
426 previousKms !== nextKms ||
427 extraChanged;
428
429 if (!replace) return undefined;
430 return {
431 action: "replace" as const,
432 deleteFirst:
433 !instanceChanged && previousId !== undefined && nextId === previousId,
434 };
435 }),
436
437 read: Effect.fn(function* ({ id, olds, output }) {
438 const env = yield* GcpEnvironment.current;
439 const instanceId = instanceIdOf(
440 olds?.instance ?? output?.instanceId ?? "",
441 );
442 if (instanceId.length === 0) return undefined;
443 const databaseId = yield* toDatabaseId(
444 id,
445 olds?.databaseId,
446 output?.databaseId,
447 );
448 const name =
449 output?.name ?? resourceName(env.project, instanceId, databaseId);
450 const existing = yield* getByName(name);
451 if (existing === undefined) return undefined;
452 return toAttrs(existing, env.project);
453 }),
454
455 list: () =>
456 Effect.gen(function* () {
457 const env = yield* GcpEnvironment.current;
458 const instances = yield* listAlchemyInstances(env.project);
459 const pages = yield* Effect.forEach(
460 instances,
461 (instance) => {
462 const parent = instance.name;
463 if (parent === undefined || parent.length === 0) {
464 return Effect.succeed([] as Database["Attributes"][]);
465 }
466 return spanner.listProjectsInstancesDatabases
467 .pages({
468 parent,
469 pageSize: 1000,
470 })
471 .pipe(
472 Stream.flatMap((page) =>
473 Stream.fromIterable(page.databases ?? []),
474 ),
475 Stream.map((database) => toAttrs(database, env.project)),
476 Stream.runCollect,
477 Effect.map((chunk) => Array.from(chunk)),
478 Effect.catchTag("NotFound", () =>
479 Effect.succeed([] as Database["Attributes"][]),
480 ),
481 );
482 },
483 { concurrency: 4 },
484 );
485 return pages.flat();
486 }),
487
488 reconcile: Effect.fn(function* ({ id, news, output }) {
489 const env = yield* GcpEnvironment.current;
490 const instanceId = instanceIdOf(news.instance);
491 const databaseId = yield* toDatabaseId(
492 id,
493 news.databaseId,
494 output?.databaseId,
495 );
496 const name = resourceName(env.project, instanceId, databaseId);
497 const dialect = normalizeDialect(news.databaseDialect);
498 const desiredProtection = news.enableDropProtection === true;
499
500 let current = yield* getByName(output?.name ?? name);
501
502 if (current === undefined) {
503 const created = yield* spanner
504 .createProjectsInstancesDatabases({
505 parent: instanceName(env.project, instanceId),
506 body: {
507 createStatement: createStatementOf(databaseId, dialect),
508 extraStatements: news.extraStatements,
509 encryptionConfig: news.encryptionConfig,
510 databaseDialect:
511 dialect === DEFAULT_DIALECT ? undefined : dialect,
512 },
513 })
514 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
515 if (created !== undefined) {
516 yield* waitForOperation(created, { alreadyExistsOk: true });
517 }
518 current = yield* waitUntilExists(name);
519 }
520
521 if (current === undefined) {
522 return yield* new DatabaseNotResolved({ name });
523 }
524
525 if (!isReadyState(current.state)) {
526 current = yield* waitUntilReady(name);
527 }
528
529 const observedProtection = current.enableDropProtection === true;
530 if (observedProtection !== desiredProtection) {
531 const patched = yield* retryConcurrentChanges(
532 spanner.patchProjectsInstancesDatabases({
533 name,
534 updateMask: "enableDropProtection",
535 body: {
536 name,
537 enableDropProtection: desiredProtection,
538 },
539 }),
540 );
541 yield* waitForOperation(patched);
542 current = yield* waitUntilReady(name);
543 }
544
545 return toAttrs(current, env.project);
546 }),
547
548 delete: Effect.fn(function* ({ output }) {
549 const current = yield* getByName(output.name);
550 if (current === undefined) return;
551
552 if (current.enableDropProtection === true) {
553 const patched = yield* retryConcurrentChanges(
554 spanner.patchProjectsInstancesDatabases({
555 name: output.name,
556 updateMask: "enableDropProtection",
557 body: {
558 name: output.name,
559 enableDropProtection: false,
560 },
561 }),
562 );
563 yield* waitForOperation(patched);
564 }
565
566 yield* retryConcurrentChanges(
567 spanner
568 .dropDatabaseProjectsInstancesDatabases({ database: output.name })
569 .pipe(Effect.catchTag("NotFound", () => Effect.void)),
570 );
571 yield* waitUntilGone(output.name);
572 }),
573 });