1import * as rdc from "@distilled.cloud/cloudflare/r2-data-catalog";
2import * as Effect from "effect/Effect";
3import * as Predicate from "effect/Predicate";
4import * as Redacted from "effect/Redacted";
5import * as Schedule from "effect/Schedule";
6
7import { Unowned } from "../../AdoptPolicy.ts";
8import { isResolved } from "../../Diff.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
12import type { Providers } from "../Providers.ts";
13
14const TypeId = "Cloudflare.R2.DataCatalog" as const;
15type TypeId = typeof TypeId;
16
17/**
18 * Whether a maintenance job (compaction or snapshot expiration) runs.
19 */
20export type MaintenanceState = "enabled" | "disabled";
21
22/**
23 * Target output file size, in MB, that catalog compaction rewrites small
24 * files into.
25 */
26export type TargetSizeMb = "64" | "128" | "256" | "512";
27
28/**
29 * Catalog-level compaction maintenance settings.
30 */
31export type Compaction = {
32 /**
33 * Whether compaction runs for tables in this catalog.
34 */
35 state?: MaintenanceState;
36 /**
37 * Target output file size, in MB.
38 * @default "128"
39 */
40 targetSizeMb?: TargetSizeMb;
41};
42
43/**
44 * Catalog-level snapshot expiration maintenance settings.
45 */
46export type SnapshotExpiration = {
47 /**
48 * Whether snapshot expiration runs for tables in this catalog.
49 */
50 state?: MaintenanceState;
51 /**
52 * Maximum age a snapshot may reach before it is expired, expressed as a
53 * duration string (e.g. `"7d"`).
54 */
55 maxSnapshotAge?: string;
56 /**
57 * Minimum number of snapshots retained per table regardless of age.
58 */
59 minSnapshotsToKeep?: number;
60};
61
62export type DataCatalogProps = {
63 /**
64 * Name of the R2 bucket to enable the Iceberg data catalog on. The bucket
65 * must already exist — pass `bucket.bucketName` from a `Cloudflare.R2.Bucket`
66 * resource to order catalog-after-bucket. Changing the bucket replaces the
67 * catalog (the old bucket's catalog is disabled; table data is untouched).
68 */
69 bucketName: string;
70 /**
71 * Compaction maintenance configuration. Only the fields you specify are
72 * enforced; omitted fields keep Cloudflare's defaults.
73 */
74 compaction?: Compaction;
75 /**
76 * Snapshot expiration maintenance configuration. Only the fields you
77 * specify are enforced; omitted fields keep Cloudflare's defaults.
78 */
79 snapshotExpiration?: SnapshotExpiration;
80 /**
81 * Cloudflare API token (with R2 read/write access) the catalog uses to
82 * run maintenance jobs against the bucket. Write-only: Cloudflare exposes
83 * only `credentialStatus: "present" | "absent"`, never the token itself.
84 * Maintenance jobs stay pending until a credential is provided.
85 */
86 token?: Redacted.Redacted<string>;
87};
88
89export type DataCatalogAttributes = {
90 /**
91 * Unique identifier of the catalog (stable across disable/enable cycles).
92 */
93 catalogId: string;
94 /**
95 * Catalog (warehouse) name, generated by Cloudflare as
96 * `{accountId}_{bucketName}`.
97 */
98 name: string;
99 /**
100 * Name of the R2 bucket backing the catalog.
101 */
102 bucketName: string;
103 /**
104 * The Cloudflare account the catalog belongs to.
105 */
106 accountId: string;
107 /**
108 * Catalog status. A reconciled catalog is always `active`.
109 */
110 status: "active" | "inactive";
111 /**
112 * Whether a maintenance credential is registered for this catalog.
113 */
114 credentialStatus: "present" | "absent" | (string & {});
115 /**
116 * Observed compaction maintenance configuration.
117 */
118 compaction:
119 | {
120 state: MaintenanceState;
121 targetSizeMb: TargetSizeMb;
122 }
123 | undefined;
124 /**
125 * Observed snapshot expiration maintenance configuration.
126 */
127 snapshotExpiration:
128 | {
129 state: MaintenanceState;
130 maxSnapshotAge: string;
131 minSnapshotsToKeep: number;
132 }
133 | undefined;
134 /**
135 * Iceberg REST catalog URI for this warehouse — point PyIceberg, Spark, or
136 * any Iceberg REST client at this endpoint.
137 */
138 catalogUri: string;
139};
140
141export type DataCatalog = Resource<
142 TypeId,
143 DataCatalogProps,
144 DataCatalogAttributes,
145 never,
146 Providers
147>;
148
149/**
150 * Apache Iceberg data catalog attached to a Cloudflare R2 bucket.
151 *
152 * R2 Data Catalog exposes an Iceberg REST catalog endpoint backed by an R2
153 * bucket, so engines like Spark, PyIceberg, and DuckDB can create and query
154 * Iceberg tables stored in R2. The catalog is a singleton per bucket: this
155 * resource enables it, keeps its maintenance configuration in sync, and
156 * disables it on destroy (table data in the bucket is never deleted).
157 * ### Enabling a catalog
158 * **Example:** Enable the catalog on an R2 bucket
159 * ```typescript
160 * const bucket = yield* Cloudflare.R2.Bucket("LakehouseBucket");
161 *
162 * const catalog = yield* Cloudflare.R2.R2DataCatalog("Lakehouse", {
163 * bucketName: bucket.bucketName,
164 * });
165 *
166 * // Point any Iceberg REST client at the warehouse:
167 * const uri = catalog.catalogUri;
168 * const warehouse = catalog.name;
169 * ```
170 *
171 * ### Maintenance
172 * **Example:** Configure compaction and snapshot expiration
173 * ```typescript
174 * const catalog = yield* Cloudflare.R2.R2DataCatalog("Lakehouse", {
175 * bucketName: bucket.bucketName,
176 * compaction: { state: "enabled", targetSizeMb: "256" },
177 * snapshotExpiration: {
178 * state: "enabled",
179 * maxSnapshotAge: "3d",
180 * minSnapshotsToKeep: 5,
181 * },
182 * });
183 * ```
184 *
185 * **Example:** Register a maintenance credential
186 * ```typescript
187 * // Maintenance jobs need an API token with R2 read/write on the bucket.
188 * const catalog = yield* Cloudflare.R2.R2DataCatalog("Lakehouse", {
189 * bucketName: bucket.bucketName,
190 * compaction: { state: "enabled" },
191 * token: maintenanceToken, // Redacted<string>
192 * });
193 * ```
194 *
195 * @see https://developers.cloudflare.com/r2/data-catalog/
196 *
197 * @resource
198 * @product R2 Data Catalog
199 * @category Storage & Databases
200 */
201export const DataCatalog = Resource<DataCatalog>(TypeId);
202
203/**
204 * Returns true if the given value is an R2DataCatalog resource.
205 */
206export const isDataCatalog = (value: unknown): value is DataCatalog =>
207 Predicate.hasProperty(value, "Type") && value.Type === TypeId;
208
209export const DataCatalogProvider = () =>
210 Provider.succeed(DataCatalog, {
211 stables: ["catalogId", "name", "accountId", "bucketName", "catalogUri"],
212 diff: Effect.fn(function* ({ olds, news, output }) {
213 const { accountId } = yield* yield* CloudflareEnvironment;
214 if (!isResolved(news)) return undefined;
215 if ((output?.accountId ?? accountId) !== accountId) {
216 return { action: "replace" } as const;
217 }
218 // The catalog is keyed to its bucket — moving buckets is a replacement
219 // (disable on the old bucket, enable on the new one). bucketName is an
220 // Input<string>; compare only once both sides are concrete strings.
221 const oldBucket =
222 output?.bucketName ??
223 (typeof olds?.bucketName === "string" ? olds.bucketName : undefined);
224 if (
225 oldBucket !== undefined &&
226 typeof news.bucketName === "string" &&
227 oldBucket !== news.bucketName
228 ) {
229 return { action: "replace" } as const;
230 }
231 return undefined;
232 }),
233 read: Effect.fn(function* ({ output, olds }) {
234 const { accountId } = yield* yield* CloudflareEnvironment;
235 const acct = output?.accountId ?? accountId;
236 const bucketName =
237 output?.bucketName ??
238 (typeof olds?.bucketName === "string" ? olds.bucketName : undefined);
239 if (bucketName === undefined) return undefined;
240
241 const observed = yield* getCatalog(acct, bucketName);
242 // `inactive` means the catalog was disabled — for reconciliation
243 // purposes the resource is gone (re-enable on the next deploy).
244 if (!observed || observed.status !== "active") return undefined;
245 const attrs = toAttributes(observed, acct);
246 // The catalog carries no ownership markers; if we have no state of our
247 // own, an active catalog may have been enabled out-of-band. Brand it
248 // `Unowned` so the engine gates takeover behind the adopt policy.
249 return output ? attrs : Unowned(attrs);
250 }),
251 list: Effect.fn(function* () {
252 const { accountId } = yield* yield* CloudflareEnvironment;
253 // R2 Data Catalog is an account-scoped collection: one warehouse per
254 // bucket that has the catalog enabled. Only `active` warehouses map to
255 // a live resource — `read` treats `inactive` (disabled) as gone.
256 return yield* rdc.listR2DataCatalogs({ accountId }).pipe(
257 Effect.map(({ warehouses }) =>
258 warehouses
259 .filter((w) => w.status === "active")
260 .map((w) => toAttributes(w, accountId)),
261 ),
262 // Accounts without R2 Data Catalog access reject the route entirely.
263 Effect.catchTag("InvalidRoute", () => Effect.succeed([])),
264 );
265 }),
266 reconcile: Effect.fn(function* ({ news, olds, output }) {
267 const { accountId } = yield* yield* CloudflareEnvironment;
268 const acct = output?.accountId ?? accountId;
269 // Inputs have been resolved to concrete strings by Plan.
270 const bucketName = news.bucketName as string;
271
272 // Observe — `output` is only a cache; a disabled or never-enabled
273 // catalog observes as missing/inactive and falls through to ensure.
274 let observed = yield* getCatalog(acct, bucketName);
275
276 // Ensure — enable is a true idempotent upsert (re-running it on an
277 // active catalog returns the same id), so no AlreadyExists race exists.
278 // A freshly-created bucket can briefly 404 the enable endpoint
279 // (`NoSuchBucket`, code 40406) — ride out that consistency lag.
280 if (!observed || observed.status !== "active") {
281 yield* rdc.enableR2DataCatalog({ accountId: acct, bucketName }).pipe(
282 Effect.retry({
283 while: (e) => e._tag === "NoSuchBucket",
284 schedule: catalogConsistencySchedule,
285 }),
286 );
287 observed = yield* rdc
288 .getR2DataCatalog({ accountId: acct, bucketName })
289 .pipe(
290 // The catalog can lag behind its own enable call.
291 Effect.retry({
292 while: (e) => e._tag === "WarehouseNotFound",
293 schedule: catalogConsistencySchedule,
294 }),
295 );
296 }
297
298 // Sync credential before maintenance — Cloudflare rejects maintenance
299 // updates when the warehouse has no credential registered.
300 // The API exposes only present/absent, so `olds` serves as the rotation
301 // hint: re-push when the token value changed or no credential is
302 // registered (adoption re-pushes; idempotent).
303 let credentialStatus = (observed.credentialStatus ?? "absent") as
304 | "present"
305 | "absent";
306 if (news.token !== undefined) {
307 const rotated =
308 olds?.token === undefined ||
309 Redacted.value(olds.token) !== Redacted.value(news.token);
310 if (credentialStatus !== "present" || rotated) {
311 yield* rdc.createCredential({
312 accountId: acct,
313 bucketName,
314 token: Redacted.value(news.token),
315 });
316 credentialStatus = "present";
317 }
318 }
319
320 // Sync maintenance — diff observed config against the fields the user
321 // actually specified; skip the API entirely on a no-op.
322 let maintenance = observed.maintenanceConfig ?? undefined;
323 const observedCompaction = maintenance?.compaction ?? undefined;
324 const observedExpiration = maintenance?.snapshotExpiration ?? undefined;
325 const compactionDirty =
326 news.compaction !== undefined &&
327 ((news.compaction.state !== undefined &&
328 news.compaction.state !== observedCompaction?.state) ||
329 (news.compaction.targetSizeMb !== undefined &&
330 news.compaction.targetSizeMb !== observedCompaction?.targetSizeMb));
331 const expirationDirty =
332 news.snapshotExpiration !== undefined &&
333 ((news.snapshotExpiration.state !== undefined &&
334 news.snapshotExpiration.state !== observedExpiration?.state) ||
335 (news.snapshotExpiration.maxSnapshotAge !== undefined &&
336 news.snapshotExpiration.maxSnapshotAge !==
337 observedExpiration?.maxSnapshotAge) ||
338 (news.snapshotExpiration.minSnapshotsToKeep !== undefined &&
339 news.snapshotExpiration.minSnapshotsToKeep !==
340 observedExpiration?.minSnapshotsToKeep));
341 if (compactionDirty || expirationDirty) {
342 const updated = yield* rdc.updateMaintenanceConfig({
343 accountId: acct,
344 bucketName,
345 compaction: news.compaction,
346 snapshotExpiration: news.snapshotExpiration,
347 });
348 maintenance = {
349 compaction: updated.compaction,
350 snapshotExpiration: updated.snapshotExpiration,
351 };
352 }
353
354 return toAttributes(
355 { ...observed, status: "active", maintenanceConfig: maintenance },
356 acct,
357 credentialStatus,
358 );
359 }),
360 delete: Effect.fn(function* ({ output }) {
361 // Disable is idempotent server-side; a never-enabled (or already
362 // disabled-and-bucket-deleted) catalog surfaces `WarehouseNotFound`.
363 yield* rdc
364 .disableR2DataCatalog({
365 accountId: output.accountId,
366 bucketName: output.bucketName,
367 })
368 .pipe(
369 Effect.catchTag(
370 ["WarehouseNotFound", "NoSuchBucket"],
371 () => Effect.void,
372 ),
373 );
374 }),
375 });
376
377// A freshly-created bucket (or freshly-enabled catalog) can briefly 404 on
378// the r2-catalog endpoints before the warehouse propagates.
379const catalogConsistencySchedule = Schedule.max([
380 Schedule.exponential(100),
381 Schedule.recurs(5),
382]);
383
384/**
385 * Read a catalog by bucket name, mapping "gone" (`WarehouseNotFound`, code
386 * 40401, or a missing bucket) to `undefined`.
387 */
388const getCatalog = (accountId: string, bucketName: string) =>
389 rdc
390 .getR2DataCatalog({ accountId, bucketName })
391 .pipe(
392 Effect.catchTag(["WarehouseNotFound", "NoSuchBucket"], () =>
393 Effect.succeed(undefined),
394 ),
395 );
396
397const toAttributes = (
398 catalog: rdc.GetR2DataCatalogResponse,
399 accountId: string,
400 credentialStatus?: "present" | "absent",
401): DataCatalogAttributes => ({
402 catalogId: catalog.id,
403 name: catalog.name,
404 bucketName: catalog.bucket,
405 accountId,
406 // Distilled widens generated string enums to open unions (`string & {}`).
407 status: catalog.status as "active" | "inactive",
408 credentialStatus: credentialStatus ?? catalog.credentialStatus ?? "absent",
409 compaction: catalog.maintenanceConfig?.compaction
410 ? {
411 state: catalog.maintenanceConfig.compaction.state as MaintenanceState,
412 targetSizeMb: catalog.maintenanceConfig.compaction
413 .targetSizeMb as TargetSizeMb,
414 }
415 : undefined,
416 snapshotExpiration: catalog.maintenanceConfig?.snapshotExpiration
417 ? {
418 state: catalog.maintenanceConfig.snapshotExpiration
419 .state as MaintenanceState,
420 maxSnapshotAge:
421 catalog.maintenanceConfig.snapshotExpiration.maxSnapshotAge,
422 minSnapshotsToKeep:
423 catalog.maintenanceConfig.snapshotExpiration.minSnapshotsToKeep,
424 }
425 : undefined,
426 catalogUri: `https://catalog.cloudflarestorage.com/${accountId}/${catalog.bucket}`,
427});