use std::{error::Error, fmt::Debug, sync::Arc, time::SystemTime};

use anyhow::Result;
use storage_backend::storage::create_backend_storage_for_test;
use tici::{ShardInfo, config::MetaServiceConfig, meta::writer::FTSIndex};
use tici_common::index::IndexKey;
use tici_grpc::ShardManifestHeader;
use tici_search_lib::{
    count_by_reader,
    ffi::{CountResult, Expr, Range, Shard},
};
use tici_shard::reader::shard_mirror_manager::ShardMirrorManager;
use tici_system::readable::ReadableSize;
use tici_test_utils::test_utils::{test_suite::E2ETestSuite, *};
use tipb::{ExprType, ScalarFuncSig};

mod basic;
mod failover;
mod partition_table;

struct Measurement {
    prev_size: u64,
    prev_rows: u64,
    last_time: SystemTime,
}

impl Default for Measurement {
    fn default() -> Self {
        Self {
            prev_size: 0,
            prev_rows: 0,
            last_time: SystemTime::now(),
        }
    }
}

impl Measurement {
    pub fn record(&mut self, current_size: u64, current_rows: u64) -> (f64, f64) {
        if current_rows < self.prev_rows || current_size < self.prev_size {
            // Maybe compaction took place
            return (0.0, 0.0);
        }
        let delta_size = current_size - self.prev_size;
        let delta_rows = current_rows - self.prev_rows;
        let current = SystemTime::now();
        let time_diff = current
            .duration_since(self.last_time)
            .unwrap()
            .as_secs_f64();
        let ans = (delta_size as f64 / time_diff, delta_rows as f64 / time_diff);
        self.prev_size = current_size;
        self.prev_rows = current_rows;
        self.last_time = current;
        let (mb_per_sec, kr_per_sec) = ans;
        (mb_per_sec / (1024.0 * 1024.0), kr_per_sec / 1024.0)
    }

    pub fn get_stats_string(&mut self, current_stats: &CheckStats, index: Arc<FTSIndex>) -> String {
        let (mb_per_sec, kr_per_sec) = self.record(current_stats.size, current_stats.rows);
        let real_speed = index.append_qps.bps();
        format!(
            "(append_speed={:.2} MB/s, speed_size={:.4} MB/s, speed_rows={:.4} k/s)",
            real_speed / 1024.0 / 1024.0,
            mb_per_sec,
            kr_per_sec,
        )
    }
}

fn get_current_mem() -> ReadableSize {
    #[cfg(target_os = "linux")]
    if let Ok(p) = procfs::process::Process::myself() {
        let k = p.stat.rss * unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
        ReadableSize(k as u64)
    } else {
        ReadableSize(0u64)
    }
    #[cfg(not(target_os = "linux"))]
    ReadableSize(0u64)
}

struct CheckStats {
    pub size: u64,
    pub rows: u64,
    pub split_count: u64,
    pub compaction_count: u64,
    pub s3_frag_len: usize,
    pub meta_frag_len: usize,
    pub shard_info_heap_size: ReadableSize,
    pub shards_from_restore: Vec<ShardInfo>,
    pub memory: ReadableSize,
}

impl Debug for CheckStats {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CheckStats")
            .field(
                "size",
                &format!("{:.6}", ReadableSize(self.size).as_mb_f64()),
            )
            .field("rows", &self.rows)
            .field("split_count", &self.split_count)
            .field("compaction_count", &self.compaction_count)
            .field("s3_frag_len", &self.s3_frag_len)
            .field("meta_frag_len", &self.meta_frag_len)
            .field(
                "shard_info_heap_size",
                &format!("{:.6}", self.shard_info_heap_size.as_mb_f64()),
            )
            .field("memory", &format!("{:.6}", self.memory.as_mb_f64()))
            .finish()
    }
}

struct Checker {
    pub expected: CheckStats,
}

impl Checker {
    pub fn new(expected_max_size: u64) -> Self {
        Self {
            expected: CheckStats {
                size: expected_max_size,
                rows: 0,
                split_count: 1,
                compaction_count: 1,
                s3_frag_len: 7,
                meta_frag_len: 1,
                shard_info_heap_size: ReadableSize(0),
                shards_from_restore: vec![],
                memory: ReadableSize(0),
            },
        }
    }

    pub fn is_ok(&self, stats: &CheckStats, test_suite: &E2ETestSuite) -> bool {
        for shard in stats.shards_from_restore.iter() {
            if shard.get_total_estimated_bytes()
                > (test_suite.get_meta_config().shard.max_size.0 as f64 * 1.5) as u64
            {
                panic!(
                    "shard is too large, shard={:?}, size={}",
                    shard,
                    ReadableSize(shard.get_total_estimated_bytes()).as_mb_f64()
                );
            }
        }
        stats.size > self.expected.size
            && stats.s3_frag_len >= self.expected.s3_frag_len
            && stats.meta_frag_len >= self.expected.meta_frag_len
            && stats.compaction_count >= self.expected.compaction_count
            && stats.split_count >= self.expected.split_count
    }

    pub async fn get_current_stat(
        test_suite: &E2ETestSuite,
        meta_cfg: &MetaServiceConfig,
        index_key: &IndexKey,
    ) -> CheckStats {
        let storage =
            create_backend_storage_for_test(&test_suite.get_one_worker_node_handle().config.s3)
                .await
                .unwrap();

        let frags = s3_list_frags(
            &storage,
            &meta_cfg.s3.bucket,
            &meta_cfg.s3.prefix,
            None,
            index_key,
        )
        .await
        .unwrap();
        let s3_frag_len = frags.len();
        // Due to compaction, meta_frag_len maybe fewer.
        let shard = test_suite
            .collect_shards_from_index(index_key)
            .await
            .first()
            .unwrap()
            .clone();
        let meta_frag_len = shard.manifest.len();
        let index = test_suite.get_index(index_key).await;
        let compaction_finish = index.stats.get_compaction_finish();
        let split_finish = index.stats.get_split_finish();
        let size = test_suite.compute_total_size(index_key).await;
        let rows = test_suite.compute_total_rows(index_key).await;
        let shard_info_heap_size =
            ReadableSize(test_suite.compute_index_memory_usage(index_key).await as u64);
        let shards_from_restore = test_suite.collect_shards_from_restore().await;
        let memory = get_current_mem();
        CheckStats {
            size,
            rows,
            split_count: split_finish,
            compaction_count: compaction_finish,
            s3_frag_len,
            meta_frag_len,
            shard_info_heap_size,
            shards_from_restore,
            memory,
        }
    }
}

pub fn build_default_query_expr() -> Expr {
    Expr {
        tp: ExprType::ScalarFunc as i32,
        val: Vec::new(),
        children: vec![
            Expr {
                tp: ExprType::ColumnRef as i32,
                val: b"handle".to_vec(),
                children: Vec::new(),
                rpn_args_len: 0,
                sig: ScalarFuncSig::Unspecified as i32,
                field_type: 0,
            },
            Expr {
                tp: ExprType::String as i32,
                val: Vec::new(),
                children: Vec::new(),
                rpn_args_len: 0,
                sig: ScalarFuncSig::Unspecified as i32,
                field_type: 0,
            },
        ],
        rpn_args_len: 0,
        sig: ScalarFuncSig::GeString as i32,
        field_type: 0,
    }
}

pub async fn make_count_request(
    reader: Arc<ShardMirrorManager>,
    shard_header: &ShardManifestHeader,
    index_key: &IndexKey,
    ranges: Vec<Range<'_>>,
) -> Result<CountResult, Box<dyn Error>> {
    let query = build_default_query_expr();
    let read_ts = 0_u64;
    let shard = Shard {
        keyspace_id: index_key.keyspace_id,
        index_id: index_key.index_id,
        shard_id: shard_header.shard_id,
        shard_epoch: shard_header.epoch,
    };

    let shard_reader = reader
        .get_shard(shard_header.shard_id)
        .ok_or_else(|| anyhow::anyhow!("Shard {} not found", shard_header.shard_id))?
        .get_reader();

    count_by_reader(shard_reader, &shard, &ranges, &query, &read_ts)
}