1import * as elasticache from "@distilled.cloud/aws/elasticache";
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";
12import { readElastiCacheTags, sameStringSet, tagsToWire } from "./internal.ts";
13
14export interface CacheClusterProps {
15 /** Cache cluster identifier. Generated deterministically when omitted. */
16 cacheClusterId?: string;
17 /** Memcached is the supported topology for this resource. */
18 engine?: "memcached";
19 engineVersion?: string;
20 /** ElastiCache node type. Must start with `cache.`, for example `cache.t4g.micro`. @default cache.t4g.micro */
21 nodeType?: `cache.${string}`;
22 /** Number of Memcached nodes. */
23 numCacheNodes?: number;
24 subnetGroupName?: string;
25 securityGroupIds?: string[];
26 parameterGroupName?: string;
27 preferredAvailabilityZones?: string[];
28 maintenanceWindow?: string;
29 notificationTopicArn?: string;
30 autoMinorVersionUpgrade?: boolean;
31 port?: number;
32 networkType?: elasticache.NetworkType;
33 ipDiscovery?: elasticache.IpDiscovery;
34 tags?: Record<string, string>;
35}
36
37export interface CacheNodeEndpoint {
38 address: string;
39 port: number;
40}
41
42export interface CacheCluster extends Resource<
43 "AWS.ElastiCache.CacheCluster",
44 CacheClusterProps,
45 {
46 cacheClusterId: string;
47 cacheClusterArn: string;
48 status: string;
49 engine: string;
50 engineVersion: string | undefined;
51 nodeType: string | undefined;
52 endpoints: CacheNodeEndpoint[];
53 transitEncryptionEnabled: boolean;
54 tags: Record<string, string>;
55 },
56 never,
57 Providers
58> {}
59
60/**
61 * A provisioned Memcached cluster. Use {@link ReplicationGroup} for Valkey or Redis.
62 *
63 * ### Creating a Memcached Cluster
64 * **Example:** Two-node cluster in a VPC
65 * ```typescript
66 * const cache = yield* CacheCluster("Cache", {
67 * subnetGroupName: subnets.subnetGroupName,
68 * securityGroupIds: [cacheSecurityGroup.groupId],
69 * numCacheNodes: 2,
70 * });
71 * ```
72 *
73 * @resource
74 */
75export const CacheCluster = Resource<CacheCluster>(
76 "AWS.ElastiCache.CacheCluster",
77);
78
79export const CacheClusterProvider = () =>
80 Provider.effect(
81 CacheCluster,
82 Effect.gen(function* () {
83 const toName = (id: string, props: Partial<CacheClusterProps>) =>
84 props.cacheClusterId
85 ? Effect.succeed(props.cacheClusterId)
86 : createPhysicalName({ id, maxLength: 50, lowercase: true });
87 const readCluster = Effect.fn(function* (name: string) {
88 const response = yield* elasticache
89 .describeCacheClusters({
90 CacheClusterId: name,
91 ShowCacheNodeInfo: true,
92 })
93 .pipe(
94 Effect.catchTag("CacheClusterNotFoundFault", () =>
95 Effect.succeed(undefined),
96 ),
97 );
98 return response?.CacheClusters?.[0];
99 });
100 const waitForAvailable = Effect.fn(function* (name: string) {
101 return yield* readCluster(name).pipe(
102 Effect.flatMap((cluster) =>
103 cluster?.CacheClusterStatus === "available"
104 ? Effect.succeed(cluster)
105 : Effect.fail(
106 new Error(`Cache cluster '${name}' is not available`),
107 ),
108 ),
109 Effect.retry({
110 schedule: Schedule.max([
111 Schedule.fixed("15 seconds"),
112 Schedule.recurs(60),
113 ]),
114 }),
115 );
116 });
117 const waitForDeleted = Effect.fn(function* (name: string) {
118 return yield* readCluster(name).pipe(
119 Effect.flatMap((cluster) =>
120 cluster === undefined
121 ? Effect.void
122 : Effect.fail(
123 new Error(`Cache cluster '${name}' is still deleting`),
124 ),
125 ),
126 Effect.retry({
127 schedule: Schedule.max([
128 Schedule.fixed("15 seconds"),
129 Schedule.recurs(60),
130 ]),
131 }),
132 );
133 });
134 const attrs = Effect.fn(function* (cluster: elasticache.CacheCluster) {
135 if (!cluster.CacheClusterId || !cluster.ARN) {
136 return yield* Effect.fail(
137 new Error("ElastiCache cache cluster is missing its id or ARN"),
138 );
139 }
140 return {
141 cacheClusterId: cluster.CacheClusterId,
142 cacheClusterArn: cluster.ARN,
143 status: cluster.CacheClusterStatus ?? "available",
144 engine: cluster.Engine ?? "memcached",
145 engineVersion: cluster.EngineVersion,
146 nodeType: cluster.CacheNodeType,
147 endpoints: (cluster.CacheNodes ?? []).flatMap((node) =>
148 node.Endpoint?.Address !== undefined &&
149 node.Endpoint.Port !== undefined
150 ? [{ address: node.Endpoint.Address, port: node.Endpoint.Port }]
151 : [],
152 ),
153 transitEncryptionEnabled: cluster.TransitEncryptionEnabled ?? false,
154 tags: yield* readElastiCacheTags(cluster.ARN),
155 };
156 });
157 const replaces = (
158 olds: Partial<CacheClusterProps>,
159 news: Partial<CacheClusterProps>,
160 ) =>
161 olds.subnetGroupName !== news.subnetGroupName ||
162 olds.port !== news.port ||
163 olds.networkType !== news.networkType ||
164 olds.engine !== news.engine;
165
166 return {
167 stables: ["cacheClusterId", "cacheClusterArn"],
168 diff: Effect.fn(function* ({ id, olds, news }) {
169 if (!isResolved(news)) return undefined;
170 if (
171 (yield* toName(id, olds ?? {})) !== (yield* toName(id, news ?? {}))
172 ) {
173 return { action: "replace" } as const;
174 }
175 if (replaces(olds ?? {}, news ?? {}))
176 return { action: "replace" } as const;
177 }),
178 read: Effect.fn(function* ({ id, olds, output }) {
179 const cluster = yield* readCluster(
180 output?.cacheClusterId ?? (yield* toName(id, olds ?? {})),
181 );
182 if (!cluster?.ARN) return undefined;
183 const result = yield* attrs(cluster);
184 return (yield* hasAlchemyTags(id, result.tags))
185 ? result
186 : Unowned(result);
187 }),
188 reconcile: Effect.fn(function* ({ id, news, output, session }) {
189 const props = news ?? {};
190 const name = output?.cacheClusterId ?? (yield* toName(id, props));
191 const desiredTags = {
192 ...(yield* createInternalTags(id)),
193 ...props.tags,
194 };
195 let cluster = yield* readCluster(name);
196 if (!cluster) {
197 yield* elasticache
198 .createCacheCluster({
199 CacheClusterId: name,
200 Engine: "memcached",
201 EngineVersion: props.engineVersion,
202 CacheNodeType: props.nodeType ?? "cache.t4g.micro",
203 NumCacheNodes: props.numCacheNodes ?? 1,
204 CacheSubnetGroupName: props.subnetGroupName,
205 SecurityGroupIds: props.securityGroupIds,
206 CacheParameterGroupName: props.parameterGroupName,
207 PreferredAvailabilityZones: props.preferredAvailabilityZones,
208 PreferredMaintenanceWindow: props.maintenanceWindow,
209 NotificationTopicArn: props.notificationTopicArn,
210 AutoMinorVersionUpgrade: props.autoMinorVersionUpgrade,
211 Port: props.port,
212 NetworkType: props.networkType,
213 IpDiscovery: props.ipDiscovery,
214 Tags: tagsToWire(desiredTags),
215 })
216 .pipe(
217 Effect.catchTag(
218 "CacheClusterAlreadyExistsFault",
219 () => Effect.void,
220 ),
221 );
222 }
223 cluster = yield* waitForAvailable(name);
224 const update: elasticache.ModifyCacheClusterMessage = {
225 CacheClusterId: name,
226 ApplyImmediately: true,
227 };
228 let changed = false;
229 const set = <K extends keyof elasticache.ModifyCacheClusterMessage>(
230 key: K,
231 desired: elasticache.ModifyCacheClusterMessage[K],
232 observed: unknown,
233 ) => {
234 if (desired !== undefined && desired !== observed) {
235 update[key] = desired;
236 changed = true;
237 }
238 };
239 set("NumCacheNodes", props.numCacheNodes, cluster.NumCacheNodes);
240 set("CacheNodeType", props.nodeType, cluster.CacheNodeType);
241 set("EngineVersion", props.engineVersion, cluster.EngineVersion);
242 set(
243 "CacheParameterGroupName",
244 props.parameterGroupName,
245 cluster.CacheParameterGroup?.CacheParameterGroupName,
246 );
247 set(
248 "PreferredMaintenanceWindow",
249 props.maintenanceWindow,
250 cluster.PreferredMaintenanceWindow,
251 );
252 set(
253 "AutoMinorVersionUpgrade",
254 props.autoMinorVersionUpgrade,
255 cluster.AutoMinorVersionUpgrade,
256 );
257 set("IpDiscovery", props.ipDiscovery, cluster.IpDiscovery);
258 if (props.securityGroupIds !== undefined) {
259 const observed = (cluster.SecurityGroups ?? [])
260 .map((group) => group.SecurityGroupId)
261 .filter((id): id is string => id !== undefined);
262 if (!sameStringSet(props.securityGroupIds, observed)) {
263 update.SecurityGroupIds = props.securityGroupIds;
264 changed = true;
265 }
266 }
267 if (changed) {
268 yield* elasticache.modifyCacheCluster(update);
269 cluster = yield* waitForAvailable(name);
270 }
271 if (cluster.ARN) {
272 const { removed, upsert } = diffTags(
273 yield* readElastiCacheTags(cluster.ARN),
274 desiredTags,
275 );
276 if (upsert.length)
277 yield* elasticache.addTagsToResource({
278 ResourceName: cluster.ARN,
279 Tags: upsert,
280 });
281 if (removed.length)
282 yield* elasticache.removeTagsFromResource({
283 ResourceName: cluster.ARN,
284 TagKeys: removed,
285 });
286 }
287 yield* session.note(name);
288 return yield* attrs(cluster);
289 }),
290 delete: Effect.fn(function* ({ output }) {
291 yield* elasticache
292 .deleteCacheCluster({ CacheClusterId: output.cacheClusterId })
293 .pipe(
294 Effect.catchTag("CacheClusterNotFoundFault", () => Effect.void),
295 Effect.retry({
296 while: (error) =>
297 error._tag === "InvalidCacheClusterStateFault",
298 schedule: Schedule.max([
299 Schedule.fixed("15 seconds"),
300 Schedule.recurs(12),
301 ]),
302 }),
303 );
304 yield* waitForDeleted(output.cacheClusterId);
305 }),
306 list: () =>
307 elasticache.describeCacheClusters
308 .pages({ ShowCacheNodeInfo: true })
309 .pipe(
310 Stream.runCollect,
311 Effect.map((pages) =>
312 Array.from(pages).flatMap((page) => page.CacheClusters ?? []),
313 ),
314 Effect.flatMap((clusters) =>
315 Effect.forEach(
316 clusters.filter((cluster) => cluster.ARN),
317 attrs,
318 ),
319 ),
320 ),
321 };
322 }),
323 );