// https://redis.io/docs/latest/develop/reference/protocol-spec/

module redis

import math.big
import net
import net.ssl
import strings

// RESP3 wrapper types
pub struct RedisBlobError {
pub:
	data []u8
}

pub struct RedisVerbatim {
pub:
	format string
	data   []u8
}

pub struct RedisMap {
pub:
	// interleaved key/value pairs: [k1, v1, k2, v2, ...]
	pairs []RedisValue
}

pub struct RedisSet {
pub:
	elements []RedisValue
}

pub struct RedisPush {
pub:
	elements []RedisValue
}

// RedisValue represents all possible RESP (Redis Serialization Protocol) data types
pub type RedisValue = bool
	| big.Integer
	| f32
	| f64
	| i64
	| []u8
	| RedisBlobError
	| RedisMap
	| RedisNull
	| RedisPush
	| RedisSet
	| map[string]RedisValue
	| []RedisValue
	| RedisVerbatim
	| string

// RedisNull represents the Redis NULL type
pub struct RedisNull {}

const cmd_buf_pre_allocate_len = 4096 // Initial buffer size for command building
const resp_buf_pre_allocate_len = 8192 // Initial buffer size for response reading
const max_skip = 64 // Max non-prefix bytes to skip when resynchronizing

// DB represents a Redis database connection
pub struct DB {
pub mut:
	version  int // RESP protocol version
	conn     &net.TcpConn = unsafe { nil } // TCP connection to Redis
	ssl_conn &ssl.SSLConn = unsafe { nil } // SSL connection to Redis
	tls      bool

	// Pre-allocated buffers to reduce memory allocations
	cmd_buf            []u8 // Buffer for building commands
	resp_buf           []u8 // Buffer for reading responses
	pipeline_mode      bool
	pipeline_buffer    []u8
	pipeline_cmd_count int
}

// Configuration options for Redis connection
@[params]
pub struct Config {
pub mut:
	host     string = '127.0.0.1' // Redis server host
	port     u16    = 6379        // Redis server port
	password string // Redis server password (optional)
	tls      bool   // Enable TLS/SSL connection
	version  int @[deprecated] // ignored - RESP version auto handled in connect
}

fn resp2_auth(password string, mut db DB) ! {
	db.version = 2
	if password.len > 0 {
		db.auth(password)!
	}
}

// connect establishes a connection to a Redis server
pub fn connect(config Config) !DB {
	mut db := DB{
		tls:      config.tls
		cmd_buf:  []u8{cap: cmd_buf_pre_allocate_len}
		resp_buf: []u8{cap: resp_buf_pre_allocate_len}
	}

	if config.tls {
		mut ssl_conn := ssl.new_ssl_conn(ssl.SSLConnectConfig{ validate: false })!
		ssl_conn.dial(config.host, int(config.port))!
		db.ssl_conn = ssl_conn
	} else {
		db.conn = net.dial_tcp('${config.host}:${config.port}')!
	}

	// Always attempt HELLO during connect() to negotiate RESP3 (and include AUTH
	// subcommand when a password is provided). If HELLO fails or the server
	// doesn't support it, fall back to RESP2 and perform AUTH via the AUTH
	// command (if a password was given).

	// default version 3
	db.version = 3

	// build HELLO 3 command; include AUTH subcommand when password present
	db.cmd_buf.clear()
	if config.password.len > 0 {
		// *4\r\n$5\r\nHELLO\r\n$1\r\n3\r\n$4\r\nAUTH\r\n$<pwlen>\r\n<pw>\r\n
		db.cmd_buf << '*4\r\n$5\r\nHELLO\r\n$1\r\n3\r\n$4\r\nAUTH\r\n$${config.password.len}\r\n${config.password}\r\n'.bytes()
	} else {
		// *2\r\n$5\r\nHELLO\r\n$1\r\n3\r\n
		db.cmd_buf << '*2\r\n$5\r\nHELLO\r\n$1\r\n3\r\n'.bytes()
	}

	// send HELLO and attempt to read response. If any step fails, fallback to RESP2 + AUTH.
	db.write_data(db.cmd_buf) or {
		// write failed (connection error?) — fallback to RESP2 and perform AUTH if needed
		resp2_auth(config.password, mut db)!
		return db
	}

	// Try to read and consume HELLO response. If HELLO fails (unknown command / error),
	// treat as not-supported and fall back to RESP2 + AUTH.
	db.read_response() or {
		resp2_auth(config.password, mut db)!
		return db
	}

	return db
}

// close terminates the connection to Redis server
pub fn (mut db DB) close() ! {
	if db.tls {
		db.ssl_conn.close()!
	} else {
		db.conn.close()!
	}
}

// Helper methods for TLS abstraction
fn (mut db DB) write_data(data []u8) ! {
	if db.tls {
		db.ssl_conn.write(data)!
	} else {
		db.conn.write(data)!
	}
}

fn (mut db DB) read_data(mut buf []u8) !int {
	if db.tls {
		return db.ssl_conn.read(mut buf)
	} else {
		return db.conn.read(mut buf)
	}
}

fn (mut db DB) read_ptr_data(ptr &u8, len int) !int {
	if db.tls {
		return db.ssl_conn.socket_read_into_ptr(ptr, len)!
	} else {
		return db.conn.read_ptr(ptr, len)!
	}
}

// auth sends an AUTH command to the server with the given password.
pub fn (mut db DB) auth(password string) ! {
	resp := db.cmd('AUTH', password)!
	match resp {
		string {
			if resp != 'OK' {
				return error('Authentication failed: ${resp}')
			}
		}
		else {
			return error('Authentication failed: unexpected response type')
		}
	}
}

// ping sends a PING command to verify server responsiveness
pub fn (mut db DB) ping() !string {
	return db.cmd('PING')! as string
}

// validate checks whether the Redis connection is still responsive.
pub fn (mut db DB) validate() !bool {
	return db.ping()! == 'PONG'
}

// reset clears any queued pipeline state before the connection is reused.
pub fn (mut db DB) reset() ! {
	db.pipeline_mode = false
	db.pipeline_cmd_count = 0
	db.cmd_buf.clear()
	db.resp_buf.clear()
	db.pipeline_buffer.clear()
}

// del deletes a `key`
pub fn (mut db DB) del(key string) !i64 {
	// *2\r\n$3\r\nDEL\r\n$6\r\ncounter\r\n
	// send cmd
	db.cmd_buf.clear()
	db.cmd_buf << '*2\r\n$3\r\nDEL\r\n$${key.len}\r\n${key}\r\n'.bytes()
	if db.pipeline_mode {
		db.pipeline_buffer << db.cmd_buf
		db.pipeline_cmd_count++
	} else {
		db.write_data(db.cmd_buf)!

		// read resp
		return db.read_response()! as i64
	}
	return 0
}

// set stores a `key`-value` pair in Redis. Supported value types: number, string, []u8
pub fn (mut db DB) set[T](key string, value T) !string {
	// *3\r\n$3\r\nSET\r\n$4\r\nname\r\n$5\r\nVlang\r\n
	db.cmd_buf.clear()
	db.cmd_buf << '*3\r\n$3\r\nSET\r\n$${key.len}\r\n${key}\r\n'.bytes()
	$if T is $int {
		val_str := value.str()
		db.cmd_buf << '$${val_str.len}\r\n${val_str}'.bytes()
	} $else $if T is string {
		db.cmd_buf << '$${value.len}\r\n${value}'.bytes()
	} $else $if T is []u8 {
		db.cmd_buf << '$${value.len}\r\n'.bytes()
		db.cmd_buf << value
	} $else {
		return error('`set()`: unsupported value type. Allowed: number, string, []u8')
	}
	db.cmd_buf << '\r\n'.bytes()
	if db.pipeline_mode {
		db.pipeline_buffer << db.cmd_buf
		db.pipeline_cmd_count++
	} else {
		db.write_data(db.cmd_buf)!
		return db.read_response()! as string
	}
	return ''
}

// get retrieves the value of a `key`. Supported return types: string, int, []u8
pub fn (mut db DB) get[T](key string) !T {
	// *2\r\n$3\r\nGET\r\n$4\r\nname\r\n
	// send cmd
	db.cmd_buf.clear()
	db.cmd_buf << '*2\r\n$3\r\nGET\r\n$${key.len}\r\n${key}\r\n'.bytes()
	if db.pipeline_mode {
		db.pipeline_buffer << db.cmd_buf
		db.pipeline_cmd_count++
	} else {
		db.write_data(db.cmd_buf)!
		resp := db.read_response()!
		match resp {
			[]u8 {
				$if T is string {
					return resp.bytestr()
				} $else $if T is $int {
					return T(resp.bytestr().i64())
				} $else $if T is []u8 {
					return resp
				}
			}
			RedisNull {
				return error('`get()`: key ${key} not found')
			}
			else {
				return error('`get()`: unexpected response type')
			}
		}

		return error('`get()`: unsupported data type')
	}
	return T{}
}

// incr increments the integer value of a `key` by 1
pub fn (mut db DB) incr(key string) !i64 {
	// *2\r\n$4\r\nINCR\r\n$6\r\ncounter\r\n
	// send cmd
	db.cmd_buf.clear()
	db.cmd_buf << '*2\r\n$4\r\nINCR\r\n$${key.len}\r\n${key}\r\n'.bytes()
	if db.pipeline_mode {
		db.pipeline_buffer << db.cmd_buf
		db.pipeline_cmd_count++
	} else {
		db.write_data(db.cmd_buf)!

		// read resp
		return db.read_response()! as i64
	}
	return 0
}

// decr decrements the integer value of a `key` by 1
pub fn (mut db DB) decr(key string) !i64 {
	// *2\r\n$4\r\nDECR\r\n$6\r\ncounter\r\n
	// send cmd
	db.cmd_buf.clear()
	db.cmd_buf << '*2\r\n$4\r\nDECR\r\n$${key.len}\r\n${key}\r\n'.bytes()
	if db.pipeline_mode {
		db.pipeline_buffer << db.cmd_buf
		db.pipeline_cmd_count++
	} else {
		db.write_data(db.cmd_buf)!

		// read resp
		return db.read_response()! as i64
	}
	return 0
}

// hset sets multiple `key`-`value` pairs in a hash. Supported value types: string, int, []u8
pub fn (mut db DB) hset[T](key string, m map[string]T) !int {
	// HSET user:1 name "John" age 30
	// *6\r\n$4\r\nHSET\r\n$6\r\nuser:1\r\n$4\r\nname\r\n$4\r\nJohn\r\n$3\r\nage\r\n$2\r\n30\r\n
	db.cmd_buf.clear()
	db.cmd_buf << '*${2 + m.len * 2}\r\n$4\r\nHSET\r\n$${key.len}\r\n${key}\r\n'.bytes()
	for k, v in m {
		db.cmd_buf << '$${k.len}\r\n${k}\r\n'.bytes()
		$if T is string {
			db.cmd_buf << '$${v.len}\r\n${v}\r\n'.bytes()
		} $else $if T is $int {
			v_str := v.str()
			db.cmd_buf << '$${v_str.len}\r\n${v_str}\r\n'.bytes()
		} $else $if T is []u8 {
			// Write bulk string header correctly (no stray '$' after the length)
			db.cmd_buf << '$${v.len}\r\n'.bytes()
			db.cmd_buf << v
			db.cmd_buf << '\r\n'.bytes()
		} $else {
			return error('`hset()`: unsupported value type. Allowed: number, string, []u8')
		}
	}
	if db.pipeline_mode {
		db.pipeline_buffer << db.cmd_buf
		db.pipeline_cmd_count++
	} else {
		db.write_data(db.cmd_buf)!
		return int(db.read_response()! as i64)
	}
	return 0
}

// hget retrieves the value of a hash field. Supported return types: string, int, []u8
pub fn (mut db DB) hget[T](key string, m_key string) !T {
	// HGET user:1 name
	// *3\r\n$4\r\nHGET\r\n$6\r\nuser:1\r\n$4\r\nname\r\n
	db.cmd_buf.clear()
	db.cmd_buf << '*3\r\n$4\r\nHGET\r\n$${key.len}\r\n${key}\r\n'.bytes()
	db.cmd_buf << '$${m_key.len}\r\n${m_key}\r\n'.bytes()
	if db.pipeline_mode {
		db.pipeline_buffer << db.cmd_buf
		db.pipeline_cmd_count++
	} else {
		db.write_data(db.cmd_buf)!
		resp := db.read_response()! as []u8
		$if T is string {
			return resp.bytestr()
		} $else $if T is $int {
			return resp.bytestr().i64()
		} $else $if T is []u8 {
			return resp
		}
		return error('`hget()`: unsupported return type. Allowed: number, string, []u8')
	}
	return T{}
}

// hgetall retrieves all fields and values of a hash. Supported value types: string, int, []u8
pub fn (mut db DB) hgetall[T](key string) !map[string]T {
	// HGETALL user:1
	// *2\r\n$7\r\nHGETALL\r\n$6\r\nuser:1\r\n
	$if T !is string && T !is $int && T !is []u8 {
		return error('`hgetall()`: unsupported value type. Allowed: number, string, []u8')
	}
	db.cmd_buf.clear()
	db.cmd_buf << '*2\r\n$7\r\nHGETALL\r\n$${key.len}\r\n${key}\r\n'.bytes()
	if db.pipeline_mode {
		db.pipeline_buffer << db.cmd_buf
		db.pipeline_cmd_count++
	} else {
		db.write_data(db.cmd_buf)!
		resp := db.read_response()!

		// normalize result into map[string]T regardless of RESP2 array, RESP3 map,
		// or RedisMap interleaved pairs.
		$if T is string {
			mut result := map[string]T{}
			match resp {
				[]RedisValue {
					if resp.len % 2 != 0 {
						return error('`hgetall()`: invalid HGETALL response format')
					}
					for i in 0 .. resp.len / 2 {
						// keys and values expected as bulk strings for RESP2
						key_val := resp[2 * i]
						val_val := resp[2 * i + 1]
						// key
						k := match key_val {
							[]u8 { key_val.bytestr() }
							string { key_val }
							else { return error('`hgetall()`: unexpected key type: ${key_val.type_name()}') }
						}

						// value
						v := match val_val {
							[]u8 { val_val.bytestr() }
							string { val_val }
							i64 { val_val.str() }
							else { return error('`hgetall()`: unexpected value type: ${val_val.type_name()}') }
						}

						result[k] = v
					}
					return result
				}
				map[string]RedisValue {
					for k, v in resp {
						val_str := match v {
							[]u8 { v.bytestr() }
							string { v }
							i64 { v.str() }
							else { return error('`hgetall()`: unexpected value type in map: ${v.type_name()}') }
						}

						result[k] = val_str
					}
					return result
				}
				RedisMap {
					pairs := resp.pairs
					if pairs.len % 2 != 0 {
						return error('`hgetall()`: invalid RedisMap response format')
					}
					for i := 0; i < pairs.len; i += 2 {
						key_val := pairs[i]
						val_val := pairs[i + 1]
						k := match key_val {
							[]u8 { key_val.bytestr() }
							string { key_val }
							else { return error('`hgetall()`: unexpected key type in RedisMap: ${key_val.type_name()}') }
						}

						v := match val_val {
							[]u8 { val_val.bytestr() }
							string { val_val }
							i64 { val_val.str() }
							else { return error('`hgetall()`: unexpected value type in RedisMap: ${val_val.type_name()}') }
						}

						result[k] = v
					}
					return result
				}
				else {
					return error('`hgetall()`: unsupported response type: ${resp.type_name()}')
				}
			}
		} $else $if T is $int {
			mut result := map[string]T{}
			match resp {
				[]RedisValue {
					if resp.len % 2 != 0 {
						return error('`hgetall()`: invalid HGETALL response format')
					}
					for i in 0 .. resp.len / 2 {
						key_val := resp[2 * i]
						val_val := resp[2 * i + 1]
						k := match key_val {
							[]u8 { key_val.bytestr() }
							string { key_val }
							else { return error('`hgetall()`: unexpected key type: ${key_val.type_name()}') }
						}

						v := match val_val {
							[]u8 { val_val.bytestr().i64() }
							string { val_val.i64() }
							i64 { val_val }
							else { return error('`hgetall()`: unexpected value type: ${val_val.type_name()}') }
						}

						result[k] = T(v)
					}
					return result
				}
				map[string]RedisValue {
					for k, v in resp {
						n := match v {
							[]u8 { v.bytestr().i64() }
							string { v.i64() }
							i64 { v }
							else { return error('`hgetall()`: unexpected value type in map: ${v.type_name()}') }
						}

						result[k] = T(n)
					}
					return result
				}
				RedisMap {
					pairs := resp.pairs
					if pairs.len % 2 != 0 {
						return error('`hgetall()`: invalid RedisMap response format')
					}
					for i := 0; i < pairs.len; i += 2 {
						key_val := pairs[i]
						val_val := pairs[i + 1]
						k := match key_val {
							[]u8 { key_val.bytestr() }
							string { key_val }
							else { return error('`hgetall()`: unexpected key type in RedisMap: ${key_val.type_name()}') }
						}

						n := match val_val {
							[]u8 { val_val.bytestr().i64() }
							string { val_val.i64() }
							i64 { val_val }
							else { return error('`hgetall()`: unexpected value type in RedisMap: ${val_val.type_name()}') }
						}

						result[k] = T(n)
					}
					return result
				}
				else {
					return error('`hgetall()`: unsupported response type: ${resp.type_name()}')
				}
			}
		} $else $if T is []u8 {
			mut result := map[string]T{}
			match resp {
				[]RedisValue {
					if resp.len % 2 != 0 {
						return error('`hgetall()`: invalid HGETALL response format')
					}
					for i in 0 .. resp.len / 2 {
						key_val := resp[2 * i]
						val_val := resp[2 * i + 1]
						k := match key_val {
							[]u8 { key_val.bytestr() }
							string { key_val }
							else { return error('`hgetall()`: unexpected key type: ${key_val.type_name()}') }
						}

						v := match val_val {
							[]u8 { val_val }
							string { val_val.bytes() }
							i64 { val_val.str().bytes() }
							else { return error('`hgetall()`: unexpected value type: ${val_val.type_name()}') }
						}

						result[k] = v
					}
					return result
				}
				map[string]RedisValue {
					for k, v in resp {
						b := match v {
							[]u8 { v }
							string { v.bytes() }
							i64 { v.str().bytes() }
							else { return error('`hgetall()`: unexpected value type in map: ${v.type_name()}') }
						}

						result[k] = b
					}
					return result
				}
				RedisMap {
					pairs := resp.pairs
					if pairs.len % 2 != 0 {
						return error('`hgetall()`: invalid RedisMap response format')
					}
					for i := 0; i < pairs.len; i += 2 {
						key_val := pairs[i]
						val_val := pairs[i + 1]
						k := match key_val {
							[]u8 { key_val.bytestr() }
							string { key_val }
							else { return error('`hgetall()`: unexpected key type in RedisMap: ${key_val.type_name()}') }
						}

						b := match val_val {
							[]u8 { val_val }
							string { val_val.bytes() }
							i64 { val_val.str().bytes() }
							else { return error('`hgetall()`: unexpected value type in RedisMap: ${val_val.type_name()}') }
						}

						result[k] = b
					}
					return result
				}
				else {
					return error('`hgetall()`: unsupported response type: ${resp.type_name()}')
				}
			}
		} $else {
			// should not happen due to compile-time check above
			return error('`hgetall()`: unsupported value type ${T.type_name()}')
		}
	}
	return map[string]T{}
}

// expire sets a `key`'s time to live in `seconds`
pub fn (mut db DB) expire(key string, seconds int) !bool {
	// *3\r\n$6\r\nEXPIRE\r\n$6\r\ncounter\r\n$3\r\n600\r\n
	// send cmd
	seconds_str := seconds.str()
	db.cmd_buf.clear()
	db.cmd_buf << '*3\r\n$6\r\nEXPIRE\r\n$${key.len}\r\n${key}\r\n'.bytes()
	db.cmd_buf << '$${seconds_str.len}\r\n${seconds_str}\r\n'.bytes()
	if db.pipeline_mode {
		db.pipeline_buffer << db.cmd_buf
		db.pipeline_cmd_count++
	} else {
		db.write_data(db.cmd_buf)!

		// read resp
		rv := db.read_response()!

		// normalize to boolean result as before
		match rv {
			i64 { return rv != 0 }
			[]u8 { return rv.bytestr().i64() != 0 }
			string { return rv.i64() != 0 }
			else { return error('`expire()`: unexpected response type: ${rv.type_name()}') }
		}
	}
	return false
}

// read_response_bulk_string handles Redis bulk string responses (format: $<length>\r\n<data>\r\n)
fn (mut db DB) read_response_bulk_string() !RedisValue {
	mut data_length := i64(-1)
	mut chunk := []u8{len: 1}

	db.resp_buf.clear()
	for {
		bytes_read := db.read_data(mut chunk) or {
			return error('`read_response_bulk_string()`: connection error ${err}')
		}
		if bytes_read == 0 {
			return error('`read_response_bulk_string()`: connection closed prematurely')
		}
		db.resp_buf << chunk[0]

		if chunk[0] == `\n` {
			break
		}
		if (chunk[0] < `0` || chunk[0] > `9`) && chunk[0] != `\r` && chunk[0] != `-` {
			return error('`read_response_bulk_string()`: invalid bulk string header')
		}
	}

	if db.resp_buf.len < 2 {
		return error('`read_response_bulk_string()`: bulk string header too short')
	}

	data_length = db.resp_buf[0..db.resp_buf.len - 2].bytestr().i64()

	// -1 -> NULL bulk string
	if data_length == -1 {
		return RedisNull{}
	}

	// If zero-length payload, read exactly the 2-byte terminator CRLF reliably
	if data_length == 0 {
		mut term := [2]u8{}
		n := db.read_ptr_data(&term[0], 2)! // read remaining terminator bytes
		match n {
			0, 1 {
				return error('`read_response_bulk_string()`: incomplete terminator for empty string')
			}
			else {
				if n > 2 {
					return error('`read_response_bulk_string()`: should never get here - more than 2 bytes read?!?')
				}
			}
		}

		if term[0] != `\r` || term[1] != `\n` {
			return error('invalid terminator for empty string')
		}
		return []u8{}
	}

	// Read payload of exactly data_length bytes
	mut data_buf := []u8{len: int(data_length)}
	mut total_read := 0
	for total_read < data_buf.len {
		mut ptr := unsafe { &data_buf[total_read] }
		n := db.read_ptr_data(ptr, data_buf.len - total_read)!
		if n == 0 && total_read < data_buf.len {
			return error('`read_response_bulk_string()`: incomplete data: read ${total_read} / ${data_buf.len} bytes')
		}
		total_read += n
	}

	// Now read the trailing CRLF terminator (2 bytes) reliably
	mut term := []u8{len: 2}
	mut term_read := 0
	for term_read < 2 {
		mut ptr := unsafe { &term[term_read] }
		n := db.read_ptr_data(ptr, 2 - term_read)!
		if n == 0 && term_read < 2 {
			return error('`read_response_bulk_string()`: incomplete terminator after payload')
		}
		term_read += n
	}
	if term[0] != `\r` || term[1] != `\n` {
		return error('`read_response_bulk_string()`: invalid data terminator')
	}

	return data_buf
}

// read_header reads a CRLF-terminated header (returns content without trailing CRLF)
fn (mut db DB) read_header() !string {
	mut chunk := []u8{len: 1}
	db.resp_buf.clear()
	for {
		bytes_read := db.read_data(mut chunk) or {
			return error('`read_header()`: connection error ${err}')
		}
		if bytes_read == 0 {
			return error('`read_header()`: connection closed prematurely')
		}
		db.resp_buf << chunk[0]
		if chunk[0] == `\n` {
			break
		}
	}
	if db.resp_buf.len < 2 {
		return error('`read_header()`: header too short')
	}
	return db.resp_buf[0..db.resp_buf.len - 2].bytestr()
}

// read_exact_payload reads exactly n bytes + trailing CRLF and return the data bytes (without CRLF)
fn (mut db DB) read_exact_payload(n int) ![]u8 {
	if n < 0 {
		return error('invalid payload length ${n}')
	}
	mut data_buf := []u8{len: n + 2}
	mut total_read := 0
	for total_read < data_buf.len {
		remaining := data_buf.len - total_read
		chunk_size := if remaining > 1 { 1 } else { remaining }
		mut chunk_ptr := unsafe { &data_buf[total_read] }

		bytes_read := db.read_ptr_data(chunk_ptr, chunk_size)!
		total_read += bytes_read

		if bytes_read == 0 && total_read < data_buf.len {
			return error('`read_exact_payload()`: incomplete data: read ${total_read} / ${data_buf.len} bytes')
		}
	}
	// must ending with CRLF
	if data_buf[n] != `\r` || data_buf[n + 1] != `\n` {
		return error('`read_exact_payload()`: invalid data terminator')
	}
	return data_buf[0..n].clone()
}

// read_resp3_boolean_payload handles RESP3 boolean (#t or #f)
fn (mut db DB) read_resp3_boolean_payload() !bool {
	s := db.read_header()!
	if s == 't' {
		return true
	}
	if s == 'f' {
		return false
	}
	return error('`read_resp3_boolean_payload()`: invalid boolean: ${s}')
}

// read_resp3_double_payload handles RESP3 double (,<double>)
fn (mut db DB) read_resp3_double_payload() !f64 {
	s := db.read_header()!
	return s.f64()
}

// read_resp3_bignum_payload handles RESP3 big number ((<number>) -> big.Integer)
fn (mut db DB) read_resp3_bignum_payload() !big.Integer {
	mut s := db.read_header()!
	// RESP3 bignum frames may be wrapped in parentheses, e.g. "(12345)".
	// Trim leading '(' and trailing ')' if present to make the numeric string safe for the parser.
	if s.len > 0 && s[0] == `(` {
		s = s[1..]
	}
	if s.len > 0 && s[s.len - 1] == `)` {
		s = s[0..s.len - 1]
	}
	return big.integer_from_string(s)!
}

// read_resp3_blob_error_payload handles RESP3 blob error (!<len>\r\n<data>\r\n)
fn (mut db DB) read_resp3_blob_error_payload() !RedisBlobError {
	header := db.read_header()!
	length := header.i64()
	if length == -1 {
		return RedisBlobError{
			data: []u8{}
		}
	}
	payload := db.read_exact_payload(int(length))!
	return RedisBlobError{
		data: payload
	}
}

// read_resp3_verbatim_payload handles RESP3 verbatim (=<len>\r\n<fmt>:<data>\r\n)
fn (mut db DB) read_resp3_verbatim_payload() !RedisVerbatim {
	header := db.read_header()!
	length := header.i64()
	if length == -1 {
		return RedisVerbatim{
			format: ''
			data:   []u8{}
		}
	}
	payload := db.read_exact_payload(int(length))!
	// split at first ':'
	idx := payload.bytestr().index(':') or { -1 }
	if idx == -1 {
		return RedisVerbatim{
			format: ''
			data:   payload
		}
	}
	fmt := payload[0..idx].bytestr()
	data := payload[idx + 1..].clone()
	return RedisVerbatim{
		format: fmt
		data:   data
	}
}

// read_resp3_map_payload handles RESP3 map (%) where header is number of key/value pairs
// Try to return map[string]RedisValue when keys are string-like, otherwise return RedisMap
fn (mut db DB) read_resp3_map_payload() !RedisValue {
	header := db.read_header()!
	count := header.i64()
	if count == -1 {
		return RedisNull{}
	}
	if count == 0 {
		return map[string]RedisValue{}
	}
	mut pairs := []RedisValue{cap: int(count) * 2}
	for _ in 0 .. count {
		key := db.read_response()!
		val := db.read_response()!
		pairs << key
		pairs << val
	}
	// attempt to convert to map[string]RedisValue
	mut kv := map[string]RedisValue{}
	for i := 0; i < pairs.len; i += 2 {
		k := pairs[i]
		v := pairs[i + 1]
		match k {
			[]u8 {
				kv[k.bytestr()] = v
			}
			string {
				kv[k] = v
			}
			else {
				// fallback: return interleaved pairs preserved as RedisMap
				return RedisMap{
					pairs: pairs
				}
			}
		}
	}
	return kv
}

// read_resp3_attr_payload handles RESP3 attributes/attrs (|) and returns a map[string]RedisValue
// Attributes are map-like and we return a map when keys are string-like. If a non-string
// key is encountered, this treats it as an error (attributes are expected to be string-keyed).
fn (mut db DB) read_resp3_attr_payload() !RedisValue {
	header := db.read_header()!
	count := header.i64()
	if count == -1 {
		return RedisNull{}
	}
	if count == 0 {
		return map[string]RedisValue{}
	}
	mut kv := map[string]RedisValue{}
	for _ in 0 .. count {
		k := db.read_response()!
		v := db.read_response()!
		match k {
			[]u8 {
				kv[k.bytestr()] = v
			}
			string {
				kv[k] = v
			}
			else {
				return error('`read_resp3_attr_payload()`: attribute key is not a string-like type')
			}
		}
	}
	return kv
}

// read_resp3_set_payload handles RESP3 set (~)
fn (mut db DB) read_resp3_set_payload() !RedisSet {
	header := db.read_header()!
	count := header.i64()
	if count == -1 {
		return RedisSet{
			elements: []RedisValue{}
		}
	}
	mut elems := []RedisValue{cap: int(count)}
	for _ in 0 .. count {
		elems << db.read_response()!
	}
	return RedisSet{
		elements: elems
	}
}

// read_resp3_push_payload handles RESP3 push (>) - array-like
fn (mut db DB) read_resp3_push_payload() !RedisPush {
	header := db.read_header()!
	count := header.i64()
	if count == -1 {
		return RedisPush{
			elements: []RedisValue{}
		}
	}
	mut elems := []RedisValue{cap: int(count)}
	for _ in 0 .. count {
		elems << db.read_response()!
	}
	return RedisPush{
		elements: elems
	}
}

// read_response_i64 handles Redis integer responses (format: :<number>\r\n)
fn (mut db DB) read_response_i64() !i64 {
	db.resp_buf.clear()
	unsafe { db.resp_buf.grow_len(resp_buf_pre_allocate_len) }
	mut total_read := 0

	for total_read < db.resp_buf.len {
		remaining := db.resp_buf.len - total_read
		chunk_size := if remaining > 1 { 1 } else { remaining }
		mut chunk_ptr := unsafe { &db.resp_buf[total_read] }

		bytes_read := db.read_ptr_data(chunk_ptr, chunk_size)!
		total_read += bytes_read

		if total_read > 2 {
			if db.resp_buf[total_read - 2] == `\r` && db.resp_buf[total_read - 1] == `\n` {
				break
			}
		}
		if bytes_read == 0 {
			return error('`read_response_i64()`: incomplete data: read ${total_read} bytes')
		}
	}
	ret_val := db.resp_buf[0..total_read - 2].bytestr().i64()
	return ret_val
}

// read_response_simple_string handles Redis simple string responses (format: +<string>\r\n)
fn (mut db DB) read_response_simple_string() !string {
	db.resp_buf.clear()
	unsafe { db.resp_buf.grow_len(resp_buf_pre_allocate_len) }
	mut total_read := 0

	for total_read < db.resp_buf.len {
		remaining := db.resp_buf.len - total_read
		chunk_size := if remaining > 1 { 1 } else { remaining }
		mut chunk_ptr := unsafe { &db.resp_buf[total_read] }

		bytes_read := db.read_ptr_data(chunk_ptr, chunk_size)!
		total_read += bytes_read

		if total_read > 2 {
			if db.resp_buf[total_read - 2] == `\r` && db.resp_buf[total_read - 1] == `\n` {
				break
			}
		}
		if bytes_read == 0 {
			return error('`read_response_simple_string()`: incomplete data: read ${total_read} bytes')
		}
	}
	return db.resp_buf[0..total_read - 2].bytestr()
}

// read_response_array handles Redis array responses (format: *<length>\r\n<elements>)
fn (mut db DB) read_response_array() !RedisValue {
	mut array_len := i64(-1)
	mut chunk := []u8{len: 1}

	db.resp_buf.clear()
	for {
		bytes_read := db.read_data(mut chunk) or {
			return error('`read_response_array()`: connection error: ${err}')
		}
		if bytes_read == 0 {
			return error('`read_response_array()`: connection closed prematurely')
		}
		db.resp_buf << chunk[0]

		if chunk[0] == `\n` {
			break
		}
		if (chunk[0] < `0` || chunk[0] > `9`) && chunk[0] != `\r` && chunk[0] != `-` {
			return error('`read_response_array()`: invalid array header')
		}
	}

	if db.resp_buf.len < 2 {
		return error('`read_response_array()`: array header too short')
	}

	array_len = db.resp_buf[0..db.resp_buf.len - 2].bytestr().i64() // 排除\r\n

	if array_len == -1 {
		return RedisNull{}
	}
	if array_len == 0 {
		return []RedisValue{}
	}

	mut elements := []RedisValue{cap: int(array_len)}
	for _ in 0 .. array_len {
		element := db.read_response() or {
			return error('`read_response_array()`: failed to read array element: ${err}')
		}
		elements << element
	}
	return elements
}

// read_response handles all types of Redis responses (RESP2 + RESP3 when enabled)
fn (mut db DB) read_response() !RedisValue {
	db.resp_buf.clear()
	unsafe { db.resp_buf.grow_len(1) }
	// Read the first non-empty, non-CR/LF prefix byte. Some transports or
	// intermediate proxies may emit stray CR/LF bytes; skip them so we parse
	// the actual RESP prefix correctly.
	for {
		read_len := db.read_data(mut db.resp_buf)!
		if read_len != 1 {
			return error('`read_response()`: empty response from server')
		}
		// Skip stray CR and LF bytes that may precede the real response prefix.
		if db.resp_buf[0] == `\r` || db.resp_buf[0] == `\n` {
			continue
		}
		break
	}

	// If the first non-CRLF byte is not a valid RESP prefix, attempt a bounded
	// resynchronization: read and discard up to `max_skip` bytes looking for a
	// valid prefix. This helps tolerate transient stray bytes while avoiding
	// silently swallowing large amounts of data.
	mut attempts := 0

	for {
		// If this byte is a known RESP prefix, proceed to parse normally.
		ch := db.resp_buf[0]
		if ch == `+` || ch == `-` || ch == `:` || ch == `$` || ch == `*` || ch == `#` || ch == `,`
			|| ch == `(` || ch == `!` || ch == `=` || ch == `%` || ch == `~` || ch == `>`
			|| ch == `|` {
			break
		}
		// Give up after bounded attempts and return diagnostics.
		if attempts >= max_skip {
			mut prefix_val := -1
			if db.resp_buf.len > 0 {
				prefix_val = int(db.resp_buf[0])
			}
			mut hex := ''
			for i in 0 .. db.resp_buf.len {
				hex += '${int(db.resp_buf[i]):02x} '
			}
			return error("`read_response()`: unknown response prefix byte=${prefix_val} data_hex=\"${hex}\" data_str=\"${db.resp_buf.bytestr()}\"")
		}
		// Read and discard one more byte from the socket and treat it as the new candidate.
		mut tmp := []u8{len: 1}
		n := db.read_data(mut tmp) or { 0 }
		if n == 0 {
			return error('`read_response()`: incomplete data during resynchronization')
		}
		db.resp_buf[0] = tmp[0]
		// skip CRLF if encountered but still count the attempt (we consumed a byte)
		attempts++
		continue
	}

	match db.resp_buf[0] {
		`+` { // Simple string
			return db.read_response_simple_string()!
		}
		`-` { // Error message
			msg := db.read_response_simple_string()!
			return error(msg)
		}
		`:` { // Integer
			return db.read_response_i64()!
		}
		`$` { // Bulk string
			return db.read_response_bulk_string()!
		}
		`*` { // Array
			return db.read_response_array()!
		}
		// RESP3-only frames (enabled when db.version >= 3)
		`#` { // Boolean
			if db.version < 3 {
				return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
			}
			return RedisValue(db.read_resp3_boolean_payload()!)
		}
		`,` { // Double
			if db.version < 3 {
				return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
			}
			return RedisValue(db.read_resp3_double_payload()!)
		}
		`(` { // Big number
			if db.version < 3 {
				return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
			}
			return RedisValue(db.read_resp3_bignum_payload()!)
		}
		`!` { // Blob error
			if db.version < 3 {
				return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
			}
			return db.read_resp3_blob_error_payload()!
		}
		`=` { // Verbatim string
			if db.version < 3 {
				return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
			}
			return db.read_resp3_verbatim_payload()!
		}
		`%` { // Map
			if db.version < 3 {
				return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
			}
			return db.read_resp3_map_payload()!
		}
		`~` { // Set
			if db.version < 3 {
				return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
			}
			return RedisValue(db.read_resp3_set_payload()!)
		}
		`>` { // Push
			if db.version < 3 {
				return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
			}
			return RedisValue(db.read_resp3_push_payload()!)
		}
		`|` { // Attr (map-like)
			if db.version < 3 {
				return error('`read_response()`: unknown response prefix: ${db.resp_buf.bytestr()}')
			}
			// Attributes are parsed like maps; reuse map parsing (attrs preserved as map[string]RedisValue)
			return db.read_resp3_map_payload()!
		}
		else {
			// Fallback: this should be unreachable because we validated prefixes above,
			// but return a helpful diagnostic if it happens.
			mut prefix_val := -1
			if db.resp_buf.len > 0 {
				prefix_val = int(db.resp_buf[0])
			}
			mut hex := ''
			for i in 0 .. db.resp_buf.len {
				hex += '${int(db.resp_buf[i]):02x} '
			}
			return error("`read_response()`: unknown response prefix byte=${prefix_val} data_hex=\"${hex}\" data_str=\"${db.resp_buf.bytestr()}\"")
		}
	}

	return error('`read_response()`: unreachable code')
}

// cmd sends a custom command to Redis server
// for example: db.cmd('SET', 'key', 'value')!
pub fn (mut db DB) cmd(cmd ...string) !RedisValue {
	mut sb := strings.new_builder(cmd.len * 20)
	sb.write_string('*${cmd.len}\r\n') // Command array header
	for arg in cmd {
		sb.write_string('$${arg.len}\r\n${arg}\r\n')
	}
	if db.pipeline_mode {
		db.pipeline_buffer << unsafe { sb.reuse_as_plain_u8_array() }
		db.pipeline_cmd_count++
	} else {
		db.write_data(unsafe { sb.reuse_as_plain_u8_array() })!
		return db.read_response()!
	}
	return RedisNull{}
}

// pipeline_start start a new pipeline
pub fn (mut db DB) pipeline_start() {
	db.pipeline_mode = true
	db.pipeline_cmd_count = 0
	db.pipeline_buffer.clear()
}

// pipeline_execute executes the cmds in pipeline at once and retrieves all responses
pub fn (mut db DB) pipeline_execute() ![]RedisValue {
	if !db.pipeline_mode {
		return error('`pipeline_execute()`: pipeline not started')
	}
	defer {
		db.pipeline_mode = false
	}
	if db.pipeline_buffer.len == 0 {
		return []RedisValue{}
	}

	db.write_data(db.pipeline_buffer)!

	mut results := []RedisValue{cap: db.pipeline_cmd_count}
	for _ in 0 .. db.pipeline_cmd_count {
		results << db.read_response()!
	}

	// reset to non-pipeline mode
	db.pipeline_mode = false
	db.pipeline_cmd_count = 0
	return results
}