use std::{
    io::{BufRead, BufReader, Read, Write},
    net::{TcpStream, ToSocketAddrs},
    time::Duration,
};

use crate::{
    BackendError, CommandExecutionResult, ConnectionTestResult, RedisCommandRequest,
    RedisConnectionRequest, RedisFieldValueEntry, RedisKeyBrowseRequest, RedisKeyBrowseResult,
    RedisKeyMetadata, RedisKeyMetadataRequest, RedisKeyTtl, RedisKeyTtlUpdate,
    RedisKeyTtlUpdateRequest, RedisKeyTtlUpdateResult, RedisResponse, RedisSortedSetEntry,
    RedisStreamEntry, RedisValueContent, RedisValueCreateInput, RedisValueCreateRequest,
    RedisValueCreateResult, RedisValueData, RedisValueReadRequest, RedisValueRecord,
    RedisValueWriteCapability, RedisValueWriteInput, RedisValueWriteRequest, TlsMode,
};

const CONNECT_TIMEOUT_MS: u64 = 1_500;
const COMMAND_TIMEOUT_MS: u64 = 3_000;
const MAX_SCAN_PAGE_SIZE: u32 = 500;

pub fn test_connection(
    request: RedisConnectionRequest,
) -> Result<ConnectionTestResult, BackendError> {
    let mut connection = open_connection(&request)?;
    let response = connection.execute(["PING"])?;
    let round_trip_status = match response {
        RespValue::SimpleString(value) => value,
        RespValue::BulkString(Some(bytes)) => String::from_utf8(bytes).map_err(|error| {
            BackendError::command_failed(
                "Redis command `PING` returned non-UTF8 payload.",
                Some(error.to_string()),
            )
        })?,
        other => {
            return Err(BackendError::command_failed(
                "Redis command `PING` returned an unexpected response.",
                Some(format!("{other:?}")),
            ));
        }
    };

    Ok(ConnectionTestResult {
        selected_database: request.target.database,
        authenticated_as: request.target.username.clone(),
        round_trip_status,
    })
}

pub fn execute_command(
    request: RedisCommandRequest,
) -> Result<CommandExecutionResult, BackendError> {
    let command = request.command.trim();
    if command.is_empty() {
        return Err(BackendError::invalid_connection_config(
            "Redis command must not be empty.",
        ));
    }

    let mut connection = open_connection(&request.connection)?;
    let response = connection
        .execute(std::iter::once(command).chain(request.arguments.iter().map(String::as_str)))?;

    Ok(CommandExecutionResult {
        command: command.to_string(),
        arguments: request.arguments,
        database: request.connection.target.database,
        response: redis_value_to_response(response),
    })
}

pub fn browse_keys(request: RedisKeyBrowseRequest) -> Result<RedisKeyBrowseResult, BackendError> {
    let page_size = validate_scan_page_size(request.page_size)?;
    let submitted_cursor = normalize_scan_cursor(&request.cursor)?;
    let pattern = normalize_scan_pattern(request.pattern);
    let mut connection = open_connection(&request.connection)?;
    let (next_cursor, key_names) = scan_keys(
        &mut connection,
        &submitted_cursor,
        pattern.as_deref(),
        page_size,
    )?;
    let database = request.connection.target.database;
    let mut keys = Vec::with_capacity(key_names.len());

    for key_name in key_names {
        keys.push(read_key_metadata(&mut connection, database, &key_name)?);
    }

    Ok(RedisKeyBrowseResult {
        database,
        submitted_cursor,
        next_cursor: next_cursor.clone(),
        has_more: next_cursor != "0",
        page_size,
        pattern,
        keys,
    })
}

pub fn inspect_key(request: RedisKeyMetadataRequest) -> Result<RedisKeyMetadata, BackendError> {
    validate_key_name(&request.key)?;

    let database = request.connection.target.database;
    let mut connection = open_connection(&request.connection)?;
    read_key_metadata(&mut connection, database, &request.key)
}

pub fn read_value(request: RedisValueReadRequest) -> Result<RedisValueRecord, BackendError> {
    validate_key_name(&request.key)?;

    let database = request.connection.target.database;
    let mut connection = open_connection(&request.connection)?;
    read_value_record(&mut connection, database, &request.key)
}

pub fn write_value(request: RedisValueWriteRequest) -> Result<RedisValueRecord, BackendError> {
    validate_key_name(&request.key)?;

    let database = request.connection.target.database;
    let mut connection = open_connection(&request.connection)?;
    let metadata = read_key_metadata(&mut connection, database, &request.key)?;

    if !metadata.exists {
        return Err(BackendError::unsupported_operation(
            "Redis value writes require an existing key in the current backend foundation.",
            Some(format!("key `{}` no longer exists", request.key)),
        ));
    }

    match (&request.value, metadata.key_type.as_str()) {
        (RedisValueWriteInput::String { value }, "string") => {
            write_string_value(&mut connection, &request.key, value, &metadata.ttl)?;
        }
        (RedisValueWriteInput::String { .. }, other) => {
            return Err(BackendError::unsupported_operation(
                "Redis value editing is currently limited to full string replacement.",
                Some(format!("key `{}` has type `{other}`", request.key)),
            ));
        }
    }

    read_value_record(&mut connection, database, &request.key)
}

pub fn create_value(
    request: RedisValueCreateRequest,
) -> Result<RedisValueCreateResult, BackendError> {
    validate_key_name(&request.key)?;
    validate_optional_ttl_millis(request.ttl_millis)?;

    let database = request.connection.target.database;
    let mut connection = open_connection(&request.connection)?;

    match &request.value {
        RedisValueCreateInput::String { value } => {
            create_string_value(&mut connection, &request.key, value, request.ttl_millis)?;
        }
    }

    Ok(RedisValueCreateResult {
        record: read_value_record(&mut connection, database, &request.key)?,
    })
}

pub fn update_key_ttl(
    request: RedisKeyTtlUpdateRequest,
) -> Result<RedisKeyTtlUpdateResult, BackendError> {
    validate_key_name(&request.key)?;

    let database = request.connection.target.database;
    let mut connection = open_connection(&request.connection)?;
    let metadata = read_key_metadata(&mut connection, database, &request.key)?;

    if !metadata.exists {
        return Err(BackendError::unsupported_operation(
            "Redis TTL updates require an existing key.",
            Some(format!("key `{}` no longer exists", request.key)),
        ));
    }

    match request.operation {
        RedisKeyTtlUpdate::Persist => {
            if !matches!(metadata.ttl, RedisKeyTtl::Persistent) {
                let response = connection.execute(["PERSIST", request.key.as_str()])?;
                expect_integer_response(response, "PERSIST", 1)?;
            }
        }
        RedisKeyTtlUpdate::ExpiresInMillis { value } => {
            if value == 0 {
                return Err(BackendError::invalid_connection_config(
                    "Redis TTL update value must be greater than zero.",
                ));
            }

            let response = connection.execute_owned(vec![
                "PEXPIRE".to_string(),
                request.key.clone(),
                value.to_string(),
            ])?;
            expect_integer_response(response, "PEXPIRE", 1)?;
        }
    }

    Ok(RedisKeyTtlUpdateResult {
        metadata: read_key_metadata(&mut connection, database, &request.key)?,
    })
}

fn open_connection(request: &RedisConnectionRequest) -> Result<RedisConnection, BackendError> {
    request.target.validate()?;

    match request.target.tls_mode {
        TlsMode::Disabled => {}
        TlsMode::Preferred | TlsMode::Required => {
            return Err(BackendError::unsupported_tls_mode(
                "TLS modes are not implemented yet in the current backend foundation.",
            ));
        }
    }

    let address = (request.target.host.as_str(), request.target.port)
        .to_socket_addrs()
        .map_err(|error| {
            BackendError::connection_failed(
                "Redis host could not be resolved. Check host and port.",
                Some(error.to_string()),
            )
        })?
        .next()
        .ok_or_else(|| {
            BackendError::connection_failed(
                "Redis host could not be resolved. Check host and port.",
                None,
            )
        })?;

    let stream = TcpStream::connect_timeout(&address, Duration::from_millis(CONNECT_TIMEOUT_MS))
        .map_err(|error| {
            BackendError::connection_failed(
                "Redis connection failed. Check host, port, and server availability.",
                Some(error.to_string()),
            )
        })?;

    stream
        .set_read_timeout(Some(Duration::from_millis(COMMAND_TIMEOUT_MS)))
        .map_err(|error| {
            BackendError::internal(
                "Failed to apply Redis read timeout.",
                Some(error.to_string()),
            )
        })?;
    stream
        .set_write_timeout(Some(Duration::from_millis(COMMAND_TIMEOUT_MS)))
        .map_err(|error| {
            BackendError::internal(
                "Failed to apply Redis write timeout.",
                Some(error.to_string()),
            )
        })?;

    let reader = BufReader::new(stream.try_clone().map_err(|error| {
        BackendError::internal("Failed to clone Redis socket.", Some(error.to_string()))
    })?);
    let mut connection = RedisConnection {
        reader,
        writer: stream,
    };

    if let Some(password) = request.password.as_deref() {
        let auth_parts = if let Some(username) = request.target.username.as_deref() {
            vec![
                "AUTH".to_string(),
                username.to_string(),
                password.to_string(),
            ]
        } else {
            vec!["AUTH".to_string(), password.to_string()]
        };

        let auth_response = connection.execute_owned(auth_parts)?;
        expect_ok(auth_response).map_err(map_auth_error)?;
    }

    if request.target.database != 0 {
        let select_response = connection.execute_owned(vec![
            "SELECT".to_string(),
            request.target.database.to_string(),
        ])?;
        expect_ok(select_response).map_err(|error| map_command_error("SELECT", error))?;
    }

    Ok(connection)
}

fn redis_value_to_response(value: RespValue) -> RedisResponse {
    match value {
        RespValue::SimpleString(value) => RedisResponse::String { value },
        RespValue::Error(value) => RedisResponse::String { value },
        RespValue::Integer(value) => RedisResponse::Integer { value },
        RespValue::BulkString(Some(bytes)) => match String::from_utf8(bytes.clone()) {
            Ok(value) => RedisResponse::String { value },
            Err(_) => RedisResponse::Binary { bytes },
        },
        RespValue::BulkString(None) | RespValue::Array(None) => RedisResponse::Null,
        RespValue::Array(Some(items)) => RedisResponse::Array {
            items: items.into_iter().map(redis_value_to_response).collect(),
        },
    }
}

fn scan_keys(
    connection: &mut RedisConnection,
    cursor: &str,
    pattern: Option<&str>,
    page_size: u32,
) -> Result<(String, Vec<String>), BackendError> {
    let mut arguments = vec!["SCAN".to_string(), cursor.to_string()];
    if let Some(pattern) = pattern {
        arguments.push("MATCH".to_string());
        arguments.push(pattern.to_string());
    }
    arguments.push("COUNT".to_string());
    arguments.push(page_size.to_string());

    let response = connection.execute_owned(arguments)?;
    parse_scan_response(response)
}

fn parse_scan_response(response: RespValue) -> Result<(String, Vec<String>), BackendError> {
    let RespValue::Array(Some(items)) = response else {
        return Err(BackendError::command_failed(
            "Redis command `SCAN` returned an unexpected response.",
            Some(format!("{response:?}")),
        ));
    };

    if items.len() != 2 {
        return Err(BackendError::command_failed(
            "Redis command `SCAN` returned an unexpected response shape.",
            Some(format!("expected 2 items, got {}", items.len())),
        ));
    }

    let mut items = items.into_iter();
    let next_cursor = parse_resp_string(items.next().expect("cursor item"))?;
    let keys = match items.next().expect("keys item") {
        RespValue::Array(Some(values)) => {
            let mut key_names = Vec::with_capacity(values.len());
            for value in values {
                key_names.push(parse_resp_string(value)?);
            }
            key_names
        }
        other => {
            return Err(BackendError::command_failed(
                "Redis command `SCAN` returned an unexpected key list payload.",
                Some(format!("{other:?}")),
            ));
        }
    };

    Ok((next_cursor, keys))
}

fn read_key_metadata(
    connection: &mut RedisConnection,
    database: i64,
    key_name: &str,
) -> Result<RedisKeyMetadata, BackendError> {
    let type_response = connection.execute(["TYPE", key_name])?;
    let key_type = parse_resp_string(type_response).map_err(|error| {
        BackendError::command_failed(
            "Redis command `TYPE` returned an unexpected response.",
            error.detail,
        )
    })?;

    if key_type == "none" {
        return Ok(RedisKeyMetadata {
            name: key_name.to_string(),
            database,
            exists: false,
            key_type,
            ttl: RedisKeyTtl::Missing,
        });
    }

    let ttl_response = connection.execute(["PTTL", key_name])?;
    let ttl = parse_pttl(ttl_response)?;

    Ok(RedisKeyMetadata {
        name: key_name.to_string(),
        database,
        exists: true,
        key_type,
        ttl,
    })
}

fn read_value_record(
    connection: &mut RedisConnection,
    database: i64,
    key_name: &str,
) -> Result<RedisValueRecord, BackendError> {
    let metadata = read_key_metadata(connection, database, key_name)?;
    let data = if metadata.exists {
        read_value_data(connection, key_name, metadata.key_type.as_str())?
    } else {
        RedisValueData::Missing
    };

    Ok(RedisValueRecord {
        write_capability: write_capability_for_metadata(&metadata),
        ttl_write_supported: metadata.exists,
        metadata,
        data,
    })
}

fn read_value_data(
    connection: &mut RedisConnection,
    key_name: &str,
    key_type: &str,
) -> Result<RedisValueData, BackendError> {
    match key_type {
        "string" => Ok(RedisValueData::String {
            value: parse_binary_safe_value(connection.execute(["GET", key_name])?, "GET")?,
        }),
        "hash" => Ok(RedisValueData::Hash {
            entries: parse_field_value_entries(
                connection.execute(["HGETALL", key_name])?,
                "HGETALL",
            )?,
        }),
        "list" => Ok(RedisValueData::List {
            items: parse_content_array(
                connection.execute(["LRANGE", key_name, "0", "-1"])?,
                "LRANGE",
            )?,
        }),
        "set" => Ok(RedisValueData::Set {
            members: parse_content_array(connection.execute(["SMEMBERS", key_name])?, "SMEMBERS")?,
        }),
        "zset" => Ok(RedisValueData::SortedSet {
            entries: parse_sorted_set_entries(
                connection.execute(["ZRANGE", key_name, "0", "-1", "WITHSCORES"])?,
                "ZRANGE",
            )?,
        }),
        "stream" => Ok(RedisValueData::Stream {
            entries: parse_stream_entries(
                connection.execute(["XRANGE", key_name, "-", "+"])?,
                "XRANGE",
            )?,
        }),
        other => Err(BackendError::unsupported_operation(
            "Redis key type is not supported by the typed value surface.",
            Some(format!("key `{key_name}` has type `{other}`")),
        )),
    }
}

fn write_capability_for_metadata(metadata: &RedisKeyMetadata) -> RedisValueWriteCapability {
    if metadata.exists && metadata.key_type == "string" {
        RedisValueWriteCapability::ReplaceString
    } else {
        RedisValueWriteCapability::None
    }
}

fn write_string_value(
    connection: &mut RedisConnection,
    key_name: &str,
    value: &str,
    ttl: &RedisKeyTtl,
) -> Result<(), BackendError> {
    let mut commands = vec![vec![
        "SET".to_string(),
        key_name.to_string(),
        value.to_string(),
        "XX".to_string(),
    ]];

    if let RedisKeyTtl::ExpiresInMillis { value } = ttl {
        commands.push(vec![
            "PEXPIRE".to_string(),
            key_name.to_string(),
            value.to_string(),
        ]);
    }

    let results = execute_transaction(connection, &commands)?;
    expect_set_write_applied(results.first().cloned().ok_or_else(|| {
        BackendError::internal(
            "Redis transaction did not return the string write result.",
            None,
        )
    })?)?;

    if matches!(ttl, RedisKeyTtl::ExpiresInMillis { .. }) {
        let ttl_result = results.get(1).cloned().ok_or_else(|| {
            BackendError::internal(
                "Redis transaction did not return the TTL restore result.",
                None,
            )
        })?;
        expect_integer_response(ttl_result, "PEXPIRE", 1)?;
    }

    Ok(())
}

fn create_string_value(
    connection: &mut RedisConnection,
    key_name: &str,
    value: &str,
    ttl_millis: Option<u64>,
) -> Result<(), BackendError> {
    let mut command = vec![
        "SET".to_string(),
        key_name.to_string(),
        value.to_string(),
        "NX".to_string(),
    ];

    if let Some(ttl_millis) = ttl_millis {
        command.push("PX".to_string());
        command.push(ttl_millis.to_string());
    }

    let response = connection.execute_owned(command)?;
    expect_set_create_applied(response, key_name)
}

fn parse_resp_string(value: RespValue) -> Result<String, BackendError> {
    match value {
        RespValue::SimpleString(value) => Ok(value),
        RespValue::BulkString(Some(bytes)) => String::from_utf8(bytes).map_err(|error| {
            BackendError::command_failed(
                "Redis returned non-UTF8 text where a string was expected.",
                Some(error.to_string()),
            )
        }),
        other => Err(BackendError::command_failed(
            "Redis returned an unexpected response type where a string was expected.",
            Some(format!("{other:?}")),
        )),
    }
}

fn parse_binary_safe_value(
    value: RespValue,
    command: &str,
) -> Result<RedisValueContent, BackendError> {
    match value {
        RespValue::SimpleString(value) => Ok(RedisValueContent::String { value }),
        RespValue::BulkString(Some(bytes)) => match String::from_utf8(bytes.clone()) {
            Ok(value) => Ok(RedisValueContent::String { value }),
            Err(_) => Ok(RedisValueContent::Binary { bytes }),
        },
        other => Err(BackendError::command_failed(
            format!("Redis command `{command}` returned an unexpected scalar payload."),
            Some(format!("{other:?}")),
        )),
    }
}

fn parse_content_array(
    response: RespValue,
    command: &str,
) -> Result<Vec<RedisValueContent>, BackendError> {
    let RespValue::Array(Some(items)) = response else {
        return Err(BackendError::command_failed(
            format!("Redis command `{command}` returned an unexpected response."),
            Some(format!("{response:?}")),
        ));
    };

    let mut values = Vec::with_capacity(items.len());
    for item in items {
        values.push(parse_binary_safe_value(item, command)?);
    }
    Ok(values)
}

fn parse_field_value_entries(
    response: RespValue,
    command: &str,
) -> Result<Vec<RedisFieldValueEntry>, BackendError> {
    let RespValue::Array(Some(items)) = response else {
        return Err(BackendError::command_failed(
            format!("Redis command `{command}` returned an unexpected response."),
            Some(format!("{response:?}")),
        ));
    };

    if items.len() % 2 != 0 {
        return Err(BackendError::command_failed(
            format!("Redis command `{command}` returned an uneven field/value list."),
            Some(format!("received {} items", items.len())),
        ));
    }

    let mut entries = Vec::with_capacity(items.len() / 2);
    let mut items = items.into_iter();
    while let Some(field) = items.next() {
        let value = items.next().expect("field/value pairs must stay even");
        entries.push(RedisFieldValueEntry {
            field: parse_binary_safe_value(field, command)?,
            value: parse_binary_safe_value(value, command)?,
        });
    }

    Ok(entries)
}

fn parse_sorted_set_entries(
    response: RespValue,
    command: &str,
) -> Result<Vec<RedisSortedSetEntry>, BackendError> {
    let RespValue::Array(Some(items)) = response else {
        return Err(BackendError::command_failed(
            format!("Redis command `{command}` returned an unexpected response."),
            Some(format!("{response:?}")),
        ));
    };

    if items.len() % 2 != 0 {
        return Err(BackendError::command_failed(
            format!("Redis command `{command}` returned an uneven member/score list."),
            Some(format!("received {} items", items.len())),
        ));
    }

    let mut entries = Vec::with_capacity(items.len() / 2);
    let mut items = items.into_iter();
    while let Some(member) = items.next() {
        let score = items.next().expect("member/score pairs must stay even");
        entries.push(RedisSortedSetEntry {
            member: parse_binary_safe_value(member, command)?,
            score: parse_resp_string(score).map_err(|error| {
                BackendError::command_failed(
                    format!("Redis command `{command}` returned a non-string score."),
                    error.detail,
                )
            })?,
        });
    }

    Ok(entries)
}

fn parse_stream_entries(
    response: RespValue,
    command: &str,
) -> Result<Vec<RedisStreamEntry>, BackendError> {
    let RespValue::Array(Some(items)) = response else {
        return Err(BackendError::command_failed(
            format!("Redis command `{command}` returned an unexpected response."),
            Some(format!("{response:?}")),
        ));
    };

    let mut entries = Vec::with_capacity(items.len());
    for item in items {
        let RespValue::Array(Some(parts)) = item else {
            return Err(BackendError::command_failed(
                format!("Redis command `{command}` returned an unexpected stream entry."),
                Some(format!("{item:?}")),
            ));
        };

        if parts.len() != 2 {
            return Err(BackendError::command_failed(
                format!("Redis command `{command}` returned an unexpected stream entry shape."),
                Some(format!("received {} items", parts.len())),
            ));
        }

        let mut parts = parts.into_iter();
        let id = parse_resp_string(parts.next().expect("stream id"))?;
        let fields = parse_field_value_entries(parts.next().expect("stream field list"), command)?;
        entries.push(RedisStreamEntry { id, fields });
    }

    Ok(entries)
}

fn parse_pttl(value: RespValue) -> Result<RedisKeyTtl, BackendError> {
    match value {
        RespValue::Integer(-2) => Ok(RedisKeyTtl::Missing),
        RespValue::Integer(-1) => Ok(RedisKeyTtl::Persistent),
        RespValue::Integer(value) if value >= 0 => Ok(RedisKeyTtl::ExpiresInMillis { value }),
        RespValue::Integer(value) => Err(BackendError::command_failed(
            "Redis command `PTTL` returned an unexpected TTL value.",
            Some(value.to_string()),
        )),
        other => Err(BackendError::command_failed(
            "Redis command `PTTL` returned an unexpected response.",
            Some(format!("{other:?}")),
        )),
    }
}

fn validate_scan_page_size(page_size: u32) -> Result<u32, BackendError> {
    if page_size == 0 || page_size > MAX_SCAN_PAGE_SIZE {
        return Err(BackendError::invalid_connection_config(format!(
            "Redis key page size must be between 1 and {MAX_SCAN_PAGE_SIZE}."
        )));
    }

    Ok(page_size)
}

fn normalize_scan_cursor(cursor: &str) -> Result<String, BackendError> {
    let trimmed = cursor.trim();
    if trimmed.is_empty() {
        return Err(BackendError::invalid_connection_config(
            "Redis scan cursor must not be empty.",
        ));
    }

    if !trimmed.bytes().all(|byte| byte.is_ascii_digit()) {
        return Err(BackendError::invalid_connection_config(
            "Redis scan cursor must be an unsigned integer string.",
        ));
    }

    Ok(trimmed.to_string())
}

fn normalize_scan_pattern(pattern: Option<String>) -> Option<String> {
    pattern.and_then(|value| {
        let trimmed = value.trim();
        if trimmed.is_empty() {
            None
        } else {
            Some(trimmed.to_string())
        }
    })
}

fn validate_key_name(key: &str) -> Result<(), BackendError> {
    if key.trim().is_empty() {
        return Err(BackendError::invalid_connection_config(
            "Redis key name must not be empty.",
        ));
    }

    Ok(())
}

fn validate_optional_ttl_millis(ttl_millis: Option<u64>) -> Result<(), BackendError> {
    if matches!(ttl_millis, Some(0)) {
        return Err(BackendError::invalid_connection_config(
            "Redis key creation TTL must be greater than zero when provided.",
        ));
    }

    Ok(())
}

fn expect_ok(value: RespValue) -> Result<(), RespValue> {
    match value {
        RespValue::SimpleString(value) if value == "OK" => Ok(()),
        other => Err(other),
    }
}

fn expect_integer_response(
    value: RespValue,
    command: &str,
    expected: i64,
) -> Result<(), BackendError> {
    match value {
        RespValue::Integer(value) if value == expected => Ok(()),
        RespValue::Integer(value) => Err(BackendError::command_failed(
            format!("Redis command `{command}` returned an unexpected integer response."),
            Some(value.to_string()),
        )),
        other => Err(BackendError::command_failed(
            format!("Redis command `{command}` returned an unexpected response."),
            Some(format!("{other:?}")),
        )),
    }
}

fn expect_set_write_applied(value: RespValue) -> Result<(), BackendError> {
    match value {
        RespValue::SimpleString(value) if value == "OK" => Ok(()),
        RespValue::BulkString(None) => Err(BackendError::command_failed(
            "Redis string write did not apply because the key changed during the update.",
            None,
        )),
        RespValue::Error(message) => Err(BackendError::command_failed(
            "Redis string write failed.",
            Some(message),
        )),
        other => Err(BackendError::command_failed(
            "Redis string write returned an unexpected response.",
            Some(format!("{other:?}")),
        )),
    }
}

fn expect_set_create_applied(value: RespValue, key_name: &str) -> Result<(), BackendError> {
    match value {
        RespValue::SimpleString(value) if value == "OK" => Ok(()),
        RespValue::BulkString(None) => Err(BackendError::already_exists(
            "Redis key creation does not overwrite an existing key.",
            Some(format!("key `{key_name}` already exists")),
        )),
        RespValue::Error(message) => Err(BackendError::command_failed(
            "Redis string key creation failed.",
            Some(message),
        )),
        other => Err(BackendError::command_failed(
            "Redis string key creation returned an unexpected response.",
            Some(format!("{other:?}")),
        )),
    }
}

fn map_auth_error(error: RespValue) -> BackendError {
    match error {
        RespValue::Error(message) => BackendError::authentication_failed(
            "Redis authentication failed. Check username and password.",
            Some(message),
        ),
        other => BackendError::authentication_failed(
            "Redis authentication failed during handshake.",
            Some(format!("{other:?}")),
        ),
    }
}

fn map_command_error(command: &str, error: RespValue) -> BackendError {
    match error {
        RespValue::Error(message) => BackendError::command_failed(
            format!("Redis command `{command}` failed."),
            Some(message),
        ),
        other => BackendError::command_failed(
            format!("Redis command `{command}` returned an unexpected response."),
            Some(format!("{other:?}")),
        ),
    }
}

fn execute_transaction(
    connection: &mut RedisConnection,
    commands: &[Vec<String>],
) -> Result<Vec<RespValue>, BackendError> {
    let multi_response = connection.execute(["MULTI"])?;
    expect_ok(multi_response).map_err(|error| map_command_error("MULTI", error))?;

    for command in commands {
        let queued_response = connection.execute_owned(command.clone())?;
        match queued_response {
            RespValue::SimpleString(value) if value == "QUEUED" => {}
            other => {
                let command_name = command.first().map(String::as_str).unwrap_or("UNKNOWN");
                return Err(map_command_error(command_name, other));
            }
        }
    }

    let exec_response = connection.execute(["EXEC"])?;
    let RespValue::Array(Some(results)) = exec_response else {
        return Err(BackendError::command_failed(
            "Redis command `EXEC` returned an unexpected response.",
            Some(format!("{exec_response:?}")),
        ));
    };

    if results.len() != commands.len() {
        return Err(BackendError::command_failed(
            "Redis command `EXEC` returned an unexpected number of results.",
            Some(format!(
                "expected {}, got {}",
                commands.len(),
                results.len()
            )),
        ));
    }

    Ok(results)
}

#[derive(Debug, Clone)]
enum RespValue {
    SimpleString(String),
    Error(String),
    Integer(i64),
    BulkString(Option<Vec<u8>>),
    Array(Option<Vec<RespValue>>),
}

struct RedisConnection {
    reader: BufReader<TcpStream>,
    writer: TcpStream,
}

impl RedisConnection {
    fn execute<'a>(
        &mut self,
        parts: impl IntoIterator<Item = &'a str>,
    ) -> Result<RespValue, BackendError> {
        self.execute_owned(parts.into_iter().map(str::to_string).collect())
    }

    fn execute_owned(&mut self, parts: Vec<String>) -> Result<RespValue, BackendError> {
        self.write_command(&parts)?;
        self.read_response()
    }

    fn write_command(&mut self, parts: &[String]) -> Result<(), BackendError> {
        let mut payload = format!("*{}\r\n", parts.len()).into_bytes();
        for part in parts {
            let bytes = part.as_bytes();
            payload.extend(format!("${}\r\n", bytes.len()).into_bytes());
            payload.extend(bytes);
            payload.extend(b"\r\n");
        }

        self.writer.write_all(&payload).map_err(|error| {
            BackendError::connection_failed(
                "Failed to write Redis command to the socket.",
                Some(error.to_string()),
            )
        })?;
        self.writer.flush().map_err(|error| {
            BackendError::connection_failed(
                "Failed to flush Redis command to the socket.",
                Some(error.to_string()),
            )
        })?;

        Ok(())
    }

    fn read_response(&mut self) -> Result<RespValue, BackendError> {
        read_resp_value(&mut self.reader).map_err(|error| {
            BackendError::connection_failed(
                "Failed to read Redis server response.",
                Some(error.to_string()),
            )
        })
    }
}

fn read_resp_value(reader: &mut BufReader<TcpStream>) -> std::io::Result<RespValue> {
    let mut marker = [0_u8; 1];
    reader.read_exact(&mut marker)?;

    match marker[0] {
        b'+' => Ok(RespValue::SimpleString(read_line(reader)?)),
        b'-' => Ok(RespValue::Error(read_line(reader)?)),
        b':' => Ok(RespValue::Integer(
            read_line(reader)?
                .parse::<i64>()
                .expect("RESP integer should parse"),
        )),
        b'$' => {
            let len = read_line(reader)?
                .parse::<isize>()
                .expect("RESP bulk len should parse");
            if len < 0 {
                return Ok(RespValue::BulkString(None));
            }

            let mut bytes = vec![0_u8; len as usize];
            reader.read_exact(&mut bytes)?;
            let mut crlf = [0_u8; 2];
            reader.read_exact(&mut crlf)?;
            Ok(RespValue::BulkString(Some(bytes)))
        }
        b'*' => {
            let len = read_line(reader)?
                .parse::<isize>()
                .expect("RESP array len should parse");
            if len < 0 {
                return Ok(RespValue::Array(None));
            }

            let mut values = Vec::with_capacity(len as usize);
            for _ in 0..len {
                values.push(read_resp_value(reader)?);
            }
            Ok(RespValue::Array(Some(values)))
        }
        other => Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            format!("Unsupported RESP marker: {}", other as char),
        )),
    }
}

fn read_line(reader: &mut BufReader<TcpStream>) -> std::io::Result<String> {
    let mut line = String::new();
    reader.read_line(&mut line)?;
    if !line.ends_with("\r\n") {
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "RESP line must end with CRLF",
        ));
    }
    line.truncate(line.len() - 2);
    Ok(line)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::{net::TcpListener, thread};

    use crate::{BackendErrorCode, ConnectionTarget};

    #[test]
    fn rejects_negative_database_before_network_access() {
        let request = RedisConnectionRequest {
            target: ConnectionTarget {
                database: -1,
                ..ConnectionTarget::default()
            },
            password: None,
        };

        let error = test_connection(request).expect_err("negative db should fail");
        assert_eq!(error.code, BackendErrorCode::InvalidConnectionConfig);
    }

    #[test]
    fn authenticates_selects_database_and_pings() {
        let server = MockRedisServer::spawn(vec![
            ServerStep::expect(["AUTH", "default-user", "s3cret"], Response::simple("OK")),
            ServerStep::expect(["SELECT", "2"], Response::simple("OK")),
            ServerStep::expect(["PING"], Response::simple("PONG")),
        ]);

        let result = test_connection(RedisConnectionRequest {
            target: ConnectionTarget {
                host: "127.0.0.1".to_string(),
                port: server.port(),
                database: 2,
                username: Some("default-user".to_string()),
                tls_mode: TlsMode::Disabled,
            },
            password: Some("s3cret".to_string()),
        })
        .expect("mock redis should accept connection");

        assert_eq!(result.selected_database, 2);
        assert_eq!(result.authenticated_as.as_deref(), Some("default-user"));
        assert_eq!(result.round_trip_status, "PONG");
    }

    #[test]
    fn maps_authentication_failures_to_stable_error_code() {
        let server = MockRedisServer::spawn(vec![ServerStep::expect(
            ["AUTH", "backend-user", "bad-password"],
            Response::error("WRONGPASS invalid username-password pair"),
        )]);

        let error = test_connection(RedisConnectionRequest {
            target: ConnectionTarget {
                host: "127.0.0.1".to_string(),
                port: server.port(),
                database: 0,
                username: Some("backend-user".to_string()),
                tls_mode: TlsMode::Disabled,
            },
            password: Some("bad-password".to_string()),
        })
        .expect_err("auth should fail");

        assert_eq!(error.code, BackendErrorCode::AuthenticationFailed);
    }

    #[test]
    fn rejects_tls_modes_until_tls_transport_is_added() {
        let error = test_connection(RedisConnectionRequest {
            target: ConnectionTarget {
                tls_mode: TlsMode::Required,
                ..ConnectionTarget::default()
            },
            password: None,
        })
        .expect_err("tls should be unsupported for now");

        assert_eq!(error.code, BackendErrorCode::UnsupportedTlsMode);
    }

    #[test]
    fn executes_basic_commands_and_preserves_array_responses() {
        let server = MockRedisServer::spawn(vec![
            ServerStep::expect(["SELECT", "4"], Response::simple("OK")),
            ServerStep::expect(
                ["LRANGE", "recent", "0", "-1"],
                Response::array([
                    Response::bulk("alpha"),
                    Response::bulk("beta"),
                    Response::bulk("gamma"),
                ]),
            ),
        ]);

        let result = execute_command(RedisCommandRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: server.port(),
                    database: 4,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            command: "LRANGE".to_string(),
            arguments: vec!["recent".to_string(), "0".to_string(), "-1".to_string()],
        })
        .expect("command should succeed");

        assert_eq!(result.database, 4);
        assert_eq!(
            result.response,
            RedisResponse::Array {
                items: vec![
                    RedisResponse::String {
                        value: "alpha".to_string()
                    },
                    RedisResponse::String {
                        value: "beta".to_string()
                    },
                    RedisResponse::String {
                        value: "gamma".to_string()
                    },
                ],
            }
        );
    }

    #[test]
    fn browses_keys_with_pattern_cursor_and_metadata() {
        let server = MockRedisServer::spawn(vec![
            ServerStep::expect(["SELECT", "4"], Response::simple("OK")),
            ServerStep::expect(
                ["SCAN", "0", "MATCH", "session:*", "COUNT", "2"],
                Response::array([
                    Response::bulk("17"),
                    Response::array([Response::bulk("session:42"), Response::bulk("session:99")]),
                ]),
            ),
            ServerStep::expect(["TYPE", "session:42"], Response::simple("hash")),
            ServerStep::expect(["PTTL", "session:42"], Response::integer(780_000)),
            ServerStep::expect(["TYPE", "session:99"], Response::simple("string")),
            ServerStep::expect(["PTTL", "session:99"], Response::integer(-1)),
        ]);

        let result = browse_keys(RedisKeyBrowseRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: server.port(),
                    database: 4,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            cursor: "0".to_string(),
            pattern: Some("session:*".to_string()),
            page_size: 2,
        })
        .expect("browse should succeed");

        assert_eq!(result.database, 4);
        assert_eq!(result.submitted_cursor, "0");
        assert_eq!(result.next_cursor, "17");
        assert!(result.has_more);
        assert_eq!(result.pattern.as_deref(), Some("session:*"));
        assert_eq!(
            result.keys,
            vec![
                RedisKeyMetadata {
                    name: "session:42".to_string(),
                    database: 4,
                    exists: true,
                    key_type: "hash".to_string(),
                    ttl: RedisKeyTtl::ExpiresInMillis { value: 780_000 },
                },
                RedisKeyMetadata {
                    name: "session:99".to_string(),
                    database: 4,
                    exists: true,
                    key_type: "string".to_string(),
                    ttl: RedisKeyTtl::Persistent,
                },
            ]
        );
    }

    #[test]
    fn inspect_key_reports_missing_key_without_failing() {
        let server = MockRedisServer::spawn(vec![ServerStep::expect(
            ["TYPE", "ghost:key"],
            Response::simple("none"),
        )]);

        let result = inspect_key(RedisKeyMetadataRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "ghost:key".to_string(),
        })
        .expect("missing key should map to metadata");

        assert_eq!(
            result,
            RedisKeyMetadata {
                name: "ghost:key".to_string(),
                database: 0,
                exists: false,
                key_type: "none".to_string(),
                ttl: RedisKeyTtl::Missing,
            }
        );
    }

    #[test]
    fn reads_string_value_with_write_capability() {
        let server = MockRedisServer::spawn(vec![
            ServerStep::expect(["TYPE", "cache:homepage"], Response::simple("string")),
            ServerStep::expect(["PTTL", "cache:homepage"], Response::integer(48_000)),
            ServerStep::expect(
                ["GET", "cache:homepage"],
                Response::bulk("{\"hero\":\"Redis GUI\"}"),
            ),
        ]);

        let result = read_value(RedisValueReadRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "cache:homepage".to_string(),
        })
        .expect("string read should succeed");

        assert_eq!(
            result.write_capability,
            RedisValueWriteCapability::ReplaceString
        );
        assert!(result.ttl_write_supported);
        assert_eq!(
            result.data,
            RedisValueData::String {
                value: RedisValueContent::String {
                    value: "{\"hero\":\"Redis GUI\"}".to_string(),
                },
            }
        );
    }

    #[test]
    fn reads_hash_value_as_field_value_entries() {
        let server = MockRedisServer::spawn(vec![
            ServerStep::expect(["TYPE", "session:42"], Response::simple("hash")),
            ServerStep::expect(["PTTL", "session:42"], Response::integer(780_000)),
            ServerStep::expect(
                ["HGETALL", "session:42"],
                Response::array([
                    Response::bulk("userId"),
                    Response::bulk("42"),
                    Response::bulk("role"),
                    Response::bulk("editor"),
                ]),
            ),
        ]);

        let result = read_value(RedisValueReadRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "session:42".to_string(),
        })
        .expect("hash read should succeed");

        assert_eq!(result.write_capability, RedisValueWriteCapability::None);
        assert_eq!(
            result.data,
            RedisValueData::Hash {
                entries: vec![
                    RedisFieldValueEntry {
                        field: RedisValueContent::String {
                            value: "userId".to_string(),
                        },
                        value: RedisValueContent::String {
                            value: "42".to_string(),
                        },
                    },
                    RedisFieldValueEntry {
                        field: RedisValueContent::String {
                            value: "role".to_string(),
                        },
                        value: RedisValueContent::String {
                            value: "editor".to_string(),
                        },
                    },
                ],
            }
        );
    }

    #[test]
    fn reads_list_set_zset_and_stream_values() {
        let list_server = MockRedisServer::spawn(vec![
            ServerStep::expect(["TYPE", "jobs:failed"], Response::simple("list")),
            ServerStep::expect(["PTTL", "jobs:failed"], Response::integer(-1)),
            ServerStep::expect(
                ["LRANGE", "jobs:failed", "0", "-1"],
                Response::array([
                    Response::bulk("payload-1"),
                    Response::bulk("payload-2"),
                    Response::bulk("payload-3"),
                ]),
            ),
        ]);
        let list_result = read_value(RedisValueReadRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: list_server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "jobs:failed".to_string(),
        })
        .expect("list read should succeed");
        assert_eq!(
            list_result.data,
            RedisValueData::List {
                items: vec![
                    RedisValueContent::String {
                        value: "payload-1".to_string(),
                    },
                    RedisValueContent::String {
                        value: "payload-2".to_string(),
                    },
                    RedisValueContent::String {
                        value: "payload-3".to_string(),
                    },
                ],
            }
        );

        let set_server = MockRedisServer::spawn(vec![
            ServerStep::expect(["TYPE", "feature:enabled"], Response::simple("set")),
            ServerStep::expect(["PTTL", "feature:enabled"], Response::integer(3_600_000)),
            ServerStep::expect(
                ["SMEMBERS", "feature:enabled"],
                Response::array([Response::bulk("search_v2"), Response::bulk("audit_log")]),
            ),
        ]);
        let set_result = read_value(RedisValueReadRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: set_server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "feature:enabled".to_string(),
        })
        .expect("set read should succeed");
        assert_eq!(
            set_result.data,
            RedisValueData::Set {
                members: vec![
                    RedisValueContent::String {
                        value: "search_v2".to_string(),
                    },
                    RedisValueContent::String {
                        value: "audit_log".to_string(),
                    },
                ],
            }
        );

        let zset_server = MockRedisServer::spawn(vec![
            ServerStep::expect(["TYPE", "leaderboard"], Response::simple("zset")),
            ServerStep::expect(["PTTL", "leaderboard"], Response::integer(-1)),
            ServerStep::expect(
                ["ZRANGE", "leaderboard", "0", "-1", "WITHSCORES"],
                Response::array([
                    Response::bulk("alice"),
                    Response::bulk("10"),
                    Response::bulk("bob"),
                    Response::bulk("7.5"),
                ]),
            ),
        ]);
        let zset_result = read_value(RedisValueReadRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: zset_server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "leaderboard".to_string(),
        })
        .expect("sorted set read should succeed");
        assert_eq!(
            zset_result.data,
            RedisValueData::SortedSet {
                entries: vec![
                    RedisSortedSetEntry {
                        member: RedisValueContent::String {
                            value: "alice".to_string(),
                        },
                        score: "10".to_string(),
                    },
                    RedisSortedSetEntry {
                        member: RedisValueContent::String {
                            value: "bob".to_string(),
                        },
                        score: "7.5".to_string(),
                    },
                ],
            }
        );

        let stream_server = MockRedisServer::spawn(vec![
            ServerStep::expect(["TYPE", "audit:stream"], Response::simple("stream")),
            ServerStep::expect(["PTTL", "audit:stream"], Response::integer(-1)),
            ServerStep::expect(
                ["XRANGE", "audit:stream", "-", "+"],
                Response::array([Response::array([
                    Response::bulk("1711537519000-0"),
                    Response::array([
                        Response::bulk("event"),
                        Response::bulk("login"),
                        Response::bulk("userId"),
                        Response::bulk("42"),
                    ]),
                ])]),
            ),
        ]);
        let stream_result = read_value(RedisValueReadRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: stream_server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "audit:stream".to_string(),
        })
        .expect("stream read should succeed");
        assert_eq!(
            stream_result.data,
            RedisValueData::Stream {
                entries: vec![RedisStreamEntry {
                    id: "1711537519000-0".to_string(),
                    fields: vec![
                        RedisFieldValueEntry {
                            field: RedisValueContent::String {
                                value: "event".to_string(),
                            },
                            value: RedisValueContent::String {
                                value: "login".to_string(),
                            },
                        },
                        RedisFieldValueEntry {
                            field: RedisValueContent::String {
                                value: "userId".to_string(),
                            },
                            value: RedisValueContent::String {
                                value: "42".to_string(),
                            },
                        },
                    ],
                }],
            }
        );
    }

    #[test]
    fn writes_string_value_without_dropping_existing_ttl() {
        let server = MockRedisServer::spawn(vec![
            ServerStep::expect(["TYPE", "cache:homepage"], Response::simple("string")),
            ServerStep::expect(["PTTL", "cache:homepage"], Response::integer(48_000)),
            ServerStep::expect(["MULTI"], Response::simple("OK")),
            ServerStep::expect(
                ["SET", "cache:homepage", "{\"hero\":\"Updated\"}", "XX"],
                Response::simple("QUEUED"),
            ),
            ServerStep::expect(
                ["PEXPIRE", "cache:homepage", "48000"],
                Response::simple("QUEUED"),
            ),
            ServerStep::expect(
                ["EXEC"],
                Response::array([Response::simple("OK"), Response::integer(1)]),
            ),
            ServerStep::expect(["TYPE", "cache:homepage"], Response::simple("string")),
            ServerStep::expect(["PTTL", "cache:homepage"], Response::integer(47_000)),
            ServerStep::expect(
                ["GET", "cache:homepage"],
                Response::bulk("{\"hero\":\"Updated\"}"),
            ),
        ]);

        let result = write_value(RedisValueWriteRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "cache:homepage".to_string(),
            value: RedisValueWriteInput::String {
                value: "{\"hero\":\"Updated\"}".to_string(),
            },
        })
        .expect("string write should succeed");

        assert_eq!(
            result.metadata.ttl,
            RedisKeyTtl::ExpiresInMillis { value: 47_000 }
        );
        assert_eq!(
            result.write_capability,
            RedisValueWriteCapability::ReplaceString
        );
    }

    #[test]
    fn rejects_writes_for_non_string_keys_before_mutation() {
        let server = MockRedisServer::spawn(vec![
            ServerStep::expect(["TYPE", "session:42"], Response::simple("hash")),
            ServerStep::expect(["PTTL", "session:42"], Response::integer(-1)),
        ]);

        let error = write_value(RedisValueWriteRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "session:42".to_string(),
            value: RedisValueWriteInput::String {
                value: "unsafe".to_string(),
            },
        })
        .expect_err("hash writes should be blocked");

        assert_eq!(error.code, BackendErrorCode::UnsupportedOperation);
    }

    #[test]
    fn creates_string_value_without_ttl() {
        let server = MockRedisServer::spawn(vec![
            ServerStep::expect(
                ["SET", "draft:key", "hello world", "NX"],
                Response::simple("OK"),
            ),
            ServerStep::expect(["TYPE", "draft:key"], Response::simple("string")),
            ServerStep::expect(["PTTL", "draft:key"], Response::integer(-1)),
            ServerStep::expect(["GET", "draft:key"], Response::bulk("hello world")),
        ]);

        let result = create_value(RedisValueCreateRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "draft:key".to_string(),
            value: RedisValueCreateInput::String {
                value: "hello world".to_string(),
            },
            ttl_millis: None,
        })
        .expect("key create should succeed");

        assert_eq!(result.record.metadata.name, "draft:key");
        assert_eq!(result.record.metadata.ttl, RedisKeyTtl::Persistent);
        assert_eq!(
            result.record.write_capability,
            RedisValueWriteCapability::ReplaceString
        );
        assert_eq!(
            result.record.data,
            RedisValueData::String {
                value: RedisValueContent::String {
                    value: "hello world".to_string(),
                },
            }
        );
    }

    #[test]
    fn creates_string_value_with_ttl() {
        let server = MockRedisServer::spawn(vec![
            ServerStep::expect(
                ["SET", "draft:ttl", "expires soon", "NX", "PX", "60000"],
                Response::simple("OK"),
            ),
            ServerStep::expect(["TYPE", "draft:ttl"], Response::simple("string")),
            ServerStep::expect(["PTTL", "draft:ttl"], Response::integer(59_000)),
            ServerStep::expect(["GET", "draft:ttl"], Response::bulk("expires soon")),
        ]);

        let result = create_value(RedisValueCreateRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "draft:ttl".to_string(),
            value: RedisValueCreateInput::String {
                value: "expires soon".to_string(),
            },
            ttl_millis: Some(60_000),
        })
        .expect("ttl create should succeed");

        assert_eq!(
            result.record.metadata.ttl,
            RedisKeyTtl::ExpiresInMillis { value: 59_000 }
        );
    }

    #[test]
    fn rejects_duplicate_key_creation_with_stable_conflict_error() {
        let server = MockRedisServer::spawn(vec![ServerStep::expect(
            ["SET", "draft:key", "hello world", "NX"],
            Response::null_bulk(),
        )]);

        let error = create_value(RedisValueCreateRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "draft:key".to_string(),
            value: RedisValueCreateInput::String {
                value: "hello world".to_string(),
            },
            ttl_millis: None,
        })
        .expect_err("duplicate create should fail");

        assert_eq!(error.code, BackendErrorCode::AlreadyExists);
    }

    #[test]
    fn rejects_zero_ttl_for_created_key_before_network_access() {
        let error = create_value(RedisValueCreateRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget::default(),
                password: None,
            },
            key: "draft:key".to_string(),
            value: RedisValueCreateInput::String {
                value: "hello world".to_string(),
            },
            ttl_millis: Some(0),
        })
        .expect_err("zero ttl should fail");

        assert_eq!(error.code, BackendErrorCode::InvalidConnectionConfig);
    }

    #[test]
    fn updates_ttl_for_existing_keys() {
        let expiring_server = MockRedisServer::spawn(vec![
            ServerStep::expect(["TYPE", "cache:homepage"], Response::simple("string")),
            ServerStep::expect(["PTTL", "cache:homepage"], Response::integer(-1)),
            ServerStep::expect(["PEXPIRE", "cache:homepage", "60000"], Response::integer(1)),
            ServerStep::expect(["TYPE", "cache:homepage"], Response::simple("string")),
            ServerStep::expect(["PTTL", "cache:homepage"], Response::integer(60_000)),
        ]);

        let ttl_result = update_key_ttl(RedisKeyTtlUpdateRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: expiring_server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "cache:homepage".to_string(),
            operation: RedisKeyTtlUpdate::ExpiresInMillis { value: 60_000 },
        })
        .expect("ttl set should succeed");

        assert_eq!(
            ttl_result.metadata.ttl,
            RedisKeyTtl::ExpiresInMillis { value: 60_000 }
        );

        let persist_server = MockRedisServer::spawn(vec![
            ServerStep::expect(["TYPE", "session:42"], Response::simple("hash")),
            ServerStep::expect(["PTTL", "session:42"], Response::integer(780_000)),
            ServerStep::expect(["PERSIST", "session:42"], Response::integer(1)),
            ServerStep::expect(["TYPE", "session:42"], Response::simple("hash")),
            ServerStep::expect(["PTTL", "session:42"], Response::integer(-1)),
        ]);

        let persist_result = update_key_ttl(RedisKeyTtlUpdateRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget {
                    host: "127.0.0.1".to_string(),
                    port: persist_server.port(),
                    database: 0,
                    username: None,
                    tls_mode: TlsMode::Disabled,
                },
                password: None,
            },
            key: "session:42".to_string(),
            operation: RedisKeyTtlUpdate::Persist,
        })
        .expect("ttl removal should succeed");

        assert_eq!(persist_result.metadata.ttl, RedisKeyTtl::Persistent);
    }

    #[test]
    fn rejects_invalid_scan_cursor_before_network_access() {
        let error = browse_keys(RedisKeyBrowseRequest {
            connection: RedisConnectionRequest {
                target: ConnectionTarget::default(),
                password: None,
            },
            cursor: "abc".to_string(),
            pattern: None,
            page_size: 100,
        })
        .expect_err("invalid scan cursor should fail");

        assert_eq!(error.code, BackendErrorCode::InvalidConnectionConfig);
    }

    struct MockRedisServer {
        port: u16,
        join_handle: Option<thread::JoinHandle<()>>,
    }

    impl MockRedisServer {
        fn spawn(steps: Vec<ServerStep>) -> Self {
            let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind mock redis");
            let port = listener.local_addr().expect("mock addr").port();
            let join_handle = thread::spawn(move || {
                let (stream, _) = listener.accept().expect("accept client");
                serve_stream(stream, steps);
            });

            Self {
                port,
                join_handle: Some(join_handle),
            }
        }

        fn port(&self) -> u16 {
            self.port
        }
    }

    impl Drop for MockRedisServer {
        fn drop(&mut self) {
            if let Some(join_handle) = self.join_handle.take() {
                join_handle.join().expect("join mock redis thread");
            }
        }
    }

    struct ServerStep {
        expected_parts: Vec<String>,
        response: Response,
    }

    impl ServerStep {
        fn expect<const N: usize>(parts: [&str; N], response: Response) -> Self {
            Self {
                expected_parts: parts.into_iter().map(str::to_string).collect(),
                response,
            }
        }
    }

    enum Response {
        Simple(String),
        Error(String),
        Integer(i64),
        Bulk(Vec<u8>),
        NullBulk,
        Array(Vec<Response>),
    }

    impl Response {
        fn simple(value: &str) -> Self {
            Self::Simple(value.to_string())
        }

        fn error(value: &str) -> Self {
            Self::Error(value.to_string())
        }

        fn integer(value: i64) -> Self {
            Self::Integer(value)
        }

        fn bulk(value: &str) -> Self {
            Self::Bulk(value.as_bytes().to_vec())
        }

        fn null_bulk() -> Self {
            Self::NullBulk
        }

        fn array<const N: usize>(items: [Response; N]) -> Self {
            Self::Array(items.into())
        }

        fn encode(self) -> Vec<u8> {
            match self {
                Self::Simple(value) => format!("+{value}\r\n").into_bytes(),
                Self::Error(value) => format!("-{value}\r\n").into_bytes(),
                Self::Integer(value) => format!(":{value}\r\n").into_bytes(),
                Self::Bulk(bytes) => {
                    let mut encoded = format!("${}\r\n", bytes.len()).into_bytes();
                    encoded.extend(bytes);
                    encoded.extend(b"\r\n");
                    encoded
                }
                Self::NullBulk => b"$-1\r\n".to_vec(),
                Self::Array(items) => {
                    let mut encoded = format!("*{}\r\n", items.len()).into_bytes();
                    for item in items {
                        encoded.extend(item.encode());
                    }
                    encoded
                }
            }
        }
    }

    fn serve_stream(mut stream: TcpStream, steps: Vec<ServerStep>) {
        stream
            .set_read_timeout(Some(Duration::from_secs(3)))
            .expect("set read timeout");
        stream
            .set_write_timeout(Some(Duration::from_secs(3)))
            .expect("set write timeout");

        let cloned = stream.try_clone().expect("clone stream");
        let mut reader = BufReader::new(cloned);

        for step in steps {
            let command = read_resp_array(&mut reader).expect("read redis command");
            assert_eq!(command, step.expected_parts);
            stream
                .write_all(&step.response.encode())
                .expect("write redis response");
            stream.flush().expect("flush redis response");
        }
    }

    fn read_resp_array(reader: &mut BufReader<TcpStream>) -> std::io::Result<Vec<String>> {
        match read_resp_value(reader)? {
            RespValue::Array(Some(values)) => Ok(values
                .into_iter()
                .map(|value| match value {
                    RespValue::SimpleString(value) => value,
                    RespValue::BulkString(Some(bytes)) => {
                        String::from_utf8(bytes).expect("utf8 bulk")
                    }
                    other => panic!("unexpected request value {other:?}"),
                })
                .collect()),
            other => panic!("expected array request, got {other:?}"),
        }
    }
}