1import * as dax from "@distilled.cloud/aws/dax";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
10import type { Providers } from "../Providers.ts";
11import { readDaxTags, sameStringSet } from "./internal.ts";
12
13export interface ClusterProps {
14 /**
15 * Name of the cluster. Must be 1-20 alphanumeric characters or hyphens.
16 * If omitted, a deterministic physical name is generated. Changing the
17 * name replaces the cluster.
18 */
19 clusterName?: string;
20 /**
21 * Compute and memory capacity of the nodes, e.g. `"dax.t3.small"` (the
22 * cheapest supported node type) or `"dax.r5.large"`. Changing the node
23 * type replaces the cluster.
24 * @default "dax.t3.small"
25 */
26 nodeType?: string;
27 /**
28 * Number of nodes in the cluster (1 = a single node; 2+ = one primary
29 * plus read replicas). AWS recommends at least 3 nodes (spanning multiple
30 * Availability Zones) for production. Updated in place via
31 * IncreaseReplicationFactor / DecreaseReplicationFactor.
32 * @default 1
33 */
34 replicationFactor?: number;
35 /**
36 * ARN of the IAM role that DAX assumes to access DynamoDB tables on your
37 * behalf. The role's trust policy must allow `dax.amazonaws.com` to assume
38 * it. Changing the role replaces the cluster.
39 */
40 iamRoleArn: string;
41 /**
42 * Human-readable description of the cluster.
43 */
44 description?: string;
45 /**
46 * Availability Zones the nodes are placed into. Must be a subset of the
47 * AZs covered by the subnet group and match `replicationFactor` in length
48 * when provided. Changing the AZs replaces the cluster.
49 * @default DAX spreads nodes across the subnet group's AZs
50 */
51 availabilityZones?: string[];
52 /**
53 * Name of the {@link SubnetGroup} the cluster's nodes are placed into.
54 * Changing the subnet group replaces the cluster.
55 * @default the default DAX subnet group
56 */
57 subnetGroupName?: string;
58 /**
59 * VPC security groups that control network access to the cluster
60 * endpoint.
61 * @default the VPC's default security group
62 */
63 securityGroupIds?: string[];
64 /**
65 * Weekly maintenance window, e.g. `"sun:23:00-mon:01:30"` (UTC).
66 */
67 preferredMaintenanceWindow?: string;
68 /**
69 * SNS topic ARN to publish cluster events to.
70 */
71 notificationTopicArn?: string;
72 /**
73 * Name of the {@link ParameterGroup} applied to the cluster.
74 * @default the default DAX parameter group
75 */
76 parameterGroupName?: string;
77 /**
78 * Whether server-side encryption (encryption at rest) is enabled.
79 * Changing this replaces the cluster.
80 * @default false
81 */
82 sseEnabled?: boolean;
83 /**
84 * Encryption in transit for the cluster endpoint — `"NONE"` or `"TLS"`.
85 * Changing this replaces the cluster.
86 * @default "NONE"
87 */
88 clusterEndpointEncryptionType?: dax.ClusterEndpointEncryptionType;
89 /**
90 * IP address type — `"ipv4"`, `"ipv6"` or `"dual_stack"`. Changing this
91 * replaces the cluster.
92 * @default "ipv4"
93 */
94 networkType?: dax.NetworkType;
95 /**
96 * User-defined tags for the cluster.
97 */
98 tags?: Record<string, string>;
99}
100
101export interface Cluster extends Resource<
102 "AWS.DAX.Cluster",
103 ClusterProps,
104 {
105 /** Name of the DAX cluster. */
106 clusterName: string;
107 /** ARN of the DAX cluster. */
108 clusterArn: string;
109 /** Current cluster status (e.g. `available`, `creating`). */
110 status: string;
111 /** Compute/memory node type of the cluster's nodes. */
112 nodeType: string;
113 /** Total number of nodes in the cluster. */
114 totalNodes: number | undefined;
115 /** Number of nodes currently in `available` status. */
116 activeNodes: number | undefined;
117 /** Hostname of the cluster discovery endpoint. */
118 discoveryEndpointAddress: string | undefined;
119 /** Port of the cluster discovery endpoint. */
120 discoveryEndpointPort: number | undefined;
121 /** Full `dax://` (or `daxs://` for TLS) discovery endpoint URL clients connect to. */
122 discoveryEndpointUrl: string | undefined;
123 /** Name of the subnet group the cluster's nodes are placed into. */
124 subnetGroupName: string | undefined;
125 /** ARN of the IAM role DAX assumes to reach DynamoDB. */
126 iamRoleArn: string | undefined;
127 /** Name of the parameter group attached to the cluster. */
128 parameterGroupName: string | undefined;
129 /** Security group IDs attached to the cluster's nodes. */
130 securityGroupIds: string[];
131 /** Endpoint encryption in transit (`NONE` or `TLS`). */
132 clusterEndpointEncryptionType: string | undefined;
133 /** Current tags on the cluster. */
134 tags: Record<string, string>;
135 },
136 never,
137 Providers
138> {}
139
140/**
141 * An Amazon DAX cluster — a fully managed, in-memory write-through cache for
142 * DynamoDB.
143 *
144 * Clusters are VPC-only and take roughly 5-10 minutes to provision; they are
145 * billed per node-hour while they exist. Place them in a {@link SubnetGroup}
146 * and give them an IAM role that DAX assumes to reach DynamoDB. Destroy
147 * clusters you are not using.
148 * ### Creating a Cluster
149 * **Example:** Single-Node Development Cluster
150 * ```typescript
151 * const role = yield* IAM.Role("DaxRole", {
152 * assumeRolePolicyDocument: {
153 * Version: "2012-10-17",
154 * Statement: [{
155 * Effect: "Allow",
156 * Principal: { Service: "dax.amazonaws.com" },
157 * Action: ["sts:AssumeRole"],
158 * }],
159 * },
160 * managedPolicyArns: ["arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess"],
161 * });
162 * const subnetGroup = yield* SubnetGroup("DaxSubnets", {
163 * subnetIds: [subnetA.subnetId, subnetB.subnetId],
164 * });
165 * const cluster = yield* Cluster("Cache", {
166 * nodeType: "dax.t3.small",
167 * replicationFactor: 1,
168 * iamRoleArn: role.roleArn,
169 * subnetGroupName: subnetGroup.subnetGroupName,
170 * });
171 * ```
172 *
173 * ### Encryption
174 * **Example:** Cluster with Encryption At Rest and In Transit
175 * ```typescript
176 * const cluster = yield* Cluster("SecureCache", {
177 * nodeType: "dax.t3.small",
178 * replicationFactor: 3,
179 * iamRoleArn: role.roleArn,
180 * subnetGroupName: subnetGroup.subnetGroupName,
181 * sseEnabled: true,
182 * clusterEndpointEncryptionType: "TLS",
183 * });
184 * ```
185 *
186 * @resource
187 */
188export const Cluster = Resource<Cluster>("AWS.DAX.Cluster");
189
190const DEFAULT_NODE_TYPE = "dax.t3.small";
191const DEFAULT_REPLICATION_FACTOR = 1;
192
193// A freshly created IAM role is not immediately assumable by dax.amazonaws.com
194// — CreateCluster rejects with "No permission to assume role" until IAM
195// propagates (typically a few seconds). Bounded retry through that window.
196const retryIamRolePropagation = <A, E extends { _tag: string }, R>(
197 self: Effect.Effect<A, E, R>,
198): Effect.Effect<A, E, R> =>
199 Effect.retry(self, {
200 while: (e) =>
201 e._tag === "InvalidParameterValueException" &&
202 ((e as { message?: string }).message?.includes(
203 "No permission to assume role",
204 ) ??
205 false),
206 schedule: Schedule.max([Schedule.fixed("5 seconds"), Schedule.recurs(10)]),
207 });
208
209export const ClusterProvider = () =>
210 Provider.effect(
211 Cluster,
212 Effect.gen(function* () {
213 const toName = (id: string, props: Partial<ClusterProps>) =>
214 props.clusterName
215 ? Effect.succeed(props.clusterName)
216 : createPhysicalName({ id, maxLength: 20, lowercase: true });
217
218 const readCluster = Effect.fn(function* (name: string) {
219 const response = yield* dax
220 .describeClusters({ ClusterNames: [name] })
221 .pipe(
222 Effect.catchTag("ClusterNotFoundFault", () =>
223 Effect.succeed(undefined),
224 ),
225 );
226 return response?.Clusters?.[0];
227 });
228
229 // Bounded readiness wait. DAX cluster provisioning/modification
230 // typically completes in 5-10 minutes; budget ~15 min (60 * 15s).
231 const waitForAvailable = Effect.fn(function* (name: string) {
232 const policy = Schedule.max([
233 Schedule.fixed("15 seconds"),
234 Schedule.recurs(60),
235 ]);
236 return yield* readCluster(name).pipe(
237 Effect.flatMap((cluster) => {
238 if (!cluster?.ClusterArn) {
239 return Effect.fail(new Error(`DAX cluster '${name}' not found`));
240 }
241 if (cluster.Status !== "available") {
242 return Effect.fail(
243 new Error(
244 `DAX cluster '${name}' not available (status: ${cluster.Status})`,
245 ),
246 );
247 }
248 return Effect.succeed(cluster);
249 }),
250 Effect.retry({ schedule: policy }),
251 );
252 });
253
254 // Wait for a cluster to leave a transitional state before delete. Ends
255 // when the cluster is available, deleting, or gone.
256 const waitUntilSettled = Effect.fn(function* (name: string) {
257 const policy = Schedule.max([
258 Schedule.fixed("15 seconds"),
259 Schedule.recurs(60),
260 ]);
261 return yield* readCluster(name).pipe(
262 Effect.flatMap((cluster) => {
263 if (
264 cluster !== undefined &&
265 cluster.Status !== "available" &&
266 cluster.Status !== "deleting"
267 ) {
268 return Effect.fail(
269 new Error(
270 `DAX cluster '${name}' still settling (status: ${cluster.Status})`,
271 ),
272 );
273 }
274 return Effect.succeed(cluster);
275 }),
276 Effect.retry({ schedule: policy }),
277 );
278 });
279
280 const toAttrs = Effect.fn(function* (cluster: dax.Cluster) {
281 if (!cluster.ClusterName || !cluster.ClusterArn) {
282 return yield* Effect.fail(
283 new Error(
284 `DAX cluster '${cluster.ClusterName}' is missing its ARN`,
285 ),
286 );
287 }
288 return {
289 clusterName: cluster.ClusterName,
290 clusterArn: cluster.ClusterArn,
291 status: cluster.Status ?? "available",
292 nodeType: cluster.NodeType ?? DEFAULT_NODE_TYPE,
293 totalNodes: cluster.TotalNodes,
294 activeNodes: cluster.ActiveNodes,
295 discoveryEndpointAddress: cluster.ClusterDiscoveryEndpoint?.Address,
296 discoveryEndpointPort: cluster.ClusterDiscoveryEndpoint?.Port,
297 discoveryEndpointUrl: cluster.ClusterDiscoveryEndpoint?.URL,
298 subnetGroupName: cluster.SubnetGroup,
299 iamRoleArn: cluster.IamRoleArn,
300 parameterGroupName: cluster.ParameterGroup?.ParameterGroupName,
301 securityGroupIds: (cluster.SecurityGroups ?? [])
302 .map((g) => g.SecurityGroupIdentifier)
303 .filter((g): g is string => g !== undefined),
304 clusterEndpointEncryptionType: cluster.ClusterEndpointEncryptionType,
305 tags: yield* readDaxTags(cluster.ClusterArn),
306 };
307 });
308
309 return {
310 stables: ["clusterName", "clusterArn"],
311
312 diff: Effect.fn(function* ({ id, olds, news }) {
313 if (!isResolved(news)) return undefined;
314 const n = news ?? { iamRoleArn: "" };
315 const o = olds ?? { iamRoleArn: "" };
316 if ((yield* toName(id, o)) !== (yield* toName(id, n))) {
317 return { action: "replace" } as const;
318 }
319 // Create-only properties force a replacement.
320 if (
321 (n.nodeType ?? DEFAULT_NODE_TYPE) !==
322 (o.nodeType ?? DEFAULT_NODE_TYPE)
323 ) {
324 return { action: "replace" } as const;
325 }
326 if (n.iamRoleArn !== o.iamRoleArn) {
327 return { action: "replace" } as const;
328 }
329 if (
330 n.subnetGroupName !== undefined &&
331 o.subnetGroupName !== undefined &&
332 n.subnetGroupName !== o.subnetGroupName
333 ) {
334 return { action: "replace" } as const;
335 }
336 if ((n.sseEnabled ?? false) !== (o.sseEnabled ?? false)) {
337 return { action: "replace" } as const;
338 }
339 if (
340 (n.clusterEndpointEncryptionType ?? "NONE") !==
341 (o.clusterEndpointEncryptionType ?? "NONE")
342 ) {
343 return { action: "replace" } as const;
344 }
345 if ((n.networkType ?? undefined) !== (o.networkType ?? undefined)) {
346 return { action: "replace" } as const;
347 }
348 if (!sameStringSet(n.availabilityZones, o.availabilityZones)) {
349 return { action: "replace" } as const;
350 }
351 }),
352
353 read: Effect.fn(function* ({ id, olds, output }) {
354 const name =
355 output?.clusterName ??
356 (yield* toName(id, olds ?? { iamRoleArn: "" }));
357 const cluster = yield* readCluster(name);
358 if (!cluster?.ClusterArn) return undefined;
359 const attrs = yield* toAttrs(cluster);
360 return (yield* hasAlchemyTags(id, attrs.tags))
361 ? attrs
362 : Unowned(attrs);
363 }),
364
365 reconcile: Effect.fn(function* ({ id, news, output, session }) {
366 const props = news!;
367 const name = output?.clusterName ?? (yield* toName(id, props));
368 const internalTags = yield* createInternalTags(id);
369 const desiredTags = { ...internalTags, ...props.tags };
370 const desiredReplicationFactor =
371 props.replicationFactor ?? DEFAULT_REPLICATION_FACTOR;
372
373 // 1. Observe — cloud state is authoritative.
374 let observed = yield* readCluster(name);
375
376 // 2. Ensure — create if missing; tolerate AlreadyExists as a race.
377 if (observed === undefined) {
378 yield* dax
379 .createCluster({
380 ClusterName: name,
381 NodeType: props.nodeType ?? DEFAULT_NODE_TYPE,
382 ReplicationFactor: desiredReplicationFactor,
383 IamRoleArn: props.iamRoleArn,
384 Description: props.description,
385 AvailabilityZones: props.availabilityZones,
386 SubnetGroupName: props.subnetGroupName,
387 SecurityGroupIds: props.securityGroupIds,
388 PreferredMaintenanceWindow: props.preferredMaintenanceWindow,
389 NotificationTopicArn: props.notificationTopicArn,
390 ParameterGroupName: props.parameterGroupName,
391 SSESpecification:
392 props.sseEnabled !== undefined
393 ? { Enabled: props.sseEnabled }
394 : undefined,
395 ClusterEndpointEncryptionType:
396 props.clusterEndpointEncryptionType,
397 NetworkType: props.networkType,
398 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
399 Key,
400 Value,
401 })),
402 })
403 .pipe(
404 retryIamRolePropagation,
405 Effect.catchTag("ClusterAlreadyExistsFault", () => Effect.void),
406 );
407 }
408
409 // Provisioning and in-flight modifications both surface as a
410 // non-available status; wait for availability (bounded) so update
411 // calls do not hit InvalidClusterStateFault.
412 observed = yield* waitForAvailable(name);
413
414 // 3. Sync — compute the update delta from OBSERVED state.
415 const update: dax.UpdateClusterRequest = { ClusterName: name };
416 let mutated = false;
417 if (
418 props.description !== undefined &&
419 props.description !== observed.Description
420 ) {
421 update.Description = props.description;
422 mutated = true;
423 }
424 if (
425 props.preferredMaintenanceWindow !== undefined &&
426 props.preferredMaintenanceWindow !==
427 observed.PreferredMaintenanceWindow
428 ) {
429 update.PreferredMaintenanceWindow =
430 props.preferredMaintenanceWindow;
431 mutated = true;
432 }
433 if (
434 props.notificationTopicArn !== undefined &&
435 props.notificationTopicArn !==
436 observed.NotificationConfiguration?.TopicArn
437 ) {
438 update.NotificationTopicArn = props.notificationTopicArn;
439 mutated = true;
440 }
441 if (
442 props.parameterGroupName !== undefined &&
443 props.parameterGroupName !==
444 observed.ParameterGroup?.ParameterGroupName
445 ) {
446 update.ParameterGroupName = props.parameterGroupName;
447 mutated = true;
448 }
449 const observedSg = (observed.SecurityGroups ?? [])
450 .map((g) => g.SecurityGroupIdentifier)
451 .filter((g): g is string => g !== undefined);
452 if (
453 props.securityGroupIds !== undefined &&
454 !sameStringSet(props.securityGroupIds, observedSg)
455 ) {
456 update.SecurityGroupIds = props.securityGroupIds;
457 mutated = true;
458 }
459 if (mutated) {
460 yield* dax.updateCluster(update);
461 observed = yield* waitForAvailable(name);
462 }
463
464 // 3b. Sync replication factor — DAX has dedicated add/remove-node
465 // APIs instead of a general update.
466 const observedFactor = observed.TotalNodes;
467 if (
468 observedFactor !== undefined &&
469 observedFactor !== desiredReplicationFactor
470 ) {
471 if (desiredReplicationFactor > observedFactor) {
472 yield* dax.increaseReplicationFactor({
473 ClusterName: name,
474 NewReplicationFactor: desiredReplicationFactor,
475 });
476 } else {
477 yield* dax.decreaseReplicationFactor({
478 ClusterName: name,
479 NewReplicationFactor: desiredReplicationFactor,
480 });
481 }
482 observed = yield* waitForAvailable(name);
483 }
484
485 // 3c. Sync tags — diff against OBSERVED cloud tags.
486 const arn = observed.ClusterArn;
487 if (arn) {
488 const observedTags = yield* readDaxTags(arn);
489 const { removed, upsert } = diffTags(observedTags, desiredTags);
490 if (upsert.length > 0) {
491 yield* dax.tagResource({ ResourceName: arn, Tags: upsert });
492 }
493 if (removed.length > 0) {
494 yield* dax.untagResource({ ResourceName: arn, TagKeys: removed });
495 }
496 }
497
498 yield* session.note(name);
499 return yield* toAttrs(observed);
500 }),
501
502 delete: Effect.fn(function* ({ output }) {
503 const name = output.clusterName;
504 // A cluster mid-create/modify rejects deletion with
505 // InvalidClusterStateFault — wait (bounded) for it to settle
506 // first. A cluster already deleting (or gone) is success.
507 const settled = yield* waitUntilSettled(name).pipe(
508 Effect.catch(() => Effect.succeed(undefined)),
509 );
510 if (settled === undefined || settled.Status === "deleting") {
511 return;
512 }
513 yield* dax.deleteCluster({ ClusterName: name }).pipe(
514 Effect.catchTag("ClusterNotFoundFault", () => Effect.void),
515 Effect.retry({
516 while: (e) => e._tag === "InvalidClusterStateFault",
517 schedule: Schedule.max([
518 Schedule.fixed("15 seconds"),
519 Schedule.recurs(20),
520 ]),
521 }),
522 // Exhausted retries: the cluster is in a state (e.g. an
523 // out-of-band delete already in flight) where deletion is
524 // already converging — treat as success like the NotFound case.
525 Effect.catchTag("InvalidClusterStateFault", () => Effect.void),
526 );
527 }),
528
529 list: () =>
530 Effect.gen(function* () {
531 // Bounded hand-rolled pagination (no distilled paginator).
532 const clusters: dax.Cluster[] = [];
533 let nextToken: string | undefined;
534 for (let page = 0; page < 20; page++) {
535 const response = yield* dax.describeClusters({
536 NextToken: nextToken,
537 });
538 clusters.push(...(response.Clusters ?? []));
539 nextToken = response.NextToken;
540 if (!nextToken) break;
541 }
542 return yield* Effect.forEach(
543 clusters.filter(
544 (cluster) =>
545 cluster.ClusterName !== undefined &&
546 cluster.ClusterArn !== undefined,
547 ),
548 (cluster) => toAttrs(cluster),
549 { concurrency: 4 },
550 );
551 }),
552 };
553 }),
554 );