pub mod buffer;
pub mod codec_v2;
pub mod convert;
pub mod number;

use anyhow::{Result, bail};
use byteorder::{BigEndian, ByteOrder};

use crate::{
    codec::{
        codec_v2::{KEYSPACE_PREFIX_LEN, get_txn_keyspace_prefix, get_u32_keyspace_id_by_key},
        number::NumberDecoder,
    },
    index::IndexKey,
    keyspace::NULLSPACEID,
};

const SIGN_MASK: u64 = 0x8000000000000000;

fn encode_int_to_cmp_uint(v: i64) -> u64 {
    v as u64 ^ SIGN_MASK
}

fn encode_int(b: &mut [u8], v: i64) {
    let u = encode_int_to_cmp_uint(v);
    BigEndian::write_u64(&mut b[..std::mem::size_of::<i64>()], u);
}

const TABLE_PREFIX_KEY: &[u8] = b"t";
pub const TABLE_PREFIX_KEYLEN: usize = std::mem::size_of_val(TABLE_PREFIX_KEY);
pub const TABLE_KEY_LEN: usize = TABLE_PREFIX_KEYLEN + std::mem::size_of::<i64>();

pub fn table_start_key(index_key: &IndexKey) -> Vec<u8> {
    if index_key.keyspace_id == NULLSPACEID {
        // t[tableID]
        let mut key = [0; TABLE_KEY_LEN];
        key[..TABLE_PREFIX_KEYLEN].copy_from_slice(TABLE_PREFIX_KEY);
        encode_int(key[TABLE_PREFIX_KEYLEN..].as_mut(), index_key.table_id);
        key.to_vec()
    } else {
        // [keyspace]t[tableID]
        let mut key = [0; KEYSPACE_PREFIX_LEN + TABLE_KEY_LEN];
        const KEYSPACE_TABLE_START: usize = KEYSPACE_PREFIX_LEN + TABLE_PREFIX_KEYLEN;
        key[..KEYSPACE_PREFIX_LEN]
            .copy_from_slice(get_txn_keyspace_prefix(index_key.keyspace_id).as_slice());
        key[KEYSPACE_PREFIX_LEN..KEYSPACE_TABLE_START].copy_from_slice(TABLE_PREFIX_KEY);
        encode_int(key[KEYSPACE_TABLE_START..].as_mut(), index_key.table_id);
        key.to_vec()
    }
}

pub fn table_end_key(index_key: &IndexKey) -> Vec<u8> {
    table_start_key(&IndexKey::new(
        index_key.keyspace_id,
        index_key.table_id + 1,
        index_key.index_id,
    ))
}

pub fn table_raw_key(index_key: &IndexKey, v: &[u8]) -> Vec<u8> {
    // [keyspace]t[tableID][v...]
    let mut r = table_start_key(index_key).to_vec();
    r.extend(v.iter());
    r
}

fn previous(v: &mut [u8]) {
    for i in (0..v.len()).rev() {
        if v[i] == 0 {
            if i == 0 {
                panic!("minimal {:?}\n", v);
            } else {
                v[i] = 255;
            }
        } else {
            v[i] -= 1;
            return;
        }
    }
}

// Returns a vec with length v.len + extra, which is immediate precedent of v.
pub fn previous_key_helper(v: &[u8], extra: usize) -> Vec<u8> {
    let mut newv = Vec::with_capacity(v.len() + extra);
    newv.extend_from_slice(v);
    newv.extend(std::iter::repeat_n(0, extra));
    previous(newv.as_mut_slice());
    newv
}

// Parse table_id from start_key
pub fn parse_table_id_from_start_key(key: &[u8]) -> Option<i64> {
    let mut buf;

    if get_u32_keyspace_id_by_key(key).is_some() {
        buf = &key[KEYSPACE_PREFIX_LEN..];
    } else {
        buf = key;
    }

    if buf.is_empty() || buf[0] != b't' {
        return None;
    }

    buf = &buf[1..];
    let table_id = buf.read_i64().ok()?;
    Some(table_id)
}

// Wrapper for parsing table_id from sharding_key
pub fn parse_table_id_from_sharding_key(key: &[u8]) -> Option<i64> {
    parse_table_id_from_start_key(key)
}

// Wrapper for parsing table_id from ranges
pub fn parse_table_id_from_ranges<R, F>(ranges: &[R], mut get_start: F) -> Result<i64>
where
    R: std::fmt::Debug,
    F: FnMut(&R) -> &[u8],
{
    let mut table_id: Option<i64> = None;

    for range in ranges {
        if let Some(parsed) = parse_table_id_from_start_key(get_start(range)) {
            match table_id {
                None => table_id = Some(parsed),
                Some(existing) => {
                    if existing != parsed {
                        bail!(
                            "Inconsistent table_id in ranges, ranges={:?}, expected={}, found={}",
                            ranges,
                            existing,
                            parsed
                        );
                    }
                }
            }
        }
    }

    table_id.ok_or_else(|| anyhow::anyhow!("Parse ranges failed, ranges={:?}", ranges))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::codec::codec_v2::*;

    #[derive(Debug)]
    struct TestRange {
        start: Vec<u8>,
    }

    #[test]
    fn test_table_id_encode() {
        assert_eq!(
            table_start_key(&IndexKey::new_nullspace(1, 0)),
            [116, 128, 0, 0, 0, 0, 0, 0, 1]
        );
        assert_eq!(
            table_start_key(&IndexKey::new_nullspace(0, 0)),
            [116, 128, 0, 0, 0, 0, 0, 0, 0]
        );
        assert_eq!(
            table_start_key(&IndexKey::new_nullspace(-1, 0)),
            [116, 127, 255, 255, 255, 255, 255, 255, 255]
        );
    }

    #[test]
    fn test_previous() {
        assert_eq!(previous_key_helper(&[1, 1, 1], 2), vec![1, 1, 0, 255, 255]);
        assert_eq!(previous_key_helper(&[1, 1, 1], 0), vec![1, 1, 0]);
        let result = std::panic::catch_unwind(|| {
            let mut v = vec![0, 0, 0];
            previous(&mut v);
        });
        assert!(result.is_err());
    }

    #[test]
    fn test_keyspace_id() {
        let encoded = table_start_key(&IndexKey::new(111, 222, 333));
        assert_eq!(encoded.len(), KEYSPACE_PREFIX_LEN + TABLE_KEY_LEN);
        assert_eq!(
            get_keyspace_table_id(encoded.as_slice()),
            Some((111u32, 222i64))
        );
        let encoded = table_start_key(&IndexKey::new_nullspace(888, 999));
        assert_eq!(get_keyspace_table_id(encoded.as_slice()), None);
    }

    #[test]
    fn test_parse_table_id_from_start_key() {
        assert_eq!(parse_table_id_from_start_key(&[]), None);
        assert_eq!(parse_table_id_from_start_key(&[116]), None);
        assert_eq!(
            parse_table_id_from_start_key(&[116, 128, 0, 0, 0, 0, 0, 0]),
            None
        );
        assert_eq!(
            parse_table_id_from_start_key(&[116, 128, 0, 0, 0, 0, 0, 0, 1]),
            Some(1)
        );
        assert_eq!(
            parse_table_id_from_start_key(&[116, 128, 0, 0, 0, 0, 0, 0, 0]),
            Some(0)
        );
        assert_eq!(
            parse_table_id_from_start_key(&[116, 127, 255, 255, 255, 255, 255, 255, 255]),
            Some(-1)
        );
        assert_eq!(
            parse_table_id_from_start_key(&[
                116, 128, 0, 0, 0, 0, 0, 2, 187, 95, 114, 128, 0, 0, 0, 0, 0, 0, 2
            ]),
            Some(699)
        );
        assert_eq!(
            parse_table_id_from_start_key(&[
                120, 0, 0, 1, 116, 128, 0, 0, 0, 0, 0, 2, 188, 95, 114, 128, 0, 0, 0, 0, 0, 0, 2
            ]),
            Some(700)
        );
        assert_eq!(
            parse_table_id_from_start_key(&[
                120, 0, 0, 1, 115, 128, 0, 0, 0, 0, 0, 2, 188, 95, 114, 128, 0, 0, 0, 0, 0, 0, 2
            ]),
            None
        );
        assert_eq!(
            parse_table_id_from_start_key(&[120, 0, 0, 1, 116, 128, 0, 0, 0, 0, 0, 2]),
            None
        );
        assert_eq!(
            parse_table_id_from_start_key(&[120, 0, 0, 1, 116, 128, 0, 0, 0, 0, 0, 2, 188]),
            Some(700)
        );
    }

    #[test]
    fn test_parse_table_id_from_ranges() {
        let ranges = vec![
            TestRange {
                start: vec![116, 128, 0, 0, 0, 0, 0, 0, 1],
            },
            TestRange {
                start: vec![116, 128, 0, 0, 0, 0, 0, 0, 1],
            },
        ];
        assert_eq!(
            parse_table_id_from_ranges(&ranges, |range| range.start.as_slice()).unwrap(),
            1
        );
    }

    #[test]
    fn test_parse_table_id_from_ranges_inconsistent() {
        let ranges = vec![
            TestRange {
                start: vec![116, 128, 0, 0, 0, 0, 0, 0, 1],
            },
            TestRange {
                start: vec![116, 128, 0, 0, 0, 0, 0, 0, 2],
            },
        ];
        let err = parse_table_id_from_ranges(&ranges, |range| range.start.as_slice())
            .unwrap_err()
            .to_string();
        assert!(err.contains("Inconsistent table_id in ranges"));
    }
}