use std::time::Duration;

use redis::{Cmd, Pipeline, ToRedisArgs, Value, cmd};
use serde::Serialize;

use crate::error::{CacheError, CacheResult};
use crate::json;
use crate::kv::{Cache, millis};

pub struct CachePipeline {
    cache: Cache,
    pipeline: Pipeline,
}

impl CachePipeline {
    pub(crate) fn new(cache: Cache) -> Self {
        Self {
            cache,
            pipeline: Pipeline::new(),
        }
    }

    pub fn pipeline(&self) -> &Pipeline {
        &self.pipeline
    }

    pub fn len(&self) -> usize {
        self.pipeline.len()
    }

    pub fn is_empty(&self) -> bool {
        self.pipeline.is_empty()
    }

    pub fn clear(&mut self) {
        self.pipeline.clear();
    }

    pub fn atomic(&mut self) -> CacheResult<&mut Self> {
        if self.cache.is_cluster() {
            return Err(CacheError::unsupported(
                "MULTI/EXEC",
                "transactions are unsupported by Redis Cluster",
            ));
        }
        self.pipeline.atomic();
        Ok(self)
    }

    pub fn add(&mut self, command: &Cmd) -> &mut Self {
        self.pipeline.add_command(command.clone());
        self
    }

    pub fn command(&mut self, name: &str) -> Cmd {
        cmd(name)
    }

    pub fn get(&mut self, key: &str) -> &mut Self {
        self.add(cmd("GET").arg(key))
    }

    pub fn mget<K: ToRedisArgs + Send + Sync>(&mut self, keys: K) -> &mut Self {
        self.add(cmd("MGET").arg(keys))
    }

    pub fn set(&mut self, key: &str, value: &[u8]) -> &mut Self {
        self.add(cmd("SET").arg(key).arg(value))
    }

    pub fn set_ex(&mut self, key: &str, value: &[u8], ttl: Duration) -> &mut Self {
        self.add(cmd("SET").arg(key).arg(value).arg("PX").arg(millis(ttl)))
    }

    pub fn set_nx_ex(&mut self, key: &str, value: &[u8], ttl: Duration) -> &mut Self {
        self.add(
            cmd("SET")
                .arg(key)
                .arg(value)
                .arg("PX")
                .arg(millis(ttl))
                .arg("NX"),
        )
    }

    pub fn del<K: ToRedisArgs + Send + Sync>(&mut self, keys: K) -> &mut Self {
        self.add(cmd("DEL").arg(keys))
    }

    pub fn unlink<K: ToRedisArgs + Send + Sync>(&mut self, keys: K) -> &mut Self {
        self.add(cmd("UNLINK").arg(keys))
    }

    pub fn exists<K: ToRedisArgs + Send + Sync>(&mut self, keys: K) -> &mut Self {
        self.add(cmd("EXISTS").arg(keys))
    }

    pub fn expire(&mut self, key: &str, ttl: Duration) -> &mut Self {
        self.add(cmd("PEXPIRE").arg(key).arg(millis(ttl)))
    }

    pub fn persist(&mut self, key: &str) -> &mut Self {
        self.add(cmd("PERSIST").arg(key))
    }

    pub fn ttl(&mut self, key: &str) -> &mut Self {
        self.add(cmd("PTTL").arg(key))
    }

    pub fn incr(&mut self, key: &str) -> &mut Self {
        self.add(cmd("INCR").arg(key))
    }

    pub fn incr_by(&mut self, key: &str, delta: i64) -> &mut Self {
        self.add(cmd("INCRBY").arg(key).arg(delta))
    }

    pub fn decr(&mut self, key: &str) -> &mut Self {
        self.add(cmd("DECR").arg(key))
    }

    pub fn decr_by(&mut self, key: &str, delta: i64) -> &mut Self {
        self.add(cmd("DECRBY").arg(key).arg(delta))
    }

    pub fn hset(&mut self, key: &str, field: &str, value: &[u8]) -> &mut Self {
        self.add(cmd("HSET").arg(key).arg(field).arg(value))
    }

    pub fn hset_multiple<V: AsRef<[u8]>>(&mut self, key: &str, fields: &[(&str, V)]) -> &mut Self {
        let mut command = cmd("HSET");
        command.arg(key);
        for (field, value) in fields {
            command.arg(*field).arg(value.as_ref());
        }
        self.add(&command)
    }

    pub fn hget(&mut self, key: &str, field: &str) -> &mut Self {
        self.add(cmd("HGET").arg(key).arg(field))
    }

    pub fn hgetall(&mut self, key: &str) -> &mut Self {
        self.add(cmd("HGETALL").arg(key))
    }

    pub fn hdel<F: ToRedisArgs + Send + Sync>(&mut self, key: &str, fields: F) -> &mut Self {
        self.add(cmd("HDEL").arg(key).arg(fields))
    }

    pub fn hexists(&mut self, key: &str, field: &str) -> &mut Self {
        self.add(cmd("HEXISTS").arg(key).arg(field))
    }

    pub fn hlen(&mut self, key: &str) -> &mut Self {
        self.add(cmd("HLEN").arg(key))
    }

    pub fn hkeys(&mut self, key: &str) -> &mut Self {
        self.add(cmd("HKEYS").arg(key))
    }

    pub fn hvals(&mut self, key: &str) -> &mut Self {
        self.add(cmd("HVALS").arg(key))
    }

    pub fn lpush<V: ToRedisArgs + Send + Sync>(&mut self, key: &str, values: V) -> &mut Self {
        self.add(cmd("LPUSH").arg(key).arg(values))
    }

    pub fn rpush<V: ToRedisArgs + Send + Sync>(&mut self, key: &str, values: V) -> &mut Self {
        self.add(cmd("RPUSH").arg(key).arg(values))
    }

    pub fn lpop(&mut self, key: &str, count: Option<usize>) -> &mut Self {
        let mut command = cmd("LPOP");
        command.arg(key);
        if let Some(count) = count {
            command.arg(count);
        }
        self.add(&command)
    }

    pub fn rpop(&mut self, key: &str, count: Option<usize>) -> &mut Self {
        let mut command = cmd("RPOP");
        command.arg(key);
        if let Some(count) = count {
            command.arg(count);
        }
        self.add(&command)
    }

    pub fn lrange(&mut self, key: &str, start: isize, stop: isize) -> &mut Self {
        self.add(cmd("LRANGE").arg(key).arg(start).arg(stop))
    }

    pub fn ltrim(&mut self, key: &str, start: isize, stop: isize) -> &mut Self {
        self.add(cmd("LTRIM").arg(key).arg(start).arg(stop))
    }

    pub fn llen(&mut self, key: &str) -> &mut Self {
        self.add(cmd("LLEN").arg(key))
    }

    pub fn sadd<V: ToRedisArgs + Send + Sync>(&mut self, key: &str, members: V) -> &mut Self {
        self.add(cmd("SADD").arg(key).arg(members))
    }

    pub fn srem<V: ToRedisArgs + Send + Sync>(&mut self, key: &str, members: V) -> &mut Self {
        self.add(cmd("SREM").arg(key).arg(members))
    }

    pub fn scard(&mut self, key: &str) -> &mut Self {
        self.add(cmd("SCARD").arg(key))
    }

    pub fn sismember(&mut self, key: &str, member: &str) -> &mut Self {
        self.add(cmd("SISMEMBER").arg(key).arg(member))
    }

    pub fn smembers(&mut self, key: &str) -> &mut Self {
        self.add(cmd("SMEMBERS").arg(key))
    }

    pub fn zadd(&mut self, key: &str, member: &str, score: f64) -> &mut Self {
        self.add(cmd("ZADD").arg(key).arg(score).arg(member))
    }

    pub fn zrem<M: ToRedisArgs + Send + Sync>(&mut self, key: &str, members: M) -> &mut Self {
        self.add(cmd("ZREM").arg(key).arg(members))
    }

    pub fn zcard(&mut self, key: &str) -> &mut Self {
        self.add(cmd("ZCARD").arg(key))
    }

    pub fn zscore(&mut self, key: &str, member: &str) -> &mut Self {
        self.add(cmd("ZSCORE").arg(key).arg(member))
    }

    pub fn zincr(&mut self, key: &str, member: &str, delta: f64) -> &mut Self {
        self.add(cmd("ZINCRBY").arg(key).arg(delta).arg(member))
    }

    pub fn zrange(&mut self, key: &str, start: isize, stop: isize) -> &mut Self {
        self.add(cmd("ZRANGE").arg(key).arg(start).arg(stop))
    }

    pub fn set_json_ex<T: Serialize>(
        &mut self,
        key: &str,
        value: &T,
        ttl: Duration,
    ) -> CacheResult<&mut Self> {
        let encoded = json::encode(key, value)?;
        Ok(self.set_ex(key, &encoded, ttl))
    }

    pub async fn execute(&self) -> CacheResult<Vec<Value>> {
        if self.pipeline.is_empty() {
            return Ok(Vec::new());
        }
        let mut conn = self.cache.conn().clone();
        self.pipeline
            .query_async(&mut conn)
            .await
            .map_err(CacheError::from)
    }
}