sccache 0.15.0

// Copyright 2026 Mozilla Foundation
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use super::*;
use crate::cache::CacheRead;
use crate::cache::disk::DiskCache;
use crate::cache::readonly::ReadOnlyStorage;
use crate::config::Config;
use crate::config::PreprocessorCacheModeConfig;
use bytes::Bytes;
use std::collections::HashMap;
use std::env;
use std::fs;
use std::io::Cursor;
use std::sync::Arc;
use std::time::Duration;
use tempfile::Builder as TempBuilder;
use tokio::runtime::Builder as RuntimeBuilder;
use tokio::sync::Mutex;
use tokio::time::sleep;

#[test]
fn test_multi_level_storage_get() {
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    let tempdir1 = TempBuilder::new()
        .prefix("sccache_test_l1_")
        .tempdir()
        .unwrap();
    let cache_dir1 = tempdir1.path().join("cache");
    fs::create_dir(&cache_dir1).unwrap();

    let tempdir2 = TempBuilder::new()
        .prefix("sccache_test_l2_")
        .tempdir()
        .unwrap();
    let cache_dir2 = tempdir2.path().join("cache");
    fs::create_dir(&cache_dir2).unwrap();

    let cache1 = DiskCache::new(
        &cache_dir1,
        1024 * 1024 * 100,
        runtime.handle(),
        PreprocessorCacheModeConfig::default(),
        CacheMode::ReadWrite,
        vec![],
    );
    let cache2 = DiskCache::new(
        &cache_dir2,
        1024 * 1024 * 100,
        runtime.handle(),
        PreprocessorCacheModeConfig::default(),
        CacheMode::ReadWrite,
        vec![],
    );

    let cache1_storage: Arc<dyn Storage> = Arc::new(cache1);
    let cache2_storage: Arc<dyn Storage> = Arc::new(cache2);

    let storage = MultiLevelStorage::new(vec![
        Arc::clone(&cache1_storage),
        Arc::clone(&cache2_storage),
    ]);

    runtime.block_on(async {
        // Write directly to level 2 (level 1 is empty)
        {
            let entry = CacheWrite::default();
            cache2_storage.put("test_key", entry).await.unwrap();
        }

        // Now try to read through multi-level storage
        match storage.get("test_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected - found at level 2
            }
            _ => panic!("Expected cache hit at level 2"),
        }

        // Try non-existent key
        match storage.get("nonexistent").await.unwrap() {
            Cache::Miss => {
                // Expected
            }
            _ => panic!("Expected cache miss"),
        }
    });
}

#[test]
fn test_multi_level_storage_backfill_on_hit() {
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    let tempdir1 = TempBuilder::new()
        .prefix("sccache_test_bf_l1_")
        .tempdir()
        .unwrap();
    let cache_dir1 = tempdir1.path().join("cache");
    fs::create_dir(&cache_dir1).unwrap();

    let tempdir2 = TempBuilder::new()
        .prefix("sccache_test_bf_l2_")
        .tempdir()
        .unwrap();
    let cache_dir2 = tempdir2.path().join("cache");
    fs::create_dir(&cache_dir2).unwrap();

    let cache1 = DiskCache::new(
        &cache_dir1,
        1024 * 1024 * 100,
        runtime.handle(),
        PreprocessorCacheModeConfig::default(),
        CacheMode::ReadWrite,
        vec![],
    );
    let cache2 = DiskCache::new(
        &cache_dir2,
        1024 * 1024 * 100,
        runtime.handle(),
        PreprocessorCacheModeConfig::default(),
        CacheMode::ReadWrite,
        vec![],
    );

    let cache1_storage: Arc<dyn Storage> = Arc::new(cache1);
    let cache2_storage: Arc<dyn Storage> = Arc::new(cache2);

    let storage = MultiLevelStorage::new(vec![
        Arc::clone(&cache1_storage),
        Arc::clone(&cache2_storage),
    ]);

    runtime.block_on(async {
        // Write directly to level 2 (level 1 is empty)
        {
            let entry = CacheWrite::default();
            cache2_storage.put("backfill_key", entry).await.unwrap();
        }

        // Verify level 1 doesn't have it yet
        match cache1_storage.get("backfill_key").await.unwrap() {
            Cache::Miss => {
                // Expected - level 1 is empty
            }
            _ => panic!("Level 1 should be empty"),
        }

        // Now read through multi-level storage - should hit level 2 and backfill to level 1
        match storage.get("backfill_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected - found at level 2
            }
            _ => panic!("Expected cache hit at level 2"),
        }

        // Give background backfill task time to complete
        sleep(Duration::from_millis(200)).await;

        // Now level 1 should have the data (backfilled)
        match cache1_storage.get("backfill_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected - backfilled from level 2
            }
            _ => panic!("Level 1 should now have the data (backfilled)"),
        }
    });
}

/// In-memory storage mock for testing multi-level backfill with remote-like backends.
///
/// This is used to test multi-level cache backfill logic without requiring:
/// - Network access to real remote services (S3, Redis, etc.)
/// - Complex mock infrastructure (channels, queues, etc.)
/// - Disk I/O operations
///
/// The mock implements both Storage trait and get_raw() to simulate real backend
/// behavior where remote caches support raw byte retrieval for efficient backfilling.
struct InMemoryStorage {
    data: Arc<Mutex<HashMap<String, Vec<u8>>>>,
    access_log: Arc<Mutex<Vec<String>>>,
}

impl InMemoryStorage {
    fn new() -> Self {
        Self {
            data: Arc::new(Mutex::new(HashMap::new())),
            access_log: Arc::new(Mutex::new(Vec::new())),
        }
    }

    fn get_access_log(&self) -> Arc<Mutex<Vec<String>>> {
        Arc::clone(&self.access_log)
    }
}

#[async_trait]
impl Storage for InMemoryStorage {
    async fn get(&self, key: &str) -> Result<Cache> {
        self.access_log.lock().await.push(format!("get:{}", key));

        let data = self.data.lock().await;
        match data.get(key) {
            Some(bytes) => {
                let cursor = Cursor::new(bytes.clone());
                match CacheRead::from(cursor) {
                    Ok(hit) => Ok(Cache::Hit(hit)),
                    Err(_) => Ok(Cache::Miss),
                }
            }
            None => Ok(Cache::Miss),
        }
    }

    async fn put(&self, key: &str, entry: CacheWrite) -> Result<Duration> {
        self.access_log.lock().await.push(format!("put:{}", key));

        let data = entry.finish()?;
        self.data.lock().await.insert(key.to_string(), data);
        Ok(Duration::ZERO)
    }

    async fn check(&self) -> Result<CacheMode> {
        Ok(CacheMode::ReadWrite)
    }

    fn location(&self) -> String {
        "InMemory".to_string()
    }

    async fn current_size(&self) -> Result<Option<u64>> {
        Ok(None)
    }

    async fn max_size(&self) -> Result<Option<u64>> {
        Ok(None)
    }

    /// Implement get_raw() to enable backfill testing with remote-like backends.
    /// This simulates the behavior of real remote backends (S3, Redis, etc.) that
    /// can efficiently return raw serialized cache entries for backfilling.
    async fn get_raw(&self, key: &str) -> Result<Option<Bytes>> {
        Ok(self.data.lock().await.get(key).cloned().map(Bytes::from))
    }

    /// Implement put_raw() to enable backfill writes during testing.
    async fn put_raw(&self, key: &str, data: Bytes) -> Result<Duration> {
        self.data
            .lock()
            .await
            .insert(key.to_string(), data.to_vec());
        Ok(Duration::ZERO)
    }
}

#[test]
fn test_disk_plus_remote_to_remote_backfill() {
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    // Create multi-level cache: Disk (L0) + Memcached (L1) + Redis (L2) + S3 (L3)
    // This simulates a real-world setup with local disk cache and multiple remote caches
    let tempdir = TempBuilder::new()
        .prefix("sccache_test_multilevel_")
        .tempdir()
        .unwrap();
    let cache_dir = tempdir.path().join("cache");
    fs::create_dir(&cache_dir).unwrap();

    let disk_cache = Arc::new(DiskCache::new(
        &cache_dir,
        1024 * 1024 * 100,
        runtime.handle(),
        PreprocessorCacheModeConfig::default(),
        CacheMode::ReadWrite,
        vec![],
    ));

    let remote_l1 = Arc::new(InMemoryStorage::new()); // Memcached-like
    let remote_l2 = Arc::new(InMemoryStorage::new()); // Redis-like
    let remote_l3 = Arc::new(InMemoryStorage::new()); // S3-like

    let storage = MultiLevelStorage::new(vec![
        disk_cache.clone() as Arc<dyn Storage>,
        remote_l1.clone() as Arc<dyn Storage>,
        remote_l2.clone() as Arc<dyn Storage>,
        remote_l3.clone() as Arc<dyn Storage>,
    ]);

    runtime.block_on(async {
        // Scenario: Data only in S3 (L3), need to backfill all the way to local disk (L0)
        {
            let entry = CacheWrite::default();
            remote_l3.put("global_key", entry).await.unwrap();
        }

        // Verify only L3 has it
        assert!(matches!(
            disk_cache.get("global_key").await.unwrap(),
            Cache::Miss
        ));
        assert!(matches!(
            remote_l1.get("global_key").await.unwrap(),
            Cache::Miss
        ));
        assert!(matches!(
            remote_l2.get("global_key").await.unwrap(),
            Cache::Miss
        ));

        // Read through multi-level storage - should hit L3 and backfill everywhere
        match storage.get("global_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected - found at L3
            }
            _ => panic!("Expected cache hit at L3"),
        }

        // Give all background backfill tasks time to complete
        // We have 3 backfill tasks (L3 -> L2, L3 -> L1, L3 -> L0)
        sleep(Duration::from_millis(400)).await;

        // Verify local disk was backfilled (closest to CPU)
        match disk_cache.get("global_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected - backfilled from L3 to disk cache
            }
            _ => panic!("Disk cache should be backfilled from L3"),
        }

        // Verify remote L1 was backfilled
        match remote_l1.get("global_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected
            }
            _ => panic!("Remote L1 should be backfilled from L3"),
        }

        // Verify remote L2 was backfilled
        match remote_l2.get("global_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected
            }
            _ => panic!("Remote L2 should be backfilled from L3"),
        }

        // Now reading should hit at L0 (disk) - fastest
        match storage.get("global_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected - immediate local disk hit
            }
            _ => panic!("Should hit at disk cache (L0)"),
        }
    });
}

#[test]
fn test_disk_plus_remotes_write_to_all() {
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    // Test write path: ensure data is written to all levels
    let tempdir = TempBuilder::new()
        .prefix("sccache_test_write_all_")
        .tempdir()
        .unwrap();
    let cache_dir = tempdir.path().join("cache");
    fs::create_dir(&cache_dir).unwrap();

    let disk_cache = Arc::new(DiskCache::new(
        &cache_dir,
        1024 * 1024 * 100,
        runtime.handle(),
        PreprocessorCacheModeConfig::default(),
        CacheMode::ReadWrite,
        vec![],
    ));

    let remote_l1 = Arc::new(InMemoryStorage::new());
    let remote_l2 = Arc::new(InMemoryStorage::new());

    let storage = MultiLevelStorage::new(vec![
        disk_cache.clone() as Arc<dyn Storage>,
        remote_l1.clone() as Arc<dyn Storage>,
        remote_l2.clone() as Arc<dyn Storage>,
    ]);

    runtime.block_on(async {
        // Write through multi-level should go to all levels
        {
            let entry = CacheWrite::default();
            storage.put("write_test_key", entry).await.unwrap();
        }

        // Give async writes time to complete
        sleep(Duration::from_millis(200)).await;

        // Verify disk cache has it
        match disk_cache.get("write_test_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected - written to disk synchronously
            }
            _ => panic!("Disk cache should have data after put"),
        }

        // Verify both remote caches have it
        match remote_l1.get("write_test_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected - written to L1 asynchronously
            }
            _ => panic!("Remote L1 should have data after put"),
        }

        match remote_l2.get("write_test_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected - written to L2 asynchronously
            }
            _ => panic!("Remote L2 should have data after put"),
        }
    });
}

#[test]
fn test_remote_to_remote_backfill() {
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    // Create three in-memory "remote" caches to simulate:
    // L0: Memcached (fast, small)
    // L1: Redis (medium, medium)
    // L2: S3 (slow, large)
    let cache_l0 = Arc::new(InMemoryStorage::new());
    let cache_l1 = Arc::new(InMemoryStorage::new());
    let cache_l2 = Arc::new(InMemoryStorage::new());

    let storage = MultiLevelStorage::new(vec![
        cache_l0.clone() as Arc<dyn Storage>,
        cache_l1.clone() as Arc<dyn Storage>,
        cache_l2.clone() as Arc<dyn Storage>,
    ]);

    runtime.block_on(async {
        // Simulate cache miss at L0 and L1, hit at L2 (typical scenario)
        {
            let entry = CacheWrite::default();
            cache_l2.put("remote_key", entry).await.unwrap();
        }

        // Verify L0 and L1 are empty (cache misses at those levels)
        match cache_l0.get("remote_key").await.unwrap() {
            Cache::Miss => {}
            _ => panic!("L0 should be empty initially"),
        }
        match cache_l1.get("remote_key").await.unwrap() {
            Cache::Miss => {}
            _ => panic!("L1 should be empty initially"),
        }

        // Read through multi-level storage - should hit L2 and backfill to L0 and L1
        match storage.get("remote_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected - found at L2
            }
            _ => panic!("Expected cache hit at L2"),
        }

        // Give background backfill tasks time to complete
        // Multiple levels means multiple concurrent spawn tasks
        sleep(Duration::from_millis(300)).await;

        // Verify L0 was backfilled from L2 (through L1)
        match cache_l0.get("remote_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected - backfilled from L2 via L1
            }
            _ => panic!("L0 should be backfilled from L2"),
        }

        // Verify L1 was backfilled from L2
        match cache_l1.get("remote_key").await.unwrap() {
            Cache::Hit(_) => {
                // Expected - backfilled from L2
            }
            _ => panic!("L1 should be backfilled from L2"),
        }
    });
}

#[test]
#[serial_test::serial(multilevel_env)]
fn test_config_validation_invalid_level_name() {
    // Test that invalid level names are rejected
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    // Set invalid level name
    unsafe {
        env::set_var("SCCACHE_MULTILEVEL_CHAIN", "disk,invalid_backend,s3");
        env::set_var("SCCACHE_DIR", "/tmp/test-cache");
    }

    let config = Config::load().unwrap();
    let result = MultiLevelStorage::from_config(&config, runtime.handle());

    // Should error with unknown cache level
    assert!(result.is_err());
    if let Err(e) = result {
        let err_msg = format!("{}", e);
        assert!(err_msg.contains("Unknown cache level") || err_msg.contains("invalid_backend"));
    }

    unsafe {
        env::remove_var("SCCACHE_MULTILEVEL_CHAIN");
        env::remove_var("SCCACHE_DIR");
    }
}

#[test]
fn test_config_validation_empty_levels() {
    // Test that empty levels list is handled
    let storage = MultiLevelStorage::new(vec![]);

    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    runtime.block_on(async {
        // Get should return miss (no levels to check)
        match storage.get("test_key").await.unwrap() {
            Cache::Miss => {} // Expected
            _ => panic!("Empty levels should always miss"),
        }
    });
}

#[test]
fn test_config_validation_single_level() {
    // Test that single level works (passthrough mode)
    let cache = Arc::new(InMemoryStorage::new());
    let storage = MultiLevelStorage::new(vec![cache.clone() as Arc<dyn Storage>]);

    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    runtime.block_on(async {
        let entry = CacheWrite::default();
        storage.put("single_key", entry).await.unwrap();

        match storage.get("single_key").await.unwrap() {
            Cache::Hit(_) => {} // Expected
            _ => panic!("Single level should work as passthrough"),
        }

        // Should not backfill since only one level
        match cache.get("single_key").await.unwrap() {
            Cache::Hit(_) => {} // Expected - data is there
            _ => panic!("Data should be in the single level"),
        }
    });
}

#[test]
#[serial_test::serial(multilevel_env)]
fn test_config_level_not_configured() {
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    // Set level without configuration
    unsafe {
        env::set_var("SCCACHE_MULTILEVEL_CHAIN", "redis");
        // Don't set SCCACHE_REDIS_ENDPOINT
        env::remove_var("SCCACHE_REDIS");
        env::remove_var("SCCACHE_REDIS_ENDPOINT");
    }

    let config = Config::load().unwrap();
    let result = MultiLevelStorage::from_config(&config, runtime.handle());

    // Should error with "not configured" or "requires" (when feature disabled)
    assert!(result.is_err());
    if let Err(e) = result {
        let err_msg = format!("{}", e);
        assert!(
            err_msg.contains("not configured")
                || err_msg.contains("missing")
                || err_msg.contains("requires"),
            "Expected error about missing config or feature, got: {}",
            err_msg
        );
    }

    unsafe {
        env::remove_var("SCCACHE_MULTILEVEL_CHAIN");
    }
}

#[test]
fn test_concurrent_reads() {
    // Test multiple simultaneous reads to different levels
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(4)
        .build()
        .unwrap();

    let cache_l0 = Arc::new(InMemoryStorage::new());
    let cache_l1 = Arc::new(InMemoryStorage::new());
    let cache_l2 = Arc::new(InMemoryStorage::new());

    let storage = Arc::new(MultiLevelStorage::new(vec![
        cache_l0.clone() as Arc<dyn Storage>,
        cache_l1.clone() as Arc<dyn Storage>,
        cache_l2.clone() as Arc<dyn Storage>,
    ]));

    runtime.block_on(async {
        // Populate different keys at different levels
        cache_l0.put("key_l0", CacheWrite::default()).await.unwrap();
        cache_l1.put("key_l1", CacheWrite::default()).await.unwrap();
        cache_l2.put("key_l2", CacheWrite::default()).await.unwrap();

        // Concurrent reads
        let storage1 = Arc::clone(&storage);
        let storage2 = Arc::clone(&storage);
        let storage3 = Arc::clone(&storage);

        let (r1, r2, r3) = tokio::join!(
            async move { storage1.get("key_l0").await },
            async move { storage2.get("key_l1").await },
            async move { storage3.get("key_l2").await },
        );

        // All should hit
        assert!(matches!(r1.unwrap(), Cache::Hit(_)));
        assert!(matches!(r2.unwrap(), Cache::Hit(_)));
        assert!(matches!(r3.unwrap(), Cache::Hit(_)));
    });
}

#[test]
fn test_concurrent_write_and_read() {
    // Test concurrent writes and reads to same key
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(4)
        .build()
        .unwrap();

    let cache_l0 = Arc::new(InMemoryStorage::new());
    let cache_l1 = Arc::new(InMemoryStorage::new());

    let storage = Arc::new(MultiLevelStorage::new(vec![
        cache_l0.clone() as Arc<dyn Storage>,
        cache_l1.clone() as Arc<dyn Storage>,
    ]));

    runtime.block_on(async {
        let storage_write = Arc::clone(&storage);
        let storage_read = Arc::clone(&storage);

        // Concurrent write and read
        let write_task = tokio::spawn(async move {
            storage_write
                .put("concurrent_key", CacheWrite::default())
                .await
        });

        let read_task = tokio::spawn(async move {
            sleep(Duration::from_millis(10)).await;
            storage_read.get("concurrent_key").await
        });

        let (write_result, read_result) = tokio::join!(write_task, read_task);

        // Write should succeed
        write_result.unwrap().unwrap();

        // Read might miss or hit depending on timing (both are valid)
        match read_result.unwrap().unwrap() {
            Cache::Hit(_) | Cache::Miss => {} // Both valid
            _ => panic!("Unexpected cache result"),
        }
    });
}

#[test]
fn test_large_data_handling() {
    // Test with large cache entries
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    let cache_l0 = Arc::new(InMemoryStorage::new());
    let cache_l1 = Arc::new(InMemoryStorage::new());

    let storage = MultiLevelStorage::new(vec![
        cache_l0.clone() as Arc<dyn Storage>,
        cache_l1.clone() as Arc<dyn Storage>,
    ]);

    runtime.block_on(async {
        // Create large entry (1MB of data)
        let mut entry = CacheWrite::new();
        let large_data = vec![0xAB; 1024 * 1024]; // 1MB of data
        entry.put_stdout(&large_data).unwrap();
        cache_l1.put("large_key", entry).await.unwrap();

        // Read through multi-level - should hit at L1
        match storage.get("large_key").await.unwrap() {
            Cache::Hit(_) => {}
            _ => panic!("Should hit at L1"),
        }

        // Wait for backfill
        sleep(Duration::from_millis(200)).await;

        // Verify L0 was backfilled
        match cache_l0.get("large_key").await.unwrap() {
            Cache::Hit(_) => {} // Expected
            _ => panic!("L0 should have backfilled data from L1"),
        }
    });
}

#[test]
fn test_storage_trait_methods() {
    // Test Storage trait methods: check(), location(), current_size(), max_size()
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    let cache_l0 = Arc::new(InMemoryStorage::new());
    let cache_l1 = Arc::new(InMemoryStorage::new());

    let storage = MultiLevelStorage::new(vec![
        cache_l0 as Arc<dyn Storage>,
        cache_l1 as Arc<dyn Storage>,
    ]);

    runtime.block_on(async {
        // Test check() - should return ReadWrite
        match storage.check().await.unwrap() {
            CacheMode::ReadWrite => {} // Expected
            _ => panic!("Expected ReadWrite mode"),
        }

        // Test location() - should return multi-level description
        let location = storage.location();
        assert!(
            location.contains("Multi-level"),
            "Location should mention Multi-level: {}",
            location
        );

        // Test current_size() - should return None or Some
        let _ = storage.current_size().await.unwrap();

        // Test max_size() - should return None or Some
        let _ = storage.max_size().await.unwrap();
    });
}

#[test]
fn test_all_levels_fail_on_put() {
    // Test behavior when all storage levels fail on write
    // In multi-level design, put() succeeds if ANY level succeeds
    // Even if all fail, it should not panic
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    // Create ReadOnly storages that will reject writes
    let cache_l0 = Arc::new(ReadOnlyStorage(Arc::new(InMemoryStorage::new())));
    let cache_l1 = Arc::new(ReadOnlyStorage(Arc::new(InMemoryStorage::new())));

    let storage = MultiLevelStorage::new(vec![
        cache_l0 as Arc<dyn Storage>,
        cache_l1 as Arc<dyn Storage>,
    ]);

    runtime.block_on(async {
        let entry = CacheWrite::new();

        // put() should complete without panic even when all levels fail
        // (writes to L0 are synchronous, L1+ are async background)
        let result = storage.put("fail_key", entry).await;

        assert!(result.is_ok(), "Put should succeed with read-only levels");
    });
}

#[test]
fn test_preprocessor_cache_mode() {
    // Test preprocessor_cache_mode_config() returns first level's config
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    let tempdir = TempBuilder::new()
        .prefix("sccache_test_preprocessor_")
        .tempdir()
        .unwrap();
    let cache_dir = tempdir.path().join("cache");
    fs::create_dir(&cache_dir).unwrap();

    let preprocessor_config = PreprocessorCacheModeConfig {
        use_preprocessor_cache_mode: true,
        ..Default::default()
    };

    let disk_cache = Arc::new(DiskCache::new(
        &cache_dir,
        1024 * 1024 * 100,
        runtime.handle(),
        preprocessor_config,
        CacheMode::ReadWrite,
        vec![],
    ));

    let cache_l1 = Arc::new(InMemoryStorage::new());

    let storage = MultiLevelStorage::new(vec![
        disk_cache as Arc<dyn Storage>,
        cache_l1 as Arc<dyn Storage>,
    ]);

    // Should return first level's config
    let config = storage.preprocessor_cache_mode_config();
    assert!(config.use_preprocessor_cache_mode);
}

#[test]
fn test_empty_levels_new() {
    // Edge case: creating MultiLevelStorage with empty vec
    // This is allowed but from_config prevents it
    let storage = MultiLevelStorage::new(vec![]);

    // Should have zero levels
    assert_eq!(storage.levels.len(), 0);

    // location() should still work
    let location = storage.location();
    assert!(location.contains("0"));
}

#[test]
fn test_preprocessor_cache_methods() {
    // Test get_preprocessor_cache_entry and put_preprocessor_cache_entry
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    let tempdir = TempBuilder::new()
        .prefix("sccache_test_prep_")
        .tempdir()
        .unwrap();
    let cache_dir = tempdir.path().join("cache");
    fs::create_dir(&cache_dir).unwrap();

    let disk_cache = Arc::new(DiskCache::new(
        &cache_dir,
        1024 * 1024 * 100,
        runtime.handle(),
        PreprocessorCacheModeConfig::default(),
        CacheMode::ReadWrite,
        vec![],
    ));

    let storage = MultiLevelStorage::new(vec![disk_cache as Arc<dyn Storage>]);

    runtime.block_on(async {
        // Test get_preprocessor_cache_entry - should return None for non-existent key
        let result = storage.get_preprocessor_cache_entry("test_key").await;
        assert!(result.is_ok());
        assert!(result.unwrap().is_none());

        // Test put_preprocessor_cache_entry
        let entry = PreprocessorCacheEntry::default();
        let result = storage
            .put_preprocessor_cache_entry("test_key", entry)
            .await;
        assert!(result.is_ok());
    });
}

#[test]
fn test_readonly_level_in_check() {
    // Test that check() properly detects read-only levels
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    let tempdir = TempBuilder::new()
        .prefix("sccache_test_ro_")
        .tempdir()
        .unwrap();
    let cache_dir = tempdir.path().join("cache");
    fs::create_dir(&cache_dir).unwrap();

    let disk_cache = DiskCache::new(
        &cache_dir,
        1024 * 1024 * 100,
        runtime.handle(),
        PreprocessorCacheModeConfig::default(),
        CacheMode::ReadWrite,
        vec![],
    );

    // Wrap in ReadOnly
    let ro_cache = Arc::new(ReadOnlyStorage(Arc::new(disk_cache)));

    let storage = MultiLevelStorage::new(vec![ro_cache as Arc<dyn Storage>]);

    runtime.block_on(async {
        // check() should detect read-only mode
        match storage.check().await.unwrap() {
            CacheMode::ReadOnly => {} // Expected
            _ => panic!("Should detect read-only mode"),
        }
    });
}

#[test]
fn test_sequential_read_order() {
    // Test that reads happen sequentially (L0, L1, L2, ...), not in parallel
    // This verifies the documented behavior: "check multiple storage backends in sequence"
    let runtime = RuntimeBuilder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    // Create three storage levels with access tracking
    let l0 = Arc::new(InMemoryStorage::new());
    let l1 = Arc::new(InMemoryStorage::new());
    let l2 = Arc::new(InMemoryStorage::new());

    let l0_log = l0.get_access_log();
    let l1_log = l1.get_access_log();
    let l2_log = l2.get_access_log();

    // Put data only in L2 (slowest level)
    let key = "test_key_12345678901234567890";
    runtime.block_on(async {
        let mut entry = CacheWrite::default();
        entry.put_stdout(b"test data").unwrap();
        l2.put(key, entry).await.unwrap();
    });

    let storage = MultiLevelStorage::new(vec![
        l0 as Arc<dyn Storage>,
        l1 as Arc<dyn Storage>,
        l2 as Arc<dyn Storage>,
    ]);

    runtime.block_on(async {
        let result = storage.get(key).await.unwrap();

        assert!(matches!(result, Cache::Hit(_)));

        // Check that all three levels were accessed in order
        let l0_accesses = l0_log.lock().await;
        let l1_accesses = l1_log.lock().await;
        let l2_accesses = l2_log.lock().await;

        // Each level should have been accessed exactly once for get
        assert_eq!(l0_accesses.len(), 1, "L0 should be checked first");
        assert_eq!(l1_accesses.len(), 1, "L1 should be checked second");
        assert_eq!(l2_accesses.len(), 2, "L2: put (setup) + get (check)");

        assert_eq!(l0_accesses[0], format!("get:{}", key));
        assert_eq!(l1_accesses[0], format!("get:{}", key));
        assert_eq!(l2_accesses[0], format!("put:{}", key)); // from setup
        assert_eq!(l2_accesses[1], format!("get:{}", key)); // from sequential check
    });
}

#[test]
fn test_read_stops_at_first_hit_not_parallel() {
    // Test that when L1 has data, L2 is NEVER accessed (proving sequential not parallel)
    let runtime = RuntimeBuilder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let l0 = Arc::new(InMemoryStorage::new());
    let l1 = Arc::new(InMemoryStorage::new());
    let l2 = Arc::new(InMemoryStorage::new());

    let l0_log = l0.get_access_log();
    let l1_log = l1.get_access_log();
    let l2_log = l2.get_access_log();

    let key = "test_key_early_hit_1234567890ab";

    // Put data in L1
    runtime.block_on(async {
        let mut entry = CacheWrite::default();
        entry.put_stdout(b"L1 data").unwrap();
        l1.put(key, entry).await.unwrap();
    });

    let storage = MultiLevelStorage::new(vec![
        l0 as Arc<dyn Storage>,
        l1 as Arc<dyn Storage>,
        l2 as Arc<dyn Storage>,
    ]);

    runtime.block_on(async {
        let result = storage.get(key).await.unwrap();

        assert!(matches!(result, Cache::Hit(_)));

        // Verify L0 and L1 were accessed, but L2 was NOT
        let l0_accesses = l0_log.lock().await;
        let l1_accesses = l1_log.lock().await;
        let l2_accesses = l2_log.lock().await;

        assert_eq!(l0_accesses.len(), 1, "L0 should be checked first");
        assert_eq!(l1_accesses.len(), 2, "L1: put (setup) + get (check)");
        assert_eq!(
            l2_accesses.len(),
            0,
            "L2 should NOT be checked (sequential read stops at first hit)"
        );
    });
}

/// Storage mock that always fails on write (for testing error handling).
///
/// Unlike ReadOnlyStorage (which is a valid mode), this returns actual errors
/// to simulate real failure scenarios like disk full, network errors, etc.
struct FailingStorage;

#[async_trait]
impl Storage for FailingStorage {
    async fn get(&self, _key: &str) -> Result<Cache> {
        Ok(Cache::Miss)
    }

    async fn put(&self, _key: &str, _entry: CacheWrite) -> Result<Duration> {
        Err(anyhow!("Intentional failure for testing"))
    }

    async fn put_raw(&self, _key: &str, _entry: Bytes) -> Result<Duration> {
        Err(anyhow!("Intentional failure for testing"))
    }

    async fn check(&self) -> Result<CacheMode> {
        Ok(CacheMode::ReadWrite) // It's RW but fails on put
    }

    fn location(&self) -> String {
        "FailingStorage".to_string()
    }

    async fn current_size(&self) -> Result<Option<u64>> {
        Ok(None)
    }

    async fn max_size(&self) -> Result<Option<u64>> {
        Ok(None)
    }

    fn preprocessor_cache_mode_config(&self) -> PreprocessorCacheModeConfig {
        PreprocessorCacheModeConfig::default()
    }

    async fn get_preprocessor_cache_entry(
        &self,
        _key: &str,
    ) -> Result<Option<Box<dyn crate::lru_disk_cache::ReadSeek>>> {
        Err(anyhow!("Intentional failure for testing"))
    }

    async fn put_preprocessor_cache_entry(
        &self,
        _key: &str,
        _entry: PreprocessorCacheEntry,
    ) -> Result<()> {
        Err(anyhow!("Intentional failure for testing"))
    }
}

#[test]
fn test_put_mode_ignore() {
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    // All levels fail with actual errors
    let cache_l0 = Arc::new(FailingStorage);
    let cache_l1 = Arc::new(FailingStorage);

    let storage = MultiLevelStorage::with_write_error_policy(
        vec![cache_l0 as Arc<dyn Storage>, cache_l1 as Arc<dyn Storage>],
        WriteErrorPolicy::Ignore,
    );

    runtime.block_on(async {
        let entry = CacheWrite::new();
        let result = storage.put("test_key", entry).await;

        assert!(
            result.is_ok(),
            "WriteErrorPolicy::Ignore should never fail, even when all levels error"
        );
    });
}

#[test]
fn test_put_mode_l0_fails_on_error() {
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    // L0 fails with actual error, L1 succeeds
    let cache_l0 = Arc::new(FailingStorage);
    let cache_l1 = Arc::new(InMemoryStorage::new());

    let storage = MultiLevelStorage::with_write_error_policy(
        vec![cache_l0 as Arc<dyn Storage>, cache_l1 as Arc<dyn Storage>],
        WriteErrorPolicy::L0,
    );

    runtime.block_on(async {
        let entry = CacheWrite::new();
        let result = storage.put("test_key", entry).await;

        assert!(
            result.is_err(),
            "WriteErrorPolicy::L0 should fail when L0 write fails"
        );
        let err_msg = result.unwrap_err().to_string();
        assert!(
            err_msg.contains("Intentional") || err_msg.contains("put_raw not implemented"),
            "Expected failure message, got: {}",
            err_msg
        );
    });
}

#[test]
fn test_put_mode_l0_succeeds_if_l0_ok() {
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    // L0 succeeds, L1 fails (shouldn't matter in L0 mode)
    let cache_l0 = Arc::new(InMemoryStorage::new());
    let cache_l1 = Arc::new(FailingStorage);

    let storage = MultiLevelStorage::with_write_error_policy(
        vec![cache_l0 as Arc<dyn Storage>, cache_l1 as Arc<dyn Storage>],
        WriteErrorPolicy::L0,
    );

    runtime.block_on(async {
        let entry = CacheWrite::new();
        let result = storage.put("test_key", entry).await;

        assert!(
            result.is_ok(),
            "WriteErrorPolicy::L0 should succeed when L0 succeeds, even if L1+ fails"
        );
    });
}

#[test]
fn test_put_mode_all_fails_on_any_error() {
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    // L0 succeeds, L1 fails
    let cache_l0 = Arc::new(InMemoryStorage::new());
    let cache_l1 = Arc::new(FailingStorage);

    let storage = MultiLevelStorage::with_write_error_policy(
        vec![cache_l0 as Arc<dyn Storage>, cache_l1 as Arc<dyn Storage>],
        WriteErrorPolicy::All,
    );

    runtime.block_on(async {
        let entry = CacheWrite::new();
        let result = storage.put("test_key", entry).await;

        // Give background L1 task time to complete and report failure
        sleep(Duration::from_millis(100)).await;

        assert!(
            result.is_err(),
            "WriteErrorPolicy::All should fail when any RW level fails"
        );
    });
}

#[test]
fn test_put_mode_all_succeeds_when_all_ok() {
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    // Both levels succeed
    let cache_l0 = Arc::new(InMemoryStorage::new());
    let cache_l1 = Arc::new(InMemoryStorage::new());

    let storage = MultiLevelStorage::with_write_error_policy(
        vec![
            cache_l0.clone() as Arc<dyn Storage>,
            cache_l1.clone() as Arc<dyn Storage>,
        ],
        WriteErrorPolicy::All,
    );

    runtime.block_on(async {
        let entry = CacheWrite::new();
        let result = storage.put("test_key", entry).await;

        // Give background tasks time to complete
        sleep(Duration::from_millis(100)).await;

        assert!(
            result.is_ok(),
            "WriteErrorPolicy::All should succeed when all levels succeed"
        );

        // Verify both levels have the data
        assert!(matches!(
            cache_l0.get("test_key").await.unwrap(),
            Cache::Hit(_)
        ));
        assert!(matches!(
            cache_l1.get("test_key").await.unwrap(),
            Cache::Hit(_)
        ));
    });
}

#[test]
fn test_put_mode_all_skips_readonly() {
    let runtime = RuntimeBuilder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()
        .unwrap();

    // L0 writable, L1 read-only (should be skipped), L2 writable
    let cache_l0 = Arc::new(InMemoryStorage::new());
    let cache_l1 = Arc::new(ReadOnlyStorage(Arc::new(InMemoryStorage::new())));
    let cache_l2 = Arc::new(InMemoryStorage::new());

    let storage = MultiLevelStorage::with_write_error_policy(
        vec![
            cache_l0.clone() as Arc<dyn Storage>,
            cache_l1 as Arc<dyn Storage>,
            cache_l2.clone() as Arc<dyn Storage>,
        ],
        WriteErrorPolicy::All,
    );

    runtime.block_on(async {
        let entry = CacheWrite::new();
        let result = storage.put("test_key", entry).await;

        // Give background tasks time to complete
        sleep(Duration::from_millis(100)).await;

        assert!(
            result.is_ok(),
            "WriteErrorPolicy::All should succeed when read-only levels are skipped"
        );

        // Verify writable levels have the data
        assert!(matches!(
            cache_l0.get("test_key").await.unwrap(),
            Cache::Hit(_)
        ));
        assert!(matches!(
            cache_l2.get("test_key").await.unwrap(),
            Cache::Hit(_)
        ));
    });
}