1import { describe, expect, it } from "alchemy-test";
2import * as Effect from "effect/Effect";
3import * as Result from "effect/Result";
4import * as Redis from "@/Redis/index.ts";
5import {
6 encodeError,
7 encodeInteger,
8 encodeNull,
9 encodeReply,
10 encodeSimpleString,
11 Parser,
12} from "@/Redis/Resp.ts";
13
14type Session = { parser: Parser; authed: boolean };
15
16type Stored =
17 | { readonly t: "string"; v: string }
18 | { readonly t: "hash"; v: Map<string, string> }
19 | { readonly t: "list"; v: string[] }
20 | { readonly t: "set"; v: Set<string> }
21 | { readonly t: "zset"; v: Map<string, number> };
22
23const asString = (value: unknown): string => String(value ?? "");
24
25const wrongType = () =>
26 encodeError(
27 "WRONGTYPE Operation against a key holding the wrong kind of value",
28 );
29
30const execute = (
31 store: Map<string, Stored>,
32 name: string,
33 args: string[],
34): Uint8Array => {
35 const cmd = name.toUpperCase();
36 if (cmd === "PING") {
37 return args[0] === undefined
38 ? encodeSimpleString("PONG")
39 : encodeReply(args[0]);
40 }
41 if (cmd === "ECHO") return encodeReply(args[0] ?? "");
42 if (cmd === "AUTH") return encodeSimpleString("OK");
43 if (cmd === "SELECT") return encodeSimpleString("OK");
44 if (cmd === "GET") {
45 const row = store.get(args[0] ?? "");
46 if (row === undefined) return encodeNull();
47 if (row.t !== "string") return wrongType();
48 return encodeReply(row.v);
49 }
50 if (cmd === "SET") {
51 store.set(args[0] ?? "", { t: "string", v: args[1] ?? "" });
52 return encodeSimpleString("OK");
53 }
54 if (cmd === "DEL") {
55 let n = 0;
56 for (const key of args) {
57 if (store.delete(key)) n++;
58 }
59 return encodeInteger(n);
60 }
61 if (cmd === "EXISTS") {
62 return encodeInteger(args.filter((key) => store.has(key)).length);
63 }
64 if (cmd === "INCR" || cmd === "INCRBY" || cmd === "DECR") {
65 const key = args[0] ?? "";
66 const delta =
67 cmd === "DECR" ? -1 : cmd === "INCR" ? 1 : Number(args[1] ?? 1);
68 const row = store.get(key);
69 if (row !== undefined && row.t !== "string") return wrongType();
70 const next = Number(row?.t === "string" ? row.v : 0) + delta;
71 store.set(key, { t: "string", v: String(next) });
72 return encodeInteger(next);
73 }
74 if (cmd === "HSET") {
75 const key = args[0] ?? "";
76 const row = store.get(key);
77 if (row !== undefined && row.t !== "hash") return wrongType();
78 const hash = row?.t === "hash" ? row.v : new Map<string, string>();
79 let added = 0;
80 for (let i = 1; i + 1 < args.length; i += 2) {
81 if (!hash.has(args[i]!)) added++;
82 hash.set(args[i]!, args[i + 1]!);
83 }
84 store.set(key, { t: "hash", v: hash });
85 return encodeInteger(added);
86 }
87 if (cmd === "HGET") {
88 const row = store.get(args[0] ?? "");
89 if (row === undefined) return encodeNull();
90 if (row.t !== "hash") return wrongType();
91 const value = row.v.get(args[1] ?? "");
92 return value === undefined ? encodeNull() : encodeReply(value);
93 }
94 if (cmd === "HGETALL") {
95 const row = store.get(args[0] ?? "");
96 if (row === undefined) return encodeReply([]);
97 if (row.t !== "hash") return wrongType();
98 const items: string[] = [];
99 for (const [field, value] of row.v) {
100 items.push(field, value);
101 }
102 return encodeReply(items);
103 }
104 if (cmd === "LPUSH" || cmd === "RPUSH") {
105 const key = args[0] ?? "";
106 const row = store.get(key);
107 if (row !== undefined && row.t !== "list") return wrongType();
108 const list = row?.t === "list" ? row.v : [];
109 const values = args.slice(1);
110 if (cmd === "LPUSH") list.unshift(...values.reverse());
111 else list.push(...values);
112 store.set(key, { t: "list", v: list });
113 return encodeInteger(list.length);
114 }
115 if (cmd === "LRANGE") {
116 const row = store.get(args[0] ?? "");
117 if (row === undefined) return encodeReply([]);
118 if (row.t !== "list") return wrongType();
119 const start = Number(args[1] ?? 0);
120 const stop = Number(args[2] ?? -1);
121 const end = stop < 0 ? row.v.length + stop + 1 : stop + 1;
122 return encodeReply(row.v.slice(start, end));
123 }
124 if (cmd === "SADD") {
125 const key = args[0] ?? "";
126 const row = store.get(key);
127 if (row !== undefined && row.t !== "set") return wrongType();
128 const set = row?.t === "set" ? row.v : new Set<string>();
129 let added = 0;
130 for (const member of args.slice(1)) {
131 if (!set.has(member)) {
132 set.add(member);
133 added++;
134 }
135 }
136 store.set(key, { t: "set", v: set });
137 return encodeInteger(added);
138 }
139 if (cmd === "SMEMBERS") {
140 const row = store.get(args[0] ?? "");
141 if (row === undefined) return encodeReply([]);
142 if (row.t !== "set") return wrongType();
143 return encodeReply([...row.v]);
144 }
145 if (cmd === "ZADD") {
146 const key = args[0] ?? "";
147 const row = store.get(key);
148 if (row !== undefined && row.t !== "zset") return wrongType();
149 const zset = row?.t === "zset" ? row.v : new Map<string, number>();
150 const member = args[2] ?? "";
151 const added = zset.has(member) ? 0 : 1;
152 zset.set(member, Number(args[1] ?? 0));
153 store.set(key, { t: "zset", v: zset });
154 return encodeInteger(added);
155 }
156 if (cmd === "ZRANGE") {
157 const row = store.get(args[0] ?? "");
158 if (row === undefined) return encodeReply([]);
159 if (row.t !== "zset") return wrongType();
160 const ranked = [...row.v.entries()].sort((a, b) => a[1] - b[1]);
161 const start = Number(args[1] ?? 0);
162 const stop = Number(args[2] ?? -1);
163 const end = stop < 0 ? ranked.length + stop + 1 : stop + 1;
164 return encodeReply(ranked.slice(start, end).map(([member]) => member));
165 }
166 return encodeError(`ERR unknown command '${name}'`);
167};
168
169const startFakeRedis = (options?: {
170 readonly password?: string;
171 readonly splitWrites?: boolean;
172}) =>
173 Effect.acquireRelease(
174 Effect.sync(() => {
175 const store = new Map<string, Stored>();
176 const sessions = new WeakMap<object, Session>();
177 const password = options?.password ?? "";
178 const splitWrites = options?.splitWrites === true;
179 const write = (
180 socket: { write(bytes: Uint8Array): unknown },
181 payload: Uint8Array,
182 ) => {
183 if (!splitWrites) {
184 socket.write(payload);
185 return;
186 }
187 for (const byte of payload) {
188 socket.write(new Uint8Array([byte]));
189 }
190 };
191 const server = Bun.listen({
192 hostname: "127.0.0.1",
193 port: 0,
194 socket: {
195 binaryType: "uint8array",
196 open(socket) {
197 sessions.set(socket, {
198 parser: new Parser(),
199 authed: password.length === 0,
200 });
201 },
202 data(socket, chunk) {
203 const session = sessions.get(socket);
204 if (session === undefined) return;
205 session.parser.push(chunk);
206 while (true) {
207 const frame = session.parser.next();
208 if (frame._tag === "Incomplete") break;
209 if (frame._tag !== "Reply" || !Array.isArray(frame.value)) {
210 write(socket, encodeError("ERR protocol"));
211 continue;
212 }
213 const [name, ...args] = frame.value.map(asString);
214 if (!session.authed) {
215 if ((name ?? "").toUpperCase() !== "AUTH") {
216 write(socket, encodeError("NOAUTH Authentication required"));
217 continue;
218 }
219 const supplied = args.length > 1 ? args[1] : args[0];
220 if (supplied !== password) {
221 write(socket, encodeError("ERR invalid password"));
222 continue;
223 }
224 session.authed = true;
225 write(socket, encodeSimpleString("OK"));
226 continue;
227 }
228 write(socket, execute(store, name ?? "", args));
229 }
230 },
231 },
232 });
233 return {
234 url: Redis.connectionUrl({
235 host: "127.0.0.1",
236 port: server.port,
237 password,
238 }),
239 server,
240 };
241 }),
242 ({ server }) =>
243 Effect.sync(() => {
244 server.stop(true);
245 }),
246 );
247
248describe("Redis protocol client", { tags: ["unit", "local"] }, () => {
249 it.effect("PING SET GET DEL over a one-shot connection", () =>
250 Effect.gen(function* () {
251 const { url } = yield* startFakeRedis();
252 expect(yield* Redis.run(url, "PING")).toBe("PONG");
253 expect(yield* Redis.run(url, "SET", ["marker", "hello"])).toBe("OK");
254 expect(yield* Redis.run(url, "GET", ["marker"])).toBe("hello");
255 expect(yield* Redis.run(url, "GET", ["missing"])).toBeNull();
256 expect(yield* Redis.run(url, "DEL", ["marker"])).toBe(1);
257 }),
258 );
259
260 it.effect("values may contain CR LF $ * and UTF-8", () =>
261 Effect.gen(function* () {
262 const { url } = yield* startFakeRedis();
263 const value = "line1\r\n$bulk\n*2\r\nšŸ˜€";
264 yield* Redis.run(url, "SET", ["k", value]);
265 expect(yield* Redis.run(url, "GET", ["k"])).toBe(value);
266 }),
267 );
268
269 it.effect("AUTH then command, and AUTH failure", () =>
270 Effect.gen(function* () {
271 const { url } = yield* startFakeRedis({ password: "s3cret" });
272 expect(yield* Redis.run(url, "PING")).toBe("PONG");
273 const denied = yield* Redis.run(
274 Redis.connectionUrl({
275 host: "127.0.0.1",
276 port: Number(new URL(url).port),
277 password: "nope",
278 }),
279 "PING",
280 ).pipe(Effect.result);
281 expect(Result.isFailure(denied)).toBe(true);
282 }),
283 );
284
285 it.effect("pipeline issues several commands on one connection", () =>
286 Effect.gen(function* () {
287 const { url } = yield* startFakeRedis();
288 const replies = yield* Redis.pipeline(url, [
289 ["PING"],
290 ["SET", "a", "1"],
291 ["INCR", "n"],
292 ["GET", "a"],
293 ]);
294 expect(replies).toEqual(["PONG", "OK", 1, "1"]);
295 }),
296 );
297
298 it.effect("session send reuses the socket", () =>
299 Effect.gen(function* () {
300 const { url } = yield* startFakeRedis();
301 const conn = yield* Redis.connect(url);
302 expect(yield* conn.send("PING")).toBe("PONG");
303 expect(yield* conn.send("SET", ["k", "v"])).toBe("OK");
304 expect(yield* conn.send("GET", ["k"])).toBe("v");
305 expect(yield* conn.send("INCR", ["n"])).toBe(1);
306 expect(yield* conn.send("INCR", ["n"])).toBe(2);
307 }),
308 );
309
310 it.effect("hashes lists sets sorted sets and typed errors", () =>
311 Effect.gen(function* () {
312 const { url } = yield* startFakeRedis();
313 const conn = yield* Redis.connect(url);
314 expect(yield* conn.send("HSET", ["h", "a", "1", "b", "2"])).toBe(2);
315 expect(yield* conn.send("HGET", ["h", "a"])).toBe("1");
316 expect(yield* conn.send("HGETALL", ["h"])).toEqual(["a", "1", "b", "2"]);
317 expect(yield* conn.send("LPUSH", ["l", "x", "y"])).toBe(2);
318 expect(yield* conn.send("LRANGE", ["l", 0, -1])).toEqual(["y", "x"]);
319 expect(yield* conn.send("SADD", ["s", "a", "a", "b"])).toBe(2);
320 const members = yield* conn.send("SMEMBERS", ["s"]);
321 expect(Array.isArray(members) ? [...members].sort() : members).toEqual([
322 "a",
323 "b",
324 ]);
325 expect(yield* conn.send("ZADD", ["z", 2, "b"])).toBe(1);
326 expect(yield* conn.send("ZADD", ["z", 1, "a"])).toBe(1);
327 expect(yield* conn.send("ZRANGE", ["z", 0, -1])).toEqual(["a", "b"]);
328 const wrong = yield* conn.send("GET", ["h"]).pipe(Effect.result);
329 expect(Result.isFailure(wrong)).toBe(true);
330 const unknown = yield* conn.send("NOPE").pipe(Effect.result);
331 expect(Result.isFailure(unknown)).toBe(true);
332 }),
333 );
334
335 it.effect("buffers replies split into one-byte TCP writes", () =>
336 Effect.gen(function* () {
337 const { url } = yield* startFakeRedis({ splitWrites: true });
338 expect(yield* Redis.run(url, "PING")).toBe("PONG");
339 yield* Redis.run(url, "SET", ["k", "hello\r\nworld"]);
340 expect(yield* Redis.run(url, "GET", ["k"])).toBe("hello\r\nworld");
341 expect(
342 yield* Redis.pipeline(url, [
343 ["HSET", "h", "f", "v"],
344 ["HGETALL", "h"],
345 ]),
346 ).toEqual([1, ["f", "v"]]);
347 }),
348 );
349
350 it.effect("SELECT from the URL path", () =>
351 Effect.gen(function* () {
352 const { url } = yield* startFakeRedis();
353 const parsed = new URL(url);
354 const withDb = `${parsed.protocol}//${parsed.username}:${parsed.password}@${parsed.hostname}:${parsed.port}/2`;
355 expect(yield* Redis.run(withDb, "PING")).toBe("PONG");
356 }),
357 );
358});