1import {
2 CacheCluster,
3 cacheClusterConnectEnvPrefix,
4 ConnectCacheCluster,
5 ConnectCacheClusterHttp,
6 ConnectReplicationGroup,
7 ConnectReplicationGroupHttp,
8 ReplicationGroup,
9 replicationGroupConnectEnvPrefix,
10 validateReplicationGroupProps,
11} from "@/AWS/ElastiCache";
12import * as AWS from "@/AWS";
13import { sameStringSet } from "@/AWS/ElastiCache/internal.ts";
14import * as Test from "@/Test/Alchemy";
15import * as ElastiCache from "@distilled.cloud/aws/elasticache";
16import { expect } from "alchemy-test";
17import * as Effect from "effect/Effect";
18import * as Output from "@/Output.ts";
19import { RuntimeContext } from "@/RuntimeContext.ts";
20import * as Schedule from "effect/Schedule";
21import {
22 assertCacheClusterGone,
23 assertReplicationGroupGone,
24 getProvisionedNetwork,
25 shareProvisionedNetwork,
26} from "./ProvisionedFixture.ts";
27
28const { test, beforeAll, afterAll } = Test.make({
29 providers: AWS.providers(),
30});
31shareProvisionedNetwork({ beforeAll, afterAll });
32test(
33 "connection names and unordered subnet IDs are stable",
34 Effect.sync(() => {
35 expect(replicationGroupConnectEnvPrefix("SessionCache")).toBe(
36 "ELASTICACHE_SESSIONCACHE",
37 );
38 expect(cacheClusterConnectEnvPrefix("SessionCache")).toBe(
39 "ELASTICACHE_SESSIONCACHE",
40 );
41 expect(
42 sameStringSet(["subnet-a", "subnet-b"], ["subnet-b", "subnet-a"]),
43 ).toBe(true);
44 expect(sameStringSet(["subnet-a"], ["subnet-b"])).toBe(false);
45 }),
46 {
47 tags: ["provider:aws", "provider:aws:elasticache", "live"],
48 timeout: 2_700_000,
49 },
50);
51
52test(
53 "invalid replication-group configuration fails before an AWS create request",
54 Effect.sync(() => {
55 const invalid = [
56 { automaticFailoverEnabled: true, replicasPerNodeGroup: 0 },
57 {
58 multiAzEnabled: true,
59 automaticFailoverEnabled: false,
60 replicasPerNodeGroup: 1,
61 },
62 { replicasPerNodeGroup: 6 },
63 { numNodeGroups: 0 },
64 { port: 0 },
65 {
66 transitEncryptionEnabled: false,
67 transitEncryptionMode: "preferred" as const,
68 },
69 ];
70 for (const props of invalid) {
71 const error = validateReplicationGroupProps({
72 description: "invalid cache",
73 engine: "valkey",
74 ...props,
75 });
76 expect(error?._tag).toBe("InvalidReplicationGroupConfiguration");
77 }
78 }),
79 {
80 tags: ["provider:aws", "provider:aws:elasticache", "live"],
81 timeout: 2_700_000,
82 },
83);
84
85const withEnv = <A, E = never, R = never>(
86 values: Record<string, string>,
87 effect: Effect.Effect<A, E, R>,
88) => {
89 const previous = Object.fromEntries(
90 Object.keys(values).map((key) => [key, process.env[key]]),
91 );
92 return Effect.gen(function* () {
93 yield* Effect.sync(() => Object.assign(process.env, values));
94 return yield* effect;
95 }).pipe(
96 Effect.ensuring(
97 Effect.sync(() => {
98 for (const [key, value] of Object.entries(previous)) {
99 if (value === undefined) delete process.env[key];
100 else process.env[key] = value;
101 }
102 }),
103 ),
104 );
105};
106
107test(
108 "provisioned cache bindings read the endpoints published to a Function",
109 Effect.gen(function* () {
110 const replicationGroup = {
111 LogicalId: "ProvisionedBinding",
112 configurationEndpointAddress: Output.asOutput(undefined),
113 configurationEndpointPort: Output.asOutput(undefined),
114 primaryEndpointAddress: Output.asOutput("primary.example"),
115 primaryEndpointPort: Output.asOutput(6379),
116 readerEndpointAddress: Output.asOutput("reader.example"),
117 readerEndpointPort: Output.asOutput(6379),
118 transitEncryptionEnabled: Output.asOutput(true),
119 } as unknown as ReplicationGroup;
120 const memcached = {
121 LogicalId: "MemcachedBinding",
122 endpoints: Output.asOutput([{ address: "node.example", port: 11211 }]),
123 transitEncryptionEnabled: Output.asOutput(false),
124 } as unknown as CacheCluster;
125 const replicationPrefix = replicationGroupConnectEnvPrefix(
126 replicationGroup.LogicalId,
127 );
128 const memcachedPrefix = cacheClusterConnectEnvPrefix(memcached.LogicalId);
129 const replicationConnect = yield* ConnectReplicationGroup(
130 replicationGroup,
131 ).pipe(Effect.provide(ConnectReplicationGroupHttp));
132 const memcachedConnect = yield* ConnectCacheCluster(memcached).pipe(
133 Effect.provide(ConnectCacheClusterHttp),
134 );
135 const [replication, cluster] = yield* withEnv(
136 {
137 [`${replicationPrefix}_HOST`]: "primary.example",
138 [`${replicationPrefix}_PORT`]: "6379",
139 [`${replicationPrefix}_TLS`]: "true",
140 [`${replicationPrefix}_READER_HOST`]: "reader.example",
141 [`${replicationPrefix}_READER_PORT`]: "6379",
142 [`${memcachedPrefix}_ENDPOINTS`]: JSON.stringify([
143 { address: "node.example", port: 11211 },
144 ]),
145 [`${memcachedPrefix}_TLS`]: "false",
146 },
147 Effect.all([replicationConnect, memcachedConnect]).pipe(
148 Effect.provide(RuntimeContext.phantom),
149 ),
150 );
151
152 expect(replication).toEqual({
153 host: "primary.example",
154 port: 6379,
155 readerHost: "reader.example",
156 readerPort: 6379,
157 tls: true,
158 });
159 expect(cluster).toEqual({
160 endpoints: [{ address: "node.example", port: 11211 }],
161 tls: false,
162 });
163 }),
164 {
165 tags: ["provider:aws", "provider:aws:elasticache", "live"],
166 timeout: 2_700_000,
167 },
168);
169
170const assertSubnetGroupGone = (name: string) =>
171 ElastiCache.describeCacheSubnetGroups({ CacheSubnetGroupName: name }).pipe(
172 Effect.flatMap(() =>
173 Effect.fail(new Error(`subnet group '${name}' still exists`)),
174 ),
175 Effect.catchTag("CacheSubnetGroupNotFoundFault", () => Effect.void),
176 Effect.retry({
177 schedule: Schedule.max([
178 Schedule.fixed("2 seconds"),
179 Schedule.recurs(10),
180 ]),
181 }),
182 );
183
184const finalSnapshotName = "alchemy-elasticache-final-snapshot";
185
186const deleteFinalSnapshot = () =>
187 ElastiCache.deleteSnapshot({ SnapshotName: finalSnapshotName }).pipe(
188 Effect.catchTag("SnapshotNotFoundFault", () => Effect.void),
189 Effect.retry({
190 while: (error) => error._tag === "InvalidSnapshotStateFault",
191 schedule: Schedule.max([
192 Schedule.fixed("15 seconds"),
193 Schedule.recurs(12),
194 ]),
195 }),
196 );
197
198const assertFinalSnapshotAvailable = () =>
199 ElastiCache.describeSnapshots({ SnapshotName: finalSnapshotName }).pipe(
200 Effect.flatMap((result) =>
201 result.Snapshots?.[0]?.SnapshotStatus === "available"
202 ? Effect.void
203 : Effect.fail(
204 new Error(`snapshot '${finalSnapshotName}' is not ready`),
205 ),
206 ),
207 Effect.retry({
208 schedule: Schedule.max([
209 Schedule.fixed("30 seconds"),
210 Schedule.recurs(30),
211 ]),
212 }),
213 );
214
215test.provider(
216 "create, update, verify, and destroy an ElastiCache subnet group",
217 (stack) =>
218 Effect.gen(function* () {
219 yield* stack.destroy();
220 const net = yield* getProvisionedNetwork;
221 const deploy = (description: string) =>
222 stack.deploy(
223 Effect.gen(function* () {
224 const group = yield* AWS.ElastiCache.SubnetGroup("Subnets", {
225 description,
226 subnetIds: net.subnetIds,
227 tags: { fixture: "elasticache-provisioned" },
228 });
229 return { group, subnetIds: net.subnetIds };
230 }),
231 );
232
233 const { group, subnetIds } = yield* deploy("alchemy cache subnets");
234 expect(group.subnetGroupArn).toContain(":subnetgroup:");
235 expect(new Set(group.subnetIds)).toEqual(new Set(subnetIds));
236
237 const described = yield* ElastiCache.describeCacheSubnetGroups({
238 CacheSubnetGroupName: group.subnetGroupName,
239 });
240 expect(
241 described.CacheSubnetGroups?.[0]?.CacheSubnetGroupDescription,
242 ).toBe("alchemy cache subnets");
243
244 const { group: updated } = yield* deploy("alchemy cache subnets v2");
245 expect(updated.subnetGroupName).toBe(group.subnetGroupName);
246 expect(updated.description).toBe("alchemy cache subnets v2");
247
248 yield* stack.destroy();
249 yield* assertSubnetGroupGone(group.subnetGroupName);
250 }),
251 {
252 tags: [
253 "provider:aws",
254 "provider:aws:ec2",
255 "provider:aws:elasticache",
256 "live",
257 ],
258 timeout: 180_000,
259 },
260);
261
262test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
263 "create, update, verify, and destroy a provisioned Valkey replication group",
264 (stack) =>
265 Effect.gen(function* () {
266 yield* stack.destroy();
267 const net = yield* getProvisionedNetwork;
268 const deploy = (description: string) =>
269 stack.deploy(
270 Effect.gen(function* () {
271 const cache = yield* ReplicationGroup("Cache", {
272 description,
273 engine: "valkey",
274 nodeType: "cache.t4g.micro",
275 subnetGroupName: net.subnetGroupName,
276 securityGroupIds: [net.securityGroupId],
277 replicasPerNodeGroup: 0,
278 transitEncryptionEnabled: true,
279 tags: { fixture: "elasticache-provisioned" },
280 });
281 return { cache };
282 }),
283 );
284 const { cache } = yield* deploy("alchemy provisioned cache");
285 expect(cache.status).toBe("available");
286 expect(cache.replicationGroupArn).toContain(":replicationgroup:");
287 expect(cache.primaryEndpointAddress).toBeDefined();
288 const observed = yield* ElastiCache.describeReplicationGroups({
289 ReplicationGroupId: cache.replicationGroupId,
290 });
291 expect(observed.ReplicationGroups?.[0]?.Status).toBe("available");
292
293 const { cache: updated } = yield* deploy("alchemy provisioned cache v2");
294 expect(updated.replicationGroupId).toBe(cache.replicationGroupId);
295 expect(updated.status).toBe("available");
296
297 yield* stack.destroy();
298 yield* assertReplicationGroupGone(cache.replicationGroupId);
299 }),
300 {
301 tags: [
302 "provider:aws",
303 "provider:aws:ec2",
304 "provider:aws:elasticache",
305 "live",
306 ],
307 timeout: 2_700_000,
308 },
309);
310
311test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
312 "create, scale, verify, and destroy a provisioned Memcached cluster",
313 (stack) =>
314 Effect.gen(function* () {
315 yield* stack.destroy();
316 const net = yield* getProvisionedNetwork;
317 const deploy = (numCacheNodes: number) =>
318 stack.deploy(
319 Effect.gen(function* () {
320 const cache = yield* CacheCluster("Cache", {
321 subnetGroupName: net.subnetGroupName,
322 securityGroupIds: [net.securityGroupId],
323 numCacheNodes,
324 tags: { fixture: "elasticache-memcached" },
325 });
326 return { cache };
327 }),
328 );
329
330 const { cache } = yield* deploy(1);
331 expect(cache.status).toBe("available");
332 expect(cache.endpoints).toHaveLength(1);
333
334 const { cache: scaled } = yield* deploy(2);
335 expect(scaled.cacheClusterId).toBe(cache.cacheClusterId);
336 expect(scaled.endpoints).toHaveLength(2);
337
338 const observed = yield* ElastiCache.describeCacheClusters({
339 CacheClusterId: scaled.cacheClusterId,
340 ShowCacheNodeInfo: true,
341 });
342 expect(observed.CacheClusters?.[0]?.NumCacheNodes).toBe(2);
343
344 yield* stack.destroy();
345 yield* assertCacheClusterGone(cache.cacheClusterId);
346 }),
347 {
348 tags: [
349 "provider:aws",
350 "provider:aws:ec2",
351 "provider:aws:elasticache",
352 "live",
353 ],
354 timeout: 2_700_000,
355 },
356);
357
358test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
359 "replace a replication group when its port changes",
360 (stack) =>
361 Effect.gen(function* () {
362 yield* stack.destroy();
363 const net = yield* getProvisionedNetwork;
364 const deploy = (port: number) =>
365 stack.deploy(
366 Effect.gen(function* () {
367 const cache = yield* ReplicationGroup("Cache", {
368 description: "alchemy replacement cache",
369 engine: "valkey",
370 nodeType: "cache.t4g.micro",
371 subnetGroupName: net.subnetGroupName,
372 securityGroupIds: [net.securityGroupId],
373 replicasPerNodeGroup: 0,
374 transitEncryptionEnabled: true,
375 port,
376 });
377 return { cache };
378 }),
379 );
380
381 const { cache } = yield* deploy(6379);
382 const { cache: replaced } = yield* deploy(6380);
383 expect(replaced.replicationGroupId).not.toBe(cache.replicationGroupId);
384 expect(replaced.primaryEndpointPort).toBe(6380);
385 yield* assertReplicationGroupGone(cache.replicationGroupId);
386
387 yield* stack.destroy();
388 yield* assertReplicationGroupGone(replaced.replicationGroupId);
389 }),
390 {
391 tags: [
392 "provider:aws",
393 "provider:aws:ec2",
394 "provider:aws:elasticache",
395 "live",
396 ],
397 timeout: 2_700_000,
398 },
399);
400
401test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
402 "add and remove a Valkey replica",
403 (stack) =>
404 Effect.gen(function* () {
405 yield* stack.destroy();
406 const net = yield* getProvisionedNetwork;
407 const deploy = (replicasPerNodeGroup: number) =>
408 stack.deploy(
409 Effect.gen(function* () {
410 const cache = yield* ReplicationGroup("Cache", {
411 description: "alchemy scaling cache",
412 engine: "valkey",
413 nodeType: "cache.t4g.micro",
414 subnetGroupName: net.subnetGroupName,
415 securityGroupIds: [net.securityGroupId],
416 replicasPerNodeGroup,
417 automaticFailoverEnabled: replicasPerNodeGroup > 0,
418 transitEncryptionEnabled: true,
419 });
420 return { cache };
421 }),
422 );
423
424 const { cache } = yield* deploy(0);
425 const { cache: scaled } = yield* deploy(1);
426 expect(scaled.replicationGroupId).toBe(cache.replicationGroupId);
427
428 const withReplica = yield* ElastiCache.describeReplicationGroups({
429 ReplicationGroupId: scaled.replicationGroupId,
430 });
431 expect(
432 withReplica.ReplicationGroups?.[0]?.NodeGroups?.[0]?.NodeGroupMembers,
433 ).toHaveLength(2);
434
435 const { cache: reduced } = yield* deploy(0);
436 const withoutReplica = yield* ElastiCache.describeReplicationGroups({
437 ReplicationGroupId: reduced.replicationGroupId,
438 });
439 expect(
440 withoutReplica.ReplicationGroups?.[0]?.NodeGroups?.[0]
441 ?.NodeGroupMembers,
442 ).toHaveLength(1);
443
444 yield* stack.destroy();
445 yield* assertReplicationGroupGone(reduced.replicationGroupId);
446 }),
447 {
448 tags: [
449 "provider:aws",
450 "provider:aws:ec2",
451 "provider:aws:elasticache",
452 "live",
453 ],
454 timeout: 2_700_000,
455 },
456);
457
458test.provider.skipIf(!process.env.AWS_TEST_SLOW)(
459 "create and clean up the requested final snapshot",
460 (stack) =>
461 Effect.gen(function* () {
462 yield* stack.destroy();
463 yield* deleteFinalSnapshot();
464 const net = yield* getProvisionedNetwork;
465 const { cache } = yield* stack.deploy(
466 Effect.gen(function* () {
467 const cache = yield* ReplicationGroup("Cache", {
468 description: "alchemy final snapshot cache",
469 engine: "valkey",
470 nodeType: "cache.t4g.micro",
471 subnetGroupName: net.subnetGroupName,
472 securityGroupIds: [net.securityGroupId],
473 replicasPerNodeGroup: 0,
474 transitEncryptionEnabled: true,
475 finalSnapshotName,
476 });
477 return { cache };
478 }),
479 );
480
481 yield* stack.destroy();
482 yield* assertReplicationGroupGone(cache.replicationGroupId);
483 yield* assertFinalSnapshotAvailable();
484 yield* deleteFinalSnapshot();
485 }).pipe(Effect.ensuring(deleteFinalSnapshot().pipe(Effect.ignore))),
486 {
487 tags: [
488 "provider:aws",
489 "provider:aws:ec2",
490 "provider:aws:elasticache",
491 "live",
492 ],
493 timeout: 2_700_000,
494 },
495);