1import * as AWS from "@/AWS";
2import * as Cause from "effect/Cause";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import { HttpServerRequest } from "effect/http/HttpServerRequest";
7import * as HttpServerResponse from "effect/http/HttpServerResponse";
8import Valkey from "iovalkey";
9import * as net from "node:net";
10import { getProvisionedNetwork } from "./ProvisionedFixture.ts";
11
12export class ProvisionedCacheDataPlaneFunction extends AWS.Lambda.Function<AWS.Lambda.Function>()(
13 "ProvisionedCacheDataPlaneFunction",
14) {}
15
16const valkeyRoundtrip = (
17 info: AWS.ElastiCache.ReplicationGroupConnectionInfo,
18 value: string,
19) =>
20 Effect.tryPromise({
21 try: async () => {
22 const client = new Valkey({
23 host: info.host,
24 port: info.port,
25 ...(info.tls ? { tls: {} } : {}),
26 connectTimeout: 10_000,
27 maxRetriesPerRequest: 2,
28 retryStrategy: (attempt) => (attempt > 2 ? null : 500),
29 });
30 try {
31 await client.set("alchemy:provisioned:valkey", value);
32 return await client.get("alchemy:provisioned:valkey");
33 } finally {
34 client.disconnect();
35 }
36 },
37 catch: (cause) => new Error(`Valkey roundtrip failed: ${String(cause)}`),
38 });
39
40const memcachedRoundtrip = (
41 endpoint: { address: string; port: number },
42 value: string,
43) =>
44 Effect.tryPromise({
45 try: () =>
46 new Promise<string>((resolve, reject) => {
47 const socket = net.createConnection(endpoint.port, endpoint.address);
48 let response = "";
49 const key = "alchemy:provisioned:memcached";
50 const payload = `set ${key} 0 30 ${Buffer.byteLength(value)}\r\n${value}\r\nget ${key}\r\n`;
51 const fail = (error: Error) => {
52 socket.destroy();
53 reject(error);
54 };
55 socket.setTimeout(10_000, () => fail(new Error("Memcached timed out")));
56 socket.once("error", fail);
57 socket.on("data", (chunk) => {
58 response += chunk.toString();
59 if (!response.endsWith("END\r\n")) return;
60 socket.end();
61 const match = response.match(/VALUE [^\r]+\r\n([^\r]+)\r\nEND\r\n$/);
62 if (match?.[1] === undefined) {
63 reject(new Error(`Unexpected Memcached response: ${response}`));
64 return;
65 }
66 resolve(match[1]);
67 });
68 socket.once("connect", () => socket.write(payload));
69 }),
70 catch: (cause) => new Error(`Memcached roundtrip failed: ${String(cause)}`),
71 });
72
73const ProvisionedCacheDataPlaneLive = ProvisionedCacheDataPlaneFunction.make(
74 Effect.succeed({
75 main: import.meta.url,
76 functionUrl: true,
77 timeout: Duration.seconds(30),
78 memorySize: 256,
79 }),
80 Effect.gen(function* () {
81 const net = yield* getProvisionedNetwork;
82 const lambdaSecurityGroup = yield* AWS.EC2.SecurityGroup(
83 "LambdaSecurityGroup",
84 {
85 vpcId: net.vpcId,
86 description: "Provisioned ElastiCache data-plane test Lambda",
87 },
88 );
89 yield* AWS.EC2.SecurityGroupRule("ValkeyIngress", {
90 groupId: net.securityGroupId,
91 type: "ingress",
92 ipProtocol: "tcp",
93 fromPort: 6379,
94 toPort: 6379,
95 referencedGroupId: lambdaSecurityGroup.groupId,
96 });
97 yield* AWS.EC2.SecurityGroupRule("MemcachedIngress", {
98 groupId: net.securityGroupId,
99 type: "ingress",
100 ipProtocol: "tcp",
101 fromPort: 11211,
102 toPort: 11211,
103 referencedGroupId: lambdaSecurityGroup.groupId,
104 });
105 const valkey = yield* AWS.ElastiCache.ReplicationGroup("Valkey", {
106 description: "Alchemy provisioned Valkey data-plane test",
107 engine: "valkey",
108 nodeType: "cache.t4g.micro",
109 subnetGroupName: net.subnetGroupName,
110 securityGroupIds: [net.securityGroupId],
111 replicasPerNodeGroup: 0,
112 transitEncryptionEnabled: false,
113 });
114 const memcached = yield* AWS.ElastiCache.CacheCluster("Memcached", {
115 nodeType: "cache.t4g.micro",
116 subnetGroupName: net.subnetGroupName,
117 securityGroupIds: [net.securityGroupId],
118 numCacheNodes: 1,
119 });
120 const valkeyConnection = yield* AWS.ElastiCache.ConnectReplicationGroup(
121 valkey,
122 {
123 subnetIds: net.subnetIds,
124 securityGroupIds: [lambdaSecurityGroup.groupId],
125 },
126 );
127 const memcachedConnection = yield* AWS.ElastiCache.ConnectCacheCluster(
128 memcached,
129 {
130 subnetIds: net.subnetIds,
131 securityGroupIds: [lambdaSecurityGroup.groupId],
132 },
133 );
134
135 return {
136 fetch: Effect.gen(function* () {
137 const request = yield* HttpServerRequest;
138 const url = new URL(request.originalUrl);
139 const value = url.searchParams.get("value") ?? "hello-provisioned";
140 if (request.method === "GET" && url.pathname === "/connection") {
141 const [valkey, memcached] = yield* Effect.all([
142 valkeyConnection,
143 memcachedConnection,
144 ]);
145 return yield* HttpServerResponse.json({ valkey, memcached });
146 }
147 if (request.method === "GET" && url.pathname === "/valkey") {
148 const info = yield* valkeyConnection;
149 const read = yield* valkeyRoundtrip(info, value);
150 return yield* HttpServerResponse.json({ value: read });
151 }
152 if (request.method === "GET" && url.pathname === "/memcached") {
153 const info = yield* memcachedConnection;
154 const endpoint = info.endpoints[0];
155 if (endpoint === undefined) {
156 return yield* Effect.fail(
157 new Error("Memcached endpoint is unavailable"),
158 );
159 }
160 const read = yield* memcachedRoundtrip(endpoint, value);
161 return yield* HttpServerResponse.json({ value: read });
162 }
163 return yield* HttpServerResponse.json(
164 { error: "Not found" },
165 { status: 404 },
166 );
167 }).pipe(
168 Effect.catchCause((cause) =>
169 HttpServerResponse.json(
170 { error: Cause.pretty(cause) },
171 { status: 500 },
172 ),
173 ),
174 ),
175 };
176 }).pipe(
177 Effect.provide(
178 Layer.mergeAll(
179 AWS.ElastiCache.ConnectReplicationGroupHttp,
180 AWS.ElastiCache.ConnectCacheClusterHttp,
181 ),
182 ),
183 ),
184);
185
186export default ProvisionedCacheDataPlaneLive;