1import * as bigquery from "@distilled.cloud/gcp/bigquery_v2";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
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";
13
14const MAX_POLICY_ID_LENGTH = 256;
15
16export type RowAccessPolicyProps = {
17 /**
18 * Dataset id (the `{dataset}` segment of
19 * `projects/{project}/datasets/{dataset}`) or a full dataset resource
20 * name. Immutable — changing it replaces the policy.
21 */
22 datasetId: string;
23 /**
24 * Table id (the `{table}` segment of
25 * `projects/{project}/datasets/{dataset}/tables/{table}`) or a full
26 * table resource name. Immutable — changing it replaces the policy.
27 */
28 tableId: string;
29 /**
30 * Policy id (the `{policy}` segment of
31 * `projects/{project}/datasets/{dataset}/tables/{table}/rowAccessPolicies/{policy}`).
32 * If omitted, a unique name is generated from the stack, stage, and
33 * logical id. Letters, numbers, and underscores; max 256 characters.
34 * Immutable — changing it replaces the policy.
35 */
36 policyId?: string;
37 /**
38 * SQL boolean expression that selects the rows this policy covers,
39 * similar to a `WHERE` clause. Sent verbatim.
40 */
41 filterPredicate: string;
42 /**
43 * Initial IAM members granted this policy (`user:`, `group:`,
44 * `serviceAccount:`, `domain:`, `allAuthenticatedUsers`, `allUsers`).
45 * Input-only — sent on create, not readable after insert.
46 */
47 grantees?: string[];
48};
49
50export type RowAccessPolicy = Resource<
51 "GCP.BigQuery.RowAccessPolicy",
52 RowAccessPolicyProps,
53 {
54 /**
55 * Resource path
56 * `projects/{project}/datasets/{dataset}/tables/{table}/rowAccessPolicies/{policy}`.
57 */
58 name: string;
59 /** Policy id (last path segment). */
60 policyId: string;
61 /** Table id. */
62 tableId: string;
63 /** Dataset id. */
64 datasetId: string;
65 /** Project id. */
66 project: string;
67 /** Filter predicate (a legacy Alchemy ownership suffix is stripped). */
68 filterPredicate: string;
69 /** Server etag. */
70 etag: string | undefined;
71 /** Creation time in milliseconds since epoch. */
72 creationTime: string | undefined;
73 /** Last-modified time in milliseconds since epoch. */
74 lastModifiedTime: string | undefined;
75 },
76 never,
77 Providers
78>;
79
80/**
81 * A Google BigQuery row-access policy — a filter predicate plus IAM
82 * members that together hide rows from unauthorized readers.
83 *
84 * Row-access policies have no labels, and Alchemy never edits the filter
85 * predicate: `list` / `pnpm nuke:gcp` return the policies on tables that
86 * carry `alchemy-*` labels, and `read` reports a policy with an explicit
87 * id that Alchemy has no state for as unowned. `datasetId`, `tableId`, and `policyId`
88 * are immutable (changing them replaces the policy). `filterPredicate`
89 * updates in place. `grantees` is create-only.
90 *
91 * ### Creating a Row Access Policy
92 * **Example:** Filter non-null rows
93 * ```typescript
94 * const dataset = yield* GCP.BigQuery.Dataset("Analytics", {
95 * forceDestroy: true,
96 * });
97 * const table = yield* GCP.BigQuery.Table("Events", {
98 * datasetId: dataset.datasetId,
99 * schema: [{ name: "nullable_field", type: "STRING" }],
100 * });
101 * const policy = yield* GCP.BigQuery.RowAccessPolicy("Visible", {
102 * datasetId: dataset.datasetId,
103 * tableId: table.tableId,
104 * filterPredicate: "nullable_field IS NOT NULL",
105 * });
106 * ```
107 *
108 * **Example:** Grant a domain
109 * ```typescript
110 * const policy = yield* GCP.BigQuery.RowAccessPolicy("Domain", {
111 * datasetId: dataset.datasetId,
112 * tableId: table.tableId,
113 * filterPredicate: "TRUE",
114 * grantees: ["domain:example.com"],
115 * });
116 * ```
117 *
118 * @resource
119 * @category BigQuery
120 */
121export const RowAccessPolicy = Resource<RowAccessPolicy>(
122 "GCP.BigQuery.RowAccessPolicy",
123);
124
125export class RowAccessPolicyNotResolved extends Data.TaggedError(
126 "GCP.BigQuery.RowAccessPolicyNotResolved",
127)<{
128 name: string;
129}> {}
130
131const lastSegment = (value: string) => {
132 const trimmed = value.replace(/\/+$/, "");
133 const parts = trimmed.split("/");
134 return parts[parts.length - 1] || trimmed;
135};
136
137const datasetIdOf = (datasetId: string) =>
138 datasetId.includes("/") ? lastSegment(datasetId) : datasetId;
139
140const tableIdOf = (tableId: string) =>
141 tableId.includes("/") ? lastSegment(tableId) : tableId;
142
143const resourceName = (
144 project: string,
145 datasetId: string,
146 tableId: string,
147 policyId: string,
148) =>
149 `projects/${project}/datasets/${datasetId}/tables/${tableId}/rowAccessPolicies/${policyId}`;
150
151const toId = (id: string, policyId: string | undefined, existing?: string) =>
152 Effect.gen(function* () {
153 if (policyId !== undefined) return policyId;
154 if (existing !== undefined) return existing;
155 const generated = yield* createPhysicalName({
156 id,
157 maxLength: MAX_POLICY_ID_LENGTH,
158 lowercase: true,
159 delimiter: "_",
160 });
161 return generated.replaceAll("-", "_");
162 });
163
164const MARKER_SUFFIX =
165 /\s+AND\s+\('\[alchemy ([^\]]+)\]'\s+IS\s+NOT\s+NULL\)\s*$/i;
166
167const parseMarkerLabels = (encoded: string): Record<string, string> => {
168 const labels: Record<string, string> = {};
169 for (const part of encoded.split(/\s+/)) {
170 const eq = part.indexOf("=");
171 if (eq > 0) {
172 labels[part.slice(0, eq)] = part.slice(eq + 1);
173 }
174 }
175 return labels;
176};
177
178const parseFilter = (
179 filterPredicate: string | undefined,
180): {
181 labels: Record<string, string>;
182 filterPredicate: string | undefined;
183} => {
184 if (filterPredicate === undefined) {
185 return { labels: {}, filterPredicate };
186 }
187 const match = filterPredicate.match(MARKER_SUFFIX);
188 if (match === null || match.index === undefined) {
189 return { labels: {}, filterPredicate };
190 }
191 const labels = parseMarkerLabels(match[1] ?? "");
192 let user = filterPredicate.slice(0, match.index).trim();
193 if (user.startsWith("(") && user.endsWith(")")) {
194 user = user.slice(1, -1);
195 }
196 return {
197 labels,
198 filterPredicate: user.length > 0 ? user : undefined,
199 };
200};
201
202const hasAlchemyDatasetLabels = (
203 labels: Record<string, string | undefined> | null | undefined,
204) =>
205 Object.keys(labels ?? {}).some((key) => key.startsWith(ALCHEMY_LABEL_PREFIX));
206
207const toAttrs = (policy: bigquery.RowAccessPolicy, project: string) => {
208 const ref = policy.rowAccessPolicyReference;
209 const policyId = ref?.policyId ?? "";
210 const datasetId = ref?.datasetId ?? "";
211 const tableId = ref?.tableId ?? "";
212 const projectId = ref?.projectId ?? project;
213 return {
214 name: resourceName(projectId, datasetId, tableId, policyId),
215 policyId,
216 tableId,
217 datasetId,
218 project: projectId,
219 filterPredicate:
220 parseFilter(policy.filterPredicate).filterPredicate ??
221 policy.filterPredicate ??
222 "",
223 etag: policy.etag,
224 creationTime: policy.creationTime,
225 lastModifiedTime: policy.lastModifiedTime,
226 };
227};
228
229const getByRef = (
230 projectId: string,
231 datasetId: string,
232 tableId: string,
233 policyId: string,
234) =>
235 bigquery
236 .getRowAccessPolicies({
237 projectId,
238 datasetId,
239 tableId,
240 policyId,
241 })
242 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
243
244const toPolicyBody = (
245 projectId: string,
246 datasetId: string,
247 tableId: string,
248 policyId: string,
249 news: RowAccessPolicyProps,
250 filter: string,
251 includeGrantees: boolean,
252): bigquery.RowAccessPolicy => {
253 const body: bigquery.RowAccessPolicy = {
254 rowAccessPolicyReference: {
255 projectId,
256 datasetId,
257 tableId,
258 policyId,
259 },
260 filterPredicate: filter,
261 };
262 if (includeGrantees && news.grantees !== undefined) {
263 body.grantees = news.grantees;
264 }
265 return body;
266};
267
268const sameFilter = (left: string, right: string) =>
269 left.replace(/\s+/g, " ").trim() === right.replace(/\s+/g, " ").trim();
270
271export const RowAccessPolicyProvider = () =>
272 Provider.succeed(RowAccessPolicy, {
273 stables: [
274 "name",
275 "policyId",
276 "tableId",
277 "datasetId",
278 "project",
279 "creationTime",
280 ],
281
282 diff: Effect.fn(function* ({ news, olds, output }) {
283 if (!isResolved(news)) return undefined;
284 const previousPolicyId = olds?.policyId ?? output?.policyId;
285 const nextPolicyId = news.policyId ?? previousPolicyId;
286 const policyIdChanged =
287 news.policyId !== undefined &&
288 previousPolicyId !== undefined &&
289 news.policyId !== previousPolicyId;
290
291 const previousDataset = olds?.datasetId ?? output?.datasetId;
292 const nextDataset = datasetIdOf(news.datasetId);
293 const datasetChanged =
294 previousDataset !== undefined &&
295 datasetIdOf(previousDataset) !== nextDataset;
296
297 const previousTable = olds?.tableId ?? output?.tableId;
298 const nextTable = tableIdOf(news.tableId);
299 const tableChanged =
300 previousTable !== undefined && tableIdOf(previousTable) !== nextTable;
301
302 if (!policyIdChanged && !datasetChanged && !tableChanged) {
303 return undefined;
304 }
305 return {
306 action: "replace" as const,
307 deleteFirst:
308 !policyIdChanged &&
309 !datasetChanged &&
310 !tableChanged &&
311 nextPolicyId !== undefined &&
312 previousPolicyId !== undefined &&
313 nextPolicyId === previousPolicyId,
314 };
315 }),
316
317 read: Effect.fn(function* ({ id, olds, output }) {
318 const env = yield* GcpEnvironment.current;
319 const policyId = yield* toId(id, olds?.policyId, output?.policyId);
320 const datasetId = datasetIdOf(olds?.datasetId ?? output?.datasetId ?? "");
321 const tableId = tableIdOf(olds?.tableId ?? output?.tableId ?? "");
322 if (datasetId.length === 0 || tableId.length === 0) return undefined;
323 const existing = yield* getByRef(
324 env.project,
325 datasetId,
326 tableId,
327 policyId,
328 );
329 if (existing === undefined) return undefined;
330 const attrs = toAttrs(existing, env.project);
331 // No labels: a generated id derives from this stack, stage, logical
332 // id and instance; an explicit id is only ours when state has it.
333 return output !== undefined || olds?.policyId === undefined
334 ? attrs
335 : Unowned(attrs);
336 }),
337
338 list: () =>
339 Effect.gen(function* () {
340 const env = yield* GcpEnvironment.current;
341 const datasets = yield* bigquery.listDatasets
342 .pages({
343 projectId: env.project,
344 maxResults: 1000,
345 all: true,
346 })
347 .pipe(
348 Stream.flatMap((page) => Stream.fromIterable(page.datasets ?? [])),
349 Stream.filter((dataset) => hasAlchemyDatasetLabels(dataset.labels)),
350 Stream.runCollect,
351 Effect.map((chunk) => Array.from(chunk)),
352 Effect.catchTag("NotFound", () =>
353 Effect.succeed([] as bigquery.DatasetListDatasetsItem[]),
354 ),
355 );
356 const pages = yield* Effect.forEach(
357 datasets,
358 (dataset) => {
359 const datasetId = dataset.datasetReference?.datasetId;
360 if (datasetId === undefined) {
361 return Effect.succeed([] as ReturnType<typeof toAttrs>[]);
362 }
363 return bigquery.listTables
364 .pages({
365 projectId: env.project,
366 datasetId,
367 maxResults: 1000,
368 })
369 .pipe(
370 Stream.flatMap((page) =>
371 Stream.fromIterable(page.tables ?? []),
372 ),
373 Stream.filter((table) => hasAlchemyDatasetLabels(table.labels)),
374 Stream.map((table) => table.tableReference?.tableId),
375 Stream.filter(
376 (tableId): tableId is string =>
377 tableId !== undefined && tableId.length > 0,
378 ),
379 Stream.runCollect,
380 Effect.flatMap((tableIds) =>
381 Effect.forEach(
382 Array.from(tableIds),
383 (tableId) =>
384 bigquery.listRowAccessPolicies
385 .pages({
386 projectId: env.project,
387 datasetId,
388 tableId,
389 pageSize: 1000,
390 })
391 .pipe(
392 Stream.flatMap((page) =>
393 Stream.fromIterable(page.rowAccessPolicies ?? []),
394 ),
395 Stream.map((policy) => toAttrs(policy, env.project)),
396 Stream.runCollect,
397 Effect.map((chunk) => Array.from(chunk)),
398 Effect.catchTag("NotFound", () =>
399 Effect.succeed([] as ReturnType<typeof toAttrs>[]),
400 ),
401 ),
402 { concurrency: 4 },
403 ),
404 ),
405 Effect.map((nested) => nested.flat()),
406 Effect.catchTag("NotFound", () =>
407 Effect.succeed([] as ReturnType<typeof toAttrs>[]),
408 ),
409 );
410 },
411 { concurrency: 4 },
412 );
413 return pages.flat();
414 }),
415
416 reconcile: Effect.fn(function* ({ id, news, output }) {
417 const env = yield* GcpEnvironment.current;
418 const policyId = yield* toId(id, news.policyId, output?.policyId);
419 const datasetId = datasetIdOf(news.datasetId);
420 const tableId = tableIdOf(news.tableId);
421 const name = resourceName(env.project, datasetId, tableId, policyId);
422 const filter = news.filterPredicate;
423
424 let current = yield* getByRef(env.project, datasetId, tableId, policyId);
425
426 if (current === undefined) {
427 const created = yield* bigquery
428 .insertRowAccessPolicies({
429 projectId: env.project,
430 datasetId,
431 tableId,
432 body: toPolicyBody(
433 env.project,
434 datasetId,
435 tableId,
436 policyId,
437 news,
438 filter,
439 true,
440 ),
441 })
442 .pipe(
443 Effect.catchTag("Conflict", () =>
444 getByRef(env.project, datasetId, tableId, policyId),
445 ),
446 );
447 current = created ?? undefined;
448 }
449
450 if (current === undefined) {
451 return yield* new RowAccessPolicyNotResolved({ name });
452 }
453
454 const observedFilter = current.filterPredicate ?? "";
455 if (!sameFilter(observedFilter, filter)) {
456 current = yield* bigquery
457 .updateRowAccessPolicies({
458 projectId: env.project,
459 datasetId,
460 tableId,
461 policyId,
462 body: toPolicyBody(
463 env.project,
464 datasetId,
465 tableId,
466 policyId,
467 news,
468 filter,
469 false,
470 ),
471 })
472 .pipe(
473 Effect.catchTag("NotFound", () =>
474 bigquery
475 .insertRowAccessPolicies({
476 projectId: env.project,
477 datasetId,
478 tableId,
479 body: toPolicyBody(
480 env.project,
481 datasetId,
482 tableId,
483 policyId,
484 news,
485 filter,
486 true,
487 ),
488 })
489 .pipe(
490 Effect.catchTag("Conflict", () =>
491 getByRef(env.project, datasetId, tableId, policyId),
492 ),
493 ),
494 ),
495 );
496 if (current === undefined) {
497 return yield* new RowAccessPolicyNotResolved({ name });
498 }
499 }
500
501 return toAttrs(current, env.project);
502 }),
503
504 delete: Effect.fn(function* ({ output }) {
505 yield* bigquery
506 .deleteRowAccessPolicies({
507 projectId: output.project,
508 datasetId: output.datasetId,
509 tableId: output.tableId,
510 policyId: output.policyId,
511 force: true,
512 })
513 .pipe(Effect.catchTag("NotFound", () => Effect.void));
514 }),
515 });