1import * as Effect from "effect/Effect";
2import * as Queue from "effect/Queue";
3import * as Scope from "effect/Scope";
4import * as Semaphore from "effect/Semaphore";
5import type { RuntimeContext } from "../RuntimeContext.ts";
6import { CommandError, ProtocolError, type UrlMissing } from "./Errors.ts";
7import { encodeCommand, Parser, type Arg, type Reply } from "./Resp.ts";
8
9export const DEFAULT_PORT = 6379;
10export const TLS_PORT = 6380;
11
12type SocketEvent =
13 | { readonly _tag: "Data"; readonly bytes: Uint8Array }
14 | { readonly _tag: "End" }
15 | { readonly _tag: "Fail"; readonly cause: unknown };
16
17interface RawSocket {
18 write(bytes: Uint8Array): number | Promise<number>;
19 end(): void;
20}
21
22export interface RedisUrl {
23 readonly hostname: string;
24 readonly port: number;
25 readonly tls: boolean;
26 readonly username: string;
27 readonly password: string;
28 readonly db: number | undefined;
29}
30
31/** Parse a `redis://` or `rediss://` URL. */
32export const parseUrl = (url: string): RedisUrl => {
33 const parsed = new URL(url);
34 const tls = parsed.protocol === "rediss:";
35 const path = parsed.pathname.startsWith("/")
36 ? parsed.pathname.slice(1)
37 : parsed.pathname;
38 const dbRaw = path.length === 0 ? undefined : Number(path);
39 return {
40 hostname: parsed.hostname,
41 port: Number(parsed.port || (tls ? TLS_PORT : DEFAULT_PORT)),
42 tls,
43 username: decodeURIComponent(parsed.username),
44 password: decodeURIComponent(parsed.password),
45 db:
46 dbRaw !== undefined && Number.isInteger(dbRaw) && dbRaw >= 0
47 ? dbRaw
48 : undefined,
49 };
50};
51
52/** Build a `redis://` URL. Encodes the password. Used by tests and TcpProxy. */
53export const connectionUrl = (input: {
54 host: string;
55 port: number;
56 password: string;
57 username?: string;
58 db?: number;
59 tls?: boolean;
60}): string => {
61 const username = input.username ?? "default";
62 const scheme = input.tls ? "rediss" : "redis";
63 const db = input.db !== undefined ? `/${input.db}` : "";
64 return `${scheme}://${username}:${encodeURIComponent(input.password)}@${input.host}:${input.port}${db}`;
65};
66
67const commandError = (command: string, cause: unknown): CommandError =>
68 cause instanceof CommandError
69 ? new CommandError({ command, cause: cause.cause })
70 : new CommandError({ command, cause });
71
72const asBytes = (data: Uint8Array): Uint8Array => new Uint8Array(data);
73
74const writeSocket = (
75 socket: RawSocket,
76 bytes: Uint8Array,
77): Effect.Effect<void, CommandError> => {
78 const written = socket.write(bytes);
79 if (typeof written === "object" && written !== null && "then" in written) {
80 return Effect.tryPromise({
81 try: () => Promise.resolve(written),
82 catch: (cause) => commandError("WRITE", cause),
83 }).pipe(Effect.asVoid);
84 }
85 return Effect.void;
86};
87
88const openBun = (
89 options: RedisUrl,
90 events: Queue.Queue<SocketEvent>,
91): Effect.Effect<RawSocket, CommandError> =>
92 Effect.callback<RawSocket, CommandError>((resume, signal) => {
93 let settled = false;
94 let socket: RawSocket | undefined;
95 const succeed = (value: RawSocket) => {
96 if (settled) return;
97 settled = true;
98 resume(Effect.succeed(value));
99 };
100 const fail = (cause: unknown) => {
101 if (settled) {
102 Queue.offerUnsafe(events, { _tag: "Fail", cause });
103 return;
104 }
105 settled = true;
106 resume(Effect.fail(commandError("CONNECT", cause)));
107 };
108 const Client = (globalThis as { Bun?: { connect: typeof Bun.connect } })
109 .Bun;
110 if (Client === undefined) {
111 fail(new Error("Bun.connect is not available"));
112 return;
113 }
114 Client.connect({
115 hostname: options.hostname,
116 port: options.port,
117 tls: options.tls,
118 socket: {
119 binaryType: "uint8array",
120 open: (opened) => {
121 socket = opened;
122 succeed(opened);
123 },
124 data: (_opened, chunk) => {
125 Queue.offerUnsafe(events, {
126 _tag: "Data",
127 bytes: asBytes(chunk),
128 });
129 },
130 error: (_opened, cause) => fail(cause),
131 connectError: (_opened, cause) => fail(cause),
132 close: () => {
133 Queue.offerUnsafe(events, { _tag: "End" });
134 },
135 end: () => {
136 Queue.offerUnsafe(events, { _tag: "End" });
137 },
138 },
139 }).catch(fail);
140
141 const abort = () => {
142 socket?.end();
143 };
144 signal.addEventListener("abort", abort, { once: true });
145 return Effect.sync(() => {
146 signal.removeEventListener("abort", abort);
147 abort();
148 });
149 });
150
151const openNode = (
152 options: RedisUrl,
153 events: Queue.Queue<SocketEvent>,
154): Effect.Effect<RawSocket, CommandError> =>
155 Effect.callback<RawSocket, CommandError>((resume, signal) => {
156 let settled = false;
157 let socket: { write(bytes: Uint8Array): boolean; end(): void } | undefined;
158 const succeed = (value: RawSocket) => {
159 if (settled) return;
160 settled = true;
161 resume(Effect.succeed(value));
162 };
163 const fail = (cause: unknown) => {
164 if (settled) {
165 Queue.offerUnsafe(events, { _tag: "Fail", cause });
166 return;
167 }
168 settled = true;
169 resume(Effect.fail(commandError("CONNECT", cause)));
170 };
171
172 const attach = (
173 nodeSocket: {
174 write(bytes: Uint8Array): boolean;
175 end(): void;
176 on(event: string, listener: (...args: any[]) => void): unknown;
177 },
178 ready: "connect" | "secureConnect",
179 ) => {
180 socket = nodeSocket;
181 nodeSocket.on(ready, () =>
182 succeed({
183 write: (bytes) => {
184 nodeSocket.write(bytes);
185 return bytes.length;
186 },
187 end: () => {
188 nodeSocket.end();
189 },
190 }),
191 );
192 nodeSocket.on("data", (chunk: Uint8Array) => {
193 Queue.offerUnsafe(events, { _tag: "Data", bytes: asBytes(chunk) });
194 });
195 nodeSocket.on("error", fail);
196 nodeSocket.on("close", () => {
197 Queue.offerUnsafe(events, { _tag: "End" });
198 });
199 };
200
201 const start = options.tls
202 ? import("node:tls").then((tls) =>
203 attach(
204 tls.connect({
205 host: options.hostname,
206 port: options.port,
207 servername: options.hostname,
208 }),
209 "secureConnect",
210 ),
211 )
212 : import("node:net").then((net) =>
213 attach(
214 net.connect({ host: options.hostname, port: options.port }),
215 "connect",
216 ),
217 );
218 start.catch(fail);
219
220 const abort = () => {
221 socket?.end();
222 };
223 signal.addEventListener("abort", abort, { once: true });
224 return Effect.sync(() => {
225 signal.removeEventListener("abort", abort);
226 abort();
227 });
228 });
229
230const openRaw = (
231 options: RedisUrl,
232 events: Queue.Queue<SocketEvent>,
233): Effect.Effect<RawSocket, CommandError> =>
234 typeof Bun !== "undefined"
235 ? openBun(options, events)
236 : openNode(options, events);
237
238const unwrap = (
239 frame: Exclude<import("./Resp.ts").ParseResult, { _tag: "Incomplete" }>,
240 command: string,
241): Effect.Effect<Reply, CommandError> => {
242 if (frame._tag === "Protocol") {
243 return Effect.fail(commandError(command, frame.error));
244 }
245 if (frame._tag === "Error") {
246 return Effect.fail(commandError(command, frame.error));
247 }
248 if (frame._tag === "Push") {
249 return Effect.succeed(frame.value);
250 }
251 return Effect.succeed(frame.value);
252};
253
254const expectStatus = (
255 reply: Reply,
256 command: string,
257): Effect.Effect<void, CommandError> => {
258 if (typeof reply === "string" && reply.toUpperCase() === "OK") {
259 return Effect.void;
260 }
261 return Effect.fail(
262 commandError(
263 command,
264 new ProtocolError({ message: `expected OK, got ${String(reply)}` }),
265 ),
266 );
267};
268
269export interface Connection {
270 readonly send: (
271 command: string,
272 args?: readonly Arg[],
273 ) => Effect.Effect<Reply, CommandError>;
274 readonly pipeline: (
275 commands: ReadonlyArray<readonly [string, ...Arg[]]>,
276 ) => Effect.Effect<readonly Reply[], CommandError>;
277}
278
279const readReplies = (
280 events: Queue.Queue<SocketEvent>,
281 parser: Parser,
282 count: number,
283 command: string,
284): Effect.Effect<Reply[], CommandError> =>
285 Effect.gen(function* () {
286 const replies: Reply[] = [];
287 while (replies.length < count) {
288 const event = yield* Queue.take(events);
289 if (event._tag === "Fail") {
290 return yield* Effect.fail(commandError(command, event.cause));
291 }
292 if (event._tag === "End") {
293 return yield* Effect.fail(
294 commandError(
295 command,
296 new ProtocolError({ message: "connection closed" }),
297 ),
298 );
299 }
300 parser.push(event.bytes);
301 while (replies.length < count) {
302 const frame = parser.next();
303 if (frame._tag === "Incomplete") break;
304 if (frame._tag === "Push") continue;
305 replies.push(yield* unwrap(frame, command));
306 }
307 }
308 return replies;
309 });
310
311/**
312 * Open an authenticated Redis session. Closes when the ambient Scope
313 * is released. Concurrent `send`/`pipeline` calls on one connection
314 * are serialized so replies stay ordered.
315 */
316export const connect = (
317 url: string,
318): Effect.Effect<Connection, CommandError, Scope.Scope> =>
319 Effect.gen(function* () {
320 const options = yield* Effect.try({
321 try: () => parseUrl(url),
322 catch: (cause) => commandError("CONNECT", cause),
323 });
324 const events = yield* Queue.unbounded<SocketEvent>();
325 const socket = yield* Effect.acquireRelease(
326 openRaw(options, events),
327 (opened) =>
328 Effect.sync(() => {
329 opened.end();
330 }).pipe(
331 Effect.flatMap(() => Queue.shutdown(events)),
332 Effect.asVoid,
333 ),
334 );
335 const parser = new Parser();
336 const lock = yield* Semaphore.make(1);
337
338 const request = (payload: Uint8Array, count: number, command: string) =>
339 lock.withPermit(
340 Effect.gen(function* () {
341 yield* writeSocket(socket, payload);
342 return yield* readReplies(events, parser, count, command);
343 }),
344 );
345
346 if (options.password.length > 0) {
347 const authArgs =
348 options.username.length > 0 && options.username !== "default"
349 ? [options.username, options.password]
350 : [options.password];
351 const replies = yield* request(
352 encodeCommand("AUTH", authArgs),
353 1,
354 "AUTH",
355 );
356 yield* expectStatus(replies[0] ?? null, "AUTH");
357 }
358
359 if (options.db !== undefined && options.db !== 0) {
360 const replies = yield* request(
361 encodeCommand("SELECT", [options.db]),
362 1,
363 "SELECT",
364 );
365 yield* expectStatus(replies[0] ?? null, "SELECT");
366 }
367
368 return {
369 send: (command, args = []) =>
370 request(encodeCommand(command, args), 1, command).pipe(
371 Effect.map((replies) => replies[0] ?? null),
372 ),
373 pipeline: (commands) => {
374 if (commands.length === 0) return Effect.succeed([]);
375 const chunks = commands.map(([name, ...args]) =>
376 encodeCommand(name, args),
377 );
378 let total = 0;
379 for (const chunk of chunks) total += chunk.length;
380 const payload = new Uint8Array(total);
381 let offset = 0;
382 for (const chunk of chunks) {
383 payload.set(chunk, offset);
384 offset += chunk.length;
385 }
386 return request(
387 payload,
388 commands.length,
389 commands[0]?.[0] ?? "PIPELINE",
390 );
391 },
392 } satisfies Connection;
393 });
394
395const scoped = <A>(
396 command: string,
397 effect: Effect.Effect<A, CommandError, Scope.Scope>,
398): Effect.Effect<A, CommandError> =>
399 Effect.scoped(effect).pipe(
400 Effect.mapError((error) => commandError(command, error)),
401 );
402
403/**
404 * Send one Redis command over RESP and close the connection.
405 */
406export const run = (
407 url: string,
408 command: string,
409 args: readonly Arg[] = [],
410): Effect.Effect<Reply, CommandError> =>
411 scoped(
412 command,
413 Effect.gen(function* () {
414 const connection = yield* connect(url);
415 return yield* connection.send(command, args);
416 }),
417 );
418
419/**
420 * Pipeline several commands on one connection, then close it.
421 */
422export const pipeline = (
423 url: string,
424 commands: ReadonlyArray<readonly [string, ...Arg[]]>,
425): Effect.Effect<readonly Reply[], CommandError> =>
426 scoped(
427 commands[0]?.[0] ?? "PIPELINE",
428 Effect.gen(function* () {
429 const connection = yield* connect(url);
430 return yield* connection.pipeline(commands);
431 }),
432 );
433
434export const command = (
435 url: Effect.Effect<string, UrlMissing, RuntimeContext>,
436 name: string,
437 args: readonly Arg[] = [],
438): Effect.Effect<Reply, CommandError | UrlMissing, RuntimeContext> =>
439 Effect.gen(function* () {
440 const resolved = yield* url;
441 return yield* run(resolved, name, args);
442 });
443
444export const commandPipeline = (
445 url: Effect.Effect<string, UrlMissing, RuntimeContext>,
446 commands: ReadonlyArray<readonly [string, ...Arg[]]>,
447): Effect.Effect<readonly Reply[], CommandError | UrlMissing, RuntimeContext> =>
448 Effect.gen(function* () {
449 const resolved = yield* url;
450 return yield* pipeline(resolved, commands);
451 });