1import * as keyspaces from "@distilled.cloud/aws/keyspaces";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
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 { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
11import type { Providers } from "../Providers.ts";
12
13export interface KeyspaceProps {
14 /**
15 * Name of the keyspace. Must be 1-48 characters of `[a-zA-Z0-9_]`. If
16 * omitted a deterministic physical name is generated. Changing the name
17 * replaces the keyspace.
18 */
19 keyspaceName?: string;
20 /**
21 * User-defined tags for the keyspace.
22 */
23 tags?: Record<string, string>;
24}
25
26export interface Keyspace extends Resource<
27 "AWS.Keyspaces.Keyspace",
28 KeyspaceProps,
29 {
30 /**
31 * The keyspace's physical name.
32 */
33 keyspaceName: string;
34 /**
35 * ARN of the keyspace.
36 */
37 keyspaceArn: string;
38 /**
39 * Replication strategy of the keyspace (`SINGLE_REGION` or
40 * `MULTI_REGION`).
41 */
42 replicationStrategy: string;
43 },
44 never,
45 Providers
46> {}
47
48/**
49 * An Amazon Keyspaces (for Apache Cassandra) keyspace — the top-level
50 * container for Cassandra tables.
51 *
52 * Keyspaces are serverless, free to create, and provisioned near-instantly,
53 * so they make excellent building blocks and test fixtures.
54 * ### Creating a Keyspace
55 * **Example:** Basic Keyspace
56 * ```typescript
57 * const keyspace = yield* Keyspace("AppData", {});
58 * ```
59 *
60 * **Example:** Named Keyspace with Tags
61 * ```typescript
62 * const keyspace = yield* Keyspace("AppData", {
63 * keyspaceName: "app_data",
64 * tags: { team: "platform" },
65 * });
66 * ```
67 *
68 * @resource
69 */
70export const Keyspace = Resource<Keyspace>("AWS.Keyspaces.Keyspace");
71
72const toTagRecord = (
73 tags: keyspaces.Tag[] | undefined,
74): Record<string, string> =>
75 Object.fromEntries((tags ?? []).map((t) => [t.key, t.value]));
76
77/**
78 * Reserved Cassandra system keyspaces that exist in every Amazon Keyspaces
79 * account and can never be deleted. Excluded from `list()` so account-wide
80 * teardown (`alchemy unsafe nuke`) doesn't loop on undeletable resources.
81 */
82const SYSTEM_KEYSPACES = new Set([
83 "system",
84 "system_schema",
85 "system_schema_mcs",
86 "system_multiregion_info",
87]);
88
89export const KeyspaceProvider = () =>
90 Provider.effect(
91 Keyspace,
92 Effect.gen(function* () {
93 const toName = (id: string, props: KeyspaceProps) =>
94 props.keyspaceName
95 ? Effect.succeed(props.keyspaceName)
96 : createPhysicalName({ id, maxLength: 48 }).pipe(
97 Effect.map((n) => n.replaceAll("-", "_")),
98 );
99
100 const readKeyspace = Effect.fn(function* (name: string) {
101 return yield* keyspaces
102 .getKeyspace({ keyspaceName: name })
103 .pipe(
104 Effect.catchTag("ResourceNotFoundException", () =>
105 Effect.succeed(undefined),
106 ),
107 );
108 });
109
110 const readTags = Effect.fn(function* (arn: string) {
111 const tags = yield* keyspaces.listTagsForResource
112 .items({ resourceArn: arn })
113 .pipe(
114 Stream.runCollect,
115 Effect.map((c) => Array.from(c)),
116 Effect.catch(() => Effect.succeed<keyspaces.Tag[]>([])),
117 );
118 return toTagRecord(tags);
119 });
120
121 return {
122 stables: ["keyspaceName", "keyspaceArn"],
123
124 diff: Effect.fn(function* ({ id, olds = {}, news }) {
125 if (!isResolved(news)) return undefined;
126 if ((yield* toName(id, olds)) !== (yield* toName(id, news))) {
127 return { action: "replace" } as const;
128 }
129 }),
130
131 read: Effect.fn(function* ({ id, olds, output }) {
132 const name = output?.keyspaceName ?? (yield* toName(id, olds ?? {}));
133 const found = yield* readKeyspace(name);
134 if (found === undefined) return undefined;
135 const attrs = {
136 keyspaceName: found.keyspaceName,
137 keyspaceArn: found.resourceArn,
138 replicationStrategy: found.replicationStrategy,
139 };
140 const tags = yield* readTags(found.resourceArn);
141 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
142 }),
143
144 reconcile: Effect.fn(function* ({ id, news = {}, output, session }) {
145 const name = output?.keyspaceName ?? (yield* toName(id, news));
146 const internalTags = yield* createInternalTags(id);
147 const desiredTags = { ...internalTags, ...news.tags };
148
149 // 1. Observe — cloud state is authoritative.
150 let observed = yield* readKeyspace(name);
151
152 // 2. Ensure — create if missing; tolerate the AlreadyExists race.
153 if (observed === undefined) {
154 yield* keyspaces
155 .createKeyspace({
156 keyspaceName: name,
157 tags: Object.entries(desiredTags).map(([key, value]) => ({
158 key,
159 value,
160 })),
161 })
162 .pipe(Effect.catchTag("ConflictException", () => Effect.void));
163 // Keyspace creation is effectively synchronous, but GetKeyspace can
164 // briefly 404 right after create; retry a few times.
165 observed = yield* readKeyspace(name).pipe(
166 Effect.flatMap((k) =>
167 k === undefined
168 ? Effect.fail(new Error(`Keyspace '${name}' not found`))
169 : Effect.succeed(k),
170 ),
171 Effect.retry({
172 schedule: Schedule.max([
173 Schedule.fixed("2 seconds"),
174 Schedule.recurs(10),
175 ]),
176 }),
177 );
178 }
179
180 // 3. Sync tags — diff against OBSERVED cloud tags.
181 const observedTags = yield* readTags(observed.resourceArn);
182 const { upsert, removed } = diffTags(observedTags, desiredTags);
183 if (upsert.length > 0) {
184 yield* keyspaces.tagResource({
185 resourceArn: observed.resourceArn,
186 tags: upsert.map((t) => ({ key: t.Key, value: t.Value })),
187 });
188 }
189 if (removed.length > 0) {
190 yield* keyspaces.untagResource({
191 resourceArn: observed.resourceArn,
192 tags: removed.map((key) => ({ key, value: "" })),
193 });
194 }
195
196 yield* session.note(name);
197 return {
198 keyspaceName: observed.keyspaceName,
199 keyspaceArn: observed.resourceArn,
200 replicationStrategy: observed.replicationStrategy,
201 };
202 }),
203
204 delete: Effect.fn(function* ({ output }) {
205 const name = output.keyspaceName;
206 yield* keyspaces.deleteKeyspace({ keyspaceName: name }).pipe(
207 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
208 // A keyspace whose tables are still deleting rejects with
209 // ConflictException; retry briefly.
210 Effect.retry({
211 while: (e) => e._tag === "ConflictException",
212 schedule: Schedule.max([
213 Schedule.fixed("3 seconds"),
214 Schedule.recurs(20),
215 ]),
216 }),
217 );
218 // Deletion is asynchronous — wait (bounded) until the keyspace is
219 // actually gone so a subsequent recreate with the same name does not
220 // hit ConflictException.
221 yield* readKeyspace(name).pipe(
222 Effect.flatMap((k) =>
223 k === undefined
224 ? Effect.void
225 : Effect.fail(new Error(`Keyspace '${name}' still deleting`)),
226 ),
227 Effect.retry({
228 schedule: Schedule.max([
229 Schedule.fixed("3 seconds"),
230 Schedule.recurs(20),
231 ]),
232 }),
233 Effect.catch(() => Effect.void),
234 );
235 }),
236
237 list: () =>
238 keyspaces.listKeyspaces.items({}).pipe(
239 Stream.runCollect,
240 Effect.map((chunk) =>
241 Array.from(chunk)
242 .filter((k) => !SYSTEM_KEYSPACES.has(k.keyspaceName))
243 .map((k) => ({
244 keyspaceName: k.keyspaceName,
245 keyspaceArn: k.resourceArn,
246 replicationStrategy: k.replicationStrategy,
247 })),
248 ),
249 ),
250 };
251 }),
252 );