1import * as keyspaces from "@distilled.cloud/aws/keyspaces";
2import * as keyspacesstreams from "@distilled.cloud/aws/keyspacesstreams";
3import type * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
7import { Unowned } from "../../AdoptPolicy.ts";
8import { isResolved } from "../../Diff.ts";
9import { createPhysicalName } from "../../PhysicalName.ts";
10import * as Provider from "../../Provider.ts";
11import { Resource } from "../../Resource.ts";
12import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
13import { toWireSeconds } from "../../Util/Duration.ts";
14import type { Providers } from "../Providers.ts";
15
16/**
17 * A Cassandra column definition. `type` is a CQL data type such as `text`,
18 * `int`, `uuid`, `timestamp`, `blob`, or a collection like `list<text>`.
19 */
20export interface KeyspacesColumn {
21 /** Column name. */
22 name: string;
23 /** CQL data type, e.g. `text`, `int`, `uuid`. */
24 type: string;
25}
26
27/**
28 * A clustering key column and its sort order.
29 */
30export interface KeyspacesClusteringKey {
31 /** Name of a column also present in `columns`. */
32 name: string;
33 /**
34 * Sort order for the clustering key.
35 * @default "ASC"
36 */
37 orderBy?: "ASC" | "DESC";
38}
39
40/**
41 * Read/write throughput mode for a table.
42 */
43export interface KeyspacesCapacity {
44 /**
45 * Billing mode.
46 * @default "PAY_PER_REQUEST"
47 */
48 throughputMode: "PAY_PER_REQUEST" | "PROVISIONED";
49 /** Provisioned read capacity units (only for `PROVISIONED`). */
50 readCapacityUnits?: number;
51 /** Provisioned write capacity units (only for `PROVISIONED`). */
52 writeCapacityUnits?: number;
53}
54
55/**
56 * Change data capture (CDC) stream configuration for a table.
57 */
58export interface KeyspacesCdcSpecification {
59 /**
60 * Whether CDC is enabled on the table.
61 */
62 status: "ENABLED" | "DISABLED";
63 /**
64 * What data is written to the table's stream for each changed row.
65 * @default "NEW_AND_OLD_IMAGES"
66 */
67 viewType?: "NEW_IMAGE" | "OLD_IMAGE" | "KEYS_ONLY" | "NEW_AND_OLD_IMAGES";
68}
69
70export interface TableProps {
71 /**
72 * Name of the parent keyspace. Changing it replaces the table.
73 */
74 keyspaceName: string;
75 /**
76 * Name of the table. Must be 1-48 characters of `[a-zA-Z0-9_]`. If omitted a
77 * deterministic physical name is generated. Changing the name replaces the
78 * table.
79 */
80 tableName?: string;
81 /**
82 * All columns in the table, including those used as partition and clustering
83 * keys. Adding new columns is applied in place; removing or retyping columns
84 * replaces the table.
85 */
86 columns: KeyspacesColumn[];
87 /**
88 * Names of the columns forming the partition key (at least one). Changing
89 * the partition key replaces the table.
90 */
91 partitionKeys: string[];
92 /**
93 * Columns forming the clustering (sort) key. Changing them replaces the
94 * table.
95 */
96 clusteringKeys?: KeyspacesClusteringKey[];
97 /**
98 * Names of static columns (shared across all rows of a partition).
99 */
100 staticColumns?: string[];
101 /**
102 * Read/write throughput mode.
103 * @default { throughputMode: "PAY_PER_REQUEST" }
104 */
105 capacity?: KeyspacesCapacity;
106 /**
107 * Enables point-in-time recovery (continuous backups).
108 * @default false
109 */
110 pointInTimeRecovery?: boolean;
111 /**
112 * Enables row-level TTL on the table. Required before per-row TTLs can be
113 * set. Enabling TTL cannot be undone without replacing the table.
114 * @default false
115 */
116 ttlEnabled?: boolean;
117 /**
118 * Default TTL applied to all rows, e.g. `"1 day"` or `Duration.hours(12)`.
119 * The API stores whole seconds. Requires `ttlEnabled`.
120 */
121 defaultTimeToLive?: Duration.Input;
122 /**
123 * Change data capture (CDC) stream configuration. Enabling CDC creates a
124 * stream that captures row-level changes for 24 hours; consume it with the
125 * `TableStreams` binding or the `keyspacesstreams` data-plane API.
126 */
127 cdcSpecification?: KeyspacesCdcSpecification;
128 /**
129 * User-defined tags for the table.
130 */
131 tags?: Record<string, string>;
132}
133
134export interface Table extends Resource<
135 "AWS.Keyspaces.Table",
136 TableProps,
137 {
138 /**
139 * Name of the keyspace containing the table.
140 */
141 keyspaceName: string;
142 /**
143 * The table's physical name.
144 */
145 tableName: string;
146 /**
147 * ARN of the table.
148 */
149 tableArn: string;
150 /**
151 * Lifecycle status of the table (e.g. `CREATING`, `ACTIVE`).
152 */
153 status: string;
154 /**
155 * ARN of the most recent CDC stream, when `cdcSpecification` is enabled.
156 */
157 latestStreamArn: string | undefined;
158 },
159 never,
160 Providers
161> {}
162
163/**
164 * An Amazon Keyspaces (for Apache Cassandra) table.
165 *
166 * Tables are serverless and provisioned asynchronously (`CREATING` ->
167 * `ACTIVE`), usually within a minute; the provider waits for `ACTIVE`
168 * (bounded) before returning. Schema mutations (adding columns, changing
169 * capacity/TTL/PITR) are applied in place; changing keys replaces the table.
170 * ### Creating a Table
171 * **Example:** Simple Key-Value Table
172 * ```typescript
173 * const table = yield* Table("Sessions", {
174 * keyspaceName: keyspace.keyspaceName,
175 * columns: [
176 * { name: "id", type: "uuid" },
177 * { name: "data", type: "text" },
178 * ],
179 * partitionKeys: ["id"],
180 * });
181 * ```
182 *
183 * **Example:** Table with a Clustering Key and TTL
184 * ```typescript
185 * const table = yield* Table("Events", {
186 * keyspaceName: keyspace.keyspaceName,
187 * columns: [
188 * { name: "device", type: "text" },
189 * { name: "ts", type: "timestamp" },
190 * { name: "payload", type: "blob" },
191 * ],
192 * partitionKeys: ["device"],
193 * clusteringKeys: [{ name: "ts", orderBy: "DESC" }],
194 * ttlEnabled: true,
195 * defaultTimeToLive: "1 day",
196 * });
197 * ```
198 *
199 * ### Change Data Capture
200 * **Example:** CDC-Enabled Table
201 * ```typescript
202 * const table = yield* Table("Orders", {
203 * keyspaceName: keyspace.keyspaceName,
204 * columns: [
205 * { name: "id", type: "uuid" },
206 * { name: "total", type: "int" },
207 * ],
208 * partitionKeys: ["id"],
209 * cdcSpecification: {
210 * status: "ENABLED",
211 * viewType: "NEW_AND_OLD_IMAGES",
212 * },
213 * });
214 * // table.latestStreamArn → consume via the TableStreams binding
215 * ```
216 *
217 * @resource
218 */
219export const Table = Resource<Table>("AWS.Keyspaces.Table");
220
221const toTagRecord = (
222 tags: keyspaces.Tag[] | undefined,
223): Record<string, string> =>
224 Object.fromEntries((tags ?? []).map((t) => [t.key, t.value]));
225
226const buildSchema = (props: TableProps): keyspaces.SchemaDefinition => ({
227 allColumns: props.columns.map((c) => ({ name: c.name, type: c.type })),
228 partitionKeys: props.partitionKeys.map((name) => ({ name })),
229 clusteringKeys: props.clusteringKeys?.map((c) => ({
230 name: c.name,
231 orderBy: c.orderBy ?? "ASC",
232 })),
233 staticColumns: props.staticColumns?.map((name) => ({ name })),
234});
235
236const activeStatuses = new Set(["ACTIVE"]);
237
238export const TableProvider = () =>
239 Provider.effect(
240 Table,
241 Effect.gen(function* () {
242 const toName = (id: string, props: Partial<TableProps>) =>
243 props.tableName
244 ? Effect.succeed(props.tableName)
245 : createPhysicalName({ id, maxLength: 48 }).pipe(
246 Effect.map((n) => n.replaceAll("-", "_")),
247 );
248
249 const readTable = Effect.fn(function* (
250 keyspaceName: string,
251 tableName: string,
252 ) {
253 return yield* keyspaces
254 .getTable({ keyspaceName, tableName })
255 .pipe(
256 Effect.catchTag("ResourceNotFoundException", () =>
257 Effect.succeed(undefined),
258 ),
259 );
260 });
261
262 const readTags = Effect.fn(function* (arn: string) {
263 const tags = yield* keyspaces.listTagsForResource
264 .items({ resourceArn: arn })
265 .pipe(
266 Stream.runCollect,
267 Effect.map((c) => Array.from(c)),
268 Effect.catch(() => Effect.succeed<keyspaces.Tag[]>([])),
269 );
270 return toTagRecord(tags);
271 });
272
273 // Resolve the most recent CDC stream's ARN (streams are labeled with
274 // their creation timestamp; the lexicographically greatest label is the
275 // latest). Only meaningful while CDC is (or was recently) enabled.
276 const readLatestStreamArn = Effect.fn(function* (
277 keyspaceName: string,
278 tableName: string,
279 ) {
280 const response = yield* keyspacesstreams
281 .listStreams({ keyspaceName, tableName })
282 .pipe(
283 Effect.catchTag("ResourceNotFoundException", () =>
284 Effect.succeed(undefined),
285 ),
286 );
287 const sorted = [...(response?.streams ?? [])].sort((a, b) =>
288 b.streamLabel.localeCompare(a.streamLabel),
289 );
290 return sorted[0]?.streamArn;
291 });
292
293 const isCdcEnabled = (table: keyspaces.GetTableResponse) => {
294 const status = table.cdcSpecification?.status;
295 return status === "ENABLED" || status === "ENABLING";
296 };
297
298 // A freshly-enabled stream can lag ListStreams visibility briefly;
299 // poll (bounded ~60s) and fall back to undefined rather than failing
300 // the reconcile.
301 const waitForStreamArn = Effect.fn(function* (
302 keyspaceName: string,
303 tableName: string,
304 ) {
305 return yield* readLatestStreamArn(keyspaceName, tableName).pipe(
306 Effect.flatMap((arn) =>
307 arn !== undefined
308 ? Effect.succeed<string | undefined>(arn)
309 : Effect.fail(
310 new Error(
311 `CDC stream for '${keyspaceName}.${tableName}' not yet visible`,
312 ),
313 ),
314 ),
315 Effect.retry({
316 schedule: Schedule.max([
317 Schedule.fixed("5 seconds"),
318 Schedule.recurs(12),
319 ]),
320 }),
321 Effect.catch(() => Effect.succeed(undefined)),
322 );
323 });
324
325 // Bounded readiness wait; Keyspaces tables typically reach ACTIVE within
326 // a minute. Budget ~5 min (60 * 5s). When `requiredColumns` is provided
327 // the wait also blocks until those columns are visible in the schema —
328 // AddColumn updates return before the table transitions out of ACTIVE and
329 // the new columns propagate to GetTable with a short lag.
330 const waitForActive = Effect.fn(function* (
331 keyspaceName: string,
332 tableName: string,
333 requiredColumns?: readonly string[],
334 ) {
335 const policy = Schedule.max([
336 Schedule.fixed("5 seconds"),
337 Schedule.recurs(60),
338 ]);
339 return yield* readTable(keyspaceName, tableName).pipe(
340 Effect.flatMap((table) => {
341 if (table === undefined) {
342 return Effect.fail(
343 new Error(`Keyspaces table '${tableName}' not found`),
344 );
345 }
346 if (!activeStatuses.has(table.status ?? "")) {
347 return Effect.fail(
348 new Error(
349 `Keyspaces table '${tableName}' not active (status: ${table.status})`,
350 ),
351 );
352 }
353 if (requiredColumns !== undefined) {
354 const present = new Set(
355 (table.schemaDefinition?.allColumns ?? []).map((c) => c.name),
356 );
357 const missing = requiredColumns.filter((c) => !present.has(c));
358 if (missing.length > 0) {
359 return Effect.fail(
360 new Error(
361 `Keyspaces table '${tableName}' missing columns: ${missing.join(", ")}`,
362 ),
363 );
364 }
365 }
366 return Effect.succeed(table);
367 }),
368 Effect.retry({ schedule: policy }),
369 );
370 });
371
372 const toCapacity = (
373 capacity: KeyspacesCapacity | undefined,
374 ): keyspaces.CapacitySpecification | undefined =>
375 capacity === undefined
376 ? undefined
377 : {
378 throughputMode: capacity.throughputMode,
379 readCapacityUnits: capacity.readCapacityUnits,
380 writeCapacityUnits: capacity.writeCapacityUnits,
381 };
382
383 return {
384 stables: ["keyspaceName", "tableName", "tableArn"],
385
386 diff: Effect.fn(function* ({ id, olds, news }) {
387 if (!isResolved(news)) return undefined;
388 const oldProps = olds as TableProps | undefined;
389 if (oldProps === undefined) return undefined;
390 if (news.keyspaceName !== oldProps.keyspaceName) {
391 return { action: "replace" } as const;
392 }
393 if ((yield* toName(id, oldProps)) !== (yield* toName(id, news))) {
394 return { action: "replace" } as const;
395 }
396 // Keys and column removals/retypes require a replacement; pure
397 // column additions are handled in reconcile via UpdateTable.
398 const newCols = new Map(news.columns.map((c) => [c.name, c.type]));
399 const removedOrRetyped = oldProps.columns.some(
400 (c) => newCols.get(c.name) !== c.type,
401 );
402 const keysChanged =
403 JSON.stringify([...news.partitionKeys].sort()) !==
404 JSON.stringify([...oldProps.partitionKeys].sort()) ||
405 JSON.stringify(
406 (news.clusteringKeys ?? [])
407 .map((c) => `${c.name}:${c.orderBy ?? "ASC"}`)
408 .sort(),
409 ) !==
410 JSON.stringify(
411 (oldProps.clusteringKeys ?? [])
412 .map((c) => `${c.name}:${c.orderBy ?? "ASC"}`)
413 .sort(),
414 ) ||
415 JSON.stringify([...(news.staticColumns ?? [])].sort()) !==
416 JSON.stringify([...(oldProps.staticColumns ?? [])].sort());
417 if (removedOrRetyped || keysChanged) {
418 return { action: "replace" } as const;
419 }
420 }),
421
422 read: Effect.fn(function* ({ id, olds, output }) {
423 const props = (olds ?? {}) as Partial<TableProps>;
424 const keyspaceName = output?.keyspaceName ?? props.keyspaceName;
425 if (keyspaceName === undefined) return undefined;
426 const tableName = output?.tableName ?? (yield* toName(id, props));
427 const found = yield* readTable(keyspaceName, tableName);
428 if (found === undefined || found.status === "DELETING") {
429 return undefined;
430 }
431 const attrs = {
432 keyspaceName: found.keyspaceName,
433 tableName: found.tableName,
434 tableArn: found.resourceArn,
435 status: found.status ?? "ACTIVE",
436 latestStreamArn: isCdcEnabled(found)
437 ? yield* readLatestStreamArn(found.keyspaceName, found.tableName)
438 : undefined,
439 };
440 const tags = yield* readTags(found.resourceArn);
441 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
442 }),
443
444 reconcile: Effect.fn(function* ({ id, news, output, session }) {
445 const props = news as TableProps;
446 const keyspaceName = props.keyspaceName;
447 const tableName = output?.tableName ?? (yield* toName(id, props));
448 // The Keyspaces API expects the default TTL in whole seconds.
449 const desiredTtlSeconds = toWireSeconds(props.defaultTimeToLive);
450 const internalTags = yield* createInternalTags(id);
451 const desiredTags = { ...internalTags, ...props.tags };
452
453 // 1. Observe — cloud state is authoritative.
454 let observed = yield* readTable(keyspaceName, tableName);
455
456 // 2. Ensure — create if missing; tolerate the AlreadyExists race.
457 if (observed === undefined) {
458 yield* keyspaces
459 .createTable({
460 keyspaceName,
461 tableName,
462 schemaDefinition: buildSchema(props),
463 capacitySpecification: toCapacity(props.capacity),
464 pointInTimeRecovery: props.pointInTimeRecovery
465 ? { status: "ENABLED" }
466 : undefined,
467 ttl: props.ttlEnabled ? { status: "ENABLED" } : undefined,
468 defaultTimeToLive: desiredTtlSeconds,
469 cdcSpecification:
470 props.cdcSpecification?.status === "ENABLED"
471 ? {
472 status: "ENABLED",
473 viewType:
474 props.cdcSpecification.viewType ??
475 "NEW_AND_OLD_IMAGES",
476 }
477 : undefined,
478 tags: Object.entries(desiredTags).map(([key, value]) => ({
479 key,
480 value,
481 })),
482 })
483 .pipe(Effect.catchTag("ConflictException", () => Effect.void));
484 }
485
486 observed = yield* waitForActive(keyspaceName, tableName);
487
488 // 3. Sync mutable aspects against observed state.
489 const update: keyspaces.UpdateTableRequest = {
490 keyspaceName,
491 tableName,
492 };
493 let needsUpdate = false;
494
495 const observedCols = new Set(
496 (observed.schemaDefinition?.allColumns ?? []).map((c) => c.name),
497 );
498 const addColumns = props.columns.filter(
499 (c) => !observedCols.has(c.name),
500 );
501 if (addColumns.length > 0) {
502 update.addColumns = addColumns.map((c) => ({
503 name: c.name,
504 type: c.type,
505 }));
506 needsUpdate = true;
507 }
508
509 const desiredCapacity = toCapacity(props.capacity);
510 if (
511 desiredCapacity !== undefined &&
512 (desiredCapacity.throughputMode !==
513 observed.capacitySpecification?.throughputMode ||
514 (desiredCapacity.throughputMode === "PROVISIONED" &&
515 (desiredCapacity.readCapacityUnits !==
516 observed.capacitySpecification?.readCapacityUnits ||
517 desiredCapacity.writeCapacityUnits !==
518 observed.capacitySpecification?.writeCapacityUnits)))
519 ) {
520 update.capacitySpecification = desiredCapacity;
521 needsUpdate = true;
522 }
523
524 const desiredPitr = props.pointInTimeRecovery
525 ? "ENABLED"
526 : "DISABLED";
527 if (
528 desiredPitr !== (observed.pointInTimeRecovery?.status ?? "DISABLED")
529 ) {
530 update.pointInTimeRecovery = { status: desiredPitr };
531 needsUpdate = true;
532 }
533
534 if (props.ttlEnabled && observed.ttl?.status !== "ENABLED") {
535 update.ttl = { status: "ENABLED" };
536 needsUpdate = true;
537 }
538 if (
539 desiredTtlSeconds !== undefined &&
540 desiredTtlSeconds !== observed.defaultTimeToLive
541 ) {
542 update.defaultTimeToLive = desiredTtlSeconds;
543 needsUpdate = true;
544 }
545
546 const desiredCdcEnabled =
547 props.cdcSpecification?.status === "ENABLED";
548 if (desiredCdcEnabled !== isCdcEnabled(observed)) {
549 update.cdcSpecification = desiredCdcEnabled
550 ? {
551 status: "ENABLED",
552 viewType:
553 props.cdcSpecification?.viewType ?? "NEW_AND_OLD_IMAGES",
554 }
555 : { status: "DISABLED" };
556 needsUpdate = true;
557 }
558
559 if (needsUpdate) {
560 yield* keyspaces.updateTable(update);
561 observed = yield* waitForActive(
562 keyspaceName,
563 tableName,
564 props.columns.map((c) => c.name),
565 );
566 }
567
568 // 3b. Sync tags — diff against OBSERVED cloud tags.
569 const observedTags = yield* readTags(observed.resourceArn);
570 const { upsert, removed } = diffTags(observedTags, desiredTags);
571 if (upsert.length > 0) {
572 yield* keyspaces.tagResource({
573 resourceArn: observed.resourceArn,
574 tags: upsert.map((t) => ({ key: t.Key, value: t.Value })),
575 });
576 }
577 if (removed.length > 0) {
578 yield* keyspaces.untagResource({
579 resourceArn: observed.resourceArn,
580 tags: removed.map((key) => ({ key, value: "" })),
581 });
582 }
583
584 yield* session.note(`${keyspaceName}.${tableName}`);
585 return {
586 keyspaceName: observed.keyspaceName,
587 tableName: observed.tableName,
588 tableArn: observed.resourceArn,
589 status: observed.status ?? "ACTIVE",
590 latestStreamArn: isCdcEnabled(observed)
591 ? yield* waitForStreamArn(
592 observed.keyspaceName,
593 observed.tableName,
594 )
595 : undefined,
596 };
597 }),
598
599 delete: Effect.fn(function* ({ output }) {
600 const { keyspaceName, tableName } = output;
601 yield* keyspaces.deleteTable({ keyspaceName, tableName }).pipe(
602 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
603 // A table still CREATING/UPDATING rejects delete with
604 // ConflictException; retry until it settles.
605 Effect.retry({
606 while: (e) => e._tag === "ConflictException",
607 schedule: Schedule.max([
608 Schedule.fixed("5 seconds"),
609 Schedule.recurs(24),
610 ]),
611 }),
612 );
613 // Wait (bounded) until the table is gone so the parent keyspace can
614 // be deleted without a ConflictException.
615 yield* readTable(keyspaceName, tableName).pipe(
616 Effect.flatMap((t) =>
617 t === undefined
618 ? Effect.void
619 : Effect.fail(new Error(`Table '${tableName}' still deleting`)),
620 ),
621 Effect.retry({
622 schedule: Schedule.max([
623 Schedule.fixed("3 seconds"),
624 Schedule.recurs(20),
625 ]),
626 }),
627 Effect.catch(() => Effect.void),
628 );
629 }),
630
631 // Table is keyed by a parent keyspace and cannot be enumerated
632 // account-wide without iterating every keyspace; treated as a
633 // sub-resource per the factory list() convention.
634 list: () => Effect.succeed([]),
635 };
636 }),
637 );