ruvio-client 0.2.0

use std::io::{BufReader, Write};

use std::net::{IpAddr, SocketAddr, TcpStream, ToSocketAddrs};

use std::time::{Duration, SystemTime, UNIX_EPOCH};

use crate::codec::{encode_strings, read_value};

use crate::error::Error;
use crate::value::RespValue;

/// How `SET` treats a key that already exists or is missing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum SetMode {
    /// Write the key either way.
    #[default]
    Always,
    /// Write only when the key is absent (`NX`).
    IfNotExists,
    /// Write only when the key exists (`XX`).
    IfExists,
}

/// Where to connect, and an optional password sent as `AUTH`.
#[derive(Debug, Clone)]
pub struct ClientOptions {
    /// Server host. Defaults to loopback.
    pub host: String,
    /// RESP port. Defaults to 6379.
    pub port: u16,
    /// ACL user. When set together with [`Self::password`], the client sends `AUTH user password`.
    pub username: Option<String>,
    /// When set, the client sends `AUTH` before [`Client::connect_with`] returns.
    pub password: Option<String>,
    /// Logical database. When not 0, the client sends `SELECT` after `AUTH`.
    pub database: u32,
    /// How long TCP connect may take.
    pub connect_timeout: Duration,
}

impl Default for ClientOptions {
    fn default() -> Self {
        Self {
            host: "127.0.0.1".to_owned(),
            port: 6379,
            username: None,
            password: None,
            database: 0,
            connect_timeout: Duration::from_secs(5),
        }
    }
}

/// One TCP connection to Ruvio.
///
/// The client speaks RESP2 and does not send `CLIENT SETINFO` or `HELLO`.
pub struct Client {
    stream: BufReader<TcpStream>,
    database: u32,
}

impl Client {
    /// Opens a connection and sends nothing until the first command.
    pub fn connect(host: impl Into<String>, port: u16) -> Result<Self, Error> {
        Self::connect_database(host, port, 0)
    }

    /// Opens a connection bound to one logical database. Sends `SELECT` when `database` is not 0.
    ///
    /// The returned client is the database handle: every command on it runs in `database`.
    pub fn connect_database(
        host: impl Into<String>,
        port: u16,
        database: u32,
    ) -> Result<Self, Error> {
        Self::connect_with(ClientOptions {
            host: host.into(),
            port,
            database,
            ..ClientOptions::default()
        })
    }

    /// Opens a connection to an IP address and port. Sends `SELECT` when `database` is not 0.
    pub fn connect_ip(ip: IpAddr, port: u16, database: u32) -> Result<Self, Error> {
        Self::connect_addr(SocketAddr::new(ip, port), database)
    }

    /// Opens a connection to a socket address. Sends `SELECT` when `database` is not 0.
    pub fn connect_addr(address: SocketAddr, database: u32) -> Result<Self, Error> {
        Self::connect_database(address.ip().to_string(), address.port(), database)
    }

    /// Opens a connection. When [`ClientOptions::password`] is set, sends `AUTH` before returning.
    /// When [`ClientOptions::database`] is not 0, sends `SELECT` after `AUTH`.
    pub fn connect_with(options: ClientOptions) -> Result<Self, Error> {
        if options.host.is_empty() {
            return Err(Error::Protocol("host is required".to_owned()));
        }

        let mut last_error = None;

        for address in (options.host.as_str(), options.port).to_socket_addrs()? {
            match TcpStream::connect_timeout(&address, options.connect_timeout) {
                Ok(stream) => {
                    stream.set_nodelay(true)?;

                    let mut client = Client {
                        stream: BufReader::new(stream),
                        database: 0,
                    };

                    if let Some(password) = options.password.clone() {
                        let reply = if let Some(username) = options.username.clone() {
                            client.execute(&["AUTH", &username, &password])
                        } else {
                            client.execute(&["AUTH", &password])
                        };

                        reply?;
                    }

                    if options.database != 0 {
                        client.select(options.database)?;
                    }

                    return Ok(client);
                }

                Err(error) => last_error = Some(error),
            }
        }

        Err(Error::Io(last_error.unwrap_or_else(|| {
            std::io::Error::new(std::io::ErrorKind::NotFound, "no addresses for host")
        })))
    }

    /// Sends `PING` and returns the simple-string reply.
    pub fn ping(&mut self) -> Result<String, Error> {
        self.text(["PING"])
    }

    /// Sends `PING message` and returns the echoed message.
    pub fn ping_message(&mut self, message: &str) -> Result<String, Error> {
        self.text(["PING", message])
    }

    /// The logical database this connection last selected.
    pub fn database(&self) -> u32 {
        self.database
    }

    /// Sends `SELECT`. Indexes run from 0 to the server's `databases` setting minus one (16 by default).
    pub fn select(&mut self, database: u32) -> Result<(), Error> {
        self.run(["SELECT".to_owned(), database.to_string()])?;
        self.database = database;

        Ok(())
    }

    /// Sends `SELECT` and returns the same connection.
    ///
    /// This is not a second TCP socket: every later command on the client uses the new database.
    pub fn into_database(mut self, database: u32) -> Result<Self, Error> {
        self.select(database)?;

        Ok(self)
    }

    /// Sets how long a read may block. `None` waits forever.
    ///
    /// A timed-out read returns [`Error::Io`], and the reply may still arrive later,
    /// so drop the client unless it was only waiting in [`Self::read_message`].
    pub fn set_read_timeout(&mut self, timeout: Option<Duration>) -> Result<(), Error> {
        self.stream.get_ref().set_read_timeout(timeout)?;

        Ok(())
    }

    /// Sends `GET` and returns the raw bulk, or `None` when the key is missing.
    pub fn get(&mut self, key: &str) -> Result<Option<Vec<u8>>, Error> {
        match self.execute(&["GET", key])? {
            RespValue::Null => Ok(None),
            RespValue::Bulk(bytes) => Ok(Some(bytes)),
            other => Err(Error::Protocol(format!("GET returned {other:?}"))),
        }
    }

    /// Sends `GET` and decodes the bulk as UTF-8. `None` when the key is missing.
    pub fn get_string(&mut self, key: &str) -> Result<Option<String>, Error> {
        match self.get(key)? {
            None => Ok(None),
            Some(bytes) => RespValue::Bulk(bytes).as_string(),
        }
    }

    /// Sends `SET`. Returns false when `NX` or `XX` skips the write.
    pub fn set(&mut self, key: &str, value: &str) -> Result<bool, Error> {
        self.set_with(key, value, None, SetMode::Always)
    }

    /// Sends `SET` with an optional millisecond expiry and `NX` or `XX`.
    pub fn set_with(
        &mut self,
        key: &str,
        value: &str,
        expiry: Option<Duration>,
        mode: SetMode,
    ) -> Result<bool, Error> {
        let expiry = expiry.map(|ttl| ["PX".to_owned(), duration_millis(ttl).to_string()]);
        let arguments = set_arguments(key, value, expiry.as_ref().map(|pair| &pair[..]), mode);

        Ok(!matches!(self.run(arguments)?, RespValue::Null))
    }

    /// Sends `SET ... PXAT`: the key expires at `expires_at`.
    pub fn set_expires_at(
        &mut self,
        key: &str,
        value: &str,
        expires_at: SystemTime,
        mode: SetMode,
    ) -> Result<bool, Error> {
        let expiry = ["PXAT".to_owned(), unix_millis(expires_at)?.to_string()];
        let arguments = set_arguments(key, value, Some(&expiry), mode);

        Ok(!matches!(self.run(arguments)?, RespValue::Null))
    }

    /// Sends `SET ... KEEPTTL`: the key keeps its current expiry.
    pub fn set_keep_ttl(&mut self, key: &str, value: &str, mode: SetMode) -> Result<bool, Error> {
        let expiry = ["KEEPTTL".to_owned()];
        let arguments = set_arguments(key, value, Some(&expiry), mode);

        Ok(!matches!(self.run(arguments)?, RespValue::Null))
    }

    /// Sends `SET ... GET` and returns the previous string, or `None` when the key was missing.
    /// With `NX` or `XX` the previous value is returned even when the write is skipped.
    pub fn set_and_get(
        &mut self,
        key: &str,
        value: &str,
        expiry: Option<Duration>,
        mode: SetMode,
    ) -> Result<Option<String>, Error> {
        let expiry = expiry.map(|ttl| ["PX".to_owned(), duration_millis(ttl).to_string()]);
        let mut arguments = set_arguments(key, value, expiry.as_ref().map(|pair| &pair[..]), mode);

        arguments.push("GET".to_owned());
        self.bulk_or_null(arguments)
    }

    /// Sends `INCR`.
    pub fn incr(&mut self, key: &str) -> Result<i64, Error> {
        self.integer(["INCR", key])
    }

    /// Sends `EXPIRE`, or `PEXPIRE` when `ttl` is not whole seconds. False when the key does not exist.
    pub fn expire(&mut self, key: &str, ttl: Duration) -> Result<bool, Error> {
        if ttl.subsec_nanos() == 0 {
            return self.flag([
                "EXPIRE".to_owned(),
                key.to_owned(),
                duration_secs(ttl).to_string(),
            ]);
        }

        self.flag([
            "PEXPIRE".to_owned(),
            key.to_owned(),
            duration_millis(ttl).to_string(),
        ])
    }

    /// Sends `DEL` and returns how many keys were removed.
    pub fn del(&mut self, keys: &[&str]) -> Result<i64, Error> {
        if keys.is_empty() {
            return Err(Error::Protocol("DEL needs at least one key".to_owned()));
        }

        self.integer(join("DEL", keys))
    }

    /// Reads the next reply without sending a command. Use this after a subscribe
    /// for `message`, `pmessage`, and `smessage` pushes.
    pub fn read_message(&mut self) -> Result<RespValue, Error> {
        match read_value(&mut self.stream)? {
            RespValue::Error(message) => Err(Error::Server(message)),
            value => Ok(value),
        }
    }

    /// Sends one command. A RESP error becomes [`Error::Server`] and the connection stays open.
    pub fn execute(&mut self, arguments: &[&str]) -> Result<RespValue, Error> {
        let values = self.execute_many(&[arguments])?;

        match values.into_iter().next() {
            Some(RespValue::Error(message)) => Err(Error::Server(message)),
            Some(value) => Ok(value),
            None => Err(Error::Protocol("missing reply".to_owned())),
        }
    }

    /// Writes every command, then reads one reply each. Error replies stay as [`RespValue::Error`].
    pub fn execute_many(&mut self, commands: &[&[&str]]) -> Result<Vec<RespValue>, Error> {
        if commands.is_empty() {
            return Err(Error::Protocol(
                "a pipeline needs at least one command".to_owned(),
            ));
        }

        let mut payload = Vec::new();

        for command in commands {
            payload.extend(encode_strings(command)?);
        }

        self.stream.get_mut().write_all(&payload)?;
        self.stream.get_mut().flush()?;

        let mut values = Vec::with_capacity(commands.len());

        for _ in 0..commands.len() {
            values.push(read_value(&mut self.stream)?);
        }

        Ok(values)
    }

    /// Writes one command and reads `replies` replies. A subscribe confirms each channel separately.
    pub(crate) fn run_replies<I, S>(
        &mut self,
        arguments: I,
        replies: usize,
    ) -> Result<Vec<RespValue>, Error>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        let owned: Vec<S> = arguments.into_iter().collect();
        let borrowed: Vec<&str> = owned.iter().map(AsRef::as_ref).collect();
        let payload = encode_strings(&borrowed)?;

        self.stream.get_mut().write_all(&payload)?;
        self.stream.get_mut().flush()?;

        let mut values = Vec::with_capacity(replies);

        for _ in 0..replies {
            values.push(self.read_message()?);
        }

        Ok(values)
    }

    pub(crate) fn run<I, S>(&mut self, arguments: I) -> Result<RespValue, Error>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        let owned: Vec<S> = arguments.into_iter().collect();
        let borrowed: Vec<&str> = owned.iter().map(AsRef::as_ref).collect();

        self.execute(&borrowed)
    }

    pub(crate) fn integer<I, S>(&mut self, arguments: I) -> Result<i64, Error>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        let (command, reply) = self.run_named(arguments)?;

        match reply {
            RespValue::Integer(value) => Ok(value),
            other => Err(unexpected(&command, &other)),
        }
    }

    pub(crate) fn flag<I, S>(&mut self, arguments: I) -> Result<bool, Error>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        Ok(self.integer(arguments)? > 0)
    }

    pub(crate) fn ok<I, S>(&mut self, arguments: I) -> Result<(), Error>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        self.run(arguments)?;

        Ok(())
    }

    pub(crate) fn text<I, S>(&mut self, arguments: I) -> Result<String, Error>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        let (command, reply) = self.run_named(arguments)?;

        reply
            .as_string()?
            .ok_or_else(|| Error::Protocol(format!("{command} returned a null reply")))
    }

    pub(crate) fn bulk_or_null<I, S>(&mut self, arguments: I) -> Result<Option<String>, Error>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        self.run(arguments)?.as_string()
    }

    pub(crate) fn strings<I, S>(&mut self, arguments: I) -> Result<Vec<String>, Error>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        let (command, reply) = self.run_named(arguments)?;

        match reply {
            RespValue::Null => Ok(Vec::new()),
            RespValue::Array(items) => items
                .iter()
                .map(|item| {
                    item.as_string()?
                        .ok_or_else(|| Error::Protocol(format!("{command} returned a null bulk")))
                })
                .collect(),
            other => Err(unexpected(&command, &other)),
        }
    }

    pub(crate) fn optional_strings<I, S>(
        &mut self,
        arguments: I,
    ) -> Result<Vec<Option<String>>, Error>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        let (command, reply) = self.run_named(arguments)?;

        match reply {
            RespValue::Null => Ok(Vec::new()),
            RespValue::Array(items) => items.iter().map(RespValue::as_string).collect(),
            other => Err(unexpected(&command, &other)),
        }
    }

    pub(crate) fn run_named<I, S>(&mut self, arguments: I) -> Result<(String, RespValue), Error>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
    {
        let owned: Vec<S> = arguments.into_iter().collect();
        let command = owned
            .first()
            .map(|first| first.as_ref().to_owned())
            .unwrap_or_default();
        let borrowed: Vec<&str> = owned.iter().map(AsRef::as_ref).collect();

        Ok((command, self.execute(&borrowed)?))
    }
}

pub(crate) fn unexpected(command: &str, reply: &RespValue) -> Error {
    Error::Protocol(format!("{command} returned {reply:?}"))
}

pub(crate) fn join(command: &str, arguments: &[&str]) -> Vec<String> {
    let mut values = Vec::with_capacity(arguments.len() + 1);

    values.push(command.to_owned());
    values.extend(arguments.iter().map(|argument| (*argument).to_owned()));

    values
}

fn set_arguments(key: &str, value: &str, expiry: Option<&[String]>, mode: SetMode) -> Vec<String> {
    let mut arguments = vec!["SET".to_owned(), key.to_owned(), value.to_owned()];

    if let Some(expiry) = expiry {
        arguments.extend_from_slice(expiry);
    }

    match mode {
        SetMode::Always => {}
        SetMode::IfNotExists => arguments.push("NX".to_owned()),
        SetMode::IfExists => arguments.push("XX".to_owned()),
    }

    arguments
}

pub(crate) fn unix_millis(time: SystemTime) -> Result<u128, Error> {
    time.duration_since(UNIX_EPOCH)
        .map(|elapsed| elapsed.as_millis())
        .map_err(|_| Error::Protocol("time is before the Unix epoch".to_owned()))
}

pub(crate) fn duration_millis(ttl: Duration) -> u128 {
    let millis = ttl.as_millis();

    if ttl.subsec_nanos().is_multiple_of(1_000_000) {
        return millis;
    }

    millis.saturating_add(1)
}

pub(crate) fn duration_secs(ttl: Duration) -> u64 {
    let seconds = ttl.as_secs();

    if ttl.subsec_nanos() == 0 {
        return seconds;
    }

    seconds.saturating_add(1)
}

#[cfg(test)]
mod tests {
    use std::io::{Read, Write};

    use std::net::{TcpListener, TcpStream};

    use std::thread;
    use std::time::Duration;

    use super::{Client, ClientOptions};

    #[test]
    fn connect_sends_nothing_until_the_first_command() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let accepted = thread::spawn(move || listener.accept().unwrap().0);
        let mut client = Client::connect("127.0.0.1", port).unwrap();
        let mut server = accepted.join().unwrap();

        thread::sleep(Duration::from_millis(50));

        server.set_nonblocking(true).unwrap();

        let mut peeked = [0_u8; 1];

        assert!(server.peek(&mut peeked).is_err());
        server.set_nonblocking(false).unwrap();

        let ping = thread::spawn(move || client.ping().unwrap());
        let mut got = [0_u8; 14];

        server.read_exact(&mut got).unwrap();

        assert_eq!(&got, b"*1\r\n$4\r\nPING\r\n");
        server.write_all(b"+PONG\r\n").unwrap();

        assert_eq!(ping.join().unwrap(), "PONG");
    }

    #[test]
    fn a_server_error_leaves_the_next_command_usable() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let accepted = thread::spawn(move || listener.accept().unwrap().0);
        let mut client = Client::connect("127.0.0.1", port).unwrap();
        let mut server = accepted.join().unwrap();
        let error = thread::spawn(move || {
            let error = client.get_string("missing").unwrap_err();
            let pong = client.ping().unwrap();

            (error.to_string(), pong)
        });

        read_some(&mut server);
        server.write_all(b"-ERR no such key\r\n").unwrap();
        read_some(&mut server);
        server.write_all(b"+PONG\r\n").unwrap();

        let (message, pong) = error.join().unwrap();

        assert_eq!(message, "ERR no such key");
        assert_eq!(pong, "PONG");
    }

    #[test]
    fn password_is_sent_as_auth() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let accepted = thread::spawn(move || listener.accept().unwrap().0);
        let connecting = thread::spawn(move || {
            Client::connect_with(ClientOptions {
                host: "127.0.0.1".to_owned(),
                port,
                password: Some("secret".to_owned()),
                ..ClientOptions::default()
            })
            .unwrap()
        });

        let mut server = accepted.join().unwrap();
        let expected = b"*2\r\n$4\r\nAUTH\r\n$6\r\nsecret\r\n";
        let mut got = vec![0_u8; expected.len()];

        server.read_exact(&mut got).unwrap();

        assert_eq!(got, expected);
        server.write_all(b"+OK\r\n").unwrap();
        connecting.join().unwrap();
    }

    #[test]
    fn database_is_selected_after_auth() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let accepted = thread::spawn(move || listener.accept().unwrap().0);
        let connecting = thread::spawn(move || {
            Client::connect_with(ClientOptions {
                host: "127.0.0.1".to_owned(),
                port,
                password: Some("secret".to_owned()),
                database: 3,
                ..ClientOptions::default()
            })
            .unwrap()
        });

        let mut server = accepted.join().unwrap();
        let auth = b"*2\r\n$4\r\nAUTH\r\n$6\r\nsecret\r\n";
        let select = b"*2\r\n$6\r\nSELECT\r\n$1\r\n3\r\n";
        let mut got = vec![0_u8; auth.len()];

        server.read_exact(&mut got).unwrap();
        assert_eq!(got, auth);
        server.write_all(b"+OK\r\n").unwrap();
        got = vec![0_u8; select.len()];
        server.read_exact(&mut got).unwrap();
        assert_eq!(got, select);
        server.write_all(b"+OK\r\n").unwrap();
        assert_eq!(connecting.join().unwrap().database(), 3);
    }

    fn read_some(stream: &mut TcpStream) {
        let mut buffer = [0_u8; 64];

        let read = stream.read(&mut buffer).unwrap();

        assert!(read > 0);
    }
}