use std::sync::mpsc;
use std::thread;

use etcd_client::{Client, Compare, CompareOp, Txn, TxnOp};

use crate::keypath::timestamp_path;
use crate::persistence::{
    StoreError, TimestampStore, decode_timestamp_nanos, encode_timestamp_nanos,
};

enum Command {
    Load {
        cluster_id: u64,
        group_id: u32,
        resp: mpsc::Sender<Result<Option<i64>, StoreError>>,
    },
    Save {
        cluster_id: u64,
        group_id: u32,
        ts_nanos: i64,
        leader_key: String,
        expected_leader_value: String,
        resp: mpsc::Sender<Result<(), StoreError>>,
    },
    Delete {
        cluster_id: u64,
        group_id: u32,
        resp: mpsc::Sender<Result<(), StoreError>>,
    },
    SetLeader {
        _leader_key: String,
        _leader_value: String,
        resp: mpsc::Sender<Result<(), StoreError>>,
    },
    Stop,
}

/// Real etcd-backed timestamp store implementation.
///
/// The implementation uses a dedicated worker thread and a current-thread tokio runtime
/// so the existing synchronous `TimestampStore` trait can remain unchanged while
/// still using the async etcd client.
#[derive(Debug)]
pub struct EtcdTimestampStore {
    tx: mpsc::Sender<Command>,
}

impl EtcdTimestampStore {
    /// Connects to etcd and starts the background worker.
    pub fn connect(endpoints: Vec<String>) -> Result<Self, StoreError> {
        if endpoints.is_empty() {
            return Err(StoreError::Internal(
                "etcd endpoints must not be empty".to_string(),
            ));
        }

        let (tx, rx) = mpsc::channel::<Command>();
        let (ready_tx, ready_rx) = mpsc::channel::<Result<(), StoreError>>();

        thread::Builder::new()
            .name("pd-tso-etcd-store".to_string())
            .spawn(move || run_worker(endpoints, rx, ready_tx))
            .map_err(|err| StoreError::Internal(format!("spawn etcd worker failed: {err}")))?;

        ready_rx
            .recv()
            .map_err(|_| StoreError::Internal("etcd worker failed before ready".to_string()))??;

        Ok(Self { tx })
    }

    fn call<T>(
        &self,
        build: impl FnOnce(mpsc::Sender<Result<T, StoreError>>) -> Command,
    ) -> Result<T, StoreError> {
        let (resp_tx, resp_rx) = mpsc::channel::<Result<T, StoreError>>();
        self.tx
            .send(build(resp_tx))
            .map_err(|_| StoreError::Internal("etcd store worker channel closed".to_string()))?;
        resp_rx
            .recv()
            .map_err(|_| StoreError::Internal("etcd store worker stopped".to_string()))?
    }
}

impl Drop for EtcdTimestampStore {
    fn drop(&mut self) {
        let _ = self.tx.send(Command::Stop);
    }
}

impl TimestampStore for EtcdTimestampStore {
    fn load_timestamp(&self, cluster_id: u64, group_id: u32) -> Result<Option<i64>, StoreError> {
        self.call(|resp| Command::Load {
            cluster_id,
            group_id,
            resp,
        })
    }

    fn save_timestamp_if_leader(
        &self,
        cluster_id: u64,
        group_id: u32,
        ts_nanos: i64,
        leader_key: &str,
        expected_leader_value: &str,
    ) -> Result<(), StoreError> {
        self.call(|resp| Command::Save {
            cluster_id,
            group_id,
            ts_nanos,
            leader_key: leader_key.to_string(),
            expected_leader_value: expected_leader_value.to_string(),
            resp,
        })
    }

    fn delete_timestamp(&self, cluster_id: u64, group_id: u32) -> Result<(), StoreError> {
        self.call(|resp| Command::Delete {
            cluster_id,
            group_id,
            resp,
        })
    }

    fn set_leader_value(&self, leader_key: &str, leader_value: &str) -> Result<(), StoreError> {
        self.call(|resp| Command::SetLeader {
            _leader_key: leader_key.to_string(),
            _leader_value: leader_value.to_string(),
            resp,
        })
    }
}

fn run_worker(
    endpoints: Vec<String>,
    rx: mpsc::Receiver<Command>,
    ready_tx: mpsc::Sender<Result<(), StoreError>>,
) {
    let rt = match tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
    {
        Ok(runtime) => runtime,
        Err(err) => {
            let _ = ready_tx.send(Err(StoreError::Internal(format!(
                "create tokio runtime failed: {err}"
            ))));
            return;
        }
    };

    let mut client = match rt.block_on(Client::connect(endpoints, None)) {
        Ok(client) => client,
        Err(err) => {
            let _ = ready_tx.send(Err(StoreError::Internal(format!(
                "connect etcd failed: {err}"
            ))));
            return;
        }
    };

    if ready_tx.send(Ok(())).is_err() {
        return;
    }

    while let Ok(cmd) = rx.recv() {
        match cmd {
            Command::Load {
                cluster_id,
                group_id,
                resp,
            } => {
                let _ = resp.send(rt.block_on(load_timestamp(&mut client, cluster_id, group_id)));
            }
            Command::Save {
                cluster_id,
                group_id,
                ts_nanos,
                leader_key,
                expected_leader_value,
                resp,
            } => {
                let _ = resp.send(rt.block_on(save_timestamp_if_leader(
                    &mut client,
                    cluster_id,
                    group_id,
                    ts_nanos,
                    &leader_key,
                    &expected_leader_value,
                )));
            }
            Command::Delete {
                cluster_id,
                group_id,
                resp,
            } => {
                let _ = resp.send(rt.block_on(delete_timestamp(&mut client, cluster_id, group_id)));
            }
            Command::SetLeader {
                _leader_key: _,
                _leader_value: _,
                resp,
            } => {
                let _ = resp.send(Err(StoreError::Internal(
                    "direct leader-key writes are disabled for EtcdTimestampStore; use lease-bound runtime election CAS".to_string(),
                )));
            }
            Command::Stop => break,
        }
    }
}

async fn load_timestamp(
    client: &mut Client,
    cluster_id: u64,
    group_id: u32,
) -> Result<Option<i64>, StoreError> {
    let key = timestamp_path(cluster_id, group_id);
    let resp = client
        .get(key, None)
        .await
        .map_err(|err| StoreError::Internal(format!("etcd get timestamp failed: {err}")))?;
    if let Some(kv) = resp.kvs().first() {
        let nanos = decode_timestamp_nanos(kv.value())?;
        Ok(Some(nanos))
    } else {
        Ok(None)
    }
}

async fn save_timestamp_if_leader(
    client: &mut Client,
    cluster_id: u64,
    group_id: u32,
    ts_nanos: i64,
    leader_key: &str,
    expected_leader_value: &str,
) -> Result<(), StoreError> {
    if expected_leader_value.is_empty() {
        return Err(StoreError::NotLeader);
    }
    if ts_nanos < 0 {
        return Err(StoreError::NegativeTimestamp);
    }

    let ts_key = timestamp_path(cluster_id, group_id);
    let new_raw = encode_timestamp_nanos(ts_nanos).to_vec();

    let previous_raw = client
        .get(ts_key.clone(), None)
        .await
        .map_err(|err| StoreError::Internal(format!("etcd pre-read timestamp failed: {err}")))?
        .kvs()
        .first()
        .map(|kv| kv.value().to_vec());

    if let Some(ref raw) = previous_raw {
        let previous_nanos = decode_timestamp_nanos(raw)?;
        if ts_nanos <= previous_nanos {
            return Err(StoreError::NonMonotonic {
                previous_nanos,
                attempted_nanos: ts_nanos,
            });
        }
    }

    let mut compares = vec![Compare::value(
        leader_key,
        CompareOp::Equal,
        expected_leader_value.as_bytes(),
    )];
    match previous_raw {
        Some(prev_raw) => compares.push(Compare::value(ts_key.clone(), CompareOp::Equal, prev_raw)),
        None => compares.push(Compare::version(ts_key.clone(), CompareOp::Equal, 0)),
    }

    let txn = Txn::new()
        .when(compares)
        .and_then(vec![TxnOp::put(ts_key, new_raw, None)]);
    let resp = client
        .txn(txn)
        .await
        .map_err(|err| StoreError::Internal(format!("etcd txn save failed: {err}")))?;
    if resp.succeeded() {
        Ok(())
    } else {
        Err(StoreError::NotLeader)
    }
}

async fn delete_timestamp(
    client: &mut Client,
    cluster_id: u64,
    group_id: u32,
) -> Result<(), StoreError> {
    let key = timestamp_path(cluster_id, group_id);
    client
        .delete(key, None)
        .await
        .map_err(|err| StoreError::Internal(format!("etcd delete timestamp failed: {err}")))?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::EtcdTimestampStore;

    #[test]
    fn empty_endpoints_are_rejected() {
        let err = EtcdTimestampStore::connect(Vec::new()).expect_err("must fail");
        assert!(err.to_string().contains("etcd endpoints must not be empty"));
    }
}