1/**
2 * RESP2 + RESP3 codec. Commands are encoded as arrays of bulk strings.
3 * Replies are parsed with a length-prefixed recursive reader so values
4 * may contain CR/LF/`$`/`*` and may arrive split across TCP frames.
5 *
6 * Spec: https://github.com/redis/redis-specifications/blob/master/protocol/RESP2.md
7 * Algorithm: the same as NodeRedis/node-redis-parser, reimplemented here
8 * so alchemy does not take a Redis client dependency.
9 */
10import { ProtocolError, ReplyError } from "./Errors.ts";
11
12export type Arg = string | number | Uint8Array;
13
14/**
15 * A decoded RESP value. Nested errors stay as {@link ReplyError} values
16 * (arrays may contain them). Top-level errors fail the command.
17 */
18export type Reply =
19 | string
20 | number
21 | boolean
22 | bigint
23 | null
24 | ReplyError
25 | readonly Reply[]
26 | { readonly [key: string]: Reply };
27
28export type ParseResult =
29 | { readonly _tag: "Incomplete" }
30 | { readonly _tag: "Reply"; readonly value: Reply }
31 | { readonly _tag: "Error"; readonly error: ReplyError }
32 | { readonly _tag: "Push"; readonly value: readonly Reply[] }
33 | { readonly _tag: "Protocol"; readonly error: ProtocolError };
34
35const encoder = new TextEncoder();
36const decoder = new TextDecoder("utf-8");
37
38const CR = 13;
39const LF = 10;
40const CRLF = new Uint8Array([CR, LF]);
41
42const PLUS = 43;
43const MINUS = 45;
44const COLON = 58;
45const DOLLAR = 36;
46const STAR = 42;
47const UNDERSCORE = 95;
48const HASH = 35;
49const COMMA = 44;
50const LPAREN = 40;
51const EQUALS = 61;
52const PERCENT = 37;
53const TILDE = 126;
54const PIPE = 124;
55const GT = 62;
56const BANG = 33;
57const DOT = 46;
58const SEMICOLON = 59;
59const QUESTION = 63;
60
61/** Redis bulk strings are capped at 512 MB. */
62const MAX_BULK = 512 * 1024 * 1024;
63const MAX_AGGREGATE = 1_000_000;
64
65export const Incomplete: Extract<ParseResult, { _tag: "Incomplete" }> = {
66 _tag: "Incomplete",
67};
68
69const concat = (parts: readonly Uint8Array[]): Uint8Array => {
70 let total = 0;
71 for (const part of parts) total += part.length;
72 const out = new Uint8Array(total);
73 let offset = 0;
74 for (const part of parts) {
75 out.set(part, offset);
76 offset += part.length;
77 }
78 return out;
79};
80
81const bytesOf = (value: string): Uint8Array => encoder.encode(value);
82
83const argBytes = (arg: Arg): Uint8Array => {
84 if (typeof arg === "string") return bytesOf(arg);
85 if (typeof arg === "number") return bytesOf(String(arg));
86 return arg;
87};
88
89export const encodeSimpleString = (value: string): Uint8Array =>
90 concat([bytesOf(`+${value}`), CRLF]);
91
92export const encodeError = (value: string): Uint8Array =>
93 concat([bytesOf(`-${value}`), CRLF]);
94
95export const encodeInteger = (value: number | bigint): Uint8Array =>
96 concat([bytesOf(`:${value}`), CRLF]);
97
98export const encodeBoolean = (value: boolean): Uint8Array =>
99 bytesOf(value ? "#t\r\n" : "#f\r\n");
100
101export const encodeDouble = (value: number): Uint8Array => {
102 if (Number.isNaN(value)) return bytesOf(",nan\r\n");
103 if (value === Infinity) return bytesOf(",inf\r\n");
104 if (value === -Infinity) return bytesOf(",-inf\r\n");
105 return concat([bytesOf(`,${value}`), CRLF]);
106};
107
108export const encodeBigNumber = (value: bigint): Uint8Array =>
109 concat([bytesOf(`(${value}`), CRLF]);
110
111export const encodeNull = (): Uint8Array => bytesOf("$-1\r\n");
112
113export const encodeNullArray = (): Uint8Array => bytesOf("*-1\r\n");
114
115export const encodeBulk = (value: string | Uint8Array): Uint8Array => {
116 const payload = typeof value === "string" ? bytesOf(value) : value;
117 return concat([bytesOf(`$${payload.length}\r\n`), payload, CRLF]);
118};
119
120export const encodeArray = (items: readonly Uint8Array[]): Uint8Array =>
121 concat([bytesOf(`*${items.length}\r\n`), ...items]);
122
123/** Encode a client command as a RESP array of bulk strings. */
124export const encodeCommand = (
125 command: string,
126 args: readonly Arg[] = [],
127): Uint8Array => {
128 const parts = [bytesOf(command), ...args.map(argBytes)];
129 const chunks: Uint8Array[] = [bytesOf(`*${parts.length}\r\n`)];
130 for (const part of parts) {
131 chunks.push(bytesOf(`$${part.length}\r\n`), part, CRLF);
132 }
133 return concat(chunks);
134};
135
136export const encodeReply = (value: Reply): Uint8Array => {
137 if (value === null) return encodeNull();
138 if (typeof value === "string") return encodeBulk(value);
139 if (typeof value === "boolean") return encodeBoolean(value);
140 if (typeof value === "bigint") {
141 if (
142 value >= BigInt(Number.MIN_SAFE_INTEGER) &&
143 value <= BigInt(Number.MAX_SAFE_INTEGER)
144 ) {
145 return encodeInteger(Number(value));
146 }
147 return encodeBigNumber(value);
148 }
149 if (typeof value === "number") {
150 return Number.isInteger(value) ? encodeInteger(value) : encodeDouble(value);
151 }
152 if (value instanceof ReplyError) return encodeError(value.message);
153 if (Array.isArray(value)) {
154 return encodeArray(value.map(encodeReply));
155 }
156 const entries = Object.entries(value);
157 const encoded: Uint8Array[] = [bytesOf(`%${entries.length}\r\n`)];
158 for (const [key, item] of entries) {
159 encoded.push(encodeSimpleString(key), encodeReply(item));
160 }
161 return concat(encoded);
162};
163
164type Line = {
165 readonly text: string;
166 readonly bytes: Uint8Array;
167 readonly next: number;
168};
169
170type Outcome =
171 | {
172 readonly ok: true;
173 readonly frame: Exclude<ParseResult, { _tag: "Incomplete" | "Protocol" }>;
174 readonly offset: number;
175 }
176 | { readonly ok: false; readonly incomplete: true }
177 | { readonly ok: false; readonly error: ProtocolError };
178
179const INCOMPLETE: Outcome = { ok: false, incomplete: true };
180
181const fail = (message: string): Outcome => ({
182 ok: false,
183 error: new ProtocolError({ message }),
184});
185
186const ok = (
187 frame: Exclude<ParseResult, { _tag: "Incomplete" | "Protocol" }>,
188 offset: number,
189): Outcome => ({ ok: true, frame, offset });
190
191const parseLine = (
192 buf: Uint8Array,
193 offset: number,
194): Line | undefined | ProtocolError => {
195 for (let i = offset; i < buf.length; i++) {
196 if (buf[i] === LF) {
197 if (i === offset || buf[i - 1] !== CR) {
198 return new ProtocolError({ message: "expected CRLF line terminator" });
199 }
200 const bytes = buf.subarray(offset, i - 1);
201 return { text: decoder.decode(bytes), bytes, next: i + 1 };
202 }
203 }
204 return undefined;
205};
206
207const parseCount = (
208 buf: Uint8Array,
209 offset: number,
210):
211 | { readonly kind: "null"; readonly next: number }
212 | { readonly kind: "stream"; readonly next: number }
213 | { readonly kind: "n"; readonly n: number; readonly next: number }
214 | { readonly kind: "incomplete" }
215 | { readonly kind: "error"; readonly error: ProtocolError } => {
216 const line = parseLine(buf, offset);
217 if (line === undefined) return { kind: "incomplete" };
218 if (line instanceof ProtocolError) return { kind: "error", error: line };
219 if (line.text === "?") return { kind: "stream", next: line.next };
220 if (line.text === "-1") return { kind: "null", next: line.next };
221 if (line.text.length === 0 || !/^-?\d+$/.test(line.text)) {
222 return {
223 kind: "error",
224 error: new ProtocolError({
225 message: `invalid length ${JSON.stringify(line.text)}`,
226 }),
227 };
228 }
229 const n = Number(line.text);
230 if (!Number.isSafeInteger(n) || n < 0) {
231 return {
232 kind: "error",
233 error: new ProtocolError({ message: `invalid length ${line.text}` }),
234 };
235 }
236 if (n > MAX_BULK) {
237 return {
238 kind: "error",
239 error: new ProtocolError({ message: `length ${n} exceeds ${MAX_BULK}` }),
240 };
241 }
242 return { kind: "n", n, next: line.next };
243};
244
245const MIN_SAFE = BigInt(Number.MIN_SAFE_INTEGER);
246const MAX_SAFE = BigInt(Number.MAX_SAFE_INTEGER);
247
248const parseIntegerValue = (text: string): number | bigint => {
249 if (text.length === 0 || !/^-?\d+$/.test(text)) {
250 throw new ProtocolError({
251 message: `invalid integer ${JSON.stringify(text)}`,
252 });
253 }
254 const value = BigInt(text);
255 if (value >= MIN_SAFE && value <= MAX_SAFE) return Number(value);
256 return value;
257};
258
259const parseErrorLine = (text: string): ReplyError => {
260 const space = text.indexOf(" ");
261 const code = space === -1 ? text : text.slice(0, space);
262 return new ReplyError({ code, message: text });
263};
264
265const asElement = (
266 frame: Exclude<ParseResult, { _tag: "Incomplete" | "Protocol" }>,
267): Reply => {
268 if (frame._tag === "Error") return frame.error;
269 if (frame._tag === "Push") return frame.value;
270 return frame.value;
271};
272
273const parseBulkBody = (buf: Uint8Array, offset: number, n: number): Outcome => {
274 const end = offset + n + 2;
275 if (end > buf.length) return INCOMPLETE;
276 if (buf[offset + n] !== CR || buf[offset + n + 1] !== LF) {
277 return fail("missing CRLF after bulk string");
278 }
279 return ok(
280 { _tag: "Reply", value: decoder.decode(buf.subarray(offset, offset + n)) },
281 end,
282 );
283};
284
285const parseStreamedString = (buf: Uint8Array, offset: number): Outcome => {
286 const chunks: Uint8Array[] = [];
287 let cursor = offset;
288 while (true) {
289 if (cursor >= buf.length) return INCOMPLETE;
290 if (buf[cursor] !== SEMICOLON) {
291 return fail("expected streamed string chunk");
292 }
293 const count = parseCount(buf, cursor + 1);
294 if (count.kind === "incomplete") return INCOMPLETE;
295 if (count.kind === "error") return { ok: false, error: count.error };
296 if (count.kind !== "n") return fail("invalid streamed string chunk length");
297 if (count.n === 0) {
298 return ok(
299 { _tag: "Reply", value: decoder.decode(concat(chunks)) },
300 count.next,
301 );
302 }
303 const body = parseBulkBody(buf, count.next, count.n);
304 if (!body.ok) return body;
305 if (body.frame._tag !== "Reply" || typeof body.frame.value !== "string") {
306 return fail("invalid streamed string chunk");
307 }
308 chunks.push(buf.subarray(count.next, count.next + count.n));
309 cursor = body.offset;
310 }
311};
312
313const parseVerbatim = (buf: Uint8Array, offset: number): Outcome => {
314 const count = parseCount(buf, offset);
315 if (count.kind === "incomplete") return INCOMPLETE;
316 if (count.kind === "error") return { ok: false, error: count.error };
317 if (count.kind !== "n") return fail("invalid verbatim string length");
318 const body = parseBulkBody(buf, count.next, count.n);
319 if (!body.ok) return body;
320 if (body.frame._tag !== "Reply" || typeof body.frame.value !== "string") {
321 return fail("invalid verbatim string");
322 }
323 const raw = body.frame.value;
324 const value = raw.length >= 4 && raw[3] === ":" ? raw.slice(4) : raw;
325 return ok({ _tag: "Reply", value }, body.offset);
326};
327
328const parseBlobError = (buf: Uint8Array, offset: number): Outcome => {
329 const count = parseCount(buf, offset);
330 if (count.kind === "incomplete") return INCOMPLETE;
331 if (count.kind === "error") return { ok: false, error: count.error };
332 if (count.kind !== "n") return fail("invalid blob error length");
333 const body = parseBulkBody(buf, count.next, count.n);
334 if (!body.ok) return body;
335 if (body.frame._tag !== "Reply" || typeof body.frame.value !== "string") {
336 return fail("invalid blob error");
337 }
338 return ok(
339 { _tag: "Error", error: parseErrorLine(body.frame.value) },
340 body.offset,
341 );
342};
343
344const parseElements = (
345 buf: Uint8Array,
346 offset: number,
347 n: number,
348):
349 | { readonly ok: true; readonly values: Reply[]; readonly offset: number }
350 | Outcome => {
351 if (n > MAX_AGGREGATE) {
352 return fail(`aggregate length ${n} exceeds ${MAX_AGGREGATE}`);
353 }
354 const values: Reply[] = [];
355 let cursor = offset;
356 for (let i = 0; i < n; i++) {
357 const next = parseFrame(buf, cursor);
358 if (!next.ok) return next;
359 values.push(asElement(next.frame));
360 cursor = next.offset;
361 }
362 return { ok: true, values, offset: cursor };
363};
364
365const parseStreamedAggregate = (
366 buf: Uint8Array,
367 offset: number,
368):
369 | { readonly ok: true; readonly values: Reply[]; readonly offset: number }
370 | Outcome => {
371 const values: Reply[] = [];
372 let cursor = offset;
373 while (true) {
374 if (cursor >= buf.length) return INCOMPLETE;
375 if (buf[cursor] === DOT) {
376 const line = parseLine(buf, cursor + 1);
377 if (line === undefined) return INCOMPLETE;
378 if (line instanceof ProtocolError) return { ok: false, error: line };
379 if (line.text.length !== 0) return fail("invalid aggregate terminator");
380 return { ok: true, values, offset: line.next };
381 }
382 const next = parseFrame(buf, cursor);
383 if (!next.ok) return next;
384 values.push(asElement(next.frame));
385 cursor = next.offset;
386 if (values.length > MAX_AGGREGATE) {
387 return fail(`aggregate length exceeds ${MAX_AGGREGATE}`);
388 }
389 }
390};
391
392const pairsToReply = (values: readonly Reply[]): Reply => {
393 if (values.length % 2 !== 0) {
394 return values;
395 }
396 const entries: Array<[Reply, Reply]> = [];
397 for (let i = 0; i < values.length; i += 2) {
398 entries.push([values[i]!, values[i + 1]!]);
399 }
400 const out: Record<string, Reply> = {};
401 for (const [key, value] of entries) {
402 if (typeof key !== "string") return values;
403 out[key] = value;
404 }
405 return out;
406};
407
408const parseAggregate = (
409 buf: Uint8Array,
410 offset: number,
411 kind: "array" | "map" | "set" | "push",
412): Outcome => {
413 const count = parseCount(buf, offset);
414 if (count.kind === "incomplete") return INCOMPLETE;
415 if (count.kind === "error") return { ok: false, error: count.error };
416 if (count.kind === "null") {
417 return ok({ _tag: "Reply", value: null }, count.next);
418 }
419
420 const parsed =
421 count.kind === "stream"
422 ? parseStreamedAggregate(buf, count.next)
423 : parseElements(buf, count.next, kind === "map" ? count.n * 2 : count.n);
424 if (!("values" in parsed)) return parsed;
425
426 if (kind === "push") {
427 return ok({ _tag: "Push", value: parsed.values }, parsed.offset);
428 }
429 if (kind === "map") {
430 return ok(
431 { _tag: "Reply", value: pairsToReply(parsed.values) },
432 parsed.offset,
433 );
434 }
435 return ok({ _tag: "Reply", value: parsed.values }, parsed.offset);
436};
437
438const parseBulk = (buf: Uint8Array, offset: number): Outcome => {
439 const count = parseCount(buf, offset);
440 if (count.kind === "incomplete") return INCOMPLETE;
441 if (count.kind === "error") return { ok: false, error: count.error };
442 if (count.kind === "null") {
443 return ok({ _tag: "Reply", value: null }, count.next);
444 }
445 if (count.kind === "stream") return parseStreamedString(buf, count.next);
446 return parseBulkBody(buf, count.next, count.n);
447};
448
449function parseFrame(buf: Uint8Array, offset: number): Outcome {
450 let cursor = offset;
451 while (cursor < buf.length && buf[cursor] === PIPE) {
452 const attrs = parseAggregate(buf, cursor + 1, "map");
453 if (!attrs.ok) return attrs;
454 cursor = attrs.offset;
455 }
456 if (cursor >= buf.length) return INCOMPLETE;
457
458 const type = buf[cursor]!;
459 const next = cursor + 1;
460 switch (type) {
461 case PLUS: {
462 const line = parseLine(buf, next);
463 if (line === undefined) return INCOMPLETE;
464 if (line instanceof ProtocolError) return { ok: false, error: line };
465 return ok({ _tag: "Reply", value: line.text }, line.next);
466 }
467 case MINUS: {
468 const line = parseLine(buf, next);
469 if (line === undefined) return INCOMPLETE;
470 if (line instanceof ProtocolError) return { ok: false, error: line };
471 return ok({ _tag: "Error", error: parseErrorLine(line.text) }, line.next);
472 }
473 case COLON: {
474 const line = parseLine(buf, next);
475 if (line === undefined) return INCOMPLETE;
476 if (line instanceof ProtocolError) return { ok: false, error: line };
477 try {
478 return ok(
479 { _tag: "Reply", value: parseIntegerValue(line.text) },
480 line.next,
481 );
482 } catch (error) {
483 return error instanceof ProtocolError
484 ? { ok: false, error }
485 : fail(String(error));
486 }
487 }
488 case DOLLAR:
489 return parseBulk(buf, next);
490 case STAR:
491 return parseAggregate(buf, next, "array");
492 case UNDERSCORE: {
493 const line = parseLine(buf, next);
494 if (line === undefined) return INCOMPLETE;
495 if (line instanceof ProtocolError) return { ok: false, error: line };
496 if (line.text.length !== 0) return fail("invalid null");
497 return ok({ _tag: "Reply", value: null }, line.next);
498 }
499 case HASH: {
500 const line = parseLine(buf, next);
501 if (line === undefined) return INCOMPLETE;
502 if (line instanceof ProtocolError) return { ok: false, error: line };
503 if (line.text === "t")
504 return ok({ _tag: "Reply", value: true }, line.next);
505 if (line.text === "f")
506 return ok({ _tag: "Reply", value: false }, line.next);
507 return fail(`invalid boolean ${JSON.stringify(line.text)}`);
508 }
509 case COMMA: {
510 const line = parseLine(buf, next);
511 if (line === undefined) return INCOMPLETE;
512 if (line instanceof ProtocolError) return { ok: false, error: line };
513 const text = line.text.toLowerCase();
514 if (text === "nan" || text === "-nan") {
515 return ok({ _tag: "Reply", value: Number.NaN }, line.next);
516 }
517 if (text === "inf")
518 return ok({ _tag: "Reply", value: Infinity }, line.next);
519 if (text === "-inf") {
520 return ok({ _tag: "Reply", value: -Infinity }, line.next);
521 }
522 if (!/^-?(?:\d+)(?:\.\d+)?(?:[eE][+-]?\d+)?$/.test(line.text)) {
523 return fail(`invalid double ${JSON.stringify(line.text)}`);
524 }
525 return ok({ _tag: "Reply", value: Number(line.text) }, line.next);
526 }
527 case LPAREN: {
528 const line = parseLine(buf, next);
529 if (line === undefined) return INCOMPLETE;
530 if (line instanceof ProtocolError) return { ok: false, error: line };
531 if (line.text.length === 0 || !/^-?\d+$/.test(line.text)) {
532 return fail(`invalid big number ${JSON.stringify(line.text)}`);
533 }
534 return ok({ _tag: "Reply", value: BigInt(line.text) }, line.next);
535 }
536 case EQUALS:
537 return parseVerbatim(buf, next);
538 case PERCENT:
539 return parseAggregate(buf, next, "map");
540 case TILDE:
541 return parseAggregate(buf, next, "set");
542 case GT:
543 return parseAggregate(buf, next, "push");
544 case BANG:
545 return parseBlobError(buf, next);
546 case QUESTION:
547 return fail("unexpected '?' type byte");
548 default:
549 return fail(
550 `unexpected type byte ${JSON.stringify(String.fromCharCode(type))} (0x${type.toString(16)})`,
551 );
552 }
553}
554
555/**
556 * Incremental RESP parser. Feed TCP chunks with {@link Parser.push} and
557 * pull complete frames with {@link Parser.next}.
558 */
559export class Parser {
560 #buffer: Uint8Array<ArrayBufferLike> = new Uint8Array(0);
561 #offset = 0;
562
563 push(chunk: Uint8Array): void {
564 if (chunk.length === 0) return;
565 if (this.#offset >= this.#buffer.length) {
566 this.#buffer = chunk;
567 this.#offset = 0;
568 return;
569 }
570 const remaining = this.#buffer.subarray(this.#offset);
571 const next = new Uint8Array(remaining.length + chunk.length);
572 next.set(remaining, 0);
573 next.set(chunk, remaining.length);
574 this.#buffer = next;
575 this.#offset = 0;
576 }
577
578 #compact(): void {
579 if (this.#offset === 0) return;
580 if (this.#offset >= this.#buffer.length) {
581 this.#buffer = new Uint8Array(0);
582 this.#offset = 0;
583 return;
584 }
585 this.#buffer = this.#buffer.slice(this.#offset);
586 this.#offset = 0;
587 }
588
589 /** Unparsed bytes still buffered. */
590 get pending(): number {
591 return this.#buffer.length - this.#offset;
592 }
593
594 next(): ParseResult {
595 const result = parseFrame(this.#buffer, this.#offset);
596 if (!result.ok) {
597 if ("incomplete" in result) {
598 this.#compact();
599 return Incomplete;
600 }
601 return { _tag: "Protocol", error: result.error };
602 }
603 this.#offset = result.offset;
604 this.#compact();
605 return result.frame;
606 }
607}
608
609/** Parse one complete frame. Incomplete or leftover bytes are protocol errors. */
610export const decode = (bytes: Uint8Array | string): ParseResult => {
611 const buf = typeof bytes === "string" ? bytesOf(bytes) : bytes;
612 const parser = new Parser();
613 parser.push(buf);
614 const result = parser.next();
615 if (result._tag === "Incomplete") {
616 return {
617 _tag: "Protocol",
618 error: new ProtocolError({ message: "incomplete RESP value" }),
619 };
620 }
621 if (parser.pending > 0 && result._tag !== "Protocol") {
622 return {
623 _tag: "Protocol",
624 error: new ProtocolError({ message: "unexpected trailing RESP bytes" }),
625 };
626 }
627 return result;
628};
629
630/** Parse every complete frame in `bytes`. Trailing incomplete data errors. */
631export const decodeAll = (bytes: Uint8Array | string): ParseResult[] => {
632 const buf = typeof bytes === "string" ? bytesOf(bytes) : bytes;
633 const parser = new Parser();
634 parser.push(buf);
635 const frames: ParseResult[] = [];
636 while (true) {
637 const result = parser.next();
638 if (result._tag === "Incomplete") {
639 if (parser.pending > 0 || frames.length === 0) {
640 frames.push({
641 _tag: "Protocol",
642 error: new ProtocolError({ message: "incomplete RESP value" }),
643 });
644 }
645 return frames;
646 }
647 frames.push(result);
648 if (result._tag === "Protocol") return frames;
649 }
650};