use std::{sync::Arc, time::Duration};

use anyhow::Result;
use arc_swap::ArcSwap;
use log::{info, warn};
use storage_backend::storage::create_backend_storage_for_test;
use tici::{
    config::MetaServiceConfig,
    connection_pool::TiDBConnectionPool,
    meta::{Bootstrap, server::MetaServiceServer, service::MetaServiceCore},
};
use tici_service_discovery::election::leadership::MockLeadership;
use tokio::time::sleep;

pub async fn build_meta_service(config: &MetaServiceConfig) -> Arc<MetaServiceCore> {
    // Create MetaServiceCore with config-based ReaderPool
    let tidb_pools = TiDBConnectionPool::new(&config.tidb_server);
    Bootstrap::new(tidb_pools.clone())
        .ensure_bootstrapped()
        .await
        .unwrap();

    let backend_storage = create_backend_storage_for_test(&config.s3).await.unwrap();

    match tokio::time::timeout(
        Duration::from_secs(3),
        tici::meta::server::build_leadership_and_discovery(config),
    )
    .await
    {
        Ok(Ok((leadership, _discovery_client))) => {
            info!("Using real etcd leadership for meta service");
            Arc::new(MetaServiceCore::new(
                ArcSwap::new(Arc::new(config.clone())),
                tidb_pools,
                leadership,
                backend_storage,
            ))
        }
        Ok(Err(e)) => {
            warn!(
                "Failed to init etcd leadership, err={}, falling back to MockLeadership",
                e
            );
            Arc::new(MetaServiceCore::new(
                ArcSwap::new(Arc::new(config.clone())),
                tidb_pools,
                MockLeadership::default(),
                backend_storage,
            ))
        }
        Err(_) => {
            warn!("Timeout initializing etcd leadership, falling back to MockLeadership");
            Arc::new(MetaServiceCore::new(
                ArcSwap::new(Arc::new(config.clone())),
                tidb_pools,
                MockLeadership::default(),
                backend_storage,
            ))
        }
    }
}

pub async fn start_meta_service(
    meta_service: Arc<MetaServiceCore>,
) -> Result<(
    tokio::sync::oneshot::Sender<()>,
    tokio::task::JoinHandle<Result<(), tonic::transport::Error>>,
)> {
    let (tx, rx) = tokio::sync::oneshot::channel::<()>();

    meta_service.start().await?;

    let addr_str: String = meta_service.config.load().server.addr.clone();
    let addr = addr_str.parse()?;

    let meta_service_clone = meta_service.clone();
    let meta_handle = tokio::spawn(MetaServiceServer::build_server(
        addr,
        meta_service_clone,
        rx,
    ));

    // Wait for meta service to start
    sleep(Duration::from_secs(2)).await;
    info!("meta service started with config-based TTL");

    Ok((tx, meta_handle))
}