use std::{
    cell::RefCell,
    collections::{BTreeMap, HashMap},
    str::from_utf8,
    sync::atomic::{AtomicU64, Ordering},
};

use fake::{
    Fake,
    faker::lorem::en::{Paragraph, Sentence},
    rand::seq::SliceRandom,
};
use rand::{Rng, SeedableRng, rngs::StdRng};
use tici_common::encode_binary_for_canal_json;

use crate::{
    utils::{key::ShardingKey, tantivy_doc::RawDoc},
    writer::cdc_log::OperationType,
};

// static TEST_ID: AtomicU64 = AtomicU64::new(1);

fn test_gen_handle() -> ShardingKey {
    let uid = uuid::Uuid::new_v4().to_string();
    ShardingKey::from_raw(hex::encode(uid.as_bytes()).as_bytes().to_vec())
}

/// Generates a handle for binary PK index (TEST_INDEX_META_BINARY_PK): random
/// raw bytes.
fn test_gen_handle_binary_pk() -> ShardingKey {
    let mut bytes = [0u8; 16];
    THREAD_RNG.with_borrow_mut(|rng| {
        rng.fill(bytes.as_mut_slice());
    });
    ShardingKey::from_raw(bytes.to_vec())
}

pub fn test_get_start_handle() -> ShardingKey {
    ShardingKey::from_raw("0".as_bytes().to_vec())
}

pub fn test_get_end_handle() -> ShardingKey {
    ShardingKey::from_raw("z".as_bytes().to_vec())
}

pub fn test_gen_handle_with_prefix_and_num(prefix: &str, num: u64) -> ShardingKey {
    ShardingKey::from_raw(format!("{}_{:07}", prefix, num).as_bytes().to_vec())
}

static VERSION: AtomicU64 = AtomicU64::new(1);

fn test_gen_version() -> u64 {
    VERSION.fetch_add(1, Ordering::SeqCst)
}

thread_local! {
    static THREAD_RNG: RefCell<StdRng> = RefCell::new(StdRng::from_seed([42; 32]));
}

fn add_rand_word(mut s: String, rng: &mut StdRng) -> String {
    let r = rng.random_range(0..10000);

    if r < 1 {
        s += " rwp00001"; // 1/10000
    }

    if r < 10 {
        s += " rwp00010"; // 1/1000
    }

    if r < 100 {
        s += " rwp00100"; // 1/100
    }

    if r < 1000 {
        s += " rwp01000"; // 1/10
    }

    if r < 9000 {
        s += " rwp09000" // 9/10
    }

    s
}

fn test_gen_sentence() -> String {
    THREAD_RNG.with_borrow_mut(|rng| add_rand_word(Sentence(5..10).fake_with_rng(rng), rng))
}

fn test_gen_paragraph() -> String {
    THREAD_RNG.with_borrow_mut(|rng| add_rand_word(Paragraph(3..5).fake_with_rng(rng), rng))
}

pub fn test_gen_insert_raw_doc() -> RawDoc {
    let h = test_gen_handle();
    RawDoc {
        handle: h.clone(),
        version: test_gen_version(),
        op_type: OperationType::Insert,
        cols: vec![
            (1, Some(test_gen_sentence())),
            (2, Some(test_gen_paragraph())),
            // We don't really care about what value store in pk in unit-tests.
            (3, Some(from_utf8(h.raw()).unwrap().to_string())),
            (4, Some(from_utf8(h.raw()).unwrap().to_string())),
        ],
        is_local_timezone: true,
    }
}

pub fn test_gen_update_raw_doc(handle: ShardingKey) -> RawDoc {
    RawDoc {
        handle: handle.clone(),
        version: test_gen_version(),
        op_type: OperationType::Update,
        cols: vec![
            (1, Some(test_gen_sentence())),
            (2, Some(test_gen_paragraph())),
            (3, Some(from_utf8(handle.raw()).unwrap().to_string())),
            (4, Some(from_utf8(handle.raw()).unwrap().to_string())),
        ],
        is_local_timezone: true,
    }
}

pub fn test_gen_delete_raw_doc(handle: ShardingKey) -> RawDoc {
    RawDoc {
        handle,
        version: test_gen_version(),
        op_type: OperationType::Delete,
        cols: vec![],
        is_local_timezone: true,
    }
}

/// Generates an insert RawDoc for TEST_INDEX_META_BINARY_PK (col 1:
/// description, col 2: binary_pk).
pub fn test_gen_insert_raw_doc_binary_pk() -> RawDoc {
    let h = test_gen_handle_binary_pk();
    RawDoc {
        handle: h.clone(),
        version: test_gen_version(),
        op_type: OperationType::Insert,
        cols: vec![
            (1, Some(test_gen_sentence())),
            (2, Some(encode_binary_for_canal_json(h.raw()))),
        ],
        is_local_timezone: true,
    }
}

/// Generates an update RawDoc for TEST_INDEX_META_BINARY_PK.
pub fn test_gen_update_raw_doc_binary_pk(handle: ShardingKey) -> RawDoc {
    RawDoc {
        handle: handle.clone(),
        version: test_gen_version(),
        op_type: OperationType::Update,
        cols: vec![
            (1, Some(test_gen_sentence())),
            (2, Some(encode_binary_for_canal_json(handle.raw()))),
        ],
        is_local_timezone: true,
    }
}

/// Generates insert/update/delete RawDocs for TEST_INDEX_META_BINARY_PK, with
/// same semantics as `test_gen_raw_docs`.
pub fn test_gen_raw_docs_binary_pk(
    inserts: usize,
    updates: usize,
    deletes: usize,
) -> (Vec<RawDoc>, BTreeMap<ShardingKey, RawDoc>) {
    let mut raw_docs = Vec::with_capacity(inserts + updates + deletes);
    let mut expected_docs = BTreeMap::new();
    for _ in 0..inserts {
        let doc = test_gen_insert_raw_doc_binary_pk();
        let old = expected_docs.insert(doc.handle.clone(), doc.clone());
        assert!(old.is_none());
        raw_docs.push(doc);
    }

    let seed = [42; 32];
    let mut rng = StdRng::from_seed(seed);
    for _ in 0..updates {
        let i = rng.random_range(0..inserts);
        let doc = test_gen_update_raw_doc_binary_pk(raw_docs[i].handle.clone());
        let old = expected_docs.insert(doc.handle.clone(), doc.clone());
        assert!(old.is_some());
        raw_docs.push(doc);
    }

    let mut handles_to_delete = raw_docs
        .iter()
        .take(inserts)
        .map(|doc| doc.handle.clone())
        .collect::<Vec<_>>();
    for _ in 0..deletes {
        let i = rng.random_range(0..handles_to_delete.len());
        let doc = test_gen_delete_raw_doc(handles_to_delete[i].clone());
        let old = expected_docs.remove(&doc.handle);
        assert!(old.is_some());
        assert_eq!(handles_to_delete.remove(i), doc.handle);
        raw_docs.push(doc);
    }

    raw_docs.shuffle(&mut rng);
    group_by_handle_sort_by_version(&mut raw_docs);

    (raw_docs, expected_docs)
}

fn group_by_handle_sort_by_version(raw_docs: &mut [RawDoc]) {
    let mut groups: HashMap<ShardingKey, Vec<usize>> = HashMap::new();
    for (idx, doc) in raw_docs.iter().enumerate() {
        groups.entry(doc.handle.clone()).or_default().push(idx);
    }

    for entries in groups.values() {
        if entries.len() <= 1 {
            continue;
        }

        let mut docs_to_sort: Vec<_> = entries.iter().map(|&idx| raw_docs[idx].clone()).collect();
        docs_to_sort.sort_by_key(|doc| doc.version);

        for (&idx, sorted_doc) in entries.iter().zip(docs_to_sort.iter()) {
            raw_docs[idx] = sorted_doc.clone();
        }
    }
}

pub fn test_gen_raw_docs(
    inserts: usize,
    updates: usize,
    deletes: usize,
) -> (Vec<RawDoc>, BTreeMap<ShardingKey, RawDoc>) {
    let mut raw_docs = Vec::with_capacity(inserts + updates + deletes);
    let mut expected_docs = BTreeMap::new();
    for _ in 0..inserts {
        let doc = test_gen_insert_raw_doc();
        let old = expected_docs.insert(doc.handle.clone(), doc.clone());
        assert!(old.is_none());
        raw_docs.push(doc);
    }

    let seed = [42; 32];
    let mut rng = StdRng::from_seed(seed);
    for _ in 0..updates {
        let i = rng.random_range(0..inserts);
        let doc = test_gen_update_raw_doc(raw_docs[i].handle.clone());
        let old = expected_docs.insert(doc.handle.clone(), doc.clone());
        assert!(old.is_some());
        raw_docs.push(doc);
    }

    let mut handles_to_delete = raw_docs
        .iter()
        .take(inserts)
        .map(|doc| doc.handle.clone())
        .collect::<Vec<_>>();
    for _ in 0..deletes {
        let i = rng.random_range(0..handles_to_delete.len());
        let doc = test_gen_delete_raw_doc(handles_to_delete[i].clone());
        let old = expected_docs.remove(&doc.handle);
        assert!(old.is_some());
        assert_eq!(handles_to_delete.remove(i), doc.handle);
        raw_docs.push(doc);
    }

    raw_docs.shuffle(&mut rng);
    group_by_handle_sort_by_version(&mut raw_docs);

    (raw_docs, expected_docs)
}

pub fn test_gen_raw_docs_numeric_only(doc_count: usize) -> Vec<RawDoc> {
    let mut raw_docs = Vec::with_capacity(doc_count);
    for i in 0..doc_count {
        let handle = test_gen_handle();
        let raw_doc = RawDoc {
            handle,
            version: 1,
            op_type: OperationType::Insert,
            cols: vec![
                (1, Some((rand::random::<u32>() % 2).to_string())),
                (2, Some((rand::random::<u32>() % 100).to_string())),
                (3, Some((rand::random::<u32>() % 10000).to_string())),
                (4, Some((rand::random::<u32>() % 1000000).to_string())),
                (5, Some((i as i64).to_string())),
                (6, Some((i as i64).to_string())),
            ],
            is_local_timezone: true,
        };
        raw_docs.push(raw_doc);
    }
    raw_docs
}

pub fn test_gen_raw_docs_for_all_types(doc_count: usize) -> Vec<RawDoc> {
    let mut raw_docs = Vec::with_capacity(doc_count);
    for i in 0..doc_count {
        let handle = test_gen_handle_with_prefix_and_num("all_type", i as u64);

        let raw_doc = RawDoc {
            handle,
            version: 1,
            op_type: OperationType::Insert,
            cols: vec![
                (1, Some((i as i64 + 1000).to_string())), // id (primary key, LONGLONG)
                (2, Some((i as i8).to_string())),         // col_tinyint (TINY)
                (3, Some((i as i64 * 2).to_string())),    // col_bigint (LONGLONG)
                (4, Some((i as f32 * -3.3).to_string())), // col_float (FLOAT)
                (5, Some((i as f64 * 3.3).to_string())),  // col_double (DOUBLE)
                (
                    6,
                    Some(format!("2023-{:02}-{:02}", (i % 11) + 1, (i % 27) + 1)),
                ), // col_date (DATE)
                (
                    7,
                    Some(format!(
                        "2023-{:02}-{:02} 10:00:00",
                        (i % 11) + 1,
                        (i % 27) + 1
                    )),
                ), // col_datetime (DATETIME)
                (8, Some(format!("varchar_value_{}", i))), // col_varchar (VARCHAR)
                (9, Some(format!("blob_value_{}", i))),   // col_blob (BLOB)
                (10, Some(format!("{:06}", (i as f64 * 3.33)))), // col_decimal (NEWDECIMAL)
            ],
            is_local_timezone: true,
        };
        raw_docs.push(raw_doc);
    }
    raw_docs
}