1import * as AWS from "@/AWS";
2import type { CacheConnectionInfo } from "@/AWS/ElastiCache";
3import * as EC2 from "@distilled.cloud/aws/ec2";
4import * as Context from "effect/Context";
5import * as Duration from "effect/Duration";
6import * as Effect from "effect/Effect";
7import * as Layer from "effect/Layer";
8import { HttpServerRequest } from "effect/http/HttpServerRequest";
9import * as HttpServerResponse from "effect/http/HttpServerResponse";
10import Valkey from "iovalkey";
11import { getDefaultVpc } from "../DefaultVpc.ts";
12
13export class ElastiCacheTestFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
14 "ElastiCacheTestFunction",
15) {}
16
17export class FixtureCache extends Context.Service<
18 FixtureCache,
19 {
20 cache: AWS.ElastiCache.ServerlessCache;
21 }
22>()("FixtureCache") {}
23
24export const FixtureCacheLive = Layer.effect(
25 FixtureCache,
26 Effect.gen(function* () {
27 // Valkey is the cheapest engine; usage limits are pinned to the service
28 // minimums (1 GB storage, 1000 ECPUs/s) purely for cost control. Subnets
29 // and security group are left to the API defaults: the account's default
30 // VPC and its DEFAULT security group — the same network the fixture
31 // Lambda attaches to below.
32 const cache = yield* AWS.ElastiCache.ServerlessCache("FixtureCache", {
33 engine: "valkey",
34 description: "alchemy elasticache fixture",
35 cacheUsageLimits: {
36 dataStorage: { maximum: 1 },
37 ecpuPerSecond: { maximum: 1000 },
38 },
39 tags: { fixture: "elasticache-serverless" },
40 });
41 return { cache };
42 }),
43);
44
45/**
46 * Resolve the default VPC's default-for-AZ subnets and its DEFAULT security
47 * group for the fixture Lambda's VPC attachment.
48 *
49 * The default group is used on purpose (same rationale as the EFS Lambda
50 * mount test): its self-referencing ingress rule allows the Lambda ENI to
51 * reach the cache endpoint on 6379 (the serverless cache sits in the same
52 * default VPC + default SG by API default), and because the default group is
53 * never deleted, destroy avoids the ~20-minute Hyperplane ENI release that
54 * blocks deleting a security group that was attached to a Lambda.
55 *
56 * Runtime-guarded: the Lambda runtime re-executes this props effect on cold
57 * start, but VPC attachment is deploy-time-only configuration and the
58 * execution role has no ec2:Describe* permissions — return `undefined` there
59 * without touching the EC2 API.
60 */
61const resolveFixtureVpc = Effect.gen(function* () {
62 if (globalThis.__ALCHEMY_RUNTIME__) {
63 return undefined;
64 }
65 const vpc = yield* getDefaultVpc;
66 const subnets = yield* EC2.describeSubnets({
67 Filters: [
68 { Name: "vpc-id", Values: [vpc.vpcId] },
69 { Name: "default-for-az", Values: ["true"] },
70 ],
71 });
72 const subnetIds = (subnets.Subnets ?? [])
73 .map((subnet) => subnet.SubnetId)
74 .filter((subnetId): subnetId is string => subnetId !== undefined)
75 .sort()
76 .slice(0, 2);
77 const groups = yield* EC2.describeSecurityGroups({
78 Filters: [
79 { Name: "vpc-id", Values: [vpc.vpcId] },
80 { Name: "group-name", Values: ["default"] },
81 ],
82 });
83 const securityGroupId = groups.SecurityGroups?.[0]?.GroupId;
84 if (subnetIds.length === 0 || securityGroupId === undefined) {
85 return yield* Effect.die(
86 new Error(
87 "default VPC is missing default-for-AZ subnets or its default security group",
88 ),
89 );
90 }
91 return { subnetIds, securityGroupIds: [securityGroupId] };
92}).pipe(Effect.orDie);
93
94/**
95 * SET then GET a key through the serverless cache with iovalkey. Serverless
96 * caches only accept TLS connections, so `tls` is always enabled from the
97 * binding's connection info. One client per request: connect, round-trip,
98 * disconnect — sockets must not be retained across invocations.
99 */
100const cacheRoundtrip = (info: CacheConnectionInfo, value: string) =>
101 Effect.tryPromise({
102 try: async () => {
103 const client = new Valkey({
104 host: info.host,
105 port: info.port,
106 ...(info.tls ? { tls: {} } : {}),
107 connectTimeout: 10_000,
108 maxRetriesPerRequest: 2,
109 // Fail fast instead of reconnecting forever — the route must answer
110 // (or error) within the Lambda's 30s timeout.
111 retryStrategy: (times) => (times > 2 ? null : 500),
112 });
113 try {
114 await client.set("alchemy:elasticache:roundtrip", value);
115 return await client.get("alchemy:elasticache:roundtrip");
116 } finally {
117 client.disconnect();
118 }
119 },
120 catch: (cause) => new Error(`valkey roundtrip failed: ${String(cause)}`),
121 });
122
123/**
124 * VPC-attached fixture Lambda exercising the full ElastiCache data plane:
125 *
126 * - `/connection` returns the endpoint the env-only Connect binding resolved
127 * from the injected environment.
128 * - `/roundtrip?value=...` SETs then GETs a key through the cache over TLS
129 * with iovalkey and returns the read-back value.
130 * - `/snapshot?name=...` takes an on-demand snapshot of the fixture cache
131 * through the cache-scoped CreateServerlessCacheSnapshot binding.
132 */
133export const ElastiCacheTestFunctionLive = ElastiCacheTestFunction.make(
134 Effect.gen(function* () {
135 const vpc = yield* resolveFixtureVpc;
136 return {
137 main: import.meta.url,
138 functionUrl: true,
139 timeout: Duration.seconds(30),
140 memorySize: 256,
141 ...(vpc === undefined ? {} : { vpc }),
142 };
143 }),
144 Effect.gen(function* () {
145 const { cache } = yield* FixtureCache;
146 const connection = yield* AWS.ElastiCache.Connect(cache);
147 const createSnapshot =
148 yield* AWS.ElastiCache.CreateServerlessCacheSnapshot(cache);
149
150 return {
151 fetch: Effect.gen(function* () {
152 const request = yield* HttpServerRequest;
153 const url = new URL(request.originalUrl);
154 const pathname = url.pathname;
155
156 if (request.method === "GET" && pathname === "/ready") {
157 return yield* HttpServerResponse.json({ ok: true });
158 }
159
160 if (request.method === "GET" && pathname === "/connection") {
161 return yield* HttpServerResponse.json(yield* connection);
162 }
163
164 if (request.method === "GET" && pathname === "/snapshot") {
165 const name = url.searchParams.get("name") ?? "alchemy-fixture-snap";
166 return yield* createSnapshot({
167 ServerlessCacheSnapshotName: name,
168 }).pipe(
169 Effect.flatMap((result) =>
170 HttpServerResponse.json({
171 name: result.ServerlessCacheSnapshot
172 ?.ServerlessCacheSnapshotName,
173 status: result.ServerlessCacheSnapshot?.Status,
174 }),
175 ),
176 // Re-run after a crashed test: the snapshot already exists.
177 Effect.catchTag("ServerlessCacheSnapshotAlreadyExistsFault", () =>
178 HttpServerResponse.json({ name, status: "exists" }),
179 ),
180 Effect.catch((error) =>
181 HttpServerResponse.json({ error: error._tag }, { status: 500 }),
182 ),
183 );
184 }
185
186 if (request.method === "GET" && pathname === "/roundtrip") {
187 const info = yield* connection;
188 const value = url.searchParams.get("value") ?? "hello-valkey";
189 return yield* cacheRoundtrip(info, value).pipe(
190 Effect.flatMap((read) => HttpServerResponse.json({ value: read })),
191 Effect.catch((error) =>
192 HttpServerResponse.json(
193 { error: String(error) },
194 { status: 500 },
195 ),
196 ),
197 );
198 }
199
200 return yield* HttpServerResponse.json(
201 { error: "Not found", method: request.method, pathname },
202 { status: 404 },
203 );
204 }).pipe(Effect.orDie),
205 };
206 }).pipe(
207 Effect.provide(
208 Layer.mergeAll(
209 AWS.ElastiCache.ConnectHttp,
210 AWS.ElastiCache.CreateServerlessCacheSnapshotHttp,
211 FixtureCacheLive,
212 ),
213 ),
214 ),
215 // Re-merge so the deploying Stack can `yield* FixtureCache` and expose the
216 // cache attributes as deploy-time outputs. Reusing the same
217 // `FixtureCacheLive` reference keeps it a single shared cache.
218).pipe(Layer.provideMerge(FixtureCacheLive));
219
220export default ElastiCacheTestFunctionLive;