1import * as memorydb from "@distilled.cloud/aws/memorydb";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
12import { toWireDays } from "../../Util/Duration.ts";
13import type { Providers } from "../Providers.ts";
14import { readMemoryDbTags, sameStringSet } from "./internal.ts";
15
16export interface ClusterProps {
17 /**
18 * Name of the cluster. Must be 1-40 alphanumeric characters. If omitted, a
19 * deterministic physical name is generated. Changing the name replaces the
20 * cluster.
21 */
22 clusterName?: string;
23 /**
24 * Compute and memory capacity of the nodes, e.g. `"db.t4g.small"` (the
25 * cheapest supported node type) or `"db.r7g.large"`.
26 * @default "db.t4g.small"
27 */
28 nodeType?: string;
29 /**
30 * Name of the {@link ACL} that authenticates connections to the cluster.
31 * Required by MemoryDB. The ACL must include a user named `default`.
32 */
33 aclName: string;
34 /**
35 * Name of the {@link SubnetGroup} the cluster's nodes are placed into.
36 * Changing the subnet group replaces the cluster.
37 * @default the default subnet group
38 */
39 subnetGroupName?: string;
40 /**
41 * VPC security groups that control network access to the cluster endpoint.
42 * @default the VPC's default security group
43 */
44 securityGroupIds?: string[];
45 /**
46 * Number of shards (partitions) in the cluster.
47 * @default 1
48 */
49 numShards?: number;
50 /**
51 * Number of read replicas per shard (0-5).
52 * @default 1
53 */
54 numReplicasPerShard?: number;
55 /**
56 * Name of the parameter group applied to the cluster.
57 * @default the engine's default parameter group
58 */
59 parameterGroupName?: string;
60 /**
61 * Human-readable description of the cluster.
62 */
63 description?: string;
64 /**
65 * Engine — `"redis"` or `"valkey"`.
66 * @default "valkey"
67 */
68 engine?: string;
69 /**
70 * Engine version, e.g. `"7.1"` (redis) or `"7.2"` (valkey).
71 * @default latest for the engine
72 */
73 engineVersion?: string;
74 /**
75 * Port the cluster accepts connections on. Changing the port replaces the
76 * cluster.
77 * @default 6379
78 */
79 port?: number;
80 /**
81 * Whether in-transit encryption (TLS) is enabled. Changing this replaces the
82 * cluster.
83 * @default true
84 */
85 tlsEnabled?: boolean;
86 /**
87 * Customer-managed KMS key for encryption at rest. Changing the key replaces
88 * the cluster.
89 * @default AWS-owned key
90 */
91 kmsKeyId?: string;
92 /**
93 * Weekly maintenance window, e.g. `"sun:23:00-mon:01:30"` (UTC).
94 */
95 maintenanceWindow?: string;
96 /**
97 * How long automatic snapshots are retained (e.g. `"7 days"` or
98 * `Duration.days(7)`). Sent to the API in whole days; `Duration.zero`
99 * disables automatic snapshots.
100 */
101 snapshotRetentionLimit?: Duration.Input;
102 /**
103 * Daily window (UTC, `HH:MM-HH:MM`) when automatic snapshots are taken.
104 */
105 snapshotWindow?: string;
106 /**
107 * SNS topic ARN to publish cluster events to.
108 */
109 snsTopicArn?: string;
110 /**
111 * Whether minor engine version upgrades are applied automatically.
112 * @default true
113 */
114 autoMinorVersionUpgrade?: boolean;
115 /**
116 * Whether data tiering (SSD) is enabled. Only supported on r6gd node types.
117 * Changing this replaces the cluster.
118 */
119 dataTiering?: boolean;
120 /**
121 * IP address type. Changing this replaces the cluster.
122 * @default "ipv4"
123 */
124 networkType?: memorydb.NetworkType;
125 /**
126 * IP discovery protocol.
127 */
128 ipDiscovery?: memorydb.IpDiscovery;
129 /**
130 * User-defined tags for the cluster.
131 */
132 tags?: Record<string, string>;
133}
134
135export interface Cluster extends Resource<
136 "AWS.MemoryDB.Cluster",
137 ClusterProps,
138 {
139 /** Name of the cluster. */
140 clusterName: string;
141 /** ARN of the cluster. */
142 clusterArn: string;
143 /** Current lifecycle status (e.g. `creating`, `available`). */
144 status: string;
145 /** Node instance type (e.g. `db.t4g.small`). */
146 nodeType: string;
147 /** Engine (`redis` or `valkey`). */
148 engine: string;
149 /** Running engine version. */
150 engineVersion: string | undefined;
151 /** DNS address of the cluster endpoint. */
152 endpointAddress: string | undefined;
153 /** Port of the cluster endpoint. */
154 endpointPort: number | undefined;
155 /** Name of the ACL attached to the cluster. */
156 aclName: string | undefined;
157 /** Name of the parameter group in use. */
158 parameterGroupName: string | undefined;
159 /** Name of the subnet group the cluster's nodes are placed in. */
160 subnetGroupName: string | undefined;
161 /** Whether in-transit encryption (TLS) is enabled. */
162 tlsEnabled: boolean | undefined;
163 /** Number of shards in the cluster. */
164 numberOfShards: number | undefined;
165 /** Tags on the cluster (user + internal Alchemy tags). */
166 tags: Record<string, string>;
167 },
168 never,
169 Providers
170> {}
171
172/**
173 * An Amazon MemoryDB cluster — a durable, Redis/Valkey-compatible in-memory
174 * database.
175 *
176 * Clusters take roughly 10-15 minutes to provision and are billed per node
177 * while they exist. They are reachable only from inside a VPC and require an
178 * {@link ACL}; place them in a {@link SubnetGroup} spanning multiple AZs for
179 * high availability. Destroy clusters you are not using.
180 * ### Creating a Cluster
181 * **Example:** Single-Shard Cluster
182 * ```typescript
183 * const user = yield* User("CacheUser", {
184 * authenticationMode: { type: "password", passwords: [cachePassword] },
185 * accessString: "on ~* +@all",
186 * });
187 * const acl = yield* ACL("CacheAcl", { userNames: [user.userName] });
188 * const subnetGroup = yield* SubnetGroup("CacheSubnets", {
189 * subnetIds: [subnetA.subnetId, subnetB.subnetId],
190 * });
191 * const cluster = yield* Cluster("Cache", {
192 * nodeType: "db.t4g.small",
193 * aclName: acl.aclName,
194 * subnetGroupName: subnetGroup.subnetGroupName,
195 * numShards: 1,
196 * numReplicasPerShard: 1,
197 * });
198 * ```
199 *
200 * @resource
201 */
202export const Cluster = Resource<Cluster>("AWS.MemoryDB.Cluster");
203
204const DEFAULT_NODE_TYPE = "db.t4g.small";
205
206export const ClusterProvider = () =>
207 Provider.effect(
208 Cluster,
209 Effect.gen(function* () {
210 const toName = (id: string, props: Partial<ClusterProps>) =>
211 props.clusterName
212 ? Effect.succeed(props.clusterName)
213 : createPhysicalName({ id, maxLength: 40, lowercase: true });
214
215 const readCluster = Effect.fn(function* (name: string) {
216 const response = yield* memorydb
217 .describeClusters({ ClusterName: name })
218 .pipe(
219 Effect.catchTag("ClusterNotFoundFault", () =>
220 Effect.succeed(undefined),
221 ),
222 );
223 return response?.Clusters?.[0];
224 });
225
226 // Bounded readiness wait. MemoryDB cluster provisioning/modification
227 // typically completes in 10-15 minutes; budget ~20 min (80 * 15s).
228 const waitForAvailable = Effect.fn(function* (name: string) {
229 const policy = Schedule.max([
230 Schedule.fixed("15 seconds"),
231 Schedule.recurs(80),
232 ]);
233 return yield* readCluster(name).pipe(
234 Effect.flatMap((cluster) => {
235 if (!cluster?.ARN) {
236 return Effect.fail(new Error(`Cluster '${name}' not found`));
237 }
238 if (cluster.Status !== "available") {
239 return Effect.fail(
240 new Error(
241 `Cluster '${name}' not available (status: ${cluster.Status})`,
242 ),
243 );
244 }
245 return Effect.succeed(cluster);
246 }),
247 Effect.retry({ schedule: policy }),
248 );
249 });
250
251 // Wait for a cluster to leave a transitional state before delete. Ends
252 // when the cluster is available, deleting, or gone.
253 const waitUntilSettled = Effect.fn(function* (name: string) {
254 const policy = Schedule.max([
255 Schedule.fixed("15 seconds"),
256 Schedule.recurs(80),
257 ]);
258 return yield* readCluster(name).pipe(
259 Effect.flatMap((cluster) => {
260 if (
261 cluster !== undefined &&
262 cluster.Status !== "available" &&
263 cluster.Status !== "deleting"
264 ) {
265 return Effect.fail(
266 new Error(
267 `Cluster '${name}' still settling (status: ${cluster.Status})`,
268 ),
269 );
270 }
271 return Effect.succeed(cluster);
272 }),
273 Effect.retry({ schedule: policy }),
274 );
275 });
276
277 const toAttrs = Effect.fn(function* (cluster: memorydb.Cluster) {
278 if (!cluster.Name || !cluster.ARN) {
279 return yield* Effect.fail(
280 new Error(`Cluster '${cluster.Name}' is missing its ARN`),
281 );
282 }
283 return {
284 clusterName: cluster.Name,
285 clusterArn: cluster.ARN,
286 status: cluster.Status ?? "available",
287 nodeType: cluster.NodeType ?? DEFAULT_NODE_TYPE,
288 engine: cluster.Engine ?? "valkey",
289 engineVersion: cluster.EngineVersion,
290 endpointAddress: cluster.ClusterEndpoint?.Address,
291 endpointPort: cluster.ClusterEndpoint?.Port,
292 aclName: cluster.ACLName,
293 parameterGroupName: cluster.ParameterGroupName,
294 subnetGroupName: cluster.SubnetGroupName,
295 tlsEnabled: cluster.TLSEnabled,
296 numberOfShards: cluster.NumberOfShards,
297 tags: yield* readMemoryDbTags(cluster.ARN),
298 };
299 });
300
301 return {
302 stables: ["clusterName", "clusterArn"],
303
304 diff: Effect.fn(function* ({ id, olds, news }) {
305 if (!isResolved(news)) return undefined;
306 const n = news ?? { aclName: "" };
307 const o = olds ?? { aclName: "" };
308 if ((yield* toName(id, o)) !== (yield* toName(id, n))) {
309 return { action: "replace" } as const;
310 }
311 // Create-only properties force a replacement.
312 if (
313 n.subnetGroupName !== undefined &&
314 o.subnetGroupName !== undefined &&
315 n.subnetGroupName !== o.subnetGroupName
316 ) {
317 return { action: "replace" } as const;
318 }
319 if ((n.port ?? undefined) !== (o.port ?? undefined)) {
320 return { action: "replace" } as const;
321 }
322 if ((n.tlsEnabled ?? undefined) !== (o.tlsEnabled ?? undefined)) {
323 return { action: "replace" } as const;
324 }
325 if ((n.kmsKeyId ?? undefined) !== (o.kmsKeyId ?? undefined)) {
326 return { action: "replace" } as const;
327 }
328 if ((n.networkType ?? undefined) !== (o.networkType ?? undefined)) {
329 return { action: "replace" } as const;
330 }
331 if ((n.dataTiering ?? undefined) !== (o.dataTiering ?? undefined)) {
332 return { action: "replace" } as const;
333 }
334 if ((n.engine ?? undefined) !== (o.engine ?? undefined)) {
335 return { action: "replace" } as const;
336 }
337 }),
338
339 read: Effect.fn(function* ({ id, olds, output }) {
340 const name =
341 output?.clusterName ?? (yield* toName(id, olds ?? { aclName: "" }));
342 const cluster = yield* readCluster(name);
343 if (!cluster?.ARN) return undefined;
344 const attrs = yield* toAttrs(cluster);
345 return (yield* hasAlchemyTags(id, attrs.tags))
346 ? attrs
347 : Unowned(attrs);
348 }),
349
350 reconcile: Effect.fn(function* ({ id, news, output, session }) {
351 const props = news!;
352 const name = output?.clusterName ?? (yield* toName(id, props));
353 const internalTags = yield* createInternalTags(id);
354 const desiredTags = { ...internalTags, ...props.tags };
355 // The wire field is whole days.
356 const snapshotRetentionDays = toWireDays(
357 props.snapshotRetentionLimit,
358 );
359
360 // 1. Observe — cloud state is authoritative.
361 let observed = yield* readCluster(name);
362
363 // 2. Ensure — create if missing; tolerate AlreadyExists as a race.
364 if (observed === undefined) {
365 yield* memorydb
366 .createCluster({
367 ClusterName: name,
368 NodeType: props.nodeType ?? DEFAULT_NODE_TYPE,
369 ACLName: props.aclName,
370 SubnetGroupName: props.subnetGroupName,
371 SecurityGroupIds: props.securityGroupIds,
372 NumShards: props.numShards,
373 NumReplicasPerShard: props.numReplicasPerShard,
374 ParameterGroupName: props.parameterGroupName,
375 Description: props.description,
376 Engine: props.engine,
377 EngineVersion: props.engineVersion,
378 Port: props.port,
379 TLSEnabled: props.tlsEnabled,
380 KmsKeyId: props.kmsKeyId,
381 MaintenanceWindow: props.maintenanceWindow,
382 SnapshotRetentionLimit: snapshotRetentionDays,
383 SnapshotWindow: props.snapshotWindow,
384 SnsTopicArn: props.snsTopicArn,
385 AutoMinorVersionUpgrade: props.autoMinorVersionUpgrade,
386 DataTiering: props.dataTiering,
387 NetworkType: props.networkType,
388 IpDiscovery: props.ipDiscovery,
389 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
390 Key,
391 Value,
392 })),
393 })
394 .pipe(
395 Effect.catchTag("ClusterAlreadyExistsFault", () => Effect.void),
396 );
397 }
398
399 // Provisioning and in-flight modifications both surface as a
400 // non-available status; wait for availability (bounded) so modify
401 // calls do not hit InvalidClusterStateFault.
402 observed = yield* waitForAvailable(name);
403
404 // 3. Sync — compute the update delta from OBSERVED state.
405 const update: memorydb.UpdateClusterRequest = { ClusterName: name };
406 let mutated = false;
407 if (
408 props.description !== undefined &&
409 props.description !== observed.Description
410 ) {
411 update.Description = props.description;
412 mutated = true;
413 }
414 const observedSg = (observed.SecurityGroups ?? [])
415 .map((g) => g.SecurityGroupId)
416 .filter((g): g is string => g !== undefined);
417 if (
418 props.securityGroupIds !== undefined &&
419 !sameStringSet(props.securityGroupIds, observedSg)
420 ) {
421 update.SecurityGroupIds = props.securityGroupIds;
422 mutated = true;
423 }
424 if (
425 props.maintenanceWindow !== undefined &&
426 props.maintenanceWindow !== observed.MaintenanceWindow
427 ) {
428 update.MaintenanceWindow = props.maintenanceWindow;
429 mutated = true;
430 }
431 if (
432 snapshotRetentionDays !== undefined &&
433 snapshotRetentionDays !== (observed.SnapshotRetentionLimit ?? 0)
434 ) {
435 update.SnapshotRetentionLimit = snapshotRetentionDays;
436 mutated = true;
437 }
438 if (
439 props.snapshotWindow !== undefined &&
440 props.snapshotWindow !== observed.SnapshotWindow
441 ) {
442 update.SnapshotWindow = props.snapshotWindow;
443 mutated = true;
444 }
445 if (
446 props.nodeType !== undefined &&
447 props.nodeType !== observed.NodeType
448 ) {
449 update.NodeType = props.nodeType;
450 mutated = true;
451 }
452 if (
453 props.engineVersion !== undefined &&
454 props.engineVersion !== observed.EngineVersion
455 ) {
456 update.EngineVersion = props.engineVersion;
457 mutated = true;
458 }
459 if (
460 props.parameterGroupName !== undefined &&
461 props.parameterGroupName !== observed.ParameterGroupName
462 ) {
463 update.ParameterGroupName = props.parameterGroupName;
464 mutated = true;
465 }
466 if (
467 props.aclName !== undefined &&
468 props.aclName !== observed.ACLName
469 ) {
470 update.ACLName = props.aclName;
471 mutated = true;
472 }
473 if (
474 props.ipDiscovery !== undefined &&
475 props.ipDiscovery !== observed.IpDiscovery
476 ) {
477 update.IpDiscovery = props.ipDiscovery;
478 mutated = true;
479 }
480 if (
481 props.numShards !== undefined &&
482 props.numShards !== observed.NumberOfShards
483 ) {
484 update.ShardConfiguration = { ShardCount: props.numShards };
485 mutated = true;
486 }
487
488 if (mutated) {
489 yield* memorydb.updateCluster(update);
490 observed = yield* waitForAvailable(name);
491 }
492
493 // 3b. Sync tags — diff against OBSERVED cloud tags.
494 const arn = observed.ARN;
495 if (arn) {
496 const observedTags = yield* readMemoryDbTags(arn);
497 const { removed, upsert } = diffTags(observedTags, desiredTags);
498 if (upsert.length > 0) {
499 yield* memorydb.tagResource({ ResourceArn: arn, Tags: upsert });
500 }
501 if (removed.length > 0) {
502 yield* memorydb.untagResource({
503 ResourceArn: arn,
504 TagKeys: removed,
505 });
506 }
507 }
508
509 yield* session.note(name);
510 return yield* toAttrs(observed);
511 }),
512
513 delete: Effect.fn(function* ({ output }) {
514 const name = output.clusterName;
515 // A cluster mid-create/modify rejects deletion with
516 // InvalidClusterStateFault — wait (bounded) for it to settle first.
517 // A cluster already deleting (or gone) is success.
518 yield* waitUntilSettled(name).pipe(Effect.catch(() => Effect.void));
519 yield* memorydb.deleteCluster({ ClusterName: name }).pipe(
520 Effect.catchTag("ClusterNotFoundFault", () => Effect.void),
521 Effect.retry({
522 while: (e) => e._tag === "InvalidClusterStateFault",
523 schedule: Schedule.max([
524 Schedule.fixed("15 seconds"),
525 Schedule.recurs(20),
526 ]),
527 }),
528 Effect.catchTag("InvalidClusterStateFault", () => Effect.void),
529 );
530 }),
531
532 list: () =>
533 memorydb.describeClusters.pages({}).pipe(
534 Stream.runCollect,
535 Effect.map((chunk) =>
536 Array.from(chunk).flatMap((page) =>
537 (page.Clusters ?? []).filter(
538 (cluster) =>
539 cluster.Name !== undefined && cluster.ARN !== undefined,
540 ),
541 ),
542 ),
543 Effect.flatMap(
544 Effect.forEach((cluster) => toAttrs(cluster), { concurrency: 4 }),
545 ),
546 ),
547 };
548 }),
549 );