ids_service 2.1.3

// SPDX-License-Identifier: MIT
// Copyright (c) ids_service contributors

//! Module implementing the ids service using [`std::collections::hash_map::DefaultHasher`].

use data_encoding::{BASE32, BASE64, BASE64URL, HEXLOWER};
use log::*;
use rand::RngExt;
use std::collections::VecDeque;
use std::collections::hash_map::DefaultHasher;
use std::fmt;
use std::hash::Hasher;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex, mpsc};
use std::thread;
use std::thread::JoinHandle;
use std::time::{Duration, SystemTime};

use super::common::{Encode, NUMBER_OF_FILL_THREAD, Service, Uids};

/// Random buffer size: 2 × 8 bytes (64-bit hash output size).
const RAND_LENGTH: usize = 16;

// ── IdsService ────────────────────────────────────────────────────────────────

/// The ids service backed by Rust's [`DefaultHasher`].
///
/// Keeps a pre-filled cache of `u64` ids and replenishes it with background threads.
/// If the cache is empty, an id is generated on-the-fly so this method never fails.
pub struct IdsService {
    ids_cache: Arc<Mutex<VecDeque<u64>>>,
    /// Condvar used to signal waiting threads when new ids are pushed into the cache.
    cache_condvar: Arc<Condvar>,
    cache_size: usize,
    number_of_threads: usize,
    threads_pool: Vec<Option<JoinHandle<()>>>,
    stop_state: Arc<AtomicBool>,
}

impl Default for IdsService {
    /// Create an instance with default parameters:
    /// * Cache size: 100 000 items
    /// * Number of threads: number of logical CPUs
    fn default() -> Self {
        let cache_size = 100_000_usize;
        IdsService {
            ids_cache: Arc::new(Mutex::new(VecDeque::with_capacity(cache_size))),
            cache_condvar: Arc::new(Condvar::new()),
            cache_size,
            number_of_threads: *NUMBER_OF_FILL_THREAD,
            threads_pool: Vec::with_capacity(*NUMBER_OF_FILL_THREAD),
            stop_state: Arc::new(AtomicBool::new(false)),
        }
    }
}

impl fmt::Debug for IdsService {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "IdsService {{ cached_ids: {}, cache_size: {}, threads: {} }}",
            self.ids_cache
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .len(),
            self.cache_size,
            self.number_of_threads,
        )
    }
}

// ── Service ───────────────────────────────────────────────────────────────────

impl Service for IdsService {
    fn start(&mut self) {
        info!("Start ids_service (rust_hash)!");
        self.stop_state.store(false, Ordering::Relaxed);

        for i in 0..self.number_of_threads {
            let ids_cache_clone = Arc::clone(&self.ids_cache);
            let condvar_clone = Arc::clone(&self.cache_condvar);
            let cache_size = self.cache_size;
            let stop_state_clone = Arc::clone(&self.stop_state);

            let handler = thread::Builder::new()
                .name(format!("ids-rh-{}", i))
                .spawn(move || {
                    trace!("Worker thread ids-rh-{} started", i);
                    loop {
                        let needs_fill = ids_cache_clone
                            .lock()
                            .unwrap_or_else(|e| e.into_inner())
                            .len()
                            < cache_size;

                        if needs_fill {
                            let id = build_id();
                            ids_cache_clone
                                .lock()
                                .unwrap_or_else(|e| e.into_inner())
                                .push_back(id);
                            condvar_clone.notify_one();
                        } else {
                            thread::sleep(Duration::from_millis(10));
                        }

                        if stop_state_clone.load(Ordering::Relaxed) {
                            trace!("Worker thread ids-rh-{} stopped", i);
                            break;
                        }
                    }
                })
                .unwrap_or_else(|_| panic!("Failed to spawn worker thread {}", i));

            self.threads_pool.push(Some(handler));
        }
    }

    fn stop(&mut self) {
        info!("Stop ids_service (rust_hash)!");
        self.stop_state.store(true, Ordering::Relaxed);
        self.cache_condvar.notify_all();
        for handle in &mut self.threads_pool {
            if let Some(h) = handle.take() {
                h.join()
                    .unwrap_or_else(|_| error!("Worker thread panicked"));
            }
        }
        self.threads_pool.clear();
        self.ids_cache
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clear();
    }
}

// ── Uids ──────────────────────────────────────────────────────────────────────

impl Uids for IdsService {
    type Id = u64;

    /// Get an id. Falls back to on-the-fly generation if the cache is empty.
    ///
    /// # Examples
    /// ```
    /// use ids_service::rust_hash::*;
    /// use ids_service::common::*;
    ///
    /// let mut ids = IdsService::default();
    /// ids.start();
    /// println!("Get an id: {}", ids.get_id().as_hex());
    /// ids.stop();
    /// ```
    fn get_id(&mut self) -> Self::Id {
        match self
            .ids_cache
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .pop_front()
        {
            Some(id) => id,
            None => {
                info!("Cache empty — generating id on-the-fly");
                build_id()
            }
        }
    }

    /// Get an id from the cache. Returns `None` if the cache is empty.
    ///
    /// # Examples
    /// ```
    /// use ids_service::rust_hash::*;
    /// use ids_service::common::*;
    ///
    /// let mut ids = IdsService::default();
    /// ids.start();
    /// let _ = ids.filled_event().recv();
    /// println!("id: {}", ids.get_id_from_cache().expect("cache non-empty").as_hex());
    /// ids.stop();
    /// ```
    fn get_id_from_cache(&mut self) -> Option<Self::Id> {
        self.ids_cache
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .pop_front()
    }
}

// ── Iterator ──────────────────────────────────────────────────────────────────

/// `IdsService` can be used as an infinite iterator of ids.
///
/// # Examples
/// ```
/// use ids_service::rust_hash::*;
/// use ids_service::common::*;
///
/// let mut ids = IdsService::default();
/// ids.start();
/// let _ = ids.filled_event().recv();
/// for id in ids.take(10) {
///     println!("{}", id.as_json());
/// }
/// ```
impl Iterator for IdsService {
    type Item = u64;

    fn next(&mut self) -> Option<Self::Item> {
        Some(self.get_id())
    }
}

// ── Encode for u64 ────────────────────────────────────────────────────────────

impl Encode for u64 {
    fn as_hex(&self) -> String {
        HEXLOWER.encode(&self.to_ne_bytes())
    }

    fn as_base64(&self) -> String {
        BASE64.encode(&self.to_ne_bytes())
    }

    fn as_base64_url(&self) -> String {
        BASE64URL.encode(&self.to_ne_bytes())
    }

    fn as_base32(&self) -> String {
        BASE32.encode(&self.to_ne_bytes())
    }

    fn as_json(&self) -> String {
        format!(
            "{{\n\"u64\" : \"{}\",\n\"base64\" : \"{}\",\n\"base32\" : \"{}\",\n\"hex\" : \"0x{}\"\n}}",
            self,
            self.as_base64(),
            self.as_base32(),
            self.as_hex(),
        )
    }
}

// ── IdsService impl ───────────────────────────────────────────────────────────

impl IdsService {
    /// Create a new `IdsService`.
    ///
    /// * `cache_size` – number of ids to keep pre-generated.
    /// * `number_of_threads` – worker threads; `None` uses the CPU count.
    ///
    /// # Examples
    /// ```
    /// use ids_service::rust_hash::*;
    /// use ids_service::common::*;
    ///
    /// let mut ids = IdsService::new(10_000, None);
    /// ids.start();
    /// println!("Get an id: {}", ids.get_id());
    /// ids.stop();
    /// ```
    pub fn new(cache_size: usize, number_of_threads: Option<usize>) -> IdsService {
        let threads = number_of_threads.unwrap_or(*NUMBER_OF_FILL_THREAD);
        IdsService {
            ids_cache: Arc::new(Mutex::new(VecDeque::with_capacity(cache_size))),
            cache_condvar: Arc::new(Condvar::new()),
            cache_size,
            number_of_threads: threads,
            threads_pool: Vec::with_capacity(threads),
            stop_state: Arc::new(AtomicBool::new(false)),
        }
    }

    /// Resize the cache and restart the service.
    pub fn set_cache_size(&mut self, new_size: usize) {
        self.cache_size = new_size;
        info!(
            "Restart ids_service (rust_hash) with new cache size {}!",
            new_size
        );
        self.stop();
        self.start();
    }

    /// Return `true` when the cache holds at least `cache_size` ids.
    pub fn is_filled(&self) -> bool {
        self.ids_cache
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .len()
            >= self.cache_size
    }

    /// Return the current number of ids in the cache.
    pub fn get_cache_len(&self) -> usize {
        self.ids_cache
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .len()
    }

    /// Block (in a background thread) until the cache is 100 % full, then
    /// send `true` on the returned channel.
    ///
    /// # Examples
    /// ```
    /// use ids_service::rust_hash::*;
    /// use ids_service::common::*;
    ///
    /// let mut ids = IdsService::default();
    /// ids.start();
    /// let _ = ids.filled_event().recv();
    /// ids.stop();
    /// ```
    pub fn filled_event(&mut self) -> mpsc::Receiver<bool> {
        self.filled_at_percent_event(100)
    }

    /// Block (in a background thread) until the cache reaches `percentage` % of
    /// `cache_size`, then send `true` on the returned channel.
    ///
    /// # Examples
    /// ```
    /// use ids_service::rust_hash::*;
    /// use ids_service::common::*;
    ///
    /// let mut ids = IdsService::default();
    /// ids.start();
    /// let _ = ids.filled_at_percent_event(20).recv();
    /// ids.stop();
    /// ```
    pub fn filled_at_percent_event(&mut self, percentage: u8) -> mpsc::Receiver<bool> {
        let percent = percentage.min(100);
        let cache_limit = self.cache_size * percent as usize / 100;

        let (tx, rx) = mpsc::channel();
        let ids_cache_clone = Arc::clone(&self.ids_cache);
        let condvar_clone = Arc::clone(&self.cache_condvar);
        let stop_state_clone = Arc::clone(&self.stop_state);

        let name = format!("filled_event_{}", self.threads_pool.len() + 1);
        let handler = thread::Builder::new()
            .name(name)
            .spawn(move || {
                let mut guard = ids_cache_clone.lock().unwrap_or_else(|e| e.into_inner());
                loop {
                    if guard.len() >= cache_limit {
                        let _ = tx.send(true);
                        break;
                    }
                    if stop_state_clone.load(Ordering::Relaxed) {
                        let _ = tx.send(false);
                        break;
                    }
                    let result = condvar_clone
                        .wait_timeout(guard, Duration::from_millis(50))
                        .unwrap_or_else(|e| e.into_inner());
                    guard = result.0;
                }
            })
            .expect("failed to spawn filled_event thread");

        self.threads_pool.push(Some(handler));
        rx
    }
}

// ── Public convenience function ───────────────────────────────────────────────

/// Generate a single id without the caching service.
///
/// # Examples
/// ```
/// use ids_service::rust_hash::create_id;
/// use ids_service::common::Encode;
///
/// println!("{}", create_id().as_hex());
/// ```
pub fn create_id() -> u64 {
    build_id()
}

// ── Internal helper ───────────────────────────────────────────────────────────

fn build_id() -> u64 {
    let mut rng = rand::rng();
    let ts = SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .expect("system clock before UNIX epoch")
        .as_nanos();
    let mut hasher = DefaultHasher::new();
    let mut arr = [0u8; RAND_LENGTH];
    rng.fill(&mut arr[..]);
    hasher.write(&arr);
    hasher.write(&ts.to_ne_bytes());
    hasher.finish()
}

// ── Tests ─────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::common::{Encode, Service, Uids};
    use std::collections::HashSet;

    fn init_log() {
        let _ = simplelog::SimpleLogger::init(
            simplelog::LevelFilter::Info,
            simplelog::Config::default(),
        );
    }

    #[test]
    fn test_build_id() {
        let id1 = build_id();
        let id2 = build_id();
        assert_ne!(id1, id2);
        assert_ne!(id1, 0);
    }

    #[test]
    fn test_new() {
        init_log();
        let mut svc = IdsService::new(1_000, None);
        svc.start();
        let _ = svc.filled_event().recv();
        assert!(svc.get_cache_len() >= 1_000);
        svc.stop();
        assert_eq!(svc.get_cache_len(), 0);
    }

    #[test]
    fn test_default() {
        let mut svc = IdsService::default();
        svc.start();
        let _ = svc.filled_event().recv();
        assert!(svc.get_cache_len() >= 100_000);
        svc.stop();
    }

    #[test]
    fn test_get_id() {
        let mut svc = IdsService::default();
        svc.start();
        let _ = svc.filled_at_percent_event(5).recv();
        assert_ne!(svc.get_id(), 0);
    }

    #[test]
    fn test_get_id_from_cache() {
        let mut svc = IdsService::default();
        svc.start();
        let _ = svc.filled_at_percent_event(5).recv();
        let r1 = svc.get_id_from_cache();
        assert!(r1.is_some());
        assert_ne!(r1.unwrap(), 0);
        svc.stop();
        assert!(svc.get_id_from_cache().is_none());
    }

    #[test]
    fn test_filled() {
        let mut svc = IdsService::default();
        svc.start();
        let _ = svc.filled_event().recv();
        assert!(svc.is_filled());
        svc.stop();
    }

    #[test]
    fn test_filled_at_percent_event() {
        let mut svc = IdsService::new(10_000, None);
        svc.start();
        let _ = svc.filled_at_percent_event(5).recv();
        assert!(svc.get_cache_len() > 200);
        svc.stop();
    }

    #[test]
    fn test_iterator_uniqueness() {
        let mut svc = IdsService::new(10_000, None);
        svc.start();
        let _ = svc.filled_event().recv();
        let n = 10_000;
        let ids: HashSet<String> = svc.take(n).map(|id| id.as_hex()).collect();
        assert_eq!(ids.len(), n);
    }

    #[test]
    fn test_encode_hex() {
        let id = build_id();
        assert_eq!(id.as_hex().len(), 16); // u64 = 8 bytes → 16 hex chars
    }

    #[test]
    fn test_encode_json() {
        let id = build_id();
        let json = id.as_json();
        assert!(json.contains("\"u64\""));
        assert!(json.contains("\"base64\""));
        assert!(json.contains("\"base32\""));
        assert!(json.contains("\"hex\""));
    }

    #[test]
    fn test_set_cache_size() {
        let mut svc = IdsService::new(1_000, None);
        svc.start();
        let _ = svc.filled_event().recv();
        svc.set_cache_size(500);
        let _ = svc.filled_event().recv();
        assert!(svc.get_cache_len() >= 500);
        svc.stop();
    }
}