1import * as dsql from "@distilled.cloud/aws/dsql";
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 * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
10import { AWSEnvironment } from "../Environment.ts";
11import type { Providers } from "../Providers.ts";
12
13export interface ClusterProps {
14 /**
15 * Enables deletion protection. While enabled the cluster cannot be deleted;
16 * the provider automatically disables it during delete so `stack.destroy()`
17 * always succeeds.
18 * @default false
19 */
20 deletionProtectionEnabled?: boolean;
21 /**
22 * ARN of a customer-managed KMS key used to encrypt the cluster at rest.
23 * Changing the key replaces the cluster.
24 * @default an AWS-owned key
25 */
26 kmsEncryptionKey?: string;
27 /**
28 * User-defined tags for the cluster.
29 */
30 tags?: Record<string, string>;
31}
32
33export interface Cluster extends Resource<
34 "AWS.DSQL.Cluster",
35 ClusterProps,
36 {
37 /** The unique cluster identifier assigned by DSQL. */
38 clusterId: string;
39 /** The ARN of the cluster. */
40 clusterArn: string;
41 /** The current status of the cluster, e.g. `ACTIVE`. */
42 status: string;
43 /**
44 * The public cluster endpoint hostname, e.g.
45 * `<clusterId>.dsql.<region>.on.aws`. Connect with a Postgres wire client
46 * using an IAM-generated auth token as the password.
47 */
48 endpoint: string;
49 /** Whether deletion protection is enabled on the cluster. */
50 deletionProtectionEnabled: boolean;
51 },
52 never,
53 Providers
54> {}
55
56/**
57 * An Amazon Aurora DSQL cluster — a serverless, distributed SQL database with
58 * active-active high availability and Postgres wire compatibility.
59 *
60 * Clusters are pay-per-use with no provisioned capacity, so they have
61 * excellent test economics. Create is asynchronous (`CREATING` -> `ACTIVE`),
62 * usually completing in under a minute; the provider waits for `ACTIVE`
63 * (bounded) before returning.
64 * ### Creating a Cluster
65 * **Example:** Basic Cluster
66 * ```typescript
67 * const cluster = yield* Cluster("AppDb", {});
68 * // connect to cluster.endpoint on port 5432 as user "admin"
69 * ```
70 *
71 * **Example:** Cluster with Deletion Protection
72 * ```typescript
73 * const cluster = yield* Cluster("AppDb", {
74 * deletionProtectionEnabled: true,
75 * });
76 * ```
77 *
78 * **Example:** Cluster with a Customer-Managed KMS Key
79 * ```typescript
80 * const cluster = yield* Cluster("AppDb", {
81 * kmsEncryptionKey: key.keyArn,
82 * });
83 * ```
84 *
85 * @resource
86 */
87export const Cluster = Resource<Cluster>("AWS.DSQL.Cluster");
88
89const activeStatuses = new Set(["ACTIVE", "IDLE"]);
90
91export const ClusterProvider = () =>
92 Provider.effect(
93 Cluster,
94 Effect.gen(function* () {
95 const endpointFor = (identifier: string, region: string) =>
96 `${identifier}.dsql.${region}.on.aws`;
97
98 const readCluster = Effect.fn(function* (identifier: string) {
99 return yield* dsql
100 .getCluster({ identifier })
101 .pipe(
102 Effect.catchTag("ResourceNotFoundException", () =>
103 Effect.succeed(undefined),
104 ),
105 );
106 });
107
108 const readTags = Effect.fn(function* (arn: string) {
109 const response = yield* dsql
110 .listTagsForResource({ resourceArn: arn })
111 .pipe(Effect.catch(() => Effect.succeed(undefined)));
112 return Object.fromEntries(
113 Object.entries(response?.tags ?? {}).filter(
114 (entry): entry is [string, string] => typeof entry[1] === "string",
115 ),
116 );
117 });
118
119 // Bounded readiness wait. DSQL clusters usually reach ACTIVE within a
120 // minute; budget ~5 min (60 * 5s) so slow provisioning still converges
121 // without risking the test wall.
122 const waitForActive = Effect.fn(function* (identifier: string) {
123 const policy = Schedule.max([
124 Schedule.fixed("5 seconds"),
125 Schedule.recurs(60),
126 ]);
127 return yield* readCluster(identifier).pipe(
128 Effect.flatMap((cluster) => {
129 if (cluster === undefined) {
130 return Effect.fail(
131 new Error(`DSQL cluster '${identifier}' not found`),
132 );
133 }
134 if (!activeStatuses.has(cluster.status)) {
135 return Effect.fail(
136 new Error(
137 `DSQL cluster '${identifier}' not active (status: ${cluster.status})`,
138 ),
139 );
140 }
141 return Effect.succeed(cluster);
142 }),
143 Effect.retry({ schedule: policy }),
144 );
145 });
146
147 const toAttrs = (
148 cluster: dsql.GetClusterOutput | dsql.CreateClusterOutput,
149 region: string,
150 ) => ({
151 clusterId: cluster.identifier,
152 clusterArn: cluster.arn,
153 status: cluster.status,
154 endpoint: cluster.endpoint ?? endpointFor(cluster.identifier, region),
155 deletionProtectionEnabled: cluster.deletionProtectionEnabled,
156 });
157
158 return {
159 stables: ["clusterId", "clusterArn", "endpoint"],
160
161 diff: Effect.fn(function* ({ olds = {}, news }) {
162 if (!isResolved(news)) return undefined;
163 // KMS key is create-only; changing it forces a replacement.
164 if (
165 (news.kmsEncryptionKey ?? undefined) !==
166 (olds.kmsEncryptionKey ?? undefined)
167 ) {
168 return { action: "replace" } as const;
169 }
170 }),
171
172 read: Effect.fn(function* ({ id, output }) {
173 if (output?.clusterId === undefined) return undefined;
174 const { region } = yield* AWSEnvironment.current;
175 const cluster = yield* readCluster(output.clusterId);
176 if (cluster === undefined || cluster.status === "DELETED") {
177 return undefined;
178 }
179 const tags = yield* readTags(cluster.arn);
180 const attrs = toAttrs(cluster, region);
181 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
182 }),
183
184 reconcile: Effect.fn(function* ({ id, news = {}, output, session }) {
185 const { region } = yield* AWSEnvironment.current;
186 const internalTags = yield* createInternalTags(id);
187 const desiredTags = { ...internalTags, ...news.tags };
188
189 // 1. Observe — cloud state is authoritative; output caches the id.
190 let observed =
191 output?.clusterId === undefined
192 ? undefined
193 : yield* readCluster(output.clusterId);
194
195 // 2. Ensure — create if missing. DSQL assigns the identifier.
196 let identifier = observed?.identifier ?? output?.clusterId;
197 if (observed === undefined) {
198 const created = yield* dsql.createCluster({
199 deletionProtectionEnabled:
200 news.deletionProtectionEnabled ?? false,
201 kmsEncryptionKey: news.kmsEncryptionKey,
202 tags: desiredTags,
203 });
204 identifier = created.identifier;
205 }
206
207 // Wait for ACTIVE so subsequent syncs do not hit ConflictException.
208 const active = yield* waitForActive(identifier!);
209 observed = active;
210
211 // 3. Sync deletion protection against observed state.
212 if (
213 news.deletionProtectionEnabled !== undefined &&
214 news.deletionProtectionEnabled !==
215 observed.deletionProtectionEnabled
216 ) {
217 yield* dsql.updateCluster({
218 identifier: identifier!,
219 deletionProtectionEnabled: news.deletionProtectionEnabled,
220 });
221 observed = yield* waitForActive(identifier!);
222 }
223
224 // 3b. Sync tags — diff against OBSERVED cloud tags.
225 const observedTags = yield* readTags(observed.arn);
226 const { upsert, removed } = diffTags(observedTags, desiredTags);
227 if (upsert.length > 0) {
228 yield* dsql.tagResource({
229 resourceArn: observed.arn,
230 tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
231 });
232 }
233 if (removed.length > 0) {
234 yield* dsql.untagResource({
235 resourceArn: observed.arn,
236 tagKeys: removed,
237 });
238 }
239
240 yield* session.note(identifier!);
241 return toAttrs(observed, region);
242 }),
243
244 delete: Effect.fn(function* ({ output }) {
245 const identifier = output.clusterId;
246 const existing = yield* readCluster(identifier);
247 if (existing === undefined) return;
248 // Deletion protection blocks delete — disable it first.
249 if (existing.deletionProtectionEnabled) {
250 yield* dsql
251 .updateCluster({
252 identifier,
253 deletionProtectionEnabled: false,
254 })
255 .pipe(
256 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
257 );
258 }
259 yield* dsql.deleteCluster({ identifier }).pipe(
260 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
261 // A cluster still CREATING rejects delete with ConflictException;
262 // retry briefly until it settles into a deletable state.
263 Effect.retry({
264 while: (e) => e._tag === "ConflictException",
265 schedule: Schedule.max([
266 Schedule.fixed("5 seconds"),
267 Schedule.recurs(24),
268 ]),
269 }),
270 );
271 }),
272
273 list: () =>
274 Effect.gen(function* () {
275 const { region } = yield* AWSEnvironment.current;
276 const summaries = yield* dsql.listClusters.items({}).pipe(
277 Stream.runCollect,
278 Effect.map((c) => Array.from(c)),
279 );
280 return yield* Effect.forEach(
281 summaries,
282 (summary) =>
283 readCluster(summary.identifier).pipe(
284 Effect.map((cluster) =>
285 cluster === undefined || cluster.status === "DELETED"
286 ? undefined
287 : toAttrs(cluster, region),
288 ),
289 ),
290 { concurrency: 4 },
291 ).pipe(
292 Effect.map((attrs) =>
293 attrs.filter(
294 (a): a is NonNullable<typeof a> => a !== undefined,
295 ),
296 ),
297 );
298 }),
299 };
300 }),
301 );