| 1 | import * as biglake from "@distilled.cloud/gcp/biglake_v1"; |
| 2 | import * as Effect from "effect/Effect"; |
| 3 | import { Unowned } from "../../AdoptPolicy.ts"; |
| 4 | import { isResolved } from "../../Diff.ts"; |
| 5 | import * as Provider from "../../Provider.ts"; |
| 6 | import { Resource } from "../../Resource.ts"; |
| 7 | import { GcpEnvironment } from "../Environment.ts"; |
| 8 | import type { Providers } from "../Providers.ts"; |
| 9 | import { |
| 10 | BiglakeNotResolved, |
| 11 | listLocations, |
| 12 | MAX_ID_LENGTH, |
| 13 | hasAlchemyLabels, |
| 14 | hasOwnershipMarker, |
| 15 | ignoreGone, |
| 16 | listCatalogs, |
| 17 | listDatabases, |
| 18 | listTables, |
| 19 | locationParent, |
| 20 | missingGet, |
| 21 | namedOf, |
| 22 | normalizeLocation, |
| 23 | parseOwnershipMarker, |
| 24 | parseResourceName, |
| 25 | replaceOnIdentity, |
| 26 | retryTransient, |
| 27 | rfc1035, |
| 28 | toPhysicalId, |
| 29 | waitUntilGone, |
| 30 | } from "./internal.ts"; |
| 31 | |
| 32 | export type CatalogProps = { |
| 33 | /** |
| 34 | * Catalog id (the `{catalog}` segment of |
| 35 | * `projects/{project}/locations/{location}/catalogs/{catalog}`). If |
| 36 | * omitted, a unique id is generated. Immutable — changing it replaces |
| 37 | * the catalog. |
| 38 | */ |
| 39 | catalogId?: string; |
| 40 | /** |
| 41 | * Location (`us-central1`, `US`, …). Immutable — changing it replaces |
| 42 | * the catalog. `US-CENTRAL1` is accepted and normalized to |
| 43 | * `us-central1`. |
| 44 | * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`) |
| 45 | */ |
| 46 | location?: string; |
| 47 | }; |
| 48 | |
| 49 | export type Catalog = Resource< |
| 50 | "GCP.BigLake.Catalog", |
| 51 | CatalogProps, |
| 52 | { |
| 53 | /** Full resource name. */ |
| 54 | name: string; |
| 55 | /** Catalog id (last path segment). */ |
| 56 | catalogId: string; |
| 57 | /** Project id. */ |
| 58 | project: string; |
| 59 | /** Location id. */ |
| 60 | location: string; |
| 61 | /** RFC3339 creation timestamp. */ |
| 62 | createTime: string | undefined; |
| 63 | /** RFC3339 last-update timestamp. */ |
| 64 | updateTime: string | undefined; |
| 65 | /** RFC3339 deletion timestamp, if the catalog is soft-deleted. */ |
| 66 | deleteTime: string | undefined; |
| 67 | /** RFC3339 expire timestamp after delete. */ |
| 68 | expireTime: string | undefined; |
| 69 | }, |
| 70 | never, |
| 71 | Providers |
| 72 | >; |
| 73 | |
| 74 | /** |
| 75 | * A BigLake Metastore catalog — the top-level container for Hive |
| 76 | * databases and tables. |
| 77 | * |
| 78 | * Catalogs have no labels, description, or annotations, and Alchemy does |
| 79 | * not write ownership into the catalog IAM policy: generated ids carry an |
| 80 | * `alc_` prefix that `list` / nuke match, and `read` reports an explicit |
| 81 | * id it has no state for as unowned. |
| 82 | * Location and catalog id are immutable. There is nothing else to |
| 83 | * update in place. |
| 84 | * |
| 85 | * ### Creating a Catalog |
| 86 | * **Example:** Generated id |
| 87 | * ```typescript |
| 88 | * const catalog = yield* GCP.BigLake.Catalog("Lake", {}); |
| 89 | * ``` |
| 90 | * |
| 91 | * **Example:** Named catalog in a multi-region |
| 92 | * ```typescript |
| 93 | * const catalog = yield* GCP.BigLake.Catalog("Lake", { |
| 94 | * catalogId: "analytics", |
| 95 | * location: "US", |
| 96 | * }); |
| 97 | * ``` |
| 98 | * |
| 99 | * @resource |
| 100 | * @category BigLake |
| 101 | */ |
| 102 | export const Catalog = Resource<Catalog>("GCP.BigLake.Catalog"); |
| 103 | |
| 104 | const OWNERSHIP_TITLE = "alchemy-ownership"; |
| 105 | const resourceName = (project: string, location: string, catalogId: string) => |
| 106 | `${locationParent(project, location)}/catalogs/${catalogId}`; |
| 107 | |
| 108 | const toAttrs = (catalog: biglake.Catalog, project: string) => { |
| 109 | const name = catalog.name ?? ""; |
| 110 | const parsed = parseResourceName(name, "catalogs"); |
| 111 | return { |
| 112 | name, |
| 113 | catalogId: parsed.id, |
| 114 | project, |
| 115 | location: parsed.location, |
| 116 | createTime: catalog.createTime, |
| 117 | updateTime: catalog.updateTime, |
| 118 | deleteTime: catalog.deleteTime, |
| 119 | expireTime: catalog.expireTime, |
| 120 | }; |
| 121 | }; |
| 122 | |
| 123 | const getByName = missingGet(biglake.getProjectsLocationsCatalogs); |
| 124 | |
| 125 | const policyHasOwnership = (policy: biglake.Policy | undefined) => |
| 126 | (policy?.bindings ?? []).some( |
| 127 | (binding) => |
| 128 | binding.condition?.title === OWNERSHIP_TITLE || |
| 129 | hasOwnershipMarker(binding.condition?.description), |
| 130 | ); |
| 131 | |
| 132 | const getPolicy = (name: string) => |
| 133 | name.length === 0 |
| 134 | ? Effect.succeed(undefined) |
| 135 | : biglake |
| 136 | .getIamPolicyProjectsCatalogs({ |
| 137 | resource: name, |
| 138 | "options.requestedPolicyVersion": 3, |
| 139 | }) |
| 140 | .pipe( |
| 141 | Effect.catchTag(["NotFound", "BigLakeResourceNotFound"], () => |
| 142 | Effect.succeed(undefined), |
| 143 | ), |
| 144 | ); |
| 145 | |
| 146 | const toCatalogId = ( |
| 147 | id: string, |
| 148 | explicit: string | undefined, |
| 149 | existing: string | undefined, |
| 150 | ) => |
| 151 | Effect.gen(function* () { |
| 152 | if (explicit !== undefined) return explicit; |
| 153 | if (existing !== undefined) return existing; |
| 154 | const generated = yield* toPhysicalId(id, undefined, undefined); |
| 155 | if (generated.startsWith("alc_")) return generated; |
| 156 | return rfc1035(`alc_${generated}`, MAX_ID_LENGTH); |
| 157 | }); |
| 158 | |
| 159 | const lastIdOwned = (name: string | undefined) => { |
| 160 | const id = parseResourceName(name ?? "", "catalogs").id; |
| 161 | return id.startsWith("alc_"); |
| 162 | }; |
| 163 | |
| 164 | const isOwnedCatalog = (catalog: biglake.Catalog, policy?: biglake.Policy) => |
| 165 | policyHasOwnership(policy) || lastIdOwned(catalog.name); |
| 166 | |
| 167 | const ownedFromPolicy = (id: string, policy: biglake.Policy | undefined) => |
| 168 | Effect.gen(function* () { |
| 169 | for (const binding of policy?.bindings ?? []) { |
| 170 | const labels = parseOwnershipMarker(binding.condition?.description); |
| 171 | if (yield* hasAlchemyLabels(id, labels)) return true; |
| 172 | } |
| 173 | return false; |
| 174 | }); |
| 175 | |
| 176 | const emptyCatalog = (name: string) => |
| 177 | Effect.gen(function* () { |
| 178 | const databases = yield* listDatabases(name); |
| 179 | yield* Effect.forEach( |
| 180 | namedOf(databases), |
| 181 | (database) => emptyAndDeleteDatabase(database.name!), |
| 182 | { concurrency: 1 }, |
| 183 | ); |
| 184 | }); |
| 185 | |
| 186 | const emptyAndDeleteDatabase = (name: string) => |
| 187 | Effect.gen(function* () { |
| 188 | const tables = yield* listTables(name); |
| 189 | yield* Effect.forEach( |
| 190 | namedOf(tables), |
| 191 | (table) => |
| 192 | ignoreGone( |
| 193 | biglake.deleteProjectsLocationsCatalogsDatabasesTables({ |
| 194 | name: table.name!, |
| 195 | }), |
| 196 | ), |
| 197 | { concurrency: 4 }, |
| 198 | ); |
| 199 | yield* ignoreGone( |
| 200 | biglake.deleteProjectsLocationsCatalogsDatabases({ name }), |
| 201 | ); |
| 202 | }); |
| 203 | |
| 204 | export const CatalogProvider = () => |
| 205 | Provider.succeed(Catalog, { |
| 206 | stables: ["name", "catalogId", "project", "location", "createTime"], |
| 207 | |
| 208 | diff: Effect.fn(function* ({ news, olds, output }) { |
| 209 | const env = yield* GcpEnvironment.current; |
| 210 | if (!isResolved(news)) return undefined; |
| 211 | return replaceOnIdentity({ |
| 212 | previousId: olds?.catalogId ?? output?.catalogId, |
| 213 | nextId: news.catalogId, |
| 214 | previousLocation: normalizeLocation( |
| 215 | olds?.location ?? output?.location, |
| 216 | env.region, |
| 217 | ), |
| 218 | nextLocation: normalizeLocation( |
| 219 | news.location ?? olds?.location ?? output?.location, |
| 220 | env.region, |
| 221 | ), |
| 222 | }); |
| 223 | }), |
| 224 | |
| 225 | read: Effect.fn(function* ({ id, olds, output }) { |
| 226 | const env = yield* GcpEnvironment.current; |
| 227 | const location = normalizeLocation( |
| 228 | olds?.location ?? output?.location, |
| 229 | env.region, |
| 230 | ); |
| 231 | const catalogId = yield* toCatalogId( |
| 232 | id, |
| 233 | olds?.catalogId, |
| 234 | output?.catalogId, |
| 235 | ); |
| 236 | const name = |
| 237 | output?.name ?? resourceName(env.project, location, catalogId); |
| 238 | const existing = yield* getByName(name); |
| 239 | if (existing === undefined) return undefined; |
| 240 | const attrs = toAttrs(existing, env.project); |
| 241 | // Catalogs have no labels. A generated `alc_` id derives from this |
| 242 | // stack, stage, logical id and instance; an explicit id is ours when |
| 243 | // state has it or an earlier Alchemy version stamped the IAM policy. |
| 244 | if (output !== undefined || olds?.catalogId === undefined) return attrs; |
| 245 | const policy = yield* getPolicy(existing.name ?? name); |
| 246 | return (yield* ownedFromPolicy(id, policy)) ? attrs : Unowned(attrs); |
| 247 | }), |
| 248 | |
| 249 | list: () => |
| 250 | Effect.gen(function* () { |
| 251 | const env = yield* GcpEnvironment.current; |
| 252 | const catalogs = (yield* Effect.forEach( |
| 253 | listLocations(env.region), |
| 254 | (location) => listCatalogs(locationParent(env.project, location)), |
| 255 | )).flat(); |
| 256 | const owned = yield* Effect.forEach( |
| 257 | namedOf(catalogs), |
| 258 | (catalog) => |
| 259 | Effect.gen(function* () { |
| 260 | const policy = yield* getPolicy(catalog.name!); |
| 261 | return isOwnedCatalog(catalog, policy) ? catalog : undefined; |
| 262 | }), |
| 263 | { concurrency: 4 }, |
| 264 | ); |
| 265 | return owned |
| 266 | .filter((item): item is biglake.Catalog => item !== undefined) |
| 267 | .map((item) => toAttrs(item, env.project)); |
| 268 | }), |
| 269 | |
| 270 | reconcile: Effect.fn(function* ({ id, news, output }) { |
| 271 | const env = yield* GcpEnvironment.current; |
| 272 | const location = normalizeLocation( |
| 273 | news.location ?? output?.location, |
| 274 | env.region, |
| 275 | ); |
| 276 | const catalogId = yield* toCatalogId( |
| 277 | id, |
| 278 | news.catalogId, |
| 279 | output?.catalogId, |
| 280 | ); |
| 281 | const parent = locationParent(env.project, location); |
| 282 | const name = |
| 283 | output?.name ?? resourceName(env.project, location, catalogId); |
| 284 | let current = yield* getByName(name); |
| 285 | |
| 286 | if (current === undefined) { |
| 287 | const created = yield* retryTransient( |
| 288 | biglake.createProjectsLocationsCatalogs({ |
| 289 | parent, |
| 290 | catalogId, |
| 291 | body: {}, |
| 292 | }), |
| 293 | ).pipe(Effect.catchTag("Conflict", () => getByName(name))); |
| 294 | current = created ?? undefined; |
| 295 | } |
| 296 | |
| 297 | if (current === undefined) { |
| 298 | return yield* new BiglakeNotResolved({ name }); |
| 299 | } |
| 300 | |
| 301 | return toAttrs(current, env.project); |
| 302 | }), |
| 303 | |
| 304 | delete: Effect.fn(function* ({ output }) { |
| 305 | if (!output.name) return; |
| 306 | yield* emptyCatalog(output.name); |
| 307 | yield* ignoreGone( |
| 308 | biglake.deleteProjectsLocationsCatalogs({ name: output.name }), |
| 309 | ); |
| 310 | yield* waitUntilGone(getByName(output.name)); |
| 311 | }), |
| 312 | }); |