1import * as elasticache from "@distilled.cloud/aws/elasticache";
2import * as Data from "effect/Data";
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 type { Providers } from "../Providers.ts";
13import { readElastiCacheTags, sameStringSet, tagsToWire } from "./internal.ts";
14
15export interface ReplicationGroupProps {
16 /** Replication group identifier. Generated deterministically when omitted. */
17 replicationGroupId?: string;
18 /** Required human-readable description. */
19 description: string;
20 /** Cache engine. */
21 engine: "valkey" | "redis";
22 engineVersion?: string;
23 /** ElastiCache node type. Must start with `cache.`, for example `cache.t4g.micro`. @default cache.t4g.micro */
24 nodeType?: `cache.${string}`;
25 subnetGroupName?: string;
26 securityGroupIds?: string[];
27 /** Shard count. Changes use online resharding. */
28 numNodeGroups?: number;
29 /** Read replicas per shard, from 0 through 5. */
30 replicasPerNodeGroup?: number;
31 automaticFailoverEnabled?: boolean;
32 multiAzEnabled?: boolean;
33 parameterGroupName?: string;
34 port?: number;
35 maintenanceWindow?: string;
36 snapshotRetentionLimit?: number;
37 snapshotWindow?: string;
38 autoMinorVersionUpgrade?: boolean;
39 /** Enables TLS for client connections. @default true */
40 transitEncryptionEnabled?: boolean;
41 transitEncryptionMode?: elasticache.TransitEncryptionMode;
42 atRestEncryptionEnabled?: boolean;
43 kmsKeyId?: string;
44 networkType?: elasticache.NetworkType;
45 ipDiscovery?: elasticache.IpDiscovery;
46 /** User group IDs for Valkey/Redis RBAC. */
47 userGroupIds?: string[];
48 /** Snapshot used only while creating a new group. */
49 snapshotName?: string;
50 snapshotArns?: string[];
51 /** Optional final snapshot created before destroy. */
52 finalSnapshotName?: string;
53 tags?: Record<string, string>;
54}
55
56export interface ReplicationGroup extends Resource<
57 "AWS.ElastiCache.ReplicationGroup",
58 ReplicationGroupProps,
59 {
60 replicationGroupId: string;
61 replicationGroupArn: string;
62 status: string;
63 engine: string;
64 engineVersion: string | undefined;
65 nodeType: string | undefined;
66 configurationEndpointAddress: string | undefined;
67 configurationEndpointPort: number | undefined;
68 primaryEndpointAddress: string | undefined;
69 primaryEndpointPort: number | undefined;
70 readerEndpointAddress: string | undefined;
71 readerEndpointPort: number | undefined;
72 transitEncryptionEnabled: boolean;
73 nodeGroupIds: string[];
74 tags: Record<string, string>;
75 },
76 never,
77 Providers
78> {}
79
80/** Invalid desired topology rejected before AWS can start a partial create. */
81export class InvalidReplicationGroupConfiguration extends Data.TaggedError(
82 "InvalidReplicationGroupConfiguration",
83)<{ message: string }> {}
84
85export const validateReplicationGroupProps = (
86 props: ReplicationGroupProps,
87): InvalidReplicationGroupConfiguration | undefined => {
88 const invalid = (message: string) =>
89 new InvalidReplicationGroupConfiguration({ message });
90 if (
91 props.port !== undefined &&
92 (!Number.isInteger(props.port) || props.port < 1 || props.port > 65_535)
93 ) {
94 return invalid("port must be an integer from 1 through 65535");
95 }
96 if (
97 props.numNodeGroups !== undefined &&
98 (!Number.isInteger(props.numNodeGroups) || props.numNodeGroups < 1)
99 ) {
100 return invalid("numNodeGroups must be an integer of at least 1");
101 }
102 if (
103 props.replicasPerNodeGroup !== undefined &&
104 (!Number.isInteger(props.replicasPerNodeGroup) ||
105 props.replicasPerNodeGroup < 0 ||
106 props.replicasPerNodeGroup > 5)
107 ) {
108 return invalid("replicasPerNodeGroup must be an integer from 0 through 5");
109 }
110 if (
111 props.automaticFailoverEnabled === true &&
112 (props.replicasPerNodeGroup ?? 0) < 1
113 ) {
114 return invalid(
115 "automaticFailoverEnabled requires replicasPerNodeGroup to be at least 1",
116 );
117 }
118 if (props.multiAzEnabled === true && (props.replicasPerNodeGroup ?? 0) < 1) {
119 return invalid(
120 "multiAzEnabled requires replicasPerNodeGroup to be at least 1",
121 );
122 }
123 if (
124 props.multiAzEnabled === true &&
125 props.automaticFailoverEnabled !== true
126 ) {
127 return invalid(
128 "multiAzEnabled requires automaticFailoverEnabled to be true",
129 );
130 }
131 if (
132 props.transitEncryptionMode !== undefined &&
133 props.transitEncryptionEnabled === false
134 ) {
135 return invalid("transitEncryptionMode requires transitEncryptionEnabled");
136 }
137 return undefined;
138};
139
140/**
141 * A provisioned Valkey or Redis OSS replication group.
142 *
143 * ### Creating a Highly Available Cache
144 * **Example:** Valkey with a replica in a VPC
145 * ```typescript
146 * const cache = yield* ReplicationGroup("Cache", {
147 * description: "application cache",
148 * engine: "valkey",
149 * subnetGroupName: subnets.subnetGroupName,
150 * securityGroupIds: [cacheSecurityGroup.groupId],
151 * replicasPerNodeGroup: 1,
152 * automaticFailoverEnabled: true,
153 * multiAzEnabled: true,
154 * transitEncryptionEnabled: true,
155 * });
156 * ```
157 *
158 * @resource
159 */
160export const ReplicationGroup = Resource<ReplicationGroup>(
161 "AWS.ElastiCache.ReplicationGroup",
162);
163
164const DEFAULT_NODE_TYPE = "cache.t4g.micro";
165
166export const ReplicationGroupProvider = () =>
167 Provider.effect(
168 ReplicationGroup,
169 Effect.gen(function* () {
170 const toName = (id: string, props: Partial<ReplicationGroupProps>) =>
171 props.replicationGroupId
172 ? Effect.succeed(props.replicationGroupId)
173 : createPhysicalName({ id, maxLength: 40, lowercase: true });
174 const readGroup = Effect.fn(function* (name: string) {
175 const response = yield* elasticache
176 .describeReplicationGroups({ ReplicationGroupId: name })
177 .pipe(
178 Effect.catchTag("ReplicationGroupNotFoundFault", () =>
179 Effect.succeed(undefined),
180 ),
181 );
182 return response?.ReplicationGroups?.[0];
183 });
184 const waitForAvailable = Effect.fn(function* (name: string) {
185 return yield* readGroup(name).pipe(
186 Effect.flatMap((group) =>
187 group?.Status === "available"
188 ? Effect.succeed(group)
189 : Effect.fail(
190 new Error(`Replication group '${name}' is not available`),
191 ),
192 ),
193 Effect.retry({
194 schedule: Schedule.max([
195 Schedule.fixed("15 seconds"),
196 // Engine upgrades can remain in "modifying" for more than
197 // fifteen minutes before ElastiCache reports "available".
198 Schedule.recurs(100),
199 ]),
200 }),
201 );
202 });
203 const waitForDeleted = Effect.fn(function* (name: string) {
204 return yield* readGroup(name).pipe(
205 Effect.flatMap((group) =>
206 group === undefined
207 ? Effect.void
208 : Effect.fail(
209 new Error(`Replication group '${name}' is still deleting`),
210 ),
211 ),
212 Effect.retry({
213 schedule: Schedule.max([
214 Schedule.fixed("15 seconds"),
215 // Multi-AZ failover deletes can remain visible longer than the
216 // normal fifteen-minute lifecycle window.
217 Schedule.recurs(100),
218 ]),
219 }),
220 );
221 });
222 const retryWhileCacheTransitioning = <A, E, R>(
223 effect: Effect.Effect<A, E, R>,
224 ) =>
225 effect.pipe(
226 Effect.retry({
227 while: (error) =>
228 (error as { _tag?: string })._tag ===
229 "InvalidReplicationGroupStateFault" ||
230 (error as { _tag?: string })._tag ===
231 "InvalidCacheClusterStateFault",
232 schedule: Schedule.max([
233 Schedule.fixed("15 seconds"),
234 // AWS can report the group as available before its member
235 // clusters accept tag changes after an in-place update.
236 Schedule.recurs(100),
237 ]),
238 }),
239 );
240 const attrs = Effect.fn(function* (group: elasticache.ReplicationGroup) {
241 if (
242 !group.ReplicationGroupId ||
243 !group.ARN ||
244 !group.Status ||
245 !group.Engine
246 ) {
247 return yield* Effect.fail(
248 new Error(
249 "ElastiCache replication group is missing its id, ARN, status, or engine",
250 ),
251 );
252 }
253 const engineVersion = group.MemberClusters?.[0]
254 ? (yield* elasticache.describeCacheClusters({
255 CacheClusterId: group.MemberClusters[0],
256 })).CacheClusters?.[0]?.EngineVersion
257 : undefined;
258 const primary = group.NodeGroups?.[0]?.PrimaryEndpoint;
259 const reader = group.NodeGroups?.[0]?.ReaderEndpoint;
260 return {
261 replicationGroupId: group.ReplicationGroupId,
262 replicationGroupArn: group.ARN,
263 status: group.Status,
264 engine: group.Engine,
265 engineVersion,
266 nodeType: group.CacheNodeType,
267 configurationEndpointAddress: group.ConfigurationEndpoint?.Address,
268 configurationEndpointPort: group.ConfigurationEndpoint?.Port,
269 primaryEndpointAddress: primary?.Address,
270 primaryEndpointPort: primary?.Port,
271 readerEndpointAddress: reader?.Address,
272 readerEndpointPort: reader?.Port,
273 transitEncryptionEnabled: group.TransitEncryptionEnabled ?? false,
274 nodeGroupIds: (group.NodeGroups ?? [])
275 .map((nodeGroup) => nodeGroup.NodeGroupId)
276 .filter((id): id is string => id !== undefined),
277 tags: yield* readElastiCacheTags(group.ARN),
278 };
279 });
280 const replacement = (
281 olds: Partial<ReplicationGroupProps>,
282 news: Partial<ReplicationGroupProps>,
283 ) =>
284 olds.subnetGroupName !== news.subnetGroupName ||
285 olds.port !== news.port ||
286 olds.kmsKeyId !== news.kmsKeyId ||
287 olds.atRestEncryptionEnabled !== news.atRestEncryptionEnabled ||
288 olds.networkType !== news.networkType ||
289 olds.snapshotName !== news.snapshotName ||
290 // AWS cannot enable cluster mode on an existing non-cluster-mode
291 // replication group, so scaling from one shard must replace it.
292 ((olds.numNodeGroups ?? 1) <= 1 && (news.numNodeGroups ?? 1) > 1) ||
293 !sameStringSet(olds.snapshotArns, news.snapshotArns);
294
295 return {
296 stables: ["replicationGroupId", "replicationGroupArn"],
297 diff: Effect.fn(function* ({ id, olds, news }) {
298 if (!isResolved(news)) return undefined;
299 if (
300 (yield* toName(id, olds ?? {})) !== (yield* toName(id, news ?? {}))
301 ) {
302 return { action: "replace" } as const;
303 }
304 if (replacement(olds ?? {}, news ?? {}))
305 return { action: "replace" } as const;
306 }),
307 read: Effect.fn(function* ({ id, olds, output }) {
308 const group = yield* readGroup(
309 output?.replicationGroupId ?? (yield* toName(id, olds ?? {})),
310 );
311 if (!group?.ARN) return undefined;
312 const result = yield* attrs(group);
313 return (yield* hasAlchemyTags(id, result.tags))
314 ? result
315 : Unowned(result);
316 }),
317 reconcile: Effect.fn(function* ({ id, news, output, session }) {
318 const props = news!;
319 const invalid = validateReplicationGroupProps(props);
320 if (invalid) return yield* Effect.fail(invalid);
321 const name = output?.replicationGroupId ?? (yield* toName(id, props));
322 const desiredTags = {
323 ...(yield* createInternalTags(id)),
324 ...props.tags,
325 };
326 let group = yield* readGroup(name);
327 if (!group) {
328 yield* elasticache
329 .createReplicationGroup({
330 ReplicationGroupId: name,
331 ReplicationGroupDescription: props.description,
332 Engine: props.engine,
333 EngineVersion: props.engineVersion,
334 CacheNodeType: props.nodeType ?? DEFAULT_NODE_TYPE,
335 CacheSubnetGroupName: props.subnetGroupName,
336 SecurityGroupIds: props.securityGroupIds,
337 NumNodeGroups: props.numNodeGroups,
338 ReplicasPerNodeGroup: props.replicasPerNodeGroup,
339 AutomaticFailoverEnabled: props.automaticFailoverEnabled,
340 MultiAZEnabled: props.multiAzEnabled,
341 CacheParameterGroupName: props.parameterGroupName,
342 Port: props.port,
343 PreferredMaintenanceWindow: props.maintenanceWindow,
344 SnapshotRetentionLimit: props.snapshotRetentionLimit,
345 SnapshotWindow: props.snapshotWindow,
346 AutoMinorVersionUpgrade: props.autoMinorVersionUpgrade,
347 TransitEncryptionEnabled:
348 props.transitEncryptionEnabled ?? true,
349 TransitEncryptionMode: props.transitEncryptionMode,
350 AtRestEncryptionEnabled: props.atRestEncryptionEnabled,
351 KmsKeyId: props.kmsKeyId,
352 NetworkType: props.networkType,
353 IpDiscovery: props.ipDiscovery,
354 UserGroupIds: props.userGroupIds,
355 SnapshotName: props.snapshotName,
356 SnapshotArns: props.snapshotArns,
357 Tags: tagsToWire(desiredTags),
358 })
359 .pipe(
360 Effect.catchTag(
361 "ReplicationGroupAlreadyExistsFault",
362 () => Effect.void,
363 ),
364 );
365 }
366 group = yield* waitForAvailable(name);
367 const update: elasticache.ModifyReplicationGroupMessage = {
368 ReplicationGroupId: name,
369 ApplyImmediately: true,
370 };
371 let changed = false;
372 const set = <
373 K extends keyof elasticache.ModifyReplicationGroupMessage,
374 >(
375 key: K,
376 desired: elasticache.ModifyReplicationGroupMessage[K],
377 observed: unknown,
378 ) => {
379 if (desired !== undefined && desired !== observed) {
380 update[key] = desired;
381 changed = true;
382 }
383 };
384 set(
385 "ReplicationGroupDescription",
386 props.description,
387 group.Description,
388 );
389 set("CacheNodeType", props.nodeType, group.CacheNodeType);
390 set("Engine", props.engine, group.Engine);
391 set(
392 "AutoMinorVersionUpgrade",
393 props.autoMinorVersionUpgrade,
394 group.AutoMinorVersionUpgrade,
395 );
396 set("PreferredMaintenanceWindow", props.maintenanceWindow, undefined);
397 set(
398 "SnapshotRetentionLimit",
399 props.snapshotRetentionLimit,
400 group.SnapshotRetentionLimit,
401 );
402 set("SnapshotWindow", props.snapshotWindow, group.SnapshotWindow);
403 set("CacheParameterGroupName", props.parameterGroupName, undefined);
404 set("IpDiscovery", props.ipDiscovery, group.IpDiscovery);
405 set(
406 "TransitEncryptionMode",
407 props.transitEncryptionMode,
408 group.TransitEncryptionMode,
409 );
410 // ElastiCache reports VPC security groups and engine version on a
411 // member cluster, not on the replication-group response.
412 const member = group.MemberClusters?.[0]
413 ? (yield* elasticache.describeCacheClusters({
414 CacheClusterId: group.MemberClusters[0],
415 })).CacheClusters?.[0]
416 : undefined;
417 const existingSecurityGroups = (member?.SecurityGroups ?? [])
418 .map((membership) => membership.SecurityGroupId)
419 .filter((id): id is string => id !== undefined);
420 set("EngineVersion", props.engineVersion, member?.EngineVersion);
421 if (
422 props.securityGroupIds &&
423 !sameStringSet(props.securityGroupIds, existingSecurityGroups)
424 ) {
425 update.SecurityGroupIds = props.securityGroupIds;
426 changed = true;
427 }
428 const existingUsers = group.UserGroupIds ?? [];
429 if (
430 props.userGroupIds &&
431 !sameStringSet(props.userGroupIds, existingUsers)
432 ) {
433 update.UserGroupIdsToAdd = props.userGroupIds.filter(
434 (id) => !existingUsers.includes(id),
435 );
436 update.UserGroupIdsToRemove = existingUsers.filter(
437 (id) => !props.userGroupIds!.includes(id),
438 );
439 changed = true;
440 }
441 if (changed) {
442 yield* elasticache.modifyReplicationGroup(update);
443 group = yield* waitForAvailable(name);
444 }
445 const currentNodeGroups = group.NodeGroups?.length ?? 1;
446 if (
447 props.numNodeGroups !== undefined &&
448 props.numNodeGroups !== currentNodeGroups
449 ) {
450 yield* elasticache.modifyReplicationGroupShardConfiguration({
451 ReplicationGroupId: name,
452 NodeGroupCount: props.numNodeGroups,
453 ApplyImmediately: true,
454 ...(props.numNodeGroups < currentNodeGroups
455 ? {
456 NodeGroupsToRemove: (group.NodeGroups ?? [])
457 .slice(props.numNodeGroups)
458 .map((node) => node.NodeGroupId)
459 .filter((id): id is string => id !== undefined),
460 }
461 : {}),
462 });
463 group = yield* waitForAvailable(name);
464 }
465 const setAutomaticFailover = Effect.fn(function* (enabled: boolean) {
466 const current = group;
467 if (!current) {
468 return yield* Effect.die(
469 `Replication group '${name}' disappeared while updating automatic failover`,
470 );
471 }
472 if ((current.AutomaticFailover === "enabled") === enabled) return;
473 yield* retryWhileCacheTransitioning(
474 elasticache.modifyReplicationGroup({
475 ReplicationGroupId: name,
476 AutomaticFailoverEnabled: enabled,
477 ApplyImmediately: true,
478 }),
479 );
480 group = yield* waitForAvailable(name);
481 });
482 const setMultiAz = Effect.fn(function* (enabled: boolean) {
483 const current = group;
484 if (!current) {
485 return yield* Effect.die(
486 `Replication group '${name}' disappeared while updating Multi-AZ`,
487 );
488 }
489 if ((current.MultiAZ === "enabled") === enabled) return;
490 yield* retryWhileCacheTransitioning(
491 elasticache.modifyReplicationGroup({
492 ReplicationGroupId: name,
493 MultiAZEnabled: enabled,
494 ApplyImmediately: true,
495 }),
496 );
497 group = yield* waitForAvailable(name);
498 });
499 if (props.replicasPerNodeGroup !== undefined) {
500 const replicaCount = Math.max(
501 ...(group.NodeGroups ?? []).map((node) =>
502 Math.max(0, (node.NodeGroupMembers?.length ?? 1) - 1),
503 ),
504 0,
505 );
506 // AWS rejects HA before the first replica exists, and rejects
507 // removing the final replica while either HA setting is on.
508 if (props.replicasPerNodeGroup === 0 && replicaCount > 0) {
509 yield* setMultiAz(false);
510 yield* setAutomaticFailover(false);
511 }
512 if (props.replicasPerNodeGroup > replicaCount) {
513 yield* elasticache.increaseReplicaCount({
514 ReplicationGroupId: name,
515 NewReplicaCount: props.replicasPerNodeGroup,
516 ApplyImmediately: true,
517 });
518 group = yield* waitForAvailable(name);
519 } else if (props.replicasPerNodeGroup < replicaCount) {
520 yield* elasticache.decreaseReplicaCount({
521 ReplicationGroupId: name,
522 NewReplicaCount: props.replicasPerNodeGroup,
523 ApplyImmediately: true,
524 });
525 group = yield* waitForAvailable(name);
526 }
527 }
528 if (props.automaticFailoverEnabled !== undefined) {
529 yield* setAutomaticFailover(props.automaticFailoverEnabled);
530 }
531 if (props.multiAzEnabled !== undefined) {
532 yield* setMultiAz(props.multiAzEnabled);
533 }
534 if (group.ARN) {
535 const { removed, upsert } = diffTags(
536 yield* retryWhileCacheTransitioning(
537 readElastiCacheTags(group.ARN),
538 ),
539 desiredTags,
540 );
541 if (upsert.length)
542 yield* retryWhileCacheTransitioning(
543 elasticache.addTagsToResource({
544 ResourceName: group.ARN,
545 Tags: upsert,
546 }),
547 );
548 if (removed.length)
549 yield* retryWhileCacheTransitioning(
550 elasticache.removeTagsFromResource({
551 ResourceName: group.ARN,
552 TagKeys: removed,
553 }),
554 );
555 // AWS can accept a tag update before ListTagsForResource is
556 // available again. Do not report the resource as reconciled
557 // until callers can observe the finished tag state.
558 yield* retryWhileCacheTransitioning(readElastiCacheTags(group.ARN));
559 }
560 yield* session.note(name);
561 return yield* attrs(group);
562 }),
563 delete: Effect.fn(function* ({ olds, output }) {
564 yield* retryWhileCacheTransitioning(
565 elasticache
566 .deleteReplicationGroup({
567 ReplicationGroupId: output.replicationGroupId,
568 FinalSnapshotIdentifier: olds.finalSnapshotName,
569 })
570 .pipe(
571 Effect.catchTag(
572 "ReplicationGroupNotFoundFault",
573 () => Effect.void,
574 ),
575 ),
576 );
577 yield* waitForDeleted(output.replicationGroupId);
578 }),
579 list: () =>
580 elasticache.describeReplicationGroups.pages({}).pipe(
581 Stream.runCollect,
582 Effect.map((pages) =>
583 Array.from(pages).flatMap((page) => page.ReplicationGroups ?? []),
584 ),
585 Effect.flatMap((groups) =>
586 Effect.forEach(
587 groups.filter((group) => group.ARN),
588 attrs,
589 ),
590 ),
591 ),
592 };
593 }),
594 );