inklog 0.3.0-rc.1

// Copyright (c) 2026 Kirky.X
// SPDX-License-Identifier: MIT
//! # Object Pool
//!
//! High-performance object pooling with two complementary strategies:
//!
//! - `ObjectPool<K, V>`: LRU-cached pool backed by oxcache (async, TTL-capable).
//! - `ThreadLocalLogRecordPool`: Per-thread pool for `LogRecord` reuse (zero-alloc hot path).
//! - `ThreadLocalStringPool`: Per-thread pool for `String` buffer reuse.
//!
//! # Construction Patterns
//!
//! This module supports two construction patterns:
//! - `new()` - Creates pool with default configuration (async, returns Result)
//! - `with_config()` - Creates pool with custom configuration (async, returns Result)
//!
//! # Usage Examples
//!
//! ```no_run
//! use inklog::ObjectPool;
//!
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
//! // Pattern 1: new() - Default configuration
//! let pool1 = ObjectPool::<String, i32>::new().await?;
//!
//! // Pattern 2: with_config() - Custom configuration
//! use inklog::ObjectPoolConfig;
//! let pool2 = ObjectPool::<String, i32>::with_config(ObjectPoolConfig {
//!     max_capacity: 2048,
//!     ttl_secs: None,
//! }).await?;
//! # Ok(())
//! # }
//! ```

use crate::InklogError;
use crate::LogRecord;
use oxcache::Cache;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use std::sync::LazyLock;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;

/// Pool configuration - configurable via InklogConfig.performance.object_pool
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct ObjectPoolConfig {
    /// Maximum capacity of the pool
    #[serde(default = "default_object_pool_max_capacity")]
    pub max_capacity: usize,
    /// Default TTL for pooled items (None = no TTL)
    pub ttl_secs: Option<u64>,
}

fn default_object_pool_max_capacity() -> usize {
    1024
}

impl Default for ObjectPoolConfig {
    fn default() -> Self {
        Self {
            max_capacity: 1024,
            ttl_secs: None,
        }
    }
}

/// Object pool using oxcache Cache
///
/// This pool provides:
/// - LRU eviction when pool is full
/// - Thread-safe operations without explicit locking
/// - Configurable capacity and TTL
/// - Internal metrics tracking
///
/// All construction and access methods are async and return `Result` to
/// propagate errors explicitly (no panic paths, no silent fallbacks).
///
/// # Cloning
///
/// `Clone` is derived intentionally: cloning an `ObjectPool` shares the
/// underlying oxcache `Cache` and stats via `Arc`. Both clones read and
/// write the same backing store. This is by design for multi-producer
/// scenarios where several tasks need access to the same pool.
#[derive(Clone)]
pub struct ObjectPool<K, V>
where
    K: oxcache::CacheKey + Send + Sync + 'static,
    V: serde::Serialize + for<'de> serde::Deserialize<'de> + Send + Sync + Clone + 'static,
{
    /// The underlying oxcache async cache
    cache: Arc<Cache<K, V>>,
    /// Metrics tracking
    stats: Arc<PoolStats>,
}

impl<K, V> ObjectPool<K, V>
where
    K: oxcache::CacheKey + Send + Sync + 'static,
    V: serde::Serialize + for<'de> serde::Deserialize<'de> + Send + Sync + Clone + 'static,
{
    /// Create a new object pool with default configuration (capacity: 1024)
    pub async fn new() -> Result<Self, InklogError> {
        Self::with_config(ObjectPoolConfig::default()).await
    }

    /// Create a new object pool with full configuration
    ///
    /// Errors are propagated as `InklogError::CacheError`; no silent
    /// `Cache::default()` fallback is used.
    pub async fn with_config(config: ObjectPoolConfig) -> Result<Self, InklogError> {
        let mut builder = Cache::builder();
        builder = builder.capacity(config.max_capacity as u64);
        if let Some(ttl_secs) = config.ttl_secs {
            builder = builder.ttl(Duration::from_secs(ttl_secs));
        }
        let cache = builder.build().await.map_err(|e| {
            let mut args = fluent_bundle::FluentArgs::new();
            args.set("err", e.to_string());
            InklogError::CacheError(crate::i18n::tr_args("cache-build_failed", args))
        })?;
        Ok(Self {
            cache: Arc::new(cache),
            stats: Arc::new(PoolStats::default()),
        })
    }

    /// Get an item from the pool by key
    pub async fn get(&self, key: &K) -> Result<Option<V>, InklogError>
    where
        K: Clone,
    {
        let result = self.cache.get(key).await.map_err(|e| {
            let mut args = fluent_bundle::FluentArgs::new();
            args.set("err", e.to_string());
            InklogError::CacheError(crate::i18n::tr_args("cache-get_failed", args))
        })?;
        if result.is_some() {
            self.stats.hits.fetch_add(1, Ordering::Relaxed);
            self.stats.items_reused.fetch_add(1, Ordering::Relaxed);
        } else {
            self.stats.misses.fetch_add(1, Ordering::Relaxed);
        }
        Ok(result)
    }

    /// Put an item into the pool with the given key
    pub async fn put(&self, key: &K, value: V) -> Result<(), InklogError>
    where
        K: Clone,
        V: Clone,
    {
        self.cache.set(key, &value).await.map_err(|e| {
            let mut args = fluent_bundle::FluentArgs::new();
            args.set("err", e.to_string());
            InklogError::CacheError(crate::i18n::tr_args("cache-set_failed", args))
        })?;
        self.stats.total_items.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    /// Get the approximate number of items that have been put into the pool.
    ///
    /// Note: This count tracks the number of `put()` calls and may over-count
    /// if the underlying oxcache evicts entries (TTL expiry, capacity pressure).
    /// It is an approximation, not an exact count of current cache contents.
    pub fn len(&self) -> usize {
        self.stats.total_items.load(Ordering::Relaxed)
    }

    /// Returns true if the pool currently holds no items.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

// ============================================================================
// Thread-Local Object Pool (High Performance)
// ============================================================================

/// High-performance thread-local pool for LogRecord.
///
/// Uses thread-local storage to eliminate lock contention entirely.
/// Each thread has its own independent pool, maximizing performance.
#[derive(Clone)]
pub struct ThreadLocalLogRecordPool {
    capacity: usize,
}

impl ThreadLocalLogRecordPool {
    /// Create a new pool with the specified capacity.
    pub fn new(capacity: usize) -> Self {
        Self { capacity }
    }

    /// Get an object from the pool, or create a new one if empty.
    pub fn get(&self) -> LogRecord {
        THREAD_LOCAL_LOG_RECORD_POOL.with(|pool| {
            let mut pool = pool.borrow_mut();
            pool.pop().unwrap_or_default()
        })
    }

    /// Return an object to the pool.
    /// If the pool is at capacity, the object is dropped.
    pub fn put(&self, record: LogRecord) {
        THREAD_LOCAL_LOG_RECORD_POOL.with(|pool| {
            let mut pool = pool.borrow_mut();
            if pool.len() < self.capacity {
                // Reset the record before pooling for reuse
                let mut record = record;
                record.reset();
                pool.push(record);
            }
            // If at capacity, the record is simply dropped
        });
    }

    /// Get the current size of the calling thread's pool.
    pub fn len(&self) -> usize {
        THREAD_LOCAL_LOG_RECORD_POOL.with(|pool| pool.borrow().len())
    }

    /// Check if the calling thread's pool is empty.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

impl Default for ThreadLocalLogRecordPool {
    fn default() -> Self {
        Self::new(1024)
    }
}

// Thread-local storage for LogRecord pool.
thread_local! {
    static THREAD_LOCAL_LOG_RECORD_POOL: std::cell::RefCell<Vec<LogRecord>> =
        std::cell::RefCell::new(Vec::with_capacity(1024));
}

/// High-performance thread-local pool for String buffers.
#[derive(Clone)]
pub struct ThreadLocalStringPool {
    capacity: usize,
}

impl ThreadLocalStringPool {
    /// Create a new pool with the specified capacity.
    pub fn new(capacity: usize) -> Self {
        Self { capacity }
    }

    /// Get a String from the pool, or create a new empty one if empty.
    pub fn get(&self) -> String {
        THREAD_LOCAL_STRING_POOL.with(|pool| {
            let mut pool = pool.borrow_mut();
            pool.pop().unwrap_or_default()
        })
    }

    /// Return a String to the pool.
    /// The String is cleared before pooling for reuse.
    pub fn put(&self, mut s: String) {
        s.clear(); // Clear contents to prevent data leaking between users
        THREAD_LOCAL_STRING_POOL.with(|pool| {
            let mut pool = pool.borrow_mut();
            if pool.len() < self.capacity {
                pool.push(s);
            }
            // If at capacity, the string is simply dropped
        });
    }

    /// Get the current size of the calling thread's pool.
    pub fn len(&self) -> usize {
        THREAD_LOCAL_STRING_POOL.with(|pool| pool.borrow().len())
    }

    /// Check if the calling thread's pool is empty.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

impl Default for ThreadLocalStringPool {
    fn default() -> Self {
        Self::new(1024)
    }
}

// Thread-local storage for String pool.
thread_local! {
    static THREAD_LOCAL_STRING_POOL: std::cell::RefCell<Vec<String>> =
        std::cell::RefCell::new(Vec::with_capacity(1024));
}

// ============================================================================
// Global Convenience Functions
// ============================================================================

static GLOBAL_LOG_RECORD_POOL: LazyLock<ThreadLocalLogRecordPool> =
    LazyLock::new(|| ThreadLocalLogRecordPool::new(1024));
static GLOBAL_STRING_POOL: LazyLock<ThreadLocalStringPool> =
    LazyLock::new(|| ThreadLocalStringPool::new(1024));

/// Get a LogRecord from the global thread-local pool.
pub fn get_log_record() -> LogRecord {
    GLOBAL_LOG_RECORD_POOL.get()
}

/// Return a LogRecord to the global thread-local pool.
pub fn put_log_record(record: LogRecord) {
    GLOBAL_LOG_RECORD_POOL.put(record)
}

/// Get a String buffer from the global thread-local pool.
pub fn get_string_buffer() -> String {
    GLOBAL_STRING_POOL.get()
}

/// Return a String buffer to the global thread-local pool.
pub fn put_string_buffer(s: String) {
    GLOBAL_STRING_POOL.put(s)
}

/// Internal pool statistics
#[derive(Debug, Default)]
struct PoolStats {
    pub(crate) total_items: AtomicUsize,
    pub(crate) hits: AtomicUsize,
    pub(crate) misses: AtomicUsize,
    pub(crate) items_reused: AtomicUsize,
}

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

    // ============================================================================
    // ObjectPool async 测试
    // ============================================================================

    #[tokio::test]
    async fn test_object_pool_new_default_capacity() {
        let pool = ObjectPool::<String, String>::new()
            .await
            .expect("default pool should build");
        assert_eq!(pool.len(), 0);
    }

    #[tokio::test]
    async fn test_object_pool_with_config() {
        let pool = ObjectPool::<String, i32>::with_config(ObjectPoolConfig {
            max_capacity: 256,
            ttl_secs: None,
        })
        .await
        .expect("pool with config should build");
        assert_eq!(pool.len(), 0);
    }

    #[tokio::test]
    async fn test_object_pool_put_and_get() {
        let pool = ObjectPool::<String, i32>::new().await.expect("build");

        pool.put(&"a".to_string(), 1).await.expect("put");
        pool.put(&"b".to_string(), 2).await.expect("put");
        pool.put(&"c".to_string(), 3).await.expect("put");

        assert_eq!(pool.get(&"a".to_string()).await.expect("get"), Some(1));
        assert_eq!(pool.get(&"b".to_string()).await.expect("get"), Some(2));
        assert_eq!(pool.get(&"c".to_string()).await.expect("get"), Some(3));
        assert_eq!(pool.get(&"missing".to_string()).await.expect("get"), None);
    }

    #[tokio::test]
    async fn test_object_pool_get_returns_result_on_cache_error() {
        // 验证 get 返回 Result，错误显性传播
        let pool = ObjectPool::<String, i32>::new().await.expect("build");
        // 正常路径返回 Ok(None) 而非 None
        let result = pool.get(&"missing".to_string()).await;
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), None);
    }

    #[tokio::test]
    async fn test_object_pool_put_returns_result() {
        let pool = ObjectPool::<String, i32>::new().await.expect("build");
        // put 返回 Result
        let result = pool.put(&"key".to_string(), 42).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_object_pool_with_ttl_config() {
        let pool = ObjectPool::<String, String>::with_config(ObjectPoolConfig {
            max_capacity: 256,
            ttl_secs: Some(60),
        })
        .await
        .expect("build with ttl");
        pool.put(&"k".to_string(), "v".to_string())
            .await
            .expect("put");
        assert_eq!(
            pool.get(&"k".to_string()).await.expect("get"),
            Some("v".to_string())
        );
    }

    #[tokio::test]
    async fn test_object_pool_is_empty() {
        // 覆盖 is_empty() 方法 (行 162-163) 和 get() 内部 cache 访问 (行 128)
        let pool = ObjectPool::<String, i32>::new().await.expect("build");
        // 新池应为空
        assert!(pool.is_empty());
        assert_eq!(pool.len(), 0);

        // get 触发 .cache.get() 路径（行 128）
        let result = pool.get(&"missing".to_string()).await.expect("get");
        assert_eq!(result, None);

        // put 后 get 验证条目存在（覆盖 .cache 访问路径）
        pool.put(&"key".to_string(), 42).await.expect("put");
        let value = pool.get(&"key".to_string()).await.expect("get");
        assert_eq!(value, Some(42));
    }

    // ============================================================================
    // ObjectPoolConfig 测试
    // ============================================================================

    #[test]
    fn test_object_pool_config_default() {
        let config = ObjectPoolConfig::default();
        assert_eq!(config.max_capacity, 1024);
        assert_eq!(config.ttl_secs, None);
    }

    #[test]
    fn test_object_pool_config_with_ttl() {
        let config = ObjectPoolConfig {
            max_capacity: 256,
            ttl_secs: Some(60),
        };
        assert_eq!(config.max_capacity, 256);
        assert_eq!(config.ttl_secs, Some(60));
    }

    // ============================================================================
    // ThreadLocalLogRecordPool 测试
    // ============================================================================

    #[test]
    fn test_thread_local_log_record_pool() {
        let pool = ThreadLocalLogRecordPool::new(10);

        // Initially may have items from other tests; drain to known state
        while !pool.is_empty() {
            let _ = pool.get();
        }
        assert!(pool.is_empty());

        // Get creates new record if pool is empty
        let record = pool.get();
        assert_eq!(record.level, "INFO");

        // Put returns record to pool
        pool.put(record);

        // Now pool should have one item
        assert!(!pool.is_empty());

        // Get should return the pooled record (reused)
        let record2 = pool.get();
        assert_eq!(record2.level, "INFO");
    }

    #[test]
    fn test_thread_local_log_record_pool_exceed_capacity() {
        let pool = ThreadLocalLogRecordPool::new(3);

        // Drain first
        while !pool.is_empty() {
            let _ = pool.get();
        }

        // Add 3 items
        for _ in 0..3 {
            let record = pool.get();
            pool.put(record);
        }

        // Add one more — pool must not grow beyond capacity
        let extra = pool.get();
        pool.put(extra);

        // Size must not exceed configured capacity
        assert!(pool.len() <= 3);
    }

    #[test]
    fn test_thread_local_log_record_pool_default_trait() {
        let pool = ThreadLocalLogRecordPool::default();
        let r1 = pool.get();
        pool.put(r1);
        assert!(!pool.is_empty());
    }

    // ============================================================================
    // ThreadLocalStringPool 测试
    // ============================================================================

    #[test]
    fn test_thread_local_string_pool() {
        let pool = ThreadLocalStringPool::new(10);

        // Drain first
        while !pool.is_empty() {
            let _ = pool.get();
        }

        assert!(pool.is_empty());

        let s = pool.get();
        assert!(s.is_empty());

        pool.put("test".to_string());

        // Get should return pooled string (cleared on put to prevent data leaks)
        let s2 = pool.get();
        assert_eq!(s2, "", "put() clears string contents");
    }

    #[test]
    fn test_thread_local_string_pool_len_and_is_empty() {
        let pool = ThreadLocalStringPool::new(10);

        // Drain first
        while !pool.is_empty() {
            let _ = pool.get();
        }
        assert!(pool.is_empty());
        assert_eq!(pool.len(), 0);

        pool.put("first".to_string());
        assert!(!pool.is_empty());
        assert_eq!(pool.len(), 1);

        pool.put("second".to_string());
        assert_eq!(pool.len(), 2);

        let s = pool.get();
        assert_eq!(s, ""); // LIFO, but cleared on put
        assert_eq!(pool.len(), 1);

        let s = pool.get();
        assert_eq!(s, "");
        assert_eq!(pool.len(), 0);
        assert!(pool.is_empty());
    }

    #[test]
    fn test_thread_local_string_pool_exceed_capacity_drops_excess() {
        let pool = ThreadLocalStringPool::new(2);

        // Drain first
        while !pool.is_empty() {
            let _ = pool.get();
        }

        pool.put("a".to_string());
        pool.put("b".to_string());
        assert_eq!(pool.len(), 2);

        // Third put should be dropped
        pool.put("c".to_string());
        assert_eq!(
            pool.len(),
            2,
            "pool should not grow beyond capacity; excess should be dropped"
        );

        // Verify retained items are empty strings (cleared on put)
        let s1 = pool.get();
        let s2 = pool.get();
        let mut remaining = vec![s1, s2];
        remaining.sort();
        assert_eq!(remaining, vec!["".to_string(), "".to_string()]);
    }

    #[test]
    fn test_thread_local_string_pool_default_trait() {
        let pool = ThreadLocalStringPool::default();
        let s = pool.get();
        assert!(s.is_empty());
        pool.put("default".to_string());
        assert!(!pool.is_empty());
    }

    // ============================================================================
    // 全局便捷函数测试
    // ============================================================================

    #[test]
    fn test_global_log_record_functions() {
        let record = get_log_record();
        assert_eq!(record.level, "INFO");

        let mut modified = record;
        modified.message = "global test".to_string();
        put_log_record(modified);

        let record2 = get_log_record();
        assert_eq!(record2.level, "INFO"); // put 会 reset

        // 验证多次调用不会 panic
        for _ in 0..5 {
            let r = get_log_record();
            put_log_record(r);
        }
    }

    #[test]
    fn test_global_string_buffer_functions() {
        let s1 = get_string_buffer();
        put_string_buffer(s1);

        let s2 = get_string_buffer();
        // Verify the API is callable and returns a String
        let _ = s2.capacity();
    }

    // ============================================================================
    // 并发测试
    // ============================================================================

    #[tokio::test(flavor = "multi_thread")]
    async fn test_thread_local_pool_concurrent_isolation() {
        use std::sync::Arc;
        use std::sync::atomic::{AtomicUsize, Ordering};

        // ThreadLocalLogRecordPool 每个线程有独立 pool，不依赖 runtime
        // 使用 tokio::task::spawn_blocking 保证在独立 OS 线程上运行（AGENTS.md 禁止 std::thread）
        let pool = Arc::new(ThreadLocalLogRecordPool::new(10));
        let total_gets = Arc::new(AtomicUsize::new(0));

        let mut handles = Vec::new();
        for _ in 0..4 {
            let pool_clone = Arc::clone(&pool);
            let counter_clone = Arc::clone(&total_gets);
            handles.push(tokio::task::spawn_blocking(move || {
                for _ in 0..5 {
                    let _record = pool_clone.get();
                    counter_clone.fetch_add(1, Ordering::Relaxed);
                }
            }));
        }
        for h in handles {
            h.await.expect("blocking task should not panic");
        }

        // 4 线程 × 5 次 get = 20 次
        assert_eq!(total_gets.load(Ordering::Relaxed), 20);
    }

    #[test]
    fn test_string_pool_clears_string_on_put() {
        // T015: put() must clear string contents before pooling to prevent data leaks
        let pool = super::ThreadLocalStringPool::new(10);
        let mut s = pool.get();
        s.push_str("sensitive data");
        pool.put(s);
        // Retrieved string should be empty, not contain previous data
        let retrieved = pool.get();
        assert_eq!(retrieved, "", "pooled string should be cleared on put");
    }
}