use anyhow::{Context, Result};
use async_trait::async_trait;
use bytes::Bytes;
use redis::AsyncCommands;

use crate::driver::Driver;

pub struct RedisDriver {
    client: redis::Client,
}

impl RedisDriver {
    pub async fn new(url: &str) -> Result<std::sync::Arc<Self>> {
        let client = redis::Client::open(url).context("redis connect")?;
        // Test connection
        let mut conn = client
            .get_multiplexed_async_connection()
            .await
            .context("redis test connection")?;
        redis::cmd("PING")
            .query_async::<_, String>(&mut conn)
            .await
            .context("redis PING")?;
        Ok(std::sync::Arc::new(Self { client }))
    }
}

#[async_trait]
impl Driver for RedisDriver {
    async fn pop(&self, queue: &str) -> Result<Option<Bytes>> {
        let mut conn = self.client.get_multiplexed_async_connection().await?;
        let result: Option<(String, Vec<u8>)> = conn.blpop(queue, 0.0).await.context("BLPOP")?;
        Ok(result.map(|(_, v)| Bytes::from(v)))
    }

    async fn push(&self, queue: &str, payload: Bytes) -> Result<()> {
        let mut conn = self.client.get_multiplexed_async_connection().await?;
        conn.rpush::<_, _, ()>(queue, payload.to_vec()).await?;
        Ok(())
    }

    async fn depth(&self, queue: &str) -> Result<u64> {
        let mut conn = self.client.get_multiplexed_async_connection().await?;
        Ok(conn.llen::<_, u64>(queue).await.unwrap_or(0))
    }
}
