use std::time::Duration;

use anyhow::Result;
use tici::meta::Bootstrap;
use tici_common::index::IndexKey;
use tici_grpc::{DropIndexRequest, ErrorCode, WorkerNodeHeartbeatRequest};
use tici_service_discovery::discovery::client::DiscoveryClient;
use tici_system::readable::ReadableDuration;
use tici_test_utils::test_utils::{
    meta_failover::{
        RunningMeta, any_meta_has_worker, cleanup_service_discovery_election, configure_meta,
        configure_worker, current_leader, make_observer, must_create_index_on_meta, pick_free_port,
        revoke_current_leader_lease,
    },
    must_finally,
    worker_handle::WorkerServiceHandle,
};

#[test]
fn test_meta_failover_rejects_old_leader_and_leaves_new_leader_empty() {
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all()
        .build()
        .unwrap();

    rt.block_on(async {
        let mut meta1 = None;
        let mut meta2 = None;
        let mut worker = None;

        let test_result: Result<()> = async {
            let meta_addr1 = format!("127.0.0.1:{}", pick_free_port()?);
            let meta_addr2 = format!("127.0.0.1:{}", pick_free_port()?);
            let worker_addr = format!("127.0.0.1:{}", pick_free_port()?);

            let mut meta_cfg1 = configure_meta(meta_addr1.clone());
            let mut meta_cfg2 = configure_meta(meta_addr2.clone());
            let mut worker_cfg = configure_worker(worker_addr.clone());
            let election_ttl = ReadableDuration::secs(3);
            let election_interval = ReadableDuration::millis(200);
            meta_cfg1.election.campaign_ttl = election_ttl;
            meta_cfg1.election.campaign_interval = election_interval;
            meta_cfg2.election.campaign_ttl = election_ttl;
            meta_cfg2.election.campaign_interval = election_interval;
            let unique_s3_prefix = if meta_cfg1.s3.prefix.is_empty() {
                format!("e2e_failover_{}", uuid::Uuid::new_v4())
            } else {
                format!(
                    "{}/e2e_failover_{}",
                    meta_cfg1.s3.prefix,
                    uuid::Uuid::new_v4()
                )
            };
            meta_cfg1.s3.prefix = unique_s3_prefix.clone();
            meta_cfg2.s3.prefix = unique_s3_prefix.clone();
            worker_cfg.s3.prefix = unique_s3_prefix;

            Bootstrap::new_with_config(&meta_cfg1)
                .clear_all_meta(true)
                .await?;
            cleanup_service_discovery_election(&meta_cfg1.server.pd_addr).await?;

            let observer = make_observer(&meta_cfg1.server.pd_addr);

            let running_meta1 = RunningMeta::start(meta_cfg1.clone()).await?;
            let running_meta2 = RunningMeta::start(meta_cfg2.clone()).await?;
            running_meta1.wait_ready().await;
            running_meta2.wait_ready().await;

            must_finally(|| {
                let observer = &observer;
                let meta_addr1 = meta_addr1.clone();
                let meta_addr2 = meta_addr2.clone();
                async move {
                    let Ok(peers) = observer.get_all_peers().await else {
                        return false;
                    };
                    peers.nodes.contains(&meta_addr1)
                        && peers.nodes.contains(&meta_addr2)
                        && matches!(
                            peers.leader.as_deref(),
                            Some(addr) if addr == meta_addr1 || addr == meta_addr2
                        )
                }
            })
            .await;

            worker = Some(WorkerServiceHandle::new(worker_cfg, None).await?);
            meta1 = Some(running_meta1);
            meta2 = Some(running_meta2);

            must_finally(|| {
                let meta1 = meta1.as_ref().expect("meta1 should be started");
                let meta2 = meta2.as_ref().expect("meta2 should be started");
                let worker_addr = worker_addr.clone();
                async move { any_meta_has_worker(meta1, meta2, &worker_addr).await }
            })
            .await;

            let old_leader_addr = current_leader(&observer).await?;
            let old_leader = if meta1
                .as_ref()
                .is_some_and(|meta| meta.addr == old_leader_addr)
            {
                meta1.as_ref().expect("meta1 missing")
            } else {
                meta2.as_ref().expect("meta2 missing")
            };

            let index_key = IndexKey::new_nullspace(42, 2048);
            must_create_index_on_meta(old_leader, &index_key).await?;

            must_finally(|| {
                let old_leader = old_leader.meta_service.clone();
                async move {
                    old_leader
                        .shard_task_manager
                        .get_all_table_ids()
                        .await
                        .contains(&(index_key.keyspace_id, index_key.table_id))
                }
            })
            .await;

            revoke_current_leader_lease(&meta_cfg1.server.pd_addr).await?;

            must_finally(|| {
                let observer = &observer;
                let old_leader_addr = old_leader_addr.clone();
                async move {
                    matches!(
                        current_leader(observer).await,
                        Ok(addr) if addr != old_leader_addr
                    )
                }
            })
            .await;

            let new_leader_addr = current_leader(&observer).await?;
            let new_leader = if meta1
                .as_ref()
                .is_some_and(|meta| meta.addr == new_leader_addr)
            {
                meta1.as_ref().expect("meta1 missing")
            } else {
                meta2.as_ref().expect("meta2 missing")
            };

            must_finally(|| {
                let old_leader_ref = old_leader;
                async move {
                    let mut client = old_leader_ref.meta_internal_client().await;
                    match client
                        .worker_node_heartbeat(WorkerNodeHeartbeatRequest::default())
                        .await
                    {
                        Ok(response) => {
                            let mut stream = response.into_inner();
                            match stream.message().await {
                                Ok(Some(resp)) => resp.status == ErrorCode::NotLeader as i32,
                                _ => false,
                            }
                        }
                        Err(_) => false,
                    }
                }
            })
            .await;

            tokio::time::sleep(Duration::from_secs(1)).await;
            assert!(
                new_leader
                    .meta_service
                    .shard_task_manager
                    .get_all_table_ids()
                    .await
                    .is_empty(),
                "new leader should still have an empty shard_task_manager without restore"
            );
            assert!(new_leader.meta_service.get_index(&index_key).await.is_err());

            let mut old_leader_client = old_leader.meta_client().await;
            let drop_response = old_leader_client
                .drop_index(DropIndexRequest {
                    table_id: index_key.table_id,
                    index_id: index_key.index_id,
                    keyspace_id: index_key.keyspace_id,
                })
                .await?
                .into_inner();
            assert_eq!(drop_response.status, ErrorCode::NotLeader as i32);

            let mut old_leader_internal_client = old_leader.meta_internal_client().await;
            let mut heartbeat_stream = old_leader_internal_client
                .worker_node_heartbeat(WorkerNodeHeartbeatRequest::default())
                .await?
                .into_inner();
            let heartbeat_response = heartbeat_stream.message().await?.unwrap();
            assert_eq!(heartbeat_response.status, ErrorCode::NotLeader as i32);
            Ok(())
        }
        .await;

        if let Some(worker) = worker.take() {
            let _ = worker.stop().await;
        }
        if let Some(meta) = meta1.as_mut() {
            let _ = meta.stop().await;
        }
        if let Some(meta) = meta2.as_mut() {
            let _ = meta.stop().await;
        }

        test_result.unwrap();
    })
}