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";
12
13/**
14 * Cache engine for a serverless cache.
15 */
16export type ServerlessCacheEngine = "valkey" | "redis" | "memcached";
17
18/**
19 * Usage limits that cap how much a serverless cache can consume. Setting
20 * low maximums is the primary cost-control lever — serverless caches bill
21 * for data stored (GB-hours) and compute (ECPUs) with a monthly floor.
22 */
23export interface ServerlessCacheUsageLimits {
24 /**
25 * Cached data storage limits in GB.
26 * The service-minimum for `maximum` is 1 GB.
27 */
28 dataStorage?: {
29 /** Maximum data storage in GB. */
30 maximum?: number;
31 /** Minimum (pre-provisioned) data storage in GB. */
32 minimum?: number;
33 };
34 /**
35 * ElastiCache Processing Unit (ECPU) rate limits.
36 * The service-minimum for `maximum` is 1000 ECPUs/second.
37 */
38 ecpuPerSecond?: {
39 /** Maximum ECPUs per second. */
40 maximum?: number;
41 /** Minimum (pre-provisioned) ECPUs per second. */
42 minimum?: number;
43 };
44}
45
46export interface ServerlessCacheProps {
47 /**
48 * Name of the serverless cache. Must be 1-40 alphanumeric characters or
49 * hyphens. If omitted, a deterministic physical name is generated.
50 * Changing the name replaces the cache.
51 */
52 serverlessCacheName?: string;
53 /**
54 * Cache engine.
55 * @default "valkey"
56 */
57 engine?: ServerlessCacheEngine;
58 /**
59 * Major engine version, e.g. `"8"` for valkey or `"7"` for redis.
60 * @default latest for the engine
61 */
62 majorEngineVersion?: string;
63 /**
64 * Human-readable description of the cache.
65 */
66 description?: string;
67 /**
68 * Usage limits capping storage (GB) and compute (ECPUs/second).
69 * Strongly recommended for cost control.
70 */
71 cacheUsageLimits?: ServerlessCacheUsageLimits;
72 /**
73 * Customer-managed KMS key for encryption at rest. Changing the key
74 * replaces the cache.
75 * @default AWS-owned key
76 */
77 kmsKeyId?: string;
78 /**
79 * VPC security groups that control network access to the cache endpoint.
80 * @default the VPC's default security group
81 */
82 securityGroupIds?: string[];
83 /**
84 * VPC subnets the cache is reachable from. Changing subnets replaces the
85 * cache.
86 * @default subnets of the account's default VPC
87 */
88 subnetIds?: string[];
89 /**
90 * User group for RBAC authentication (valkey/redis only).
91 */
92 userGroupId?: string;
93 /**
94 * Days to retain automatic daily snapshots. 0 disables automatic
95 * snapshots.
96 * @default 0
97 */
98 snapshotRetentionLimit?: number;
99 /**
100 * Daily time window (UTC, `HH:MM`) when automatic snapshots are taken.
101 */
102 dailySnapshotTime?: string;
103 /**
104 * IP discovery network type. Changing this replaces the cache.
105 * @default "ipv4"
106 */
107 networkType?: elasticache.NetworkType;
108 /**
109 * ARNs of snapshots to seed the cache from at creation (create-only).
110 */
111 snapshotArnsToRestore?: string[];
112 /**
113 * User-defined tags for the cache.
114 */
115 tags?: Record<string, string>;
116}
117
118export interface ServerlessCache extends Resource<
119 "AWS.ElastiCache.ServerlessCache",
120 ServerlessCacheProps,
121 {
122 /** The name of the serverless cache. */
123 serverlessCacheName: string;
124 /** The ARN of the serverless cache. */
125 serverlessCacheArn: string;
126 /** The cache status (e.g. `creating`, `available`, `modifying`). */
127 status: string;
128 /** The cache engine (`valkey`, `redis`, or `memcached`). */
129 engine: string;
130 /** The major engine version (e.g. `8`). */
131 majorEngineVersion: string | undefined;
132 /** The full engine version the cache is running. */
133 fullEngineVersion: string | undefined;
134 /** The DNS hostname of the primary endpoint. */
135 endpointAddress: string;
136 /** The port of the primary endpoint (6379 for valkey/redis, 11211 for memcached). */
137 endpointPort: number;
138 /** The DNS hostname of the reader endpoint, when the engine exposes one. */
139 readerEndpointAddress: string | undefined;
140 /** The port of the reader endpoint, when the engine exposes one. */
141 readerEndpointPort: number | undefined;
142 /** The IDs of the security groups associated with the cache. */
143 securityGroupIds: string[];
144 /** The IDs of the subnets the cache is deployed in. */
145 subnetIds: string[];
146 /** The tags applied to the cache. */
147 tags: Record<string, string>;
148 },
149 never,
150 Providers
151> {}
152
153/**
154 * An Amazon ElastiCache serverless cache (valkey, redis, or memcached).
155 *
156 * Serverless caches scale storage and compute automatically and are only
157 * reachable from inside a VPC. They are metered while they exist (with a
158 * monthly minimum), so set `cacheUsageLimits` and destroy caches you are
159 * not using.
160 * ### Creating a Serverless Cache
161 * **Example:** Valkey Cache with Cost-Control Limits
162 * ```typescript
163 * const cache = yield* ServerlessCache("SessionCache", {
164 * engine: "valkey",
165 * cacheUsageLimits: {
166 * dataStorage: { maximum: 1 },
167 * ecpuPerSecond: { maximum: 1000 },
168 * },
169 * });
170 * ```
171 *
172 * **Example:** Redis Cache in Specific Subnets
173 * ```typescript
174 * const cache = yield* ServerlessCache("Cache", {
175 * engine: "redis",
176 * majorEngineVersion: "7",
177 * subnetIds: [subnetA.subnetId, subnetB.subnetId],
178 * securityGroupIds: [cacheSecurityGroup.securityGroupId],
179 * });
180 * ```
181 *
182 * ### Connecting from a Lambda Function
183 * **Example:** Bind Connection Info into a Function
184 * ```typescript
185 * const connect = yield* ElastiCache.Connect(cache);
186 * // inside a handler:
187 * const { host, port, tls } = yield* connect;
188 * ```
189 *
190 * @resource
191 */
192export const ServerlessCache = Resource<ServerlessCache>(
193 "AWS.ElastiCache.ServerlessCache",
194);
195
196const toTagRecord = (
197 tags: Array<{ Key?: string; Value?: string }> | undefined,
198): Record<string, string> =>
199 Object.fromEntries(
200 (tags ?? [])
201 .filter(
202 (tag): tag is { Key: string; Value: string } =>
203 typeof tag.Key === "string" && typeof tag.Value === "string",
204 )
205 .map((tag) => [tag.Key, tag.Value]),
206 );
207
208const sameStringSet = (
209 a: readonly string[] | undefined,
210 b: readonly string[] | undefined,
211): boolean => {
212 const left = [...(a ?? [])].sort();
213 const right = [...(b ?? [])].sort();
214 return left.length === right.length && left.every((v, i) => v === right[i]);
215};
216
217/**
218 * Convert user-facing camelCase usage limits into the wire shape. Serverless
219 * data storage is always expressed in GB.
220 */
221const toWireUsageLimits = (
222 limits: ServerlessCacheUsageLimits | undefined,
223): elasticache.CacheUsageLimits | undefined =>
224 limits === undefined
225 ? undefined
226 : {
227 ...(limits.dataStorage
228 ? {
229 DataStorage: {
230 Maximum: limits.dataStorage.maximum,
231 Minimum: limits.dataStorage.minimum,
232 Unit: "GB",
233 },
234 }
235 : {}),
236 ...(limits.ecpuPerSecond
237 ? {
238 ECPUPerSecond: {
239 Maximum: limits.ecpuPerSecond.maximum,
240 Minimum: limits.ecpuPerSecond.minimum,
241 },
242 }
243 : {}),
244 };
245
246/**
247 * True when the desired usage limits differ from the observed ones for any
248 * field the user actually specified.
249 */
250const usageLimitsDiffer = (
251 desired: elasticache.CacheUsageLimits | undefined,
252 observed: elasticache.CacheUsageLimits | undefined,
253): boolean => {
254 if (desired === undefined) return false;
255 // Only compare fields the user actually specified — the service fills in
256 // defaults (units, minimums) that must not trigger spurious modify calls.
257 if (desired.DataStorage) {
258 if (
259 (desired.DataStorage.Maximum !== undefined &&
260 desired.DataStorage.Maximum !== observed?.DataStorage?.Maximum) ||
261 (desired.DataStorage.Minimum !== undefined &&
262 desired.DataStorage.Minimum !== observed?.DataStorage?.Minimum)
263 ) {
264 return true;
265 }
266 }
267 if (desired.ECPUPerSecond) {
268 if (
269 (desired.ECPUPerSecond.Maximum !== undefined &&
270 desired.ECPUPerSecond.Maximum !== observed?.ECPUPerSecond?.Maximum) ||
271 (desired.ECPUPerSecond.Minimum !== undefined &&
272 desired.ECPUPerSecond.Minimum !== observed?.ECPUPerSecond?.Minimum)
273 ) {
274 return true;
275 }
276 }
277 return false;
278};
279
280export const ServerlessCacheProvider = () =>
281 Provider.effect(
282 ServerlessCache,
283 Effect.gen(function* () {
284 const toName = (id: string, props: ServerlessCacheProps) =>
285 props.serverlessCacheName
286 ? Effect.succeed(props.serverlessCacheName)
287 : createPhysicalName({ id, maxLength: 40, lowercase: true });
288
289 const readCache = Effect.fn(function* (name: string) {
290 const response = yield* elasticache
291 .describeServerlessCaches({ ServerlessCacheName: name })
292 .pipe(
293 Effect.catchTag("ServerlessCacheNotFoundFault", () =>
294 Effect.succeed(undefined),
295 ),
296 );
297 return response?.ServerlessCaches?.[0];
298 });
299
300 const readTags = Effect.fn(function* (arn: string) {
301 const response = yield* elasticache
302 .listTagsForResource({ ResourceName: arn })
303 .pipe(Effect.catch(() => Effect.succeed(undefined)));
304 return toTagRecord(response?.TagList);
305 });
306
307 // Bounded readiness wait. Serverless cache provisioning/modification
308 // typically completes in 1-3 minutes; budget ~10 min (60 * 10s) like
309 // the RDS cluster wait so slow regions still converge.
310 const waitForCache = Effect.fn(function* (name: string) {
311 const readinessPolicy = Schedule.max([
312 Schedule.fixed("10 seconds"),
313 Schedule.recurs(60),
314 ]);
315 return yield* readCache(name).pipe(
316 Effect.flatMap((cache) => {
317 if (!cache?.ARN) {
318 return Effect.fail(
319 new Error(`Serverless cache '${name}' not found`),
320 );
321 }
322 if (cache.Status !== "available") {
323 return Effect.fail(
324 new Error(
325 `Serverless cache '${name}' not available (status: ${cache.Status})`,
326 ),
327 );
328 }
329 return Effect.succeed(cache);
330 }),
331 Effect.retry({ schedule: readinessPolicy }),
332 );
333 });
334
335 // Wait for a cache to leave a transitional state before delete. Ends
336 // when the cache is available, deleting, or gone.
337 const waitUntilSettled = Effect.fn(function* (name: string) {
338 const settlePolicy = Schedule.max([
339 Schedule.fixed("10 seconds"),
340 Schedule.recurs(60),
341 ]);
342 return yield* readCache(name).pipe(
343 Effect.flatMap((cache) => {
344 if (
345 cache !== undefined &&
346 cache.Status !== "available" &&
347 cache.Status !== "deleting"
348 ) {
349 return Effect.fail(
350 new Error(
351 `Serverless cache '${name}' still settling (status: ${cache.Status})`,
352 ),
353 );
354 }
355 return Effect.succeed(cache);
356 }),
357 Effect.retry({ schedule: settlePolicy }),
358 );
359 });
360
361 const toAttrs = Effect.fn(function* (cache: elasticache.ServerlessCache) {
362 if (
363 !cache.ServerlessCacheName ||
364 !cache.ARN ||
365 !cache.Endpoint?.Address ||
366 cache.Endpoint.Port === undefined
367 ) {
368 return yield* Effect.fail(
369 new Error(
370 `Serverless cache '${cache.ServerlessCacheName}' is missing its ARN or endpoint (status: ${cache.Status})`,
371 ),
372 );
373 }
374 return {
375 serverlessCacheName: cache.ServerlessCacheName,
376 serverlessCacheArn: cache.ARN,
377 status: cache.Status ?? "available",
378 engine: cache.Engine ?? "valkey",
379 majorEngineVersion: cache.MajorEngineVersion,
380 fullEngineVersion: cache.FullEngineVersion,
381 endpointAddress: cache.Endpoint.Address,
382 endpointPort: cache.Endpoint.Port,
383 readerEndpointAddress: cache.ReaderEndpoint?.Address,
384 readerEndpointPort: cache.ReaderEndpoint?.Port,
385 securityGroupIds: [...(cache.SecurityGroupIds ?? [])],
386 subnetIds: [...(cache.SubnetIds ?? [])],
387 tags: yield* readTags(cache.ARN),
388 };
389 });
390
391 return {
392 stables: ["serverlessCacheName", "serverlessCacheArn"],
393
394 diff: Effect.fn(function* ({ id, olds, news }) {
395 if (!isResolved(news)) return undefined;
396 if (
397 (yield* toName(id, olds ?? {})) !== (yield* toName(id, news ?? {}))
398 ) {
399 return { action: "replace" } as const;
400 }
401 // Create-only properties force a replacement.
402 if (
403 news?.subnetIds !== undefined &&
404 olds?.subnetIds !== undefined &&
405 !sameStringSet(news.subnetIds, olds.subnetIds)
406 ) {
407 return { action: "replace" } as const;
408 }
409 if ((news?.kmsKeyId ?? undefined) !== (olds?.kmsKeyId ?? undefined)) {
410 return { action: "replace" } as const;
411 }
412 if (
413 (news?.networkType ?? undefined) !==
414 (olds?.networkType ?? undefined)
415 ) {
416 return { action: "replace" } as const;
417 }
418 }),
419
420 read: Effect.fn(function* ({ id, olds, output }) {
421 const name =
422 output?.serverlessCacheName ?? (yield* toName(id, olds ?? {}));
423 const cache = yield* readCache(name);
424 // A cache that is still provisioning has no endpoint yet; report
425 // it as missing so reconcile (which tolerates the AlreadyExists
426 // race and waits for availability) converges it.
427 if (
428 !cache?.ARN ||
429 !cache.Endpoint?.Address ||
430 cache.Endpoint.Port === undefined
431 ) {
432 return undefined;
433 }
434 const attrs = yield* toAttrs(cache);
435 return (yield* hasAlchemyTags(id, attrs.tags))
436 ? attrs
437 : Unowned(attrs);
438 }),
439
440 reconcile: Effect.fn(function* ({ id, news = {}, output, session }) {
441 const name = output?.serverlessCacheName ?? (yield* toName(id, news));
442 const internalTags = yield* createInternalTags(id);
443 const desiredTags = { ...internalTags, ...news.tags };
444 const desiredLimits = toWireUsageLimits(news.cacheUsageLimits);
445
446 // 1. Observe — cloud state is authoritative.
447 let observed = yield* readCache(name);
448
449 // 2. Ensure — create if missing; tolerate AlreadyExists as a race.
450 if (observed === undefined) {
451 yield* elasticache
452 .createServerlessCache({
453 ServerlessCacheName: name,
454 Engine: news.engine ?? "valkey",
455 MajorEngineVersion: news.majorEngineVersion,
456 Description: news.description,
457 CacheUsageLimits: desiredLimits,
458 KmsKeyId: news.kmsKeyId,
459 SecurityGroupIds: news.securityGroupIds,
460 SubnetIds: news.subnetIds,
461 UserGroupId: news.userGroupId,
462 SnapshotRetentionLimit: news.snapshotRetentionLimit,
463 DailySnapshotTime: news.dailySnapshotTime,
464 NetworkType: news.networkType,
465 SnapshotArnsToRestore: news.snapshotArnsToRestore,
466 Tags: Object.entries(desiredTags).map(([Key, Value]) => ({
467 Key,
468 Value,
469 })),
470 })
471 .pipe(
472 Effect.catchTag(
473 "ServerlessCacheAlreadyExistsFault",
474 () => Effect.void,
475 ),
476 );
477 }
478
479 // Provisioning and in-flight modifications both surface as a
480 // non-available status; wait for the cache to become available
481 // (bounded) so sync/modify calls do not hit
482 // InvalidServerlessCacheStateFault.
483 observed = yield* waitForCache(name);
484
485 // 3. Sync — compute the modify delta from OBSERVED state.
486 const modify: elasticache.ModifyServerlessCacheRequest = {};
487 if (
488 news.description !== undefined &&
489 news.description !== observed.Description
490 ) {
491 modify.Description = news.description;
492 }
493 if (usageLimitsDiffer(desiredLimits, observed.CacheUsageLimits)) {
494 modify.CacheUsageLimits = desiredLimits;
495 }
496 if (
497 news.securityGroupIds !== undefined &&
498 !sameStringSet(news.securityGroupIds, observed.SecurityGroupIds)
499 ) {
500 modify.SecurityGroupIds = news.securityGroupIds;
501 }
502 if (
503 news.userGroupId !== undefined &&
504 news.userGroupId !== observed.UserGroupId
505 ) {
506 modify.UserGroupId = news.userGroupId;
507 } else if (
508 news.userGroupId === undefined &&
509 observed.UserGroupId !== undefined
510 ) {
511 modify.RemoveUserGroup = true;
512 }
513 if (
514 news.snapshotRetentionLimit !== undefined &&
515 news.snapshotRetentionLimit !==
516 (observed.SnapshotRetentionLimit ?? 0)
517 ) {
518 modify.SnapshotRetentionLimit = news.snapshotRetentionLimit;
519 }
520 if (
521 news.dailySnapshotTime !== undefined &&
522 news.dailySnapshotTime !== observed.DailySnapshotTime
523 ) {
524 modify.DailySnapshotTime = news.dailySnapshotTime;
525 }
526 if (news.engine !== undefined && news.engine !== observed.Engine) {
527 // valkey <-> redis upgrades are supported in-place by the API.
528 modify.Engine = news.engine;
529 if (news.majorEngineVersion !== undefined) {
530 modify.MajorEngineVersion = news.majorEngineVersion;
531 }
532 } else if (
533 news.majorEngineVersion !== undefined &&
534 news.majorEngineVersion !== observed.MajorEngineVersion
535 ) {
536 modify.MajorEngineVersion = news.majorEngineVersion;
537 }
538
539 if (Object.keys(modify).length > 0) {
540 yield* elasticache.modifyServerlessCache({
541 ServerlessCacheName: name,
542 ...modify,
543 });
544 observed = yield* waitForCache(name);
545 }
546
547 // 3b. Sync tags — diff against OBSERVED cloud tags.
548 const arn = observed.ARN;
549 if (arn) {
550 const observedTags = yield* readTags(arn);
551 const { removed, upsert } = diffTags(observedTags, desiredTags);
552 if (upsert.length > 0) {
553 yield* elasticache.addTagsToResource({
554 ResourceName: arn,
555 Tags: upsert,
556 });
557 }
558 if (removed.length > 0) {
559 yield* elasticache.removeTagsFromResource({
560 ResourceName: arn,
561 TagKeys: removed,
562 });
563 }
564 }
565
566 // 4. Return fresh attributes.
567 yield* session.note(name);
568 return yield* toAttrs(observed);
569 }),
570
571 delete: Effect.fn(function* ({ output }) {
572 const name = output.serverlessCacheName;
573 // A cache mid-create/modify rejects deletion with
574 // InvalidServerlessCacheStateFault — wait (bounded) for it to
575 // settle first. A cache already deleting (or gone) is success.
576 yield* waitUntilSettled(name);
577 yield* elasticache
578 .deleteServerlessCache({ ServerlessCacheName: name })
579 .pipe(
580 Effect.catchTag(
581 "ServerlessCacheNotFoundFault",
582 () => Effect.void,
583 ),
584 Effect.catchTag(
585 "InvalidServerlessCacheStateFault",
586 () =>
587 // Already deleting — deletion is in progress.
588 Effect.void,
589 ),
590 );
591 }),
592
593 list: () =>
594 elasticache.describeServerlessCaches.pages({}).pipe(
595 Stream.runCollect,
596 Effect.map((chunk) =>
597 Array.from(chunk).flatMap((page) =>
598 (page.ServerlessCaches ?? []).filter(
599 (cache) =>
600 cache.ServerlessCacheName !== undefined &&
601 cache.ARN !== undefined &&
602 cache.Endpoint?.Address !== undefined &&
603 cache.Endpoint.Port !== undefined,
604 ),
605 ),
606 ),
607 Effect.flatMap(
608 Effect.forEach((cache) => toAttrs(cache), { concurrency: 4 }),
609 ),
610 ),
611 };
612 }),
613 );