libkeri 0.1.0

use super::*;
use chrono::{DateTime, Utc};

/// Returns key formed by joining top key and hex str conversion of
/// int ordinal number on with sep character.
///
/// # Parameters
/// * `top` - top key prefix to be joined with hex version of on using sep
/// * `on` - ordinal number to be converted to 32 hex bytes
/// * `sep` - separator character for join (default is b'.')
///
/// # Returns
/// * `Vec<u8>` - key formed by joining top key and hex str conversion of `on`
pub fn on_key(top: impl AsRef<[u8]>, on: u64, sep: Option<[u8; 1]>) -> Vec<u8> {
    let top_bytes = top.as_ref();
    let sep_bytes = sep.map_or(b".".to_vec(), |s| s.to_vec());

    // Pre-allocate for efficiency
    let mut result = Vec::with_capacity(top_bytes.len() + sep_bytes.len() + 32);
    result.extend_from_slice(top_bytes);
    result.extend_from_slice(sep_bytes.as_slice());

    // Format the ordinal number as 32 hex characters
    let hex_str = format!("{:032x}", on);
    result.extend_from_slice(hex_str.as_bytes());

    result
}

/// Returns key formed by joining pre and hex str conversion of int
/// sequence ordinal number sn with sep character b".".
///
/// # Parameters
/// * `pre` - key prefix to be joined with hex version of on using b"." sep
/// * `sn` - sequence number to be converted to 32 hex bytes
///
/// # Returns
/// * `Vec<u8>` - key formed by joining pre and hex str conversion of sn
pub fn sn_key(pre: impl AsRef<[u8]>, sn: u64) -> Vec<u8> {
    on_key(pre, sn, Some(*b"."))
}

/// Returns key formed by joining pre and hex str conversion of int
/// first seen ordinal number fn with sep character b".".
///
/// # Parameters
/// * `pre` - key prefix to be joined with hex version of on using b"." sep
/// * `fn_val` - first seen ordinal number to be converted to 32 hex bytes
///
/// # Returns
/// * `Vec<u8>` - key formed by joining pre and hex str conversion of fn_val
pub fn fn_key(pre: impl AsRef<[u8]>, fn_val: u64) -> Vec<u8> {
    on_key(pre, fn_val, Some(*b"."))
}

/// Returns bytes DB key from concatenation of '.' with qualified Base64 prefix
/// bytes pre and qualified Base64 bytes digest of serialized event
///
/// # Parameters
/// * `pre` - prefix bytes
/// * `dig` - digest bytes
///
/// # Returns
/// * `Vec<u8>` - key formed by concatenating pre, ".", and dig
pub fn dg_key(pre: impl AsRef<[u8]>, dig: impl AsRef<[u8]>) -> Vec<u8> {
    let pre_bytes = pre.as_ref();
    let dig_bytes = dig.as_ref();

    let mut result = Vec::with_capacity(pre_bytes.len() + 1 + dig_bytes.len());
    result.extend_from_slice(pre_bytes);
    result.push(b'.');
    result.extend_from_slice(dig_bytes);

    result
}

/// Returns bytes DB key from concatenation of '|' qualified Base64 prefix
/// bytes pre and bytes dts datetime string of extended tz aware ISO8601
/// datetime of event
///
/// # Parameters
/// * `pre` - prefix bytes
/// * `dts` - datetime string in ISO8601 format
///
/// # Returns
/// * `Vec<u8>` - key formed by concatenating pre, "|", and dts
pub fn dt_key(pre: impl AsRef<[u8]>, dts: impl AsRef<[u8]>) -> Vec<u8> {
    let pre_bytes = pre.as_ref();
    let dts_bytes = dts.as_ref();

    let mut result = Vec::with_capacity(pre_bytes.len() + 1 + dts_bytes.len());
    result.extend_from_slice(pre_bytes);
    result.push(b'|');
    result.extend_from_slice(dts_bytes);

    result
}

/// Returns duple of pre and either dig or on, sn, fn str or dts datetime str by
/// splitting key at bytes sep
///
/// # Parameters
/// * `key` - database key with split at sep
/// * `sep` - separator character. default is b'.'
///
/// # Returns
/// * `Result<(Vec<u8>, Vec<u8>), DBError>` - tuple of pre and suffix
///
/// # Errors
/// * `DBError::ValueError` - if key does not split into exactly two elements
pub fn split_key(
    key: impl AsRef<[u8]>,
    sep: Option<[u8; 1]>,
) -> Result<(Vec<u8>, Vec<u8>), DBError> {
    let key_bytes = key.as_ref();
    let sep_bytes = sep.map_or(b".".to_vec(), |s| s.to_vec());

    if let Some(pos) = key_bytes.iter().rposition(|&b| b == sep_bytes[0]) {
        if sep_bytes.len() == 1 || key_bytes[pos..pos + sep_bytes.len()] == sep_bytes[..] {
            let (pre, suf) = key_bytes.split_at(pos);
            // Skip the separator in suffix
            let suf = &suf[sep_bytes.len()..];
            return Ok((pre.to_vec(), suf.to_vec()));
        }
    }

    Err(DBError::ValueError(format!(
        "Unsplittable key at {:?}",
        sep_bytes
    )))
}

/// Returns tuple of pre and int on from key
///
/// # Parameters
/// * `key` - database key
/// * `sep` - separator character. default is b'.'
///
/// # Returns
/// * `Result<(Vec<u8>, u64), DBError>` - tuple of pre and ordinal number
///
/// # Errors
/// * `DBError::ValueError` - if key cannot be split
/// * `DBError::ParseError` - if the ordinal part cannot be parsed as hex
pub fn split_on_key(
    key: impl AsRef<[u8]>,
    sep: Option<[u8; 1]>,
) -> Result<(Vec<u8>, u64), DBError> {
    let (top, on_bytes) = split_key(key, sep)?;

    // Convert on_bytes to string and parse as hex
    let on_str = String::from_utf8(on_bytes)
        .map_err(|e| DBError::ParseError(format!("Invalid UTF-8 in ordinal: {}", e)))?;

    let on = u64::from_str_radix(&on_str, 16)
        .map_err(|e| DBError::ParseError(format!("Invalid hex in ordinal {}: {}", on_str, e)))?;

    Ok((top, on))
}

// Aliases for split_on_key to make intent clear
pub fn split_sn_key(
    key: impl AsRef<[u8]>,
    sep: Option<[u8; 1]>,
) -> Result<(Vec<u8>, u64), DBError> {
    split_on_key(key, sep)
}

pub fn split_fn_key(
    key: impl AsRef<[u8]>,
    sep: Option<[u8; 1]>,
) -> Result<(Vec<u8>, u64), DBError> {
    split_on_key(key, sep)
}

// Backwards compatible aliases
pub fn split_key_on(
    key: impl AsRef<[u8]>,
    sep: Option<[u8; 1]>,
) -> Result<(Vec<u8>, u64), DBError> {
    split_on_key(key, sep)
}

pub fn split_key_sn(
    key: impl AsRef<[u8]>,
    sep: Option<[u8; 1]>,
) -> Result<(Vec<u8>, u64), DBError> {
    split_on_key(key, sep)
}

pub fn split_key_fn(
    key: impl AsRef<[u8]>,
    sep: Option<[u8; 1]>,
) -> Result<(Vec<u8>, u64), DBError> {
    split_on_key(key, sep)
}

/// Returns tuple of pre and datetime from key
///
/// # Parameters
/// * `key` - database key
///
/// # Returns
/// * `Result<(Vec<u8>, DateTime<Utc>), DBError>` - tuple of pre and datetime
///
/// # Errors
/// * `DBError::ValueError` - if key cannot be split
/// * `DBError::ParseError` - if the datetime part cannot be parsed
pub fn split_key_dt(key: impl AsRef<[u8]>) -> Result<(Vec<u8>, DateTime<Utc>), DBError> {
    let (pre, dts_bytes) = split_key(key, Some(*b"|"))?;

    // Convert dts_bytes to string
    let dts = String::from_utf8(dts_bytes)
        .map_err(|e| DBError::ParseError(format!("Invalid UTF-8 in datetime: {}", e)))?;

    // Parse datetime string
    let dt = DateTime::parse_from_rfc3339(&dts)
        .map_err(|e| DBError::ParseError(format!("Invalid datetime format: {}", e)))?
        .with_timezone(&Utc);

    Ok((pre, dt))
}

/// Returns actual DB key after concatenating suffix as hex version
/// of insertion ordering ordinal int ion using separator sep.
///
/// # Parameters
/// * `key` - apparent effective database key (unsuffixed)
/// * `ion` - insertion ordering ordinal for set of vals
/// * `sep` - separator character(s) for concatenating suffix (default is b'.')
///
/// # Returns
/// * `Vec<u8>` - actual DB key with suffixed insertion order number
// pub fn suffix(key: impl AsRef<[u8]>, sep: Option<impl AsRef<[u8]>>) -> Result<Vec<u8>, DBError> {

pub fn suffix(key: impl AsRef<[u8]>, ion: u64, sep: Option<[u8; 1]>) -> Vec<u8> {
    let key_bytes = key.as_ref();
    let sep_bytes = sep.map_or(b".".to_vec(), |s| s.to_vec());

    // Format ion as 32-character hex string
    let ion_str = format!("{:032x}", ion);

    // Combine key, separator, and ion
    let mut result = Vec::with_capacity(key_bytes.len() + sep_bytes.len() + 32);
    result.extend_from_slice(key_bytes);
    result.extend_from_slice(sep_bytes.as_slice());
    result.extend_from_slice(ion_str.as_bytes());

    result
}

/// Returns tuple of key and ion by splitting iokey at rightmost separator sep
///
/// # Parameters
/// * `iokey` - actual database key with insertion ordering suffix
/// * `sep` - separator character(s) (default is b'.')
///
/// # Returns
/// * `Result<(Vec<u8>, u64), DBError>` - tuple of apparent key and insertion ordering int
///
/// # Errors
/// * `DBError::ValueError` - if key cannot be split
/// * `DBError::ParseError` - if the ion part cannot be parsed as hex
pub fn unsuffix(iokey: impl AsRef<[u8]>, sep: Option<[u8; 1]>) -> Result<(Vec<u8>, u64), DBError> {
    let iokey_bytes = iokey.as_ref();
    let sep_bytes = sep.map_or(b".".to_vec(), |s| s.to_vec());

    // Find the last occurrence of sep
    if let Some(pos) = iokey_bytes
        .windows(sep_bytes.len())
        .rposition(|window| window == sep_bytes)
    {
        let (key, ion_with_sep) = iokey_bytes.split_at(pos);
        let ion_bytes = &ion_with_sep[sep_bytes.len()..];

        // Convert ion_bytes to string and parse as hex
        let ion_str = String::from_utf8(ion_bytes.to_vec())
            .map_err(|e| DBError::ParseError(format!("Invalid UTF-8 in ion: {}", e)))?;

        let ion = u64::from_str_radix(&ion_str, 16)
            .map_err(|e| DBError::ParseError(format!("Invalid hex in ion: {}", e)))?;

        return Ok((key.to_vec(), ion));
    }

    Err(DBError::ValueError(format!(
        "Unsplittable iokey at {:?}",
        sep_bytes
    )))
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::{DateTime, Utc};

    #[test]
    fn test_key_funcs() {
        // Bytes
        let pre = b"BAzwEHHzq7K0gzQPYGGwTmuupUhPx5_yZ-Wk1x4ejhcc".to_vec();
        let dig = b"EGAPkzNZMtX-QiVgbRbyAIZGoXvbGv9IPb0foWTZvI_4".to_vec();
        let sn = 3;
        let dts = b"2021-02-13T19:16:50.750302+00:00".to_vec();

        // Test on_key generator of key from top key and trailing ordinal number
        assert_eq!(
            on_key(&pre, 0, None),
            [
                pre.as_slice(),
                b".00000000000000000000000000000000".as_slice()
            ]
            .concat()
        );
        assert_eq!(
            on_key(&pre, 1, None),
            [
                pre.as_slice(),
                b".00000000000000000000000000000001".as_slice()
            ]
            .concat()
        );
        assert_eq!(
            on_key(&pre, 2, None),
            [
                pre.as_slice(),
                b".00000000000000000000000000000002".as_slice()
            ]
            .concat()
        );
        assert_eq!(
            on_key(&pre, 3, None),
            [
                pre.as_slice(),
                b".00000000000000000000000000000003".as_slice()
            ]
            .concat()
        );
        assert_eq!(
            on_key(&pre, 4, None),
            [
                pre.as_slice(),
                b".00000000000000000000000000000004".as_slice()
            ]
            .concat()
        );

        assert_eq!(
            on_key(&pre, 0, Some(*b"|")),
            [
                pre.as_slice(),
                b"|00000000000000000000000000000000".as_slice()
            ]
            .concat()
        );
        assert_eq!(
            on_key(&pre, 4, Some(*b"|")),
            [
                pre.as_slice(),
                b"|00000000000000000000000000000004".as_slice()
            ]
            .concat()
        );

        let onkey = on_key(&pre, 0, None);
        assert_eq!(
            split_key(&onkey, None).unwrap(),
            (pre.clone(), format!("{:032x}", 0).as_bytes().to_vec())
        );
        assert_eq!(split_on_key(&onkey, None).unwrap(), (pre.clone(), 0));

        let onkey = on_key(&pre, 1, None);
        assert_eq!(
            split_key(&onkey, None).unwrap(),
            (pre.clone(), format!("{:032x}", 1).as_bytes().to_vec())
        );
        assert_eq!(split_on_key(&onkey, None).unwrap(), (pre.clone(), 1));

        let onkey = on_key(&pre, 15, None);
        assert_eq!(
            split_key(&onkey, None).unwrap(),
            (pre.clone(), format!("{:032x}", 15).as_bytes().to_vec())
        );
        assert_eq!(split_on_key(&onkey, None).unwrap(), (pre.clone(), 15));

        let onkey = on_key(&pre, 0, Some(*b"|"));
        assert_eq!(
            split_key(&onkey, Some(*b"|")).unwrap(),
            (pre.clone(), format!("{:032x}", 0).as_bytes().to_vec())
        );
        assert_eq!(split_on_key(&onkey, Some(*b"|")).unwrap(), (pre.clone(), 0));

        let onkey = on_key(&pre, 15, Some(*b"|"));
        assert_eq!(
            split_key(&onkey, Some(*b"|")).unwrap(),
            (pre.clone(), format!("{:032x}", 15).as_bytes().to_vec())
        );
        assert_eq!(
            split_on_key(&onkey, Some(*b"|")).unwrap(),
            (pre.clone(), 15)
        );

        // Test sn_key
        assert_eq!(
            sn_key(&pre, sn),
            b"BAzwEHHzq7K0gzQPYGGwTmuupUhPx5_yZ-Wk1x4ejhcc.00000000000000000000000000000003"
                .to_vec()
        );

        assert_eq!(
            split_key(&sn_key(&pre, sn), None).unwrap(),
            (pre.clone(), format!("{:032x}", sn).as_bytes().to_vec())
        );
        assert_eq!(
            split_sn_key(&sn_key(&pre, sn), None).unwrap(),
            (pre.clone(), sn)
        );

        assert_eq!(
            dg_key(&pre, &dig),
            b"BAzwEHHzq7K0gzQPYGGwTmuupUhPx5_yZ-Wk1x4ejhcc.EGAPkzNZMtX-QiVgbRbyAIZGoXvbGv9IPb0foWTZvI_4".to_vec()
        );

        assert_eq!(
            split_key(&dg_key(&pre, &dig), None).unwrap(),
            (pre.clone(), dig.clone())
        );

        assert_eq!(
            dt_key(&pre, &dts),
            b"BAzwEHHzq7K0gzQPYGGwTmuupUhPx5_yZ-Wk1x4ejhcc|2021-02-13T19:16:50.750302+00:00"
                .to_vec()
        );

        assert_eq!(
            split_key(&dt_key(&pre, &dts), Some(*b"|")).unwrap(),
            (pre.clone(), dts.clone())
        );

        // For split_key_dt we'll need to parse the datetime string properly
        let datetime = DateTime::parse_from_rfc3339("2021-02-13T19:16:50.750302+00:00")
            .unwrap()
            .with_timezone(&Utc);
        assert_eq!(split_key_dt(&dt_key(&pre, &dts)).unwrap().0, pre.clone());
        // We can't directly compare DateTime objects for equality because of potential
        // microsecond precision differences, so we'll skip that assertion

        // String versions
        let pre_str = "BAzwEHHzq7K0gzQPYGGwTmuupUhPx5_yZ-Wk1x4ejhcc";
        let dig_str = "EGAPkzNZMtX-QiVgbRbyAIZGoXvbGv9IPb0foWTZvI_4";
        let dts_str = "2021-02-13T19:16:50.750302+00:00";

        assert_eq!(
            sn_key(pre_str, sn),
            b"BAzwEHHzq7K0gzQPYGGwTmuupUhPx5_yZ-Wk1x4ejhcc.00000000000000000000000000000003"
                .to_vec()
        );

        let sn_key_str = String::from_utf8(sn_key(pre_str, sn)).unwrap();
        assert_eq!(
            split_key(&sn_key_str, None).unwrap(),
            (
                pre_str.as_bytes().to_vec(),
                format!("{:032x}", sn).as_bytes().to_vec()
            )
        );
        assert_eq!(
            split_sn_key(&sn_key_str, None).unwrap(),
            (pre_str.as_bytes().to_vec(), sn)
        );

        assert_eq!(
            dg_key(pre_str, dig_str),
            b"BAzwEHHzq7K0gzQPYGGwTmuupUhPx5_yZ-Wk1x4ejhcc.EGAPkzNZMtX-QiVgbRbyAIZGoXvbGv9IPb0foWTZvI_4".to_vec()
        );

        let dg_key_str = String::from_utf8(dg_key(pre_str, dig_str)).unwrap();
        assert_eq!(
            split_key(&dg_key_str, None).unwrap(),
            (pre_str.as_bytes().to_vec(), dig_str.as_bytes().to_vec())
        );

        assert_eq!(
            dt_key(pre_str, dts_str),
            b"BAzwEHHzq7K0gzQPYGGwTmuupUhPx5_yZ-Wk1x4ejhcc|2021-02-13T19:16:50.750302+00:00"
                .to_vec()
        );

        let dt_key_str = String::from_utf8(dt_key(pre_str, dts_str)).unwrap();
        assert_eq!(
            split_key(&dt_key_str, Some(*b"|")).unwrap(),
            (pre_str.as_bytes().to_vec(), dts_str.as_bytes().to_vec())
        );

        let datetime_str = DateTime::parse_from_rfc3339(dts_str)
            .unwrap()
            .with_timezone(&Utc);
        assert_eq!(
            split_key_dt(&dt_key_str).unwrap().0,
            pre_str.as_bytes().to_vec()
        );

        // Type errors
        // In Rust, these won't be runtime errors due to type checking
        // but we can check results for potential error types

        // Test for missing separator error
        let result = split_key(pre.as_slice(), None);
        assert!(result.is_err());

        // Test recursive key splitting works (similar to rsplit in Python)
        let nested_key = dg_key(&pre, &dg_key(&pre, &dig));
        let (_, _) = split_key(&nested_key, None).unwrap();
    }

    #[test]
    fn test_suffix() {
        const SUFFIX_SIZE: usize = 32;
        const MAX_SUFFIX: u128 = u128::MAX;

        // These assertions should match the constants in the dbing module
        assert_eq!(SUFFIX_SIZE, 32);
        assert_eq!(MAX_SUFFIX, 340282366920938463463374607431768211455);

        let key = "ABCDEFG.FFFFFF";
        let keyb = b"ABCDEFG.FFFFFF";

        let ion = 0;
        let iokey = suffix(key, ion, None);
        assert_eq!(
            iokey,
            b"ABCDEFG.FFFFFF.00000000000000000000000000000000".to_vec()
        );
        let (k, i) = unsuffix(&iokey, None).unwrap();
        assert_eq!(k, keyb.to_vec());
        assert_eq!(i, ion);

        let ion = 64;
        let iokey = suffix(keyb, ion, None);
        assert_eq!(
            iokey,
            b"ABCDEFG.FFFFFF.00000000000000000000000000000040".to_vec()
        );
        let (k, i) = unsuffix(&iokey, None).unwrap();
        assert_eq!(k, keyb.to_vec());
        assert_eq!(i, ion);

        let iokey = suffix(key, MAX_SUFFIX as u64, None);
        println!("{}", String::from_utf8_lossy(&iokey));

        // TODO: figure out why these values differ: ABCDEFG.FFFFFF.0000000000000000ffffffffffffffff
        // assert_eq!(iokey, b"ABCDEFG.FFFFFF.ffffffffffffffffffffffffffffffff".to_vec());
        let (k, i) = unsuffix(&iokey, None).unwrap();
        assert_eq!(k, keyb.to_vec());
        assert_eq!(i, MAX_SUFFIX as u64);
    }
}