mod cold_tier_tests {
use super::*;
use http_body_util::BodyExt;

fn cold_settings() -> Settings {
    Settings {
        enabled: true,
        max_entries: 1,
        max_bytes: 256 * 1024,
        max_object_bytes: 64 * 1024,
        cold_enabled: true,
        cold_path: "cache/cold-test".into(),
        cold_max_bytes: 2 * 1024 * 1024,
        cold_max_entries: 16,
        cold_after_seconds: 300,
        ..Settings::default()
    }
}

async fn fill(cache: &Arc<Cache>, settings: &Settings, path: &str, body: &'static str) -> Ticket {
    let ticket = cache
        .ticket_scoped(
            11,
            "cold-test",
            "http",
            "cold.example",
            "origin@127.0.0.1:8080",
            path,
            &HeaderMap::new(),
        )
        .unwrap();
    let mut response = Response::new(Body::from(body));
    response.headers_mut().insert(
        "cache-control",
        HeaderValue::from_static("public, max-age=120"),
    );
    response
        .headers_mut()
        .insert("etag", HeaderValue::from_static("\"cold-v1\""));
    let plan = plan(StatusCode::OK, response.headers(), settings, Duration::ZERO).unwrap();
    cache
        .wrap(response, ticket.clone(), plan, settings)
        .into_body()
        .collect()
        .await
        .unwrap();
    ticket
}

async fn wait_for_cold(cache: &Cache, expected: usize) {
    for _ in 0..100 {
        if cache.view().cold_entries >= expected {
            return;
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    panic!("cold tier did not persist in time");
}

#[tokio::test]
async fn selective_purge_removes_matching_cold_object() {
    let dir = tempfile::tempdir().unwrap();
    let settings = cold_settings();
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        shutdown.clone(),
    );
    let first = fill(&cache, &settings, "/one?v=1", "first-payload").await;
    let _second = fill(&cache, &settings, "/two", "second-payload").await;
    wait_for_cold(&cache, 1).await;
    assert_eq!(cache.view().cold_entries, 1);
    let result = cache
        .invalidate(InvalidationRequest {
            operation: InvalidationOperation::Purge,
            selector: InvalidationSelector::Exact,
            value: "/one?v=1".into(),
            host: Some("cold.example".into()),
            vhost: Some("cold-test".into()),
            ttl_seconds: None,
            confirm_wide: false,
        })
        .unwrap();
    assert_eq!(result.affected_entries, 1);
    assert_eq!(cache.view().cold_entries, 0);
    assert!(cache.get_tiered(&first, &settings).await.is_none());
    shutdown.cancel();
}

#[tokio::test]
async fn aged_transformed_css_remains_fresh_when_demoted_to_cold() {
    let dir = tempfile::tempdir().unwrap();
    let settings = cold_settings();
    let shutdown = CancellationToken::new();
    let cache = Cache::start(11, Some(dir.path().to_path_buf()), settings.clone(), shutdown.clone());
    let ticket = cache.ticket_scoped(
        11, "cold-test", "http", "cold.example", "origin@127.0.0.1:8080",
        "/aged-theme.css", &HeaderMap::new(),
    ).unwrap();
    let body = b"body { color: red; margin: 0px; }";
    let mut origin = Response::new(Body::from(Bytes::from_static(body)));
    origin.headers_mut().insert("cache-control", HeaderValue::from_static("public, max-age=2592000"));
    origin.headers_mut().insert("age", HeaderValue::from_static("1553483"));
    origin.headers_mut().insert("date", httpdate::fmt_http_date(SystemTime::now()).parse().unwrap());
    origin.headers_mut().insert("content-type", HeaderValue::from_static("text/css"));
    origin.headers_mut().insert("content-length", HeaderValue::from_str(&body.len().to_string()).unwrap());
    origin.headers_mut().insert("etag", HeaderValue::from_static("\"origin\""));
    let origin_plan = plan(StatusCode::OK, origin.headers(), &settings, Duration::ZERO).unwrap();
    let transformed = crate::transform::apply(origin, "/aged-theme.css",
        &crate::transform::Settings { enabled: true, css: true, ..Default::default() }, false
    ).await.unwrap();
    let transformed_plan = plan(StatusCode::OK, transformed.headers(), &settings, Duration::ZERO).unwrap();
    cache.wrap(transformed, ticket.clone(), origin_plan.restrict(transformed_plan), &settings)
        .into_body().collect().await.unwrap();
    assert_eq!(cache.view().entries, 1);
    let _another = fill(&cache, &settings, "/another", "another-payload").await;
    wait_for_cold(&cache, 1).await;
    let row = cache.dashboard(Some("cold-test"), 0, 20).objects
        .into_iter().find(|v| v.path == "/aged-theme.css").unwrap();
    assert_eq!(row.tier, "cold");
    assert!(row.transformed);
    assert!(row.remaining_ttl_seconds >= 86_000);
    shutdown.cancel();
    drop(cache);
    let shutdown = CancellationToken::new();
    let cache = Cache::start(11, Some(dir.path().to_path_buf()), settings.clone(), shutdown.clone());
    let hit = cache.get_tiered(&ticket, &settings).await.expect("aged transformed COLD should HIT after restart");
    assert!(hit.transformed);
    assert!(hit.remaining_ttl >= 86_000);
    let transformed_body = hit.response.into_body().collect().await.unwrap().to_bytes();
    assert_ne!(transformed_body.as_ref(), &body[..]);
    shutdown.cancel();
}

#[tokio::test]
async fn lru_demotion_persists_and_survives_restart() {
    let dir = tempfile::tempdir().unwrap();
    let settings = cold_settings();
    settings.validate().unwrap();
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        shutdown.clone(),
    );
    let first = fill(&cache, &settings, "/one", "first-payload").await;
    let _second = fill(&cache, &settings, "/two", "second-payload").await;
    wait_for_cold(&cache, 1).await;
    assert_eq!(cache.view().entries, 1);
    assert_eq!(cache.view().cold_entries, 1);
    let cold = cache.dashboard(Some("cold-test"), 0, 50);
    assert!(cold
        .objects
        .iter()
        .any(|row| row.path == "/one" && row.tier == "cold"));

    shutdown.cancel();
    drop(cache);

    let restart_shutdown = CancellationToken::new();
    let restarted = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        restart_shutdown.clone(),
    );
    assert_eq!(restarted.view().cold_entries, 1);
    let hit = restarted
        .get_tiered(&first, &settings)
        .await
        .expect("cold object should be readable after restart");
    let payload = hit.response.into_body().collect().await.unwrap().to_bytes();
    assert_eq!(payload, Bytes::from_static(b"first-payload"));
    assert_eq!(restarted.view().cold_hits, 1);
    assert_eq!(restarted.view().cold_promotions, 1);
    restart_shutdown.cancel();
}

#[tokio::test]
async fn dashboard_lists_retained_expired_cold_objects_without_permitting_fresh_hits() {
    let dir = tempfile::tempdir().unwrap();
    let settings = cold_settings();
    let shutdown = CancellationToken::new();
    let cache = Cache::start(11, Some(dir.path().to_path_buf()), settings.clone(), shutdown.clone());
    let first = fill(&cache, &settings, "/one", "first-payload").await;
    fill(&cache, &settings, "/two", "second-payload").await;
    wait_for_cold(&cache, 1).await;
    {
        let mut cold = cache.cold.lock().unwrap();
        for entry in cold.entries.values_mut() {
            entry.stored_at = SystemTime::now() - Duration::from_secs(2 * 60 * 60);
        }
    }
    let dashboard = cache.dashboard(Some("cold-test"), 0, 50);
    assert_eq!(dashboard.matching_objects, 2);
    let row = dashboard.objects.iter().find(|row| row.path == "/one").unwrap();
    assert_eq!(row.tier, "cold");
    assert_eq!(row.remaining_ttl_seconds, 0);
    assert!(row.age_seconds >= 2 * 60 * 60);
    assert_eq!(row.storage_bytes, cache.view().cold_bytes);
    assert!(cache.get_tiered(&first, &settings).await.is_none());
    let mut stale = first;
    stale.request_headers.insert("cache-control", HeaderValue::from_static("max-stale"));
    let hit = cache.get_tiered(&stale, &settings).await.expect("explicit max-stale can reuse the retained disk object");
    assert!(hit.stale);
    assert_eq!(hit.response.into_body().collect().await.unwrap().to_bytes(), "first-payload");
    shutdown.cancel();
}

#[tokio::test]
async fn cold_restart_preserves_transformed_validators_stale_metadata_and_no_transform() {
    let dir = tempfile::tempdir().unwrap();
    let mut settings = cold_settings();
    settings.max_entries = 8;
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        shutdown.clone(),
    );

    let transformed_ticket = cache
        .ticket_scoped(
            11,
            "cold-test",
            "http",
            "cold.example",
            "origin@127.0.0.1:8080",
            "/transformed.json",
            &HeaderMap::new(),
        )
        .unwrap();
    let mut transformed = Response::new(Body::from(r#"{"a":1}"#));
    transformed.headers_mut().insert(
        "cache-control",
        HeaderValue::from_static(
            "public, max-age=120, stale-while-revalidate=30, stale-if-error=90",
        ),
    );
    transformed.headers_mut().insert(
        "etag",
        HeaderValue::from_static("\"rustproxy-representation\""),
    );
    transformed
        .headers_mut()
        .insert("content-length", HeaderValue::from_static("7"));
    transformed
        .extensions_mut()
        .insert(crate::transform::OriginValidators {
            etag: Some(HeaderValue::from_static("\"origin-v1\"")),
            last_modified: Some(HeaderValue::from_static("Mon, 26 Jan 2026 23:11:40 GMT")),
        });
    transformed
        .extensions_mut()
        .insert(crate::transform::Applied {
            kind: crate::transform::Kind::Json,
            original_bytes: 9,
            transformed_bytes: 7,
        });
    let transformed_plan = plan(
        StatusCode::OK,
        transformed.headers(),
        &settings,
        Duration::ZERO,
    )
    .unwrap();
    cache
        .wrap(
            transformed,
            transformed_ticket.clone(),
            transformed_plan,
            &settings,
        )
        .into_body()
        .collect()
        .await
        .unwrap();

    let protected_ticket = cache
        .ticket_scoped(
            11,
            "cold-test",
            "http",
            "cold.example",
            "origin@127.0.0.1:8080",
            "/protected.css",
            &HeaderMap::new(),
        )
        .unwrap();
    let original_body = "body { color: red; }";
    let mut protected = Response::new(Body::from(original_body));
    protected.headers_mut().insert(
        "cache-control",
        HeaderValue::from_static("public, max-age=120, immutable, no-transform"),
    );
    protected
        .headers_mut()
        .insert("etag", HeaderValue::from_static("\"origin-css\""));
    protected.extensions_mut().insert(crate::compression::Skip);
    let protected_plan = plan(
        StatusCode::OK,
        protected.headers(),
        &settings,
        Duration::ZERO,
    )
    .unwrap();
    cache
        .wrap(
            protected,
            protected_ticket.clone(),
            protected_plan,
            &settings,
        )
        .into_body()
        .collect()
        .await
        .unwrap();

    assert_eq!(cache.flush_hot_to_cold().await, 2);
    wait_for_cold(&cache, 2).await;
    shutdown.cancel();
    drop(cache);

    let restart_shutdown = CancellationToken::new();
    let restarted = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        restart_shutdown.clone(),
    );
    assert_eq!(restarted.view().cold_entries, 2);
    // COLD entries intentionally keep validators and headers on disk to bound RAM.
    // Inspect them through the same verified metadata loader used on COLD hits.
    let (key, transformed) = {
        let cold = restarted.cold.lock().unwrap();
        cold.entries
            .iter()
            .find(|(_, entry)| entry.path == "/transformed.json")
            .map(|(key, entry)| (*key, entry.clone()))
            .expect("transformed cold entry")
    };
    assert_eq!(transformed.stale_while_revalidate, 30);
    assert_eq!(transformed.stale_if_error, 90);
    assert!(transformed.has_validator);
    assert!(transformed.transformed);
    let metadata = read_cold_metadata(&key, &transformed)
        .await
        .expect("persisted transformed metadata");
    assert_eq!(metadata.origin_etag.as_ref().unwrap(), "\"origin-v1\"");
    assert_eq!(
        metadata.origin_last_modified.as_ref().unwrap(),
        "Mon, 26 Jan 2026 23:11:40 GMT"
    );
    assert_eq!(metadata.headers["etag"], "\"rustproxy-representation\"");

    let transformed_hit = restarted
        .get_tiered(&transformed_ticket, &settings)
        .await
        .expect("transformed object after restart");
    assert_eq!(
        transformed_hit.response.headers()["etag"],
        "\"rustproxy-representation\""
    );
    assert!(transformed_hit.transformed);
    assert!(transformed_hit
        .response
        .extensions()
        .get::<crate::transform::CachedTransformed>()
        .is_some());
    assert_eq!(
        transformed_hit
            .response
            .into_body()
            .collect()
            .await
            .unwrap()
            .to_bytes(),
        Bytes::from_static(br#"{"a":1}"#)
    );

    let protected_hit = restarted
        .get_tiered(&protected_ticket, &settings)
        .await
        .expect("no-transform object after restart");
    assert_eq!(protected_hit.response.headers()["etag"], "\"origin-css\"");
    assert!(crate::transform::no_transform(
        protected_hit.response.headers()
    ));
    assert!(protected_hit
        .response
        .extensions()
        .get::<crate::compression::Skip>()
        .is_some());
    assert_eq!(
        protected_hit
            .response
            .into_body()
            .collect()
            .await
            .unwrap()
            .to_bytes(),
        Bytes::from_static(b"body { color: red; }")
    );
    restart_shutdown.cancel();
}

#[tokio::test]
async fn cold_policy_can_disable_lru_demotion() {
    let dir = tempfile::tempdir().unwrap();
    let mut settings = cold_settings();
    settings.cold_on_lru = false;
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        shutdown.clone(),
    );
    fill(&cache, &settings, "/one", "first-payload").await;
    fill(&cache, &settings, "/two", "second-payload").await;
    tokio::time::sleep(Duration::from_millis(30)).await;
    let view = cache.view();
    assert_eq!(view.entries, 1);
    assert_eq!(view.cold_entries, 0);
    assert_eq!(view.evictions, 1);
    shutdown.cancel();
}

#[tokio::test]
async fn cold_policy_can_disable_idle_demotion() {
    let dir = tempfile::tempdir().unwrap();
    let mut settings = cold_settings();
    settings.max_entries = 8;
    settings.cold_on_idle = false;
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        shutdown.clone(),
    );
    fill(&cache, &settings, "/idle", "idle-payload").await;
    {
        let mut store = cache.store.lock().unwrap();
        for entry in store.entries.values_mut() {
            entry.last_access = Instant::now() - Duration::from_secs(600);
        }
    }
    cache.demote_idle();
    assert_eq!(cache.view().entries, 1);
    assert_eq!(cache.view().cold_entries, 0);

    settings.cold_on_idle = true;
    cache.configure(settings.clone());
    cache.demote_idle();
    wait_for_cold(&cache, 1).await;
    assert_eq!(cache.view().entries, 0);
    shutdown.cancel();
}

#[tokio::test]
async fn cold_policy_can_demote_by_total_hot_residency_age() {
    let dir = tempfile::tempdir().unwrap();
    let mut settings = cold_settings();
    settings.max_entries = 8;
    settings.cold_on_idle = false;
    settings.cold_on_age = true;
    settings.cold_max_hot_seconds = 5;
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        shutdown.clone(),
    );
    fill(&cache, &settings, "/aged", "aged-payload").await;
    {
        let mut store = cache.store.lock().unwrap();
        for entry in store.entries.values_mut() {
            entry.resident_since = Instant::now() - Duration::from_secs(60);
            entry.last_access = Instant::now();
        }
    }
    cache.demote_idle();
    wait_for_cold(&cache, 1).await;
    assert_eq!(cache.view().entries, 0);
    assert_eq!(cache.view().cold_entries, 1);
    shutdown.cancel();
}

#[tokio::test]
async fn graceful_shutdown_flush_moves_hot_objects_and_releases_resident_bytes() {
    let dir = tempfile::tempdir().unwrap();
    let mut settings = cold_settings();
    settings.max_entries = 8;
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        shutdown.clone(),
    );
    fill(&cache, &settings, "/one", "first-payload").await;
    fill(&cache, &settings, "/two", "second-payload").await;
    assert_eq!(cache.view().entries, 2);
    assert!(cache.view().resident_bytes > 0);
    assert_eq!(cache.view().cold_entries, 0);

    assert_eq!(cache.flush_hot_to_cold().await, 2);
    let view = cache.view();
    assert_eq!(view.entries, 0);
    assert_eq!(view.resident_bytes, 0);
    assert_eq!(view.hot_logical_bytes, 0);
    assert_eq!(
        view.transient_bytes, 0,
        "persisted ColdEntry must not retain the body allocation"
    );
    assert_eq!(view.cold_entries, 2);
    shutdown.cancel();

    let restart_shutdown = CancellationToken::new();
    let restarted = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        restart_shutdown.clone(),
    );
    assert_eq!(restarted.view().cold_entries, 2);
    restart_shutdown.cancel();
}

#[tokio::test]
async fn managed_update_flush_ignores_shutdown_toggle_when_cold_tier_is_enabled() {
    let dir = tempfile::tempdir().unwrap();
    let mut settings = cold_settings();
    settings.max_entries = 8;
    settings.cold_on_shutdown = false;
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        shutdown.clone(),
    );
    fill(&cache, &settings, "/update", "update-payload").await;
    assert_eq!(cache.flush_hot_to_cold().await, 0);
    assert_eq!(cache.view().entries, 1);
    assert_eq!(cache.flush_hot_to_cold_for_update().await, 1);
    let view = cache.view();
    assert_eq!(view.entries, 0);
    assert_eq!(view.resident_bytes, 0);
    assert_eq!(view.cold_entries, 1);
    shutdown.cancel();
}

#[tokio::test]
async fn graceful_shutdown_flush_can_be_disabled() {
    let dir = tempfile::tempdir().unwrap();
    let mut settings = cold_settings();
    settings.max_entries = 8;
    settings.cold_on_shutdown = false;
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        shutdown.clone(),
    );
    fill(&cache, &settings, "/one", "first-payload").await;
    assert_eq!(cache.flush_hot_to_cold().await, 0);
    assert_eq!(cache.view().entries, 1);
    assert_eq!(cache.view().cold_entries, 0);
    shutdown.cancel();
}

#[tokio::test]
async fn cold_cache_uses_two_hex_character_shards() {
    let dir = tempfile::tempdir().unwrap();
    let settings = cold_settings();
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        shutdown.clone(),
    );
    fill(&cache, &settings, "/one", "first-payload").await;
    fill(&cache, &settings, "/two", "second-payload").await;
    wait_for_cold(&cache, 1).await;
    let root = dir.path().join(&settings.cold_path);
    let mut shards = Vec::new();
    for item in std::fs::read_dir(&root).unwrap() {
        let item = item.unwrap();
        let name = item.file_name().to_string_lossy().to_string();
        assert!(
            item.path().is_dir(),
            "cold root must contain shard directories only"
        );
        assert_eq!(name.len(), 2);
        assert!(name.bytes().all(|b| b.is_ascii_hexdigit()));
        shards.push(item.path());
    }
    assert!(!shards.is_empty());
    assert!(shards.iter().any(|shard| {
        std::fs::read_dir(shard)
            .unwrap()
            .filter_map(|entry| entry.ok())
            .any(|entry| {
                let name = entry.file_name().to_string_lossy().to_string();
                name.ends_with(".json") || name.ends_with(".body")
            })
    }));
    shutdown.cancel();
}

#[tokio::test]
async fn legacy_flat_cold_cache_is_not_migrated() {
    let dir = tempfile::tempdir().unwrap();
    let settings = cold_settings();
    let root = dir.path().join(&settings.cold_path);
    std::fs::create_dir_all(&root).unwrap();
    std::fs::write(root.join("legacy.body"), b"old").unwrap();
    std::fs::write(root.join("legacy.json"), b"{}").unwrap();
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings,
        shutdown.clone(),
    );
    assert_eq!(cache.view().cold_entries, 0);
    assert!(!root.join("legacy.body").exists());
    assert!(!root.join("legacy.json").exists());
    shutdown.cancel();
}

#[tokio::test]
async fn manual_cold_cleanup_reports_removed_objects_and_freed_bytes() {
    let dir = tempfile::tempdir().unwrap();
    let mut settings = cold_settings();
    settings.max_entries = 1;
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        shutdown.clone(),
    );
    fill(&cache, &settings, "/one", "first-payload").await;
    fill(&cache, &settings, "/two", "second-payload").await;
    fill(&cache, &settings, "/three", "third-payload").await;
    wait_for_cold(&cache, 2).await;
    let job = cache
        .start_cold_cleanup(ColdCleanupRequest {
            unused_for_seconds: None,
            oldest: Some(1),
            hits_zero: false,
        })
        .unwrap();
    for _ in 0..100 {
        let current = cache.cold_cleanup_job(job.id).unwrap();
        if current.status == "done" {
            assert_eq!(current.removed_entries, 1);
            assert!(current.freed_bytes > 0);
            assert_eq!(cache.view().cold_entries, 1);
            shutdown.cancel();
            return;
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    panic!("cold cleanup did not finish");
}

#[tokio::test]
async fn manual_cold_cleanup_is_single_flight() {
    let dir = tempfile::tempdir().unwrap();
    let settings = cold_settings();
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings,
        shutdown.clone(),
    );
    let _held = cache.cleanup_slots.clone().try_acquire_owned().unwrap();
    let result = cache.start_cold_cleanup(ColdCleanupRequest {
        unused_for_seconds: None,
        oldest: Some(1),
        hits_zero: false,
    });
    assert!(result.is_err());
    shutdown.cancel();
}

#[tokio::test]
async fn cold_cleanup_skips_an_entry_while_a_read_is_pinned() {
    let dir = tempfile::tempdir().unwrap();
    let mut settings = cold_settings();
    settings.max_entries = 1;
    let shutdown = CancellationToken::new();
    let cache = Cache::start(
        11,
        Some(dir.path().to_path_buf()),
        settings.clone(),
        shutdown.clone(),
    );
    fill(&cache, &settings, "/one", "first-payload").await;
    fill(&cache, &settings, "/two", "second-payload").await;
    wait_for_cold(&cache, 1).await;

    let key = {
        let mut cold = cache.cold.lock().unwrap();
        let key = *cold.entries.keys().next().unwrap();
        cold.entries.get_mut(&key).unwrap().pinned_reads = 1;
        key
    };
    let request = ColdCleanupRequest {
        unused_for_seconds: None,
        oldest: Some(1),
        hits_zero: false,
    };
    let result = cache
        .remove_cold_candidates(Some(&request), None, None, 1)
        .await;
    assert_eq!(result.removed_entries, 0);
    assert!(cache.cold.lock().unwrap().entries.contains_key(&key));

    cache
        .cold
        .lock()
        .unwrap()
        .entries
        .get_mut(&key)
        .unwrap()
        .pinned_reads = 0;
    let result = cache
        .remove_cold_candidates(Some(&request), None, None, 1)
        .await;
    assert_eq!(result.removed_entries, 1);
    assert!(!cache.cold.lock().unwrap().entries.contains_key(&key));
    shutdown.cancel();
}

#[test]
fn old_cold_read_guard_does_not_unpin_a_replacement_with_the_same_key() {
    let cache = Cache::new(1);
    let key = [7_u8; 32];
    let template = ColdEntry {
        base_key: key,
        vary: Vec::new(),
        revision: 1,
        scope: "app".to_owned(),
        scheme: "https".to_owned(),
        authority: "example.test".to_owned(),
        domain: "example.test".to_owned(),
        path: "/asset".to_owned(),
        target: "https://example.test/asset".to_owned(),
        tags: Vec::new(),
        created_at: SystemTime::now(),
        body_path: PathBuf::from("old.body"),
        meta_path: PathBuf::from("old.json"),
        body_bytes: 1,
        body_sha256: [0u8; 32],
        disk_bytes: 2,
        stored_at: SystemTime::now(),
        last_access: SystemTime::now(),
        age_at_store_seconds: 0,
        resident_age_at_store_seconds: 0,
        ttl: 60,
        stale_while_revalidate: 0,
        stale_if_error: 0,
        must_revalidate: false,
        proxy_revalidate: false,
        has_validator: false,
        hits: 0,
        transformed: false,
        sequence: 1,
        generation: 1,
        skip_compression: false,
        requires_revalidation: false,
        pinned_reads: 1,
    };
    cache
        .cold
        .lock()
        .unwrap()
        .entries
        .insert(key, template.clone());
    let old_guard = ColdReadGuard {
        cache: &cache,
        key,
        generation: 1,
    };

    let mut replacement = template;
    replacement.sequence = 2;
    replacement.generation = 2;
    replacement.pinned_reads = 1;
    cache.cold.lock().unwrap().entries.insert(key, replacement);
    drop(old_guard);

    let cold = cache.cold.lock().unwrap();
    let replacement = cold.entries.get(&key).unwrap();
    assert_eq!(replacement.generation, 2);
    assert_eq!(replacement.pinned_reads, 1);
}

#[test]
fn cold_path_cannot_escape_data_directory() {
    let mut settings = Settings::default();
    settings.cold_path = "../outside".into();
    assert!(settings.validate().is_err());
    settings.cold_path = "/tmp/outside".into();
    assert!(settings.validate().is_err());
    settings.cold_path = "cache/cold".into();
    assert!(settings.validate().is_ok());
}

#[tokio::test]
async fn cold_scope_fingerprint_preserves_unmodified_vhost_and_rejects_stale_restart() {
    let dir = tempfile::tempdir().unwrap();
    let settings = cold_settings();
    let shutdown = CancellationToken::new();
    let mut fingerprints = HashMap::new();
    fingerprints.insert("cold-test".to_owned(), 77);
    let cache = Cache::start_with_scope_revisions(
        11, Some(dir.path().to_path_buf()), settings.clone(), shutdown.clone(),
        fingerprints.clone(),
    );
    fill(&cache, &settings, "/one", "stable-contents").await;
    fill(&cache, &settings, "/two", "second-contents").await;
    wait_for_cold(&cache, 1).await;
    shutdown.cancel();
    drop(cache);

    let restart = CancellationToken::new();
    let same = Cache::start_with_scope_revisions(
        11, Some(dir.path().to_path_buf()), settings.clone(), restart.clone(),
        fingerprints.clone(),
    );
    assert_eq!(same.view().cold_entries, 1, "operational saves retain COLD");
    restart.cancel();
    drop(same);

    fingerprints.insert("cold-test".to_owned(), 78);
    let restart = CancellationToken::new();
    let changed = Cache::start_with_scope_revisions(
        11, Some(dir.path().to_path_buf()), settings.clone(), restart.clone(), fingerprints,
    );
    assert_eq!(changed.view().cold_entries, 0, "old COLD must never survive a content revision");
    restart.cancel();
}
}