armour 0.30.27

use std::{
    cell::Cell,
    fmt::Debug,
    marker::PhantomData,
    mem::size_of,
    str::FromStr,
    sync::{
        LazyLock,
        atomic::{AtomicU8, Ordering},
    },
    time::{Duration, SystemTime},
};

use rapira::{Rapira, RapiraError};
use rend::NonZeroU64_be;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
#[cfg(feature = "sled")]
use sled::InlineArray;
use zerocopy::{Immutable, IntoBytes, KnownLayout, TryFromBytes};

use super::{
    ArmourError, IdStr, Result,
    enc::IdHasher,
    fuid::{
        SHARD_BITS, SHARD_ID_MAX, SHARD_INSTANCE_ID_MAX, SHARD_THREAD_BITS, SHARD_THREAD_ID_MAX,
    },
    num_ops::g4bits,
};
use crate::{Cid, GetType, KeyScheme, KeyType, Typ};

const LOW_ID_BITS: u32 = 56;

/// different from the unix timestamp epoch in secs
///
/// 2023-04-09 0:00:00 UTC
const TS_DIFF: u64 = 0x6432_0000;

const TS_BITS: u32 = 32;
const TS_MAX: u64 = (1 << TS_BITS) - 1;
const TS_SHIFT: u32 = LOW_ID_BITS - TS_BITS;
const NANOS_DELIM: u128 = 100_000_000;

const SEQ_BITS: u32 = LOW_ID_BITS - TS_BITS - SHARD_BITS;
const SEQ_MAX: u64 = (1 << SEQ_BITS) - 1;

// 5 + 16 = 21
const SHARD_INST_SHIFT: u32 = SHARD_THREAD_BITS + SEQ_BITS;

#[derive(Debug, Clone, Copy)]
struct SeqForMs {
    /// timestamp in 0.1 secs
    ts: u64,
    seq: u64,
}

impl SeqForMs {
    #[cfg(feature = "std")]
    #[inline]
    fn with_ts(ts: u64) -> Self {
        SeqForMs { ts, seq: 0 }
    }

    #[cold]
    fn wait_inc(self, ts: u64, duration: Duration) -> Self {
        info!(?duration, "wait_inc");
        const NANOS: u32 = NANOS_DELIM as u32;
        let nanos = duration.subsec_nanos() % NANOS;
        let wait_nanos = NANOS - nanos;
        std::thread::sleep(Duration::from_nanos(wait_nanos as u64));

        SeqForMs::with_ts(ts + 1)
    }

    #[inline]
    fn increment(mut self, ts: u64, duration: Duration) -> Self {
        if self.seq == SEQ_MAX {
            self.wait_inc(ts, duration)
        } else {
            self.seq += 1;
            self
        }
    }
}

static NEW_EPOCH: LazyLock<SystemTime> =
    LazyLock::new(|| SystemTime::UNIX_EPOCH + Duration::from_secs(TS_DIFF));

// for check timestamp correctness
static NOW_PLUS_ONE_YEAR: LazyLock<u64> = LazyLock::new(|| {
    let now = SystemTime::now();
    #[allow(clippy::unwrap_used)]
    let now = now.duration_since(SystemTime::UNIX_EPOCH).unwrap();
    let now = now + Duration::from_secs(86400 * 365);
    let now = now.as_nanos();
    let now = (now / NANOS_DELIM) as u64;
    if now > TS_MAX { TS_MAX } else { now }
});

thread_local! {
    static SEQ_CHECK: Cell<SeqForMs> = Cell::new(SeqForMs::with_ts(0));
    static THREAD_ID: Cell<u8> = Cell::new(THREAD_SEQ.fetch_add(1, Ordering::Relaxed));
}

static THREAD_SEQ: AtomicU8 = AtomicU8::new(0);

type DbBytes = [u8; 7];
type MemoryBytes = [u8; 8];

/// - 56 bit id in NonZeroU64 [--zero(8)--|--timestamp(32)--|--shard_id(8)--|--sequence_id(16)--]
/// - optimized for InlineArray type
/// - first byte is 0x00
/// - 32 bit - timestamp (in 0.1 seconds) started from 2023-04-09 0:00:00 UTC, 13 years available (up to 2036)
/// - 8 bit - shard_id ([datacenter_id / instance_id] with thread_id), 256 values
/// - 16 bit - random sequence id (unique for thread / per thread), 65_536 values
/// - Blowfish for encryption
/// - ZBASE32 for encoding into 13 length string
/// - create ~ 1us (~ 1k in 1ms or 655_360 in 1s) in 1 thread
///
/// Notes:
/// - checks timestamp and shard_id when deserialized from string
/// - LowId is inlined for InlineArray type
/// - use only individually, it does not make sense to use in the composition
#[derive(Hash, IntoBytes, FromBytes, Immutable, KnownLayout)]
#[cfg_attr(
    feature = "rkyv",
    derive(rkyv::Archive, rkyv::Serialize, rkyv::Deserialize)
)]
#[repr(transparent)]
pub struct LowId<H>(pub NonZeroU64_be, PhantomData<H>);

impl<T> Clone for LowId<T> {
    fn clone(&self) -> Self {
        *self
    }
}

impl<T> Copy for LowId<T> {}

impl<H> LowId<H> {
    pub fn from_thread() -> Self {
        let thread_id = THREAD_ID.get();
        Self::new(thread_id as u64)
    }

    /// - 3 bit - instance id (datacenter id / shard id) (max 7)
    /// - 5 bit - thread id (max 31)
    #[cfg(feature = "std")]
    #[inline]
    pub fn with_shard_thread(instance_id: u64, thread_id: u64) -> Self {
        assert!(instance_id <= SHARD_INSTANCE_ID_MAX);
        assert!(thread_id <= SHARD_THREAD_ID_MAX);

        let shard_id = (instance_id << SHARD_THREAD_BITS) | thread_id;
        Self::new(shard_id)
    }

    /// - shard_id < 256
    #[cfg(feature = "std")]
    pub fn new(shard_id: u64) -> Self {
        assert!(shard_id <= SHARD_ID_MAX);

        let now = SystemTime::now();
        #[allow(clippy::unwrap_used)]
        let duration = now.duration_since(*NEW_EPOCH).unwrap();

        let timestamp = duration.as_nanos();
        let millis100 = (timestamp / NANOS_DELIM) as u64;

        let id = millis100 << TS_SHIFT;

        let id = id | (shard_id << SEQ_BITS);

        let mut seq_check = SEQ_CHECK.get();

        if seq_check.ts == millis100 {
            seq_check = seq_check.increment(millis100, duration);
        } else {
            seq_check = SeqForMs::with_ts(millis100);
        }
        SEQ_CHECK.set(seq_check);

        let id = id | seq_check.seq;

        Self(unsafe { NonZeroU64_be::new_unchecked(id) }, PhantomData)
    }

    /// timestamp in 0.1 secs (1 unit = 100 ms)
    #[inline]
    pub fn timestamp(&self) -> u64 {
        let ts = self.0.get() >> TS_SHIFT;
        ts + (TS_DIFF * 10)
    }

    pub fn date(&self) -> time::OffsetDateTime {
        let ts = self.timestamp();
        let ts = ts as i128;
        let ts = ts * 1_000_000 * 100;
        let dt = time::OffsetDateTime::from_unix_timestamp_nanos(ts);
        dt.expect("invalid timestamp")
    }

    #[inline]
    pub fn get(&self) -> u64 {
        self.0.get()
    }

    #[inline]
    pub fn to_le_bytes(self) -> MemoryBytes {
        self.0.get().to_le_bytes()
    }

    #[inline]
    pub fn to_be_bytes(self) -> [u8; 8] {
        zerocopy::transmute!(self)
    }

    #[inline]
    pub fn from_be_bytes(bytes: [u8; 8]) -> Result<Self> {
        zerocopy::try_transmute!(bytes).map_err(|_| ArmourError::NonZeroError)
    }

    #[inline]
    pub fn instance_id(&self) -> u64 {
        let ts = self.0.get() >> SHARD_INST_SHIFT;
        ts & SHARD_INSTANCE_ID_MAX
    }

    #[inline]
    pub fn thread_id(&self) -> u64 {
        let ts = self.0.get() >> SEQ_BITS;
        ts & SHARD_THREAD_ID_MAX
    }

    #[inline]
    pub fn shard_id(&self) -> u64 {
        let ts = self.0.get() >> SEQ_BITS;
        ts & SHARD_ID_MAX
    }

    #[inline]
    fn seq(&self) -> u64 {
        let ts = self.0.get();
        ts & SEQ_MAX
    }

    /// id: {id:#x}; {id:#b}; {dt}-{instance_id}-{thread_id}-{seq}
    #[cfg(feature = "std")]
    pub fn format<W>(&self, w: &mut W) -> Result<(), core::fmt::Error>
    where
        W: std::fmt::Write,
    {
        let dt = self.date();
        let instance_id = self.instance_id();
        let thread_id = self.thread_id();
        let seq = self.seq();
        let id = self.0.get();
        write!(
            w,
            "id: {id:#x}; {id:#b}; {dt}-{instance_id}-{thread_id}-{seq}"
        )
    }

    /// return prefix for date [--timestamp(5)--|0x00(1)|0x0000(2)]
    /// ms - unix timestamp in milliseconds
    pub fn date_prefix(ms: u64) -> [u8; 8] {
        let ts = (ms - (TS_DIFF * 1000)) << TS_SHIFT;
        ts.to_be_bytes()
    }
}

impl<H: IdHasher> LowId<H> {
    pub fn ser(&self) -> IdStr {
        let u: u64 = zerocopy::transmute!(*self);
        H::ser(u)
    }

    pub fn deser(id: &str) -> Result<Self, ArmourError> {
        let id = H::deser(id)?;
        zerocopy::try_transmute!(id).map_err(|_| ArmourError::NonZeroError)
    }

    pub fn check(&self) -> Result<(), ArmourError> {
        let ts = self.0.get() >> TS_SHIFT;
        // timestamp cannot be in the future
        if ts > *NOW_PLUS_ONE_YEAR {
            return Err(ArmourError::IdDecodeError);
        }
        Ok(())
    }
}

impl<T> PartialOrd for LowId<T> {
    fn partial_cmp(&self, other: &Self) -> Option<core::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

impl<T> Ord for LowId<T> {
    fn cmp(&self, other: &Self) -> core::cmp::Ordering {
        self.0.cmp(&other.0)
    }
}

impl<T> PartialEq for LowId<T> {
    fn eq(&self, other: &Self) -> bool {
        self.0 == other.0
    }
}

impl<T> Eq for LowId<T> {}

/// only check if non-zero, not check valid byte mask
impl<H> Rapira for LowId<H> {
    const STATIC_SIZE: Option<usize> = Some(size_of::<MemoryBytes>());
    const MIN_SIZE: usize = size_of::<MemoryBytes>();

    fn size(&self) -> usize {
        size_of::<MemoryBytes>()
    }

    fn check_bytes(slice: &mut &[u8]) -> rapira::Result<()> {
        let bytes: &[u8] = slice.get(..8).ok_or(RapiraError::SliceLen)?;

        if bytes == [0u8; 8] {
            return Err(RapiraError::NonZero);
        }

        *slice = unsafe { slice.get_unchecked(8..) };
        Ok(())
    }

    fn from_slice(slice: &mut &[u8]) -> rapira::Result<Self>
    where
        Self: Sized,
    {
        let bytes = <[u8; 8]>::from_slice(slice)?;
        let id = zerocopy::try_transmute!(bytes).map_err(|_| RapiraError::NonZero)?;
        Ok(id)
    }

    fn convert_to_bytes(&self, slice: &mut [u8], cursor: &mut usize) {
        let bytes: &[u8; 8] = zerocopy::transmute_ref!(self);
        bytes.convert_to_bytes(slice, cursor);
    }
}

impl<H> GetType for LowId<H> {
    const TYPE: Typ = Typ::LowId;
}

impl<H> Cid for LowId<H> {
    type B = InlineArray;
    const TY: KeyScheme = KeyScheme::Typed(&[KeyType::Array(size_of::<DbBytes>())]);
    // 24 bit timestamp in 0.1 sec represent 19 days
    // другими словами, первые 8 бит - уникальные для группы. Одна группа будет содержать идентификаторы, созданные в окне из 19 дней. В году будет 19 групп.
    // 8 bit - 256
    // 24 bit - 16_777_216
    const GROUP_BITS: u32 = 8;

    fn encode(&self) -> Self::B {
        let bytes: [u8; 8] = zerocopy::transmute!(*self);
        InlineArray::from(&bytes[1..])
    }

    #[inline]
    fn encode_owned(self) -> Self::B {
        let bytes: [u8; 8] = zerocopy::transmute!(self);
        InlineArray::from(&bytes[1..])
    }

    fn decode(bytes: &Self::B) -> Result<Self> {
        let mut u = [0; 8];
        u[1..].copy_from_slice(bytes);
        zerocopy::try_transmute!(u).map_err(|_| ArmourError::NonZeroError)
    }

    /// 19 days one group
    #[inline]
    fn group_id(&self) -> u32 {
        let id = (self.0.get() >> 24) as u32;
        g4bits(id, Self::GROUP_BITS)
    }
}

impl<H: IdHasher> std::fmt::Display for LowId<H> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.ser())
    }
}

impl<H> Debug for LowId<H> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let id = self.0.get();
        let id = format!("{id:#X}");
        f.debug_tuple("LowId").field(&id).finish()
    }
}

impl<T: IdHasher> TryFrom<&str> for LowId<T> {
    type Error = ArmourError;
    fn try_from(val: &str) -> Result<Self, Self::Error> {
        Self::deser(val)
    }
}

impl<T: IdHasher> FromStr for LowId<T> {
    type Err = ArmourError;
    fn from_str(s: &str) -> Result<Self, Self::Err> {
        Self::deser(s)
    }
}

#[cfg(feature = "std")]
impl<T: IdHasher> Serialize for LowId<T> {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        let s = self.ser();
        serializer.serialize_str(&s)
    }
}

#[cfg(feature = "std")]
impl<'de, T: IdHasher> Deserialize<'de> for LowId<T> {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        use serde::de::Error;

        let s: &str = Deserialize::deserialize(deserializer)?;
        let a = LowId::<T>::deser(s).map_err(|err| {
            error!("id value error: {err}");
            D::Error::custom("id value error")
        })?;
        Ok(a)
    }
}

#[cfg(feature = "ts-rs")]
impl<H> ts_rs::TS for LowId<H> {
    type WithoutGenerics = LowId<()>;
    type OptionInnerType = Self;
    fn name() -> String {
        "LowId".to_owned()
    }
    fn decl_concrete() -> String {
        format!("type {} = {};", Self::name(), Self::inline())
    }
    fn decl() -> String {
        let inline = <LowId<()> as ::ts_rs::TS>::inline();
        format!("type {} = {};", Self::name(), inline)
    }
    fn inline() -> String {
        "string".to_owned()
    }
    fn inline_flattened() -> String {
        panic!("{} cannot be flattened", Self::name())
    }
    fn output_path() -> Option<std::path::PathBuf> {
        Some(std::path::PathBuf::from("low-id.ts"))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::collections::enc::Cipher;

    #[derive(Debug, PartialOrd, Ord, PartialEq, Eq, Clone, Copy, Hash)]
    pub struct Hasher;

    impl IdHasher for Hasher {
        const HASHER: Cipher = Cipher::new(
            "_mKbKGF2IrkGvIJvl97HuCgWjgt6QRZ7Ye8DHBQ2anvyi18BdMz8uN6Ej3YJApooY6qDu0obqq4",
        );
    }

    #[test]
    fn test_creation() {
        for i in 0..=7 {
            for j in 0..=31 {
                let id = LowId::<Hasher>::with_shard_thread(i, j);
                assert_eq!(id.instance_id(), i);
                assert_eq!(id.thread_id(), j);
            }
        }

        for i in 0..=255 {
            let id = LowId::<Hasher>::new(i);
            assert_eq!(id.shard_id(), i);
        }
    }

    #[test]
    fn test_ser_deser() {
        for _ in 0..100_000 {
            let id = LowId::<Hasher>::new(111);
            let _ = id.timestamp();
            let _ = id.date();
            let id_str = id.ser();
            #[allow(clippy::unwrap_used)]
            let deserialized = LowId::<Hasher>::deser(&id_str).unwrap();
            assert_eq!(id, deserialized);
        }
    }

    #[test]
    fn test_encode_decode() {
        for _ in 0..100_000 {
            let id = LowId::<Hasher>::new(255);
            let encoded = id.encode();
            assert!(encoded.len() == 7);
            #[allow(clippy::unwrap_used)]
            let decoded = LowId::<Hasher>::decode(&encoded).unwrap();
            assert_eq!(id, decoded);
        }
    }
}