//! # cachly – Official Rust SDK for cachly.dev
//!
//! Managed Valkey/Redis cache · DSGVO-compliant · German servers · 30s provisioning
//!
//! ## Quick start
//!
//! ```rust,no_run
//! use cachly::CachlyClient;
//!
//! #[tokio::main]
//! async fn main() -> cachly::Result<()> {
//!     let cache = CachlyClient::new("redis://:password@my-instance.cachly.dev:6379").await?;
//!
//!     // Set a value with TTL
//!     cache.set("user:42", &serde_json::json!({"name": "Alice"}), Some(300)).await?;
//!
//!     // Get a value
//!     let val: Option<serde_json::Value> = cache.get("user:42").await?;
//!
//!     // Get-or-set pattern
//!     let data: String = cache
//!         .get_or_set("expensive:key", || async { Ok("computed".to_string()) }, Some(60))
//!         .await?;
//!
//!     Ok(())
//! }
//! ```
//!
//! ## Semantic cache with pgvector API (HNSW O(log n))
//!
//! Enable the `vector-api` feature in `Cargo.toml`:
//! ```toml
//! cachly = { version = "*", features = ["vector-api"] }
//! ```
//! Then pass `vector_url` to [`SemanticOptions`]:
//! ```rust,no_run
//! # use cachly::{CachlyClient, SemanticOptions};
//! # #[tokio::main]
//! # async fn main() -> cachly::Result<()> {
//! let client = CachlyClient::new("redis://:pw@host:6379").await?;
//! let sem = client.semantic_with_vector(
//!     |text: String| async move { Ok::<Vec<f64>, cachly::CachlyError>(vec![0.1, 0.2]) },
//!     std::env::var("CACHLY_VECTOR_URL").ok(),
//! );
//! # Ok(())
//! # }
//! ```

use std::future::Future;

use redis::{aio::ConnectionManager, AsyncCommands, Client};
use serde::{de::DeserializeOwned, Serialize};
use thiserror::Error;
use uuid::Uuid;

// ── Error ─────────────────────────────────────────────────────────────────────

/// All errors returned by the cachly SDK.
#[derive(Debug, Error)]
pub enum CachlyError {
    #[error("Redis error: {0}")]
    Redis(#[from] redis::RedisError),
    #[error("Serialization error: {0}")]
    Serde(#[from] serde_json::Error),
    #[error("Embed function error: {0}")]
    Embed(String),
    #[error("HTTP error: {0}")]
    Http(String),
}

/// Convenience alias.
pub type Result<T> = std::result::Result<T, CachlyError>;

// ── SemanticConfidence ────────────────────────────────────────────────────────

/// Three-level quality band for a semantic cache hit.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SemanticConfidence {
    /// similarity ≥ high_confidence_threshold (default 0.97). Serve directly.
    High,
    /// similarity ≥ threshold. Consider A/B logging.
    Medium,
    /// Cache miss.
    Uncertain,
}

// ── SemanticResult ────────────────────────────────────────────────────────────

/// Returned by [`SemanticCache::get_or_set`].
#[derive(Debug)]
pub struct SemanticResult<T> {
    pub value: T,
    pub hit: bool,
    pub similarity: Option<f64>,
    /// Only set on a cache hit.
    pub confidence: Option<SemanticConfidence>,
}

// ── Internal helpers ──────────────────────────────────────────────────────────

const DEFAULT_FILLER_WORDS: &[&str] = &[
    // EN
    "please", "hey", "hi", "hello",
    "could you", "can you", "would you", "will you",
    "just", "quickly", "briefly", "simply",
    "tell me", "show me", "give me", "help me", "assist me",
    "explain to me", "describe to me",
    "i need", "i want", "i would like", "i'd like", "i'm looking for",
    // DE
    "bitte", "mal eben", "schnell", "kurz", "einfach",
    "kannst du", "könntest du", "könnten sie", "würden sie", "würdest du",
    "hallo", "hi", "hey",
    "sag mir", "zeig mir", "gib mir", "hilf mir", "erkläre mir", "erklär mir",
    "ich brauche", "ich möchte", "ich hätte gerne", "ich suche",
    // FR
    "s'il vous plaît", "svp", "stp", "bonjour", "salut", "allô",
    "pouvez-vous", "pourriez-vous", "peux-tu", "pourrais-tu",
    "dis-moi", "dites-moi", "montre-moi", "montrez-moi",
    "j'ai besoin de", "je voudrais", "je cherche", "je souhaite",
    "expliquez-moi", "explique-moi", "aidez-moi", "aide-moi",
    // ES
    "por favor", "hola", "oye",
    "puedes", "podrías", "podría usted", "me puedes", "me podrías",
    "dime", "dígame", "muéstrame", "muéstreme", "dame", "deme",
    "necesito", "quisiera", "me gustaría", "quiero saber",
    "ayúdame", "ayúdeme", "explícame", "explíqueme",
    // IT
    "per favore", "perfavore", "ciao", "salve", "ehi",
    "potresti", "mi potresti", "potrebbe", "mi potrebbe",
    "dimmi", "mi dica", "mostrami", "dammi", "mi dia",
    "ho bisogno di", "vorrei", "mi piacerebbe",
    "aiutami", "mi aiuti", "spiegami", "mi spieghi",
    // PT
    "por favor", "olá", "oi", "ei",
    "pode", "poderia", "você poderia", "você pode", "podes",
    "me diga", "diga-me", "me mostre", "mostre-me", "me dê", "dê-me",
    "preciso de", "gostaria de", "quero saber", "estou procurando",
    "me ajude", "ajude-me", "explique-me", "me explique",
];

/// Strip filler words, lowercase, collapse whitespace. +8–12% hit-rate uplift.
fn normalize_prompt(text: &str, filler_words: Option<&[&str]>) -> String {
    let mut s = text.trim().to_lowercase();
    let words = filler_words.unwrap_or(DEFAULT_FILLER_WORDS);
    for fw in words {
        s = s.replace(*fw, "");
    }
    // Collapse spaces
    while s.contains("  ") {
        s = s.replace("  ", " ");
    }
    s = s.trim().to_string();
    // Normalise trailing punctuation
    let stripped = s.trim_end_matches(|c| c == '?' || c == '!');
    if stripped.is_empty() { return s; }
    format!("{stripped}?")
}

/// Returns the confidence band for a similarity score.
fn confidence_band(sim: f64, threshold: f64, high_threshold: f64) -> SemanticConfidence {
    if sim >= high_threshold { SemanticConfidence::High }
    else if sim >= threshold { SemanticConfidence::Medium }
    else { SemanticConfidence::Uncertain }
}

/// Derive the val key from an emb key: `{ns}:emb:{uuid}` → `{ns}:val:{uuid}`.
fn val_key(emb_key: &str) -> String {
    let last_colon = emb_key.rfind(':').unwrap_or(emb_key.len());
    let uuid_part = &emb_key[last_colon..]; // ":uuid-str"
    let ns_type = &emb_key[..last_colon]; // "{ns}:emb"
    let second_last_colon = ns_type.rfind(':').unwrap_or(ns_type.len());
    let ns = &ns_type[..second_last_colon]; // "{ns}"
    format!("{ns}:val{uuid_part}")
}

/// Extract the UUID (last segment) from any key: `{ns}:emb:{uuid}` → `{uuid}`.
fn extract_id(key: &str) -> &str {
    key.rfind(':').map(|i| &key[i + 1..]).unwrap_or(key)
}

/// Extract the namespace from an emb/val key: `{ns}:emb:{uuid}` → `{ns}`.
fn extract_namespace(key: &str) -> &str {
    let last = key.rfind(':').unwrap_or(key.len());
    let ns_type = &key[..last];
    let second = ns_type.rfind(':').unwrap_or(ns_type.len());
    &ns_type[..second]
}

/// Cursor-based SCAN; collects all matching keys without blocking the server.
async fn scan_all(conn: &mut ConnectionManager, pattern: &str) -> Result<Vec<String>> {
    let mut keys = Vec::new();
    let mut cursor: u64 = 0;
    loop {
        let (new_cursor, batch): (u64, Vec<String>) = redis::cmd("SCAN")
            .arg(cursor)
            .arg("MATCH")
            .arg(pattern)
            .arg("COUNT")
            .arg(100)
            .query_async(conn)
            .await?;
        keys.extend(batch);
        cursor = new_cursor;
        if cursor == 0 {
            break;
        }
    }
    Ok(keys)
}

// ── HTTP helper (requires feature "vector-api") ───────────────────────────────

#[cfg(feature = "vector-api")]
async fn http_request(
    method: reqwest::Method,
    url: &str,
    body: Option<serde_json::Value>,
) -> Result<serde_json::Value> {
    let client = reqwest::Client::builder()
        .timeout(std::time::Duration::from_secs(10))
        .build()
        .map_err(|e| CachlyError::Http(e.to_string()))?;

    let mut req = client.request(method, url);
    if let Some(b) = body {
        req = req.json(&b);
    }
    let resp = req
        .send()
        .await
        .map_err(|e| CachlyError::Http(e.to_string()))?;

    let status = resp.status();
    let text = resp
        .text()
        .await
        .map_err(|e| CachlyError::Http(e.to_string()))?;

    if !status.is_success() {
        return Err(CachlyError::Http(format!(
            "cachly api returned {status}: {text}"
        )));
    }
    let val: serde_json::Value = serde_json::from_str(&text)?;
    Ok(val)
}

// ── §7 int8 Quantization ──────────────────────────────────────────────────────

/// Scalar int8-quantized representation of a float64 embedding vector.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct QuantizedEmbedding {
    pub values: Vec<i8>,
    pub min: f64,
    pub max: f64,
}

/// Scalar-quantize a float64 embedding to int8 range [-128, 127] (§7).
/// <1% cosine-similarity quality loss; reduces API payload ~8x.
pub fn quantize_embedding(vec: &[f64]) -> QuantizedEmbedding {
    if vec.is_empty() {
        return QuantizedEmbedding { values: vec![], min: 0.0, max: 0.0 };
    }
    let (mut min_val, mut max_val) = (vec[0], vec[0]);
    for &v in &vec[1..] {
        if v < min_val { min_val = v; }
        if v > max_val { max_val = v; }
    }
    let rng = max_val - min_val;
    let values = if rng > 0.0 {
        let scale = 255.0 / rng;
        vec.iter().map(|&v| {
            let q = (scale * (v - min_val)).round() as i64 - 128;
            q.max(-128).min(127) as i8
        }).collect()
    } else {
        vec![0i8; vec.len()]
    };
    QuantizedEmbedding { values, min: min_val, max: max_val }
}

// ── SemanticCache ─────────────────────────────────────────────────────────────

/// Cache LLM responses by *meaning*, not just exact key.
///
/// When `vector_url` is set → uses cachly pgvector API (HNSW O(log n)) with
/// graceful SCAN fallback if the API is unreachable.
///
/// # Example
/// ```rust,no_run
/// use cachly::{CachlyClient, SemanticOptions};
///
/// # #[tokio::main]
/// # async fn main() -> cachly::Result<()> {
/// let client = CachlyClient::new("redis://:pw@host:6379").await?;
///
/// let embed_fn = |text: String| async move {
///     Ok::<Vec<f64>, cachly::CachlyError>(vec![0.1, 0.2, 0.3])
/// };
///
/// let sem = client.semantic(embed_fn);
/// let result = sem.get_or_set(
///     "What is Cachly?",
///     || async { Ok("Cachly is a managed Redis-compatible cache.".to_string()) },
///     SemanticOptions { threshold: 0.90, ttl: Some(3600), namespace: None, vector_url: None },
/// ).await?;
/// println!("hit={} val={}", result.hit, result.value);
/// # Ok(())
/// # }
/// ```
pub struct SemanticCache<EF> {
    conn: ConnectionManager,
    embed_fn: EF,
}

/// Options for a [`SemanticCache::get_or_set`] call.
#[derive(Debug, Clone, Default)]
pub struct SemanticOptions {
    /// Cosine-similarity threshold (0.0–1.0). Default should be ~0.85.
    pub threshold: f64,
    /// TTL in seconds for new cache entries. `None` = no expiry.
    pub ttl: Option<u64>,
    /// Redis key namespace prefix. `None` = `"cachly:sem"`.
    pub namespace: Option<String>,
    /// Optional pgvector API base URL: `https://api.cachly.dev/v1/sem/{token}`.
    pub vector_url: Option<String>,
    /// Strip filler words and lowercase before embedding (default: true).
    /// +8–12% semantic hit-rate uplift at zero quality cost.
    pub normalize_prompt: Option<bool>,
    /// Similarity ≥ this → `SemanticConfidence::High`. Default: 0.97.
    pub high_confidence_threshold: Option<f64>,
    /// §1 – replace `threshold` with the server-side F1-calibrated value.
    /// Requires `vector_url`. Falls back to `threshold` on API error.
    pub use_adaptive_threshold: bool,
    /// §4 – auto-detect namespace from the prompt using text heuristics.
    /// Ignored when `namespace` is explicitly set. Default: `false`.
    pub auto_namespace: bool,
    /// §7 – quantize embedding before sending to the pgvector API.
    /// `Some("int8")` reduces JSON payload ~8x with <1% quality loss.
    /// `None` or `Some("float32")` = full precision (default).
    pub quantize: Option<String>,
    /// §3 – enable Hybrid BM25+Vector RRF fusion search.
    /// Passes `hybrid: true` and the normalised prompt to the pgvector API.
    /// Requires `vector_url`. Default: `false`.
    pub use_hybrid: bool,
}

// ── §4 Namespace Auto-Detection ───────────────────────────────────────────────

/// Classify a prompt into one of 5 semantic namespaces using text heuristics.
///
/// Overhead: < 0.1 ms. No embedding required.
///
/// Returns one of:
/// `cachly:sem:code`, `cachly:sem:translation`, `cachly:sem:summary`,
/// `cachly:sem:qa`, `cachly:sem:creative`
pub fn detect_namespace(prompt: &str) -> &'static str {
    let s = prompt.trim().to_lowercase();
    let s = s.as_str();

    // Code
    for kw in &["function ", "def ", "class ", "import ", "const ", "let ", "var ",
                "return ", " => ", "void ", "public class", "func ", "#include", "package ",
                "struct {", "interface {", "async def", "lambda ", "#!/"] {
        if s.contains(kw) { return "cachly:sem:code"; }
    }
    // Translation
    for kw in &["translate", "übersetze", "auf deutsch", "auf englisch",
                "in english", "in german", "ins deutsche", "ins englische", "übersetz",
                "traduce", "traduis", "vertaal"] {
        if s.contains(kw) { return "cachly:sem:translation"; }
    }
    // Summary
    for kw in &["summarize", "summarise", "summary", "zusammenfass", "tl;dr", "tldr",
                "key points", "stichpunkte", "fasse zusammen", "give me a brief",
                "kurze zusammenfassung", "in a nutshell"] {
        if s.contains(kw) { return "cachly:sem:summary"; }
    }
    // Q&A
    for prefix in &["what ", "who ", "where ", "when ", "why ", "how ", "which ",
                    "is ", "are ", "was ", "were ", "does ", "do ", "did ",
                    "can ", "could ", "would ", "should ", "will ",
                    "wer ", "wie ", "wo ", "wann ", "warum ", "welche", "wieso "] {
        if s.starts_with(prefix) { return "cachly:sem:qa"; }
    }
    if s.trim_end().ends_with('?') { return "cachly:sem:qa"; }

    "cachly:sem:creative"
}

// ── §8 Cache-Warming ──────────────────────────────────────────────────────────

/// One entry to pre-warm into the semantic cache (§8).
pub struct WarmupEntry<F, Fut, T>
where
    F: Fn() -> Fut,
    Fut: std::future::Future<Output = Result<T>>,
    T: Serialize,
{
    pub prompt: String,
    pub fn_: F,
    /// Optional namespace override; `None` falls back to `SemanticOptions::namespace`.
    pub namespace: Option<String>,
}

/// Result of a [`SemanticCache::warmup`] call (§8).
#[derive(Debug, Clone)]
pub struct WarmupResult {
    pub warmed: usize,
    pub skipped: usize,
}

// ── New Result Types (SDK Feature Gap) ────────────────────────────────────────

/// Cache statistics returned by [`SemanticCache::stats`].
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default)]
pub struct CacheStats {
    pub hits: u64,
    pub misses: u64,
    pub hit_rate: f64,
    pub total: u64,
    pub namespaces: Vec<serde_json::Value>,
}

/// One entry for bulk indexing via [`SemanticCache::batch_index`].
#[derive(Debug, Clone, serde::Serialize)]
pub struct BatchIndexEntry {
    pub id: String,
    pub prompt: String,
    pub embedding: Vec<f64>,
    pub namespace: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub expires_at: Option<String>,
}

/// Result of [`SemanticCache::batch_index`].
#[derive(Debug, Clone, serde::Deserialize)]
pub struct BatchIndexResult {
    pub indexed: u32,
    pub skipped: u32,
}

/// A single guardrail violation.
#[derive(Debug, Clone, serde::Deserialize)]
pub struct GuardrailViolation {
    #[serde(rename = "type")]
    pub violation_type: String,
    pub pattern: String,
    pub action: String,
}

/// Result of [`SemanticCache::check_guardrail`].
#[derive(Debug, Clone, serde::Deserialize)]
pub struct GuardrailCheckResult {
    pub safe: bool,
    pub violations: Vec<GuardrailViolation>,
}

/// Tags associated with a cache key.
#[derive(Debug, Clone, serde::Deserialize)]
pub struct TagsResult {
    pub key: String,
    pub tags: Vec<String>,
    #[serde(default = "default_true")]
    pub ok: bool,
}

fn default_true() -> bool { true }

/// Result of [`CachlyClient::invalidate_tag`].
#[derive(Debug, Clone, serde::Deserialize)]
pub struct InvalidateTagResult {
    pub tag: String,
    pub keys_deleted: u32,
    #[serde(default)]
    pub keys: Vec<String>,
    pub duration_ms: u64,
}

/// One entry in the SWR (Stale-While-Revalidate) registry.
#[derive(Debug, Clone, serde::Deserialize)]
pub struct SwrEntry {
    pub key: String,
    pub fetcher_hint: Option<String>,
    pub stale_for: Option<String>,
    pub refresh_at: Option<String>,
}

/// Result of [`CachlyClient::swr_check`].
#[derive(Debug, Clone, serde::Deserialize)]
pub struct SwrCheckResult {
    pub stale_keys: Vec<SwrEntry>,
    pub count: u32,
    pub checked_at: String,
}

/// Result of [`CachlyClient::bulk_warmup`].
#[derive(Debug, Clone, serde::Deserialize)]
pub struct BulkWarmupResult {
    pub warmed: u32,
    pub skipped: u32,
    pub duration_ms: u64,
}

/// Result of [`SemanticCache::snapshot_warmup`].
#[derive(Debug, Clone, serde::Deserialize)]
pub struct SnapshotWarmupResult {
    pub warmed: u32,
    pub duration_ms: u64,
}

/// LLM proxy statistics.
#[derive(Debug, Clone, serde::Deserialize, Default)]
pub struct LlmProxyStatsResult {
    pub total_requests: u64,
    pub cache_hits: u64,
    pub cache_misses: u64,
    pub estimated_saved_usd: f64,
    pub avg_latency_ms_cached: u64,
    pub avg_latency_ms_uncached: u64,
}

/// A message received via [`PubSubClient::subscribe`].
#[derive(Debug, Clone, serde::Deserialize)]
pub struct PubSubMessage {
    pub channel: String,
    pub message: String,
    pub at: String,
}

/// A workflow checkpoint.
#[derive(Debug, Clone, serde::Deserialize, Default)]
pub struct WorkflowCheckpoint {
    pub id: String,
    pub run_id: String,
    pub step_index: i32,
    pub step_name: String,
    pub agent_name: String,
    pub status: String,
    pub state: Option<String>,
    pub output: Option<String>,
    pub duration_ms: Option<u64>,
    pub created_at: Option<String>,
}

/// Summary of a workflow run.
#[derive(Debug, Clone, serde::Deserialize)]
pub struct WorkflowRun {
    pub run_id: String,
    pub steps: u32,
    pub latest_status: String,
    #[serde(default)]
    pub checkpoints: Vec<WorkflowCheckpoint>,
}

/// One entry for [`CachlyClient::bulk_warmup`].
#[derive(Debug, Clone, serde::Serialize)]
pub struct BulkWarmupEntry {
    pub key: String,
    pub value: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub ttl: Option<u64>,
}

impl<EF, EFut> SemanticCache<EF>
where
    EF: Fn(String) -> EFut,
    EFut: Future<Output = Result<Vec<f64>>>,
{
    // ── §1 Adaptive Threshold ─────────────────────────────────────────────────

    /// Record whether a cache hit was accepted as correct (§1).
    /// Requires the `vector-api` feature and `opts.vector_url`. No-op otherwise.
    #[cfg(feature = "vector-api")]
    pub async fn feedback(
        &self,
        hit_id: &str,
        accepted: bool,
        similarity: f64,
        namespace: Option<&str>,
        vector_url: &str,
    ) -> Result<()> {
        let ns = namespace.unwrap_or("cachly:sem");
        http_request(
            reqwest::Method::POST,
            &format!("{vector_url}/feedback"),
            Some(serde_json::json!({
                "hit_id": hit_id,
                "accepted": accepted,
                "similarity": similarity,
                "namespace": ns,
            })),
        )
        .await
        .map(|_| ())
        .or_else(|_| Ok(())) // best-effort
    }

    /// Return the server-side F1-calibrated threshold for a namespace (§1).
    /// Falls back to `fallback` (typically 0.85) on error or when no data exists.
    #[cfg(feature = "vector-api")]
    pub async fn adaptive_threshold(
        &self,
        namespace: Option<&str>,
        vector_url: &str,
        fallback: f64,
    ) -> f64 {
        let ns = namespace.unwrap_or("cachly:sem");
        match http_request(
            reqwest::Method::GET,
            &format!("{vector_url}/threshold?namespace={ns}"),
            None,
        )
        .await
        {
            Ok(resp) => resp.get("threshold").and_then(|v| v.as_f64()).unwrap_or(fallback),
            Err(_) => fallback,
        }
    }

    /// Return a cached response for semantically similar prompts, or call `f`
    /// and cache the result.
    ///
    /// When `opts.vector_url` is set and the `vector-api` feature is enabled,
    /// uses the pgvector API (HNSW O(log n)) with graceful SCAN fallback.
    pub async fn get_or_set<T, F, Fut>(
        &self,
        prompt: &str,
        f: F,
        opts: SemanticOptions,
    ) -> Result<SemanticResult<T>>
    where
        T: Serialize + DeserializeOwned,
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<T>>,
    {
        // Normalise prompt before embedding (default: true).
        let should_norm = opts.normalize_prompt.unwrap_or(true);
        let text_for_embed: std::borrow::Cow<str> = if should_norm {
            std::borrow::Cow::Owned(normalize_prompt(prompt, None))
        } else {
            std::borrow::Cow::Borrowed(prompt)
        };

        let threshold = if opts.threshold == 0.0 { 0.85 } else { opts.threshold };
        let high_threshold = opts.high_confidence_threshold.unwrap_or(0.97);
        // §4 – auto-detect namespace when requested and no explicit namespace was set.
        let ns: std::borrow::Cow<str> = match (&opts.namespace, opts.auto_namespace) {
            (Some(n), _) => std::borrow::Cow::Borrowed(n.as_str()),
            (None, true) => std::borrow::Cow::Borrowed(detect_namespace(prompt)),
            (None, false) => std::borrow::Cow::Borrowed("cachly:sem"),
        };
        let ns = ns.as_ref();

        // Try pgvector API when vector_url is set and feature is enabled.
        #[cfg(feature = "vector-api")]
        if let Some(ref vurl) = opts.vector_url {
            // §1 – fetch adaptive threshold when requested.
            let effective_threshold = if opts.use_adaptive_threshold {
                self.adaptive_threshold(Some(ns), vurl, threshold).await
            } else {
                threshold
            };

            match self.get_or_set_via_api(&text_for_embed, vurl, &opts, effective_threshold).await {
                Ok((Some(hit_id), sim)) => {
                    let vk = format!("{ns}:val:{hit_id}");
                    let mut conn = self.conn.clone();
                    let raw: Option<String> = conn.get(&vk).await?;
                    if let Some(raw) = raw {
                        if let Ok(value) = serde_json::from_str::<T>(&raw) {
                            let conf = confidence_band(sim, threshold, high_threshold);
                            // Stats – fire-and-forget
                            let mut sc = self.conn.clone();
                            let stats_key = format!("{ns}:stats:hits");
                            tokio::spawn(async move { let _: Result<i64, _> = sc.incr(&stats_key, 1).await; });
                            return Ok(SemanticResult { value, hit: true, similarity: Some(sim), confidence: Some(conf) });
                        }
                    }
                    // orphaned pgvector entry – fall through to miss
                }
                Ok((None, _)) => {
                    // Cache miss via API – stats + index.
                    let mut sc = self.conn.clone();
                    let stats_key = format!("{ns}:stats:misses");
                    tokio::spawn(async move { let _: Result<i64, _> = sc.incr(&stats_key, 1).await; });
                    return self.cache_miss_via_api(&text_for_embed, f, vurl, &opts).await;
                }
                Err(_) => {
                    // API unreachable → graceful degradation: use SCAN.
                }
            }
        }

        // SCAN fallback.
        self.get_or_set_via_scan(&text_for_embed, prompt, f, opts).await
    }

    /// pgvector API search only – returns `(Some(id), similarity)` on hit, `(None, 0.0)` on miss.
    #[cfg(feature = "vector-api")]
    async fn get_or_set_via_api(
        &self,
        normalized_prompt: &str,
        vector_url: &str,
        opts: &SemanticOptions,
        effective_threshold: f64,
    ) -> Result<(Option<String>, f64)> {
        let ns = opts.namespace.as_deref().unwrap_or("cachly:sem");
        let query_embed = (self.embed_fn)(normalized_prompt.to_owned()).await?;

        // §7 – build search body with optional int8 quantization.
        let mut search_body = if opts.quantize.as_deref() == Some("int8") {
            serde_json::json!({
                "embedding_q8": quantize_embedding(&query_embed),
                "namespace": ns,
                "threshold": effective_threshold,
            })
        } else {
            serde_json::json!({
                "embedding": query_embed,
                "namespace": ns,
                "threshold": effective_threshold,
            })
        };

        // §3 – hybrid BM25+Vector RRF: include prompt and set hybrid=true.
        if opts.use_hybrid && !normalized_prompt.is_empty() {
            search_body["hybrid"] = serde_json::json!(true);
            search_body["prompt"] = serde_json::json!(normalized_prompt);
        }

        let resp = http_request(
            reqwest::Method::POST,
            &format!("{vector_url}/search"),
            Some(search_body),
        )
        .await?;

        let found = resp.get("found").and_then(|v| v.as_bool()).unwrap_or(false);
        if found {
            let id = resp.get("id").and_then(|v| v.as_str()).map(str::to_owned);
            let sim = resp.get("similarity").and_then(|v| v.as_f64()).unwrap_or(0.0);
            return Ok((id, sim));
        }
        Ok((None, 0.0))
    }

    /// Called on API cache miss: run fn, store val in Valkey, index in pgvector.
    #[cfg(feature = "vector-api")]
    async fn cache_miss_via_api<T, F, Fut>(
        &self,
        normalized_prompt: &str,
        f: F,
        vector_url: &str,
        opts: &SemanticOptions,
    ) -> Result<SemanticResult<T>>
    where
        T: Serialize + DeserializeOwned,
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<T>>,
    {
        let ns = opts.namespace.as_deref().unwrap_or("cachly:sem");
        let query_embed = (self.embed_fn)(normalized_prompt.to_owned()).await?;
        let value = f().await?;
        let id = Uuid::new_v4().to_string();
        let v_key = format!("{ns}:val:{id}");
        let val_payload = serde_json::to_string(&value)?;

        let mut conn = self.conn.clone();
        if let Some(ttl) = opts.ttl {
            let _: () = redis::cmd("SETEX").arg(&v_key).arg(ttl).arg(&val_payload).query_async(&mut conn).await?;
        } else {
            let _: () = conn.set(&v_key, &val_payload).await?;
        }

        let mut index_body = if opts.quantize.as_deref() == Some("int8") {
            serde_json::json!({
                "id":           id,
                "embedding_q8": quantize_embedding(&query_embed),
                "prompt":       normalized_prompt,
                "namespace":    ns,
            })
        } else {
            serde_json::json!({
                "id":        id,
                "embedding": query_embed,
                "prompt":    normalized_prompt,
                "namespace": ns,
            })
        };
        if let Some(ttl) = opts.ttl {
            let exp = chrono_expires_at(ttl);
            index_body["expires_at"] = serde_json::Value::String(exp);
        }
        let _ = http_request(
            reqwest::Method::POST,
            &format!("{vector_url}/entries"),
            Some(index_body),
        )
        .await; // best-effort

        Ok(SemanticResult { value, hit: false, similarity: None, confidence: None })
    }

    /// Linear SCAN fallback.
    async fn get_or_set_via_scan<T, F, Fut>(
        &self,
        normalized_prompt: &str,
        original_prompt: &str,
        f: F,
        opts: SemanticOptions,
    ) -> Result<SemanticResult<T>>
    where
        T: Serialize + DeserializeOwned,
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<T>>,
    {
        let ns = opts.namespace.as_deref().unwrap_or("cachly:sem");
        let threshold = if opts.threshold == 0.0 { 0.85 } else { opts.threshold };
        let high_threshold = opts.high_confidence_threshold.unwrap_or(0.97);
        let query_embed = (self.embed_fn)(normalized_prompt.to_owned()).await?;
        let mut conn = self.conn.clone();

        let emb_keys = scan_all(&mut conn, &format!("{ns}:emb:*")).await?;

        let mut best_sim = -1.0_f64;
        let mut best_val_key: Option<String> = None;

        for emb_key in &emb_keys {
            let raw: Option<String> = conn.get(emb_key).await?;
            let Some(raw) = raw else { continue };
            let Ok(entry) = serde_json::from_str::<serde_json::Value>(&raw) else { continue };
            let Some(emb_arr) = entry.get("embedding").and_then(|v| v.as_array()) else { continue };
            let stored: Vec<f64> = emb_arr.iter().filter_map(|v| v.as_f64()).collect();
            let sim = cosine_similarity(&query_embed, &stored);
            if sim > best_sim {
                best_sim = sim;
                best_val_key = Some(val_key(emb_key));
            }
        }

        if best_sim >= threshold {
            if let Some(vk) = best_val_key {
                let raw: Option<String> = conn.get(&vk).await?;
                if let Some(raw) = raw {
                    if let Ok(value) = serde_json::from_str::<T>(&raw) {
                        let conf = confidence_band(best_sim, threshold, high_threshold);
                        // Stats – fire-and-forget
                        let mut sc = self.conn.clone();
                        let stats_key = format!("{ns}:stats:hits");
                        tokio::spawn(async move { let _: std::result::Result<i64, _> = sc.incr(&stats_key, 1).await; });
                        return Ok(SemanticResult { value, hit: true, similarity: Some(best_sim), confidence: Some(conf) });
                    }
                }
                // orphaned emb – fall through to miss
            }
        }

        // Cache miss
        let mut sc = self.conn.clone();
        let stats_key = format!("{ns}:stats:misses");
        tokio::spawn(async move { let _: std::result::Result<i64, _> = sc.incr(&stats_key, 1).await; });
        let value = f().await?;
        let id = Uuid::new_v4().to_string();
        let emb_key = format!("{ns}:emb:{id}");
        let v_key = format!("{ns}:val:{id}");

        let emb_payload = serde_json::json!({
            "embedding":       query_embed,
            "prompt":          normalized_prompt,
            "original_prompt": if original_prompt.is_empty() { normalized_prompt } else { original_prompt },
        })
        .to_string();
        let val_payload = serde_json::to_string(&value)?;

        if let Some(ttl) = opts.ttl {
            let _: () = redis::cmd("SETEX").arg(&v_key).arg(ttl).arg(&val_payload).query_async(&mut conn).await?;
            let _: () = redis::cmd("SETEX").arg(&emb_key).arg(ttl).arg(&emb_payload).query_async(&mut conn).await?;
        } else {
            let _: () = conn.set(&v_key, &val_payload).await?;
            let _: () = conn.set(&emb_key, &emb_payload).await?;
        }

        Ok(SemanticResult { value, hit: false, similarity: None, confidence: None })
    }

    /// Remove a single semantic cache entry by its emb key.
    ///
    /// In API mode: deletes val from Valkey + calls DELETE on the API.
    /// In SCAN mode: deletes both emb and val from Valkey.
    pub async fn invalidate(&self, key: &str, vector_url: Option<&str>) -> Result<bool> {
        let id = extract_id(key);
        let ns = extract_namespace(key);
        let vk = format!("{ns}:val:{id}");
        let mut conn = self.conn.clone();

        #[cfg(feature = "vector-api")]
        if let Some(vurl) = vector_url {
            let _: usize = conn.del(&[vk.as_str()]).await?;
            let url = format!("{vurl}/entries/{id}");
            return match http_request(reqwest::Method::DELETE, &url, None).await {
                Ok(_) => Ok(true),
                Err(_) => Ok(false),
            };
        }

        let deleted: usize = conn.del(&[key, vk.as_str()]).await?;
        Ok(deleted > 0)
    }

    /// List all cached prompts with their emb Redis keys.
    ///
    /// In API mode: queries pgvector API.
    /// In SCAN mode: scans Valkey emb keys.
    pub async fn entries(
        &self,
        namespace: Option<&str>,
        vector_url: Option<&str>,
    ) -> Result<Vec<(String, String)>> {
        let ns = namespace.unwrap_or("cachly:sem");

        #[cfg(feature = "vector-api")]
        if let Some(vurl) = vector_url {
            let url = format!("{vurl}/entries?namespace={ns}");
            if let Ok(resp) = http_request(reqwest::Method::GET, &url, None).await {
                let mut result = Vec::new();
                if let Some(data) = resp.get("data").and_then(|v| v.as_array()) {
                    for item in data {
                        let id = item.get("id").and_then(|v| v.as_str()).unwrap_or("").to_owned();
                        let prompt = item.get("prompt").and_then(|v| v.as_str()).unwrap_or("").to_owned();
                        if !id.is_empty() {
                            result.push((format!("{ns}:emb:{id}"), prompt));
                        }
                    }
                }
                return Ok(result);
            }
            // Fallback to SCAN on error.
        }

        let mut conn = self.conn.clone();
        let emb_keys = scan_all(&mut conn, &format!("{ns}:emb:*")).await?;
        let mut result = Vec::new();
        for emb_key in emb_keys {
            let raw: Option<String> = conn.get(&emb_key).await?;
            let Some(raw) = raw else { continue };
            let Ok(entry) = serde_json::from_str::<serde_json::Value>(&raw) else { continue };
            // Prefer original_prompt (un-normalised, for display); fall back to prompt.
            let prompt = entry.get("original_prompt")
                .and_then(|v| v.as_str()).filter(|s| !s.is_empty())
                .or_else(|| entry.get("prompt").and_then(|v| v.as_str()))
                .unwrap_or("").to_owned();
            result.push((emb_key, prompt));
        }
        Ok(result)
    }

    /// Delete all semantic cache entries in a namespace.
    /// Returns count of logical entries (emb keys) deleted.
    pub async fn flush(&self, namespace: Option<&str>, vector_url: Option<&str>) -> Result<usize> {
        let ns = namespace.unwrap_or("cachly:sem");

        #[cfg(feature = "vector-api")]
        if let Some(vurl) = vector_url {
            let url = format!("{vurl}/flush?namespace={ns}");
            let deleted = match http_request(reqwest::Method::DELETE, &url, None).await {
                Ok(resp) => resp.get("deleted").and_then(|v| v.as_u64()).unwrap_or(0) as usize,
                Err(_) => 0,
            };
            // Purge val keys from Valkey.
            let mut conn = self.conn.clone();
            let val_keys = scan_all(&mut conn, &format!("{ns}:val:*")).await?;
            if !val_keys.is_empty() {
                let _: usize = conn.del(val_keys).await?;
            }
            return Ok(deleted);
        }

        let mut conn = self.conn.clone();
        let emb_keys = scan_all(&mut conn, &format!("{ns}:emb:*")).await?;
        let val_keys = scan_all(&mut conn, &format!("{ns}:val:*")).await?;
        let emb_count = emb_keys.len();
        let mut all: Vec<String> = emb_keys;
        all.extend(val_keys);
        if !all.is_empty() {
            let _: usize = conn.del(all).await?;
        }
        Ok(emb_count)
    }

    /// Return the number of entries in the semantic cache namespace.
    pub async fn size(&self, namespace: Option<&str>, vector_url: Option<&str>) -> Result<usize> {
        let ns = namespace.unwrap_or("cachly:sem");

        #[cfg(feature = "vector-api")]
        if let Some(vurl) = vector_url {
            let url = format!("{vurl}/size?namespace={ns}");
            if let Ok(resp) = http_request(reqwest::Method::GET, &url, None).await {
                return Ok(resp.get("size").and_then(|v| v.as_u64()).unwrap_or(0) as usize);
            }
            // Fallback to SCAN on error.
        }

        let mut conn = self.conn.clone();
        Ok(scan_all(&mut conn, &format!("{ns}:emb:*")).await?.len())
    }

    // ── §8 Cache-Warming ─────────────────────────────────────────────────────

    /// §8 – Pre-warm the semantic cache with a list of prompt/fn pairs.
    ///
    /// For each entry, [`get_or_set`] is called with a high threshold (0.98) so
    /// already-cached entries are returned without calling `f` again.
    ///
    /// # Arguments
    /// - `entries`: slice of `(prompt, namespace_override, fn)` tuples
    ///   where `fn: FnOnce() -> Future<Output = Result<T>>`
    /// - `opts`: base [`SemanticOptions`] applied to every entry
    ///
    /// # Returns
    /// [`WarmupResult`] with `warmed` (new) and `skipped` (already cached) counts.
    pub async fn warmup<T, F, Fut>(
        &self,
        entries: Vec<(String, Option<String>, F)>,
        opts: SemanticOptions,
    ) -> Result<WarmupResult>
    where
        T: Serialize + DeserializeOwned,
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<T>>,
    {
        let mut warmed = 0usize;
        let mut skipped = 0usize;
        let warm_threshold = if opts.threshold == 0.0 { 0.98 } else { opts.threshold };

        for (prompt, ns_override, f) in entries {
            let mut entry_opts = opts.clone();
            entry_opts.threshold = warm_threshold;
            if let Some(ns) = ns_override {
                entry_opts.namespace = Some(ns);
            } else if entry_opts.auto_namespace && entry_opts.namespace.is_none() {
                entry_opts.namespace = Some(detect_namespace(&prompt).to_owned());
            }
            match self.get_or_set::<T, F, Fut>(&prompt, f, entry_opts).await {
                Ok(r) if r.hit => skipped += 1,
                Ok(_)          => warmed  += 1,
                Err(_)         => skipped += 1,
            }
        }
        Ok(WarmupResult { warmed, skipped })
    }

    /// §8 – Import prompts from a JSONL file and warm the cache with the results.
    ///
    /// Each line must be a JSON object. The prompt is extracted from the field named by
    /// `prompt_field` (default: `"prompt"`). `response_fn` is called for every cache miss.
    /// `batch_size` controls how many lines are processed per iteration (default: 50).
    /// The similarity threshold in `opts` defaults to 0.98 when set to 0.
    ///
    /// # Errors
    /// Returns `CachlyError::Embed` when the file cannot be read.
    pub async fn import_from_log<T, R, Fut>(
        &self,
        file_path: &str,
        response_fn: R,
        opts: SemanticOptions,
        prompt_field: &str,
        _batch_size: usize,
    ) -> Result<WarmupResult>
    where
        T: Serialize + DeserializeOwned,
        R: Fn(String) -> Fut,
        Fut: Future<Output = Result<T>>,
    {
        let content = tokio::fs::read_to_string(file_path).await
            .map_err(|e| CachlyError::Embed(e.to_string()))?;

        let field = if prompt_field.is_empty() { "prompt" } else { prompt_field };
        let warm_opts = SemanticOptions {
            threshold: if opts.threshold == 0.0 { 0.98 } else { opts.threshold },
            ..opts
        };

        let mut warmed = 0usize;
        let mut skipped = 0usize;

        for line in content.lines() {
            let line = line.trim();
            if line.is_empty() { continue; }
            let obj: serde_json::Value = match serde_json::from_str(line) {
                Ok(v) => v,
                Err(_) => { skipped += 1; continue; }
            };
            let prompt = match obj.get(field).and_then(|v| v.as_str()) {
                Some(p) if !p.is_empty() => p.to_owned(),
                _ => { skipped += 1; continue; }
            };
            let p = prompt.clone();
            let fut = response_fn(prompt.clone());
            match self.get_or_set::<T, _, Fut>(&p, || fut, warm_opts.clone()).await {
                Ok(r) if r.hit => skipped += 1,
                Ok(_)          => warmed  += 1,
                Err(_)         => skipped += 1,
            }
        }
        Ok(WarmupResult { warmed, skipped })
    }

    // ── New API methods (SDK Feature Gap) ─────────────────────────────────────

    /// Set the F1-calibrated similarity threshold for a namespace.
    ///
    /// `POST /v1/sem/:token/threshold`
    #[cfg(feature = "vector-api")]
    pub async fn set_threshold(&self, namespace: &str, threshold: f64, vector_url: &str) -> Result<()> {
        http_request(
            reqwest::Method::POST,
            &format!("{vector_url}/threshold"),
            Some(serde_json::json!({ "namespace": namespace, "threshold": threshold })),
        )
        .await
        .map(|_| ())
    }

    /// Return cache statistics for a namespace.
    ///
    /// `GET /v1/sem/:token/stats?namespace=…`
    #[cfg(feature = "vector-api")]
    pub async fn stats(&self, namespace: &str, vector_url: &str) -> Result<CacheStats> {
        let url = format!("{vector_url}/stats?namespace={namespace}");
        let resp = http_request(reqwest::Method::GET, &url, None).await?;
        Ok(serde_json::from_value(resp).unwrap_or_default())
    }

    /// SSE-streaming semantic search – returns text chunks.
    ///
    /// `POST /v1/sem/:token/search/stream`
    #[cfg(feature = "vector-api")]
    pub async fn stream_search(
        &self,
        prompt: &str,
        vector_url: &str,
        namespace: Option<&str>,
        threshold: f64,
    ) -> Result<Vec<String>> {
        let ns = namespace.unwrap_or("cachly:sem");
        let embed = (self.embed_fn)(prompt.to_owned()).await?;
        let body = serde_json::json!({
            "embedding": embed,
            "namespace": ns,
            "threshold": threshold,
            "prompt": prompt,
        });
        let client = reqwest::Client::builder()
            .timeout(std::time::Duration::from_secs(60))
            .build()
            .map_err(|e| CachlyError::Http(e.to_string()))?;
        let text = client
            .post(&format!("{vector_url}/search/stream"))
            .header("Accept", "text/event-stream")
            .json(&body)
            .send()
            .await
            .map_err(|e| CachlyError::Http(e.to_string()))?
            .text()
            .await
            .map_err(|e| CachlyError::Http(e.to_string()))?;

        let mut chunks = Vec::new();
        for line in text.lines() {
            if let Some(data) = line.strip_prefix("data:") {
                let data = data.trim();
                if data.is_empty() || data == "{}" { continue; }
                if let Ok(v) = serde_json::from_str::<serde_json::Value>(data) {
                    if let Some(t) = v.get("text").and_then(|t| t.as_str()) {
                        if !t.is_empty() { chunks.push(t.to_owned()); }
                    }
                }
            }
        }
        Ok(chunks)
    }

    /// Bulk-index up to 500 semantic cache entries.
    ///
    /// `POST /v1/sem/:token/entries/batch`
    #[cfg(feature = "vector-api")]
    pub async fn batch_index(
        &self,
        entries: Vec<BatchIndexEntry>,
        vector_url: &str,
    ) -> Result<BatchIndexResult> {
        if entries.len() > 500 {
            return Err(CachlyError::Embed("batch_index: max 500 entries per request".into()));
        }
        let resp = http_request(
            reqwest::Method::POST,
            &format!("{vector_url}/entries/batch"),
            Some(serde_json::json!({ "entries": entries })),
        )
        .await?;
        Ok(serde_json::from_value(resp).map_err(|e| CachlyError::Serde(e))?)
    }

    /// Create a new vector index.
    ///
    /// `POST /v1/sem/:token/indexes`
    #[cfg(feature = "vector-api")]
    pub async fn create_index(
        &self,
        namespace: &str,
        dimensions: u32,
        model: &str,
        metric: &str,
        hybrid_enabled: bool,
        vector_url: &str,
    ) -> Result<serde_json::Value> {
        http_request(
            reqwest::Method::POST,
            &format!("{vector_url}/indexes"),
            Some(serde_json::json!({
                "namespace": namespace, "dimensions": dimensions,
                "model": model, "metric": metric, "hybrid_enabled": hybrid_enabled,
            })),
        )
        .await
    }

    /// Delete an index and all its entries.
    ///
    /// `DELETE /v1/sem/:token/indexes/:namespace`
    #[cfg(feature = "vector-api")]
    pub async fn delete_index(&self, namespace: &str, vector_url: &str) -> Result<serde_json::Value> {
        http_request(
            reqwest::Method::DELETE,
            &format!("{vector_url}/indexes/{namespace}"),
            None,
        )
        .await
    }

    /// Attach JSONB metadata to a semantic cache entry.
    ///
    /// `POST /v1/sem/:token/metadata`
    #[cfg(feature = "vector-api")]
    pub async fn set_metadata(
        &self,
        entry_id: &str,
        metadata: serde_json::Value,
        vector_url: &str,
    ) -> Result<serde_json::Value> {
        http_request(
            reqwest::Method::POST,
            &format!("{vector_url}/metadata"),
            Some(serde_json::json!({ "entry_id": entry_id, "metadata": metadata })),
        )
        .await
    }

    /// Semantic search with metadata filter.
    ///
    /// `POST /v1/sem/:token/search/filtered`
    #[cfg(feature = "vector-api")]
    pub async fn filtered_search(
        &self,
        prompt: &str,
        vector_url: &str,
        namespace: Option<&str>,
        threshold: f64,
        filter: serde_json::Value,
        limit: u32,
    ) -> Result<serde_json::Value> {
        let ns = namespace.unwrap_or("cachly:sem");
        let embed = (self.embed_fn)(normalize_prompt(prompt, None)).await?;
        http_request(
            reqwest::Method::POST,
            &format!("{vector_url}/search/filtered"),
            Some(serde_json::json!({
                "prompt": prompt, "embedding": embed,
                "namespace": ns, "threshold": threshold,
                "filter": filter, "limit": limit,
            })),
        )
        .await
    }

    /// Configure content-safety guardrails.
    ///
    /// `POST /v1/sem/:token/guardrails`
    #[cfg(feature = "vector-api")]
    pub async fn set_guardrail(
        &self,
        namespace: &str,
        pii_action: &str,
        toxic_action: &str,
        toxic_threshold: f64,
        vector_url: &str,
    ) -> Result<serde_json::Value> {
        http_request(
            reqwest::Method::POST,
            &format!("{vector_url}/guardrails"),
            Some(serde_json::json!({
                "namespace": namespace, "pii_action": pii_action,
                "toxic_action": toxic_action, "toxic_threshold": toxic_threshold,
            })),
        )
        .await
    }

    /// Remove guardrail configuration for a namespace.
    ///
    /// `DELETE /v1/sem/:token/guardrails/:namespace`
    #[cfg(feature = "vector-api")]
    pub async fn delete_guardrail(&self, namespace: &str, vector_url: &str) -> Result<serde_json::Value> {
        http_request(
            reqwest::Method::DELETE,
            &format!("{vector_url}/guardrails/{namespace}"),
            None,
        )
        .await
    }

    /// Check text against configured guardrails.
    ///
    /// `POST /v1/sem/:token/guardrails/check`
    #[cfg(feature = "vector-api")]
    pub async fn check_guardrail(
        &self,
        text: &str,
        namespace: &str,
        vector_url: &str,
    ) -> Result<GuardrailCheckResult> {
        let resp = http_request(
            reqwest::Method::POST,
            &format!("{vector_url}/guardrails/check"),
            Some(serde_json::json!({ "text": text, "namespace": namespace })),
        )
        .await?;
        Ok(serde_json::from_value(resp).map_err(|e| CachlyError::Serde(e))?)
    }

    /// Re-warm the semantic cache from existing entries.
    ///
    /// `POST /v1/sem/:token/warmup/snapshot`
    #[cfg(feature = "vector-api")]
    pub async fn snapshot_warmup(
        &self,
        namespace: &str,
        limit: u32,
        vector_url: &str,
    ) -> Result<SnapshotWarmupResult> {
        let resp = http_request(
            reqwest::Method::POST,
            &format!("{vector_url}/warmup/snapshot"),
            Some(serde_json::json!({ "namespace": namespace, "limit": limit })),
        )
        .await?;
        Ok(serde_json::from_value(resp).map_err(|e| CachlyError::Serde(e))?)
    }
}

/// Main client for cachly.dev managed Valkey/Redis instances.
///
/// Uses a [`ConnectionManager`] for automatic reconnects.
pub struct CachlyClient {
    conn: ConnectionManager,
    /// Optional KV Batch API URL (`CACHLY_BATCH_URL` from the dashboard).
    /// Format: `https://api.cachly.dev/v1/cache/{token}`
    batch_url: Option<String>,
    /// Optional Pub/Sub API URL. Format: `https://api.cachly.dev/v1/pubsub/{token}`
    pubsub_url: Option<String>,
    /// Optional Workflow API URL. Format: `https://api.cachly.dev/v1/workflow/{token}`
    workflow_url: Option<String>,
    /// Optional LLM Proxy stats URL. Format: `https://api.cachly.dev/v1/llm-proxy/{token}`
    llm_proxy_url: Option<String>,
    /// Optional Edge Cache Management API URL. Format: `https://api.cachly.dev/v1/edge/{token}`
    edge_api_url: Option<String>,
}

// ── Batch API types ───────────────────────────────────────────────────────────

/// A single operation in a [`CachlyClient::batch`] call.
#[derive(Debug, Clone, serde::Serialize)]
pub struct BatchOp {
    /// Operation type: `"get"` | `"set"` | `"del"` | `"exists"` | `"ttl"`
    pub op: String,
    /// Cache key (required).
    pub key: String,
    /// Value (required for `"set"`).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub value: Option<String>,
    /// TTL in seconds for `"set"` ops. `None` = no expiry.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub ttl: Option<u64>,
}

impl BatchOp {
    pub fn get(key: impl Into<String>) -> Self { Self { op: "get".into(), key: key.into(), value: None, ttl: None } }
    pub fn set(key: impl Into<String>, value: impl Into<String>, ttl: Option<u64>) -> Self {
        Self { op: "set".into(), key: key.into(), value: Some(value.into()), ttl }
    }
    pub fn del(key: impl Into<String>) -> Self { Self { op: "del".into(), key: key.into(), value: None, ttl: None } }
    pub fn exists(key: impl Into<String>) -> Self { Self { op: "exists".into(), key: key.into(), value: None, ttl: None } }
    pub fn ttl(key: impl Into<String>) -> Self { Self { op: "ttl".into(), key: key.into(), value: None, ttl: None } }
}

/// Result of a single operation in a [`CachlyClient::batch`] call.
#[derive(Debug, Clone)]
pub struct BatchOpResult {
    /// Value for `"get"` ops. `None` = key not found.
    pub value: Option<String>,
    /// `true` if a `"get"` key existed.
    pub found: bool,
    /// `true` for successful `"set"` and `"del"` (≥1 key deleted).
    pub ok: bool,
    /// Result of `"exists"` op.
    pub exists: bool,
    /// Remaining TTL in seconds for `"ttl"` op. `-1` = no expiry, `-2` = not found.
    pub ttl_seconds: i64,
    /// Per-op error from the server (non-fatal).
    pub error: Option<String>,
}

// ── LockHandle ────────────────────────────────────────────────────────────────

/// A handle returned by a successful [`CachlyClient::lock`] call.
/// Call [`release`](LockHandle::release) in a `defer`/`finally` block to free the lock early.
/// The lock auto-expires after the configured TTL even if `release` is never called.
pub struct LockHandle {
    conn: ConnectionManager,
    lock_key: String,
    token: String,
}

impl LockHandle {
    /// Unique fencing token for this lock acquisition.
    pub fn token(&self) -> &str { &self.token }

    /// Release the lock atomically. No-op if the lock has already expired.
    pub async fn release(&mut self) -> Result<()> {
        const SCRIPT: &str = r#"
            if redis.call("get", KEYS[1]) == ARGV[1] then
                return redis.call("del", KEYS[1])
            else
                return 0
            end
        "#;
        let mut conn = self.conn.clone();
        redis::Script::new(SCRIPT)
            .key(&self.lock_key)
            .arg(&self.token)
            .invoke_async::<_, i64>(&mut conn)
            .await?;
        Ok(())
    }
}

impl CachlyClient {
    /// Create a client from a Redis connection URL.
    ///
    /// Format: `redis://:password@host:port`
    pub async fn new(url: &str) -> Result<Self> {
        let client = Client::open(url)?;
        let conn = ConnectionManager::new(client).await?;
        Ok(CachlyClient { conn, batch_url: None, pubsub_url: None, workflow_url: None, llm_proxy_url: None, edge_api_url: None })
    }

    /// Set the KV Batch API URL (from dashboard: `CACHLY_BATCH_URL`).
    /// When set, [`batch`](CachlyClient::batch) sends all ops in one HTTP request.
    pub fn with_batch_url(mut self, url: impl Into<String>) -> Self {
        self.batch_url = Some(url.into().trim_end_matches('/').to_string());
        self
    }

    /// Set the Pub/Sub API URL.
    pub fn with_pubsub_url(mut self, url: impl Into<String>) -> Self {
        self.pubsub_url = Some(url.into().trim_end_matches('/').to_string());
        self
    }

    /// Set the Workflow API URL.
    pub fn with_workflow_url(mut self, url: impl Into<String>) -> Self {
        self.workflow_url = Some(url.into().trim_end_matches('/').to_string());
        self
    }

    /// Set the LLM Proxy stats URL.
    pub fn with_llm_proxy_url(mut self, url: impl Into<String>) -> Self {
        self.llm_proxy_url = Some(url.into().trim_end_matches('/').to_string());
        self
    }

    /// Set the Edge Cache Management API URL.
    /// When set, [`edge`](CachlyClient::edge) returns an [`EdgeCacheClient`].
    pub fn with_edge_api_url(mut self, url: impl Into<String>) -> Self {
        self.edge_api_url = Some(url.into().trim_end_matches('/').to_string());
        self
    }

    /// Return an [`EdgeCacheClient`] for managing Cloudflare Edge Cache.
    /// Returns `None` if `edge_api_url` was not configured.
    #[cfg(feature = "vector-api")]
    pub fn edge(&self) -> Option<EdgeCacheClient> {
        self.edge_api_url.as_ref().map(|url| EdgeCacheClient { base_url: url.clone() })
    }

    // ── Basic operations ──────────────────────────────────────────────────────

    /// Get a value by key. Returns `None` when not found.
    pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>> {
        let mut conn = self.conn.clone();
        let raw: Option<String> = conn.get(key).await?;
        match raw {
            None => Ok(None),
            Some(s) => {
                let v: T = serde_json::from_str(&s)
                    .or_else(|_| serde_json::from_value(serde_json::Value::String(s)))?;
                Ok(Some(v))
            }
        }
    }

    /// Set a value. Pass `ttl` (seconds) for automatic expiry.
    pub async fn set<T: Serialize>(&self, key: &str, value: &T, ttl: Option<u64>) -> Result<()> {
        let mut conn = self.conn.clone();
        let serialized = serde_json::to_string(value)?;
        if let Some(secs) = ttl {
            let _: () = redis::cmd("SETEX")
                .arg(key)
                .arg(secs)
                .arg(&serialized)
                .query_async(&mut conn)
                .await?;
        } else {
            let _: () = conn.set(key, serialized).await?;
        }
        Ok(())
    }

    /// Delete one or more keys. Returns the number of keys deleted.
    pub async fn del(&self, keys: &[&str]) -> Result<usize> {
        let mut conn = self.conn.clone();
        Ok(conn.del(keys.to_vec()).await?)
    }

    /// Check whether a key exists.
    pub async fn exists(&self, key: &str) -> Result<bool> {
        let mut conn = self.conn.clone();
        Ok(conn.exists(key).await?)
    }

    /// Update TTL for an existing key.
    pub async fn expire(&self, key: &str, seconds: u64) -> Result<bool> {
        let mut conn = self.conn.clone();
        Ok(conn.expire(key, seconds as i64).await?)
    }

    /// Return the remaining TTL for a key in seconds.
    /// Returns `-1` if the key has no TTL, `-2` if the key does not exist.
    pub async fn ttl(&self, key: &str) -> Result<i64> {
        let mut conn = self.conn.clone();
        let v: i64 = redis::cmd("TTL").arg(key).query_async(&mut conn).await?;
        Ok(v)
    }

    /// Check connectivity to the cache server.
    /// Returns `true` if the server responds to PING.
    pub async fn ping(&self) -> bool {
        let mut conn = self.conn.clone();
        redis::cmd("PING")
            .query_async::<_, String>(&mut conn)
            .await
            .map(|r| r.eq_ignore_ascii_case("pong"))
            .unwrap_or(false)
    }

    /// Atomic counter increment. Returns the new value.
    pub async fn incr(&self, key: &str) -> Result<i64> {
        let mut conn = self.conn.clone();
        Ok(conn.incr(key, 1).await?)
    }

    /// Get-or-set: return cached value or call `f`, cache and return result.
    pub async fn get_or_set<T, F, Fut>(
        &self,
        key: &str,
        f: F,
        ttl: Option<u64>,
    ) -> Result<T>
    where
        T: Serialize + DeserializeOwned,
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<T>>,
    {
        if let Some(v) = self.get::<T>(key).await? {
            return Ok(v);
        }
        let value = f().await?;
        self.set(key, &value, ttl).await?;
        Ok(value)
    }

    // ── Bulk operations ───────────────────────────────────────────────────────

    /// Set multiple key-value pairs in a single pipeline round-trip.
    /// Supports per-key TTL – unlike native MSET which has no expiry option.
    ///
    /// `items` is a slice of `(key, value, ttl_secs)` tuples.
    pub async fn mset_bulk<T: Serialize>(&self, items: &[(&str, T, Option<u64>)]) -> Result<()> {
        if items.is_empty() {
            return Ok(());
        }
        let mut conn = self.conn.clone();
        let mut pipe = redis::pipe();
        for (key, value, ttl) in items.iter() {
            let serialized = serde_json::to_string(value)?;
            if let Some(secs) = ttl {
                pipe.cmd("SETEX").arg(*key).arg(*secs).arg(&serialized).ignore();
            } else {
                pipe.set(*key, &serialized).ignore();
            }
        }
        pipe.query_async::<_, ()>(&mut conn).await?;
        Ok(())
    }

    /// Retrieve multiple keys in one round-trip (native MGET).
    /// Returns a `Vec` in the same order as `keys`; missing keys are `None`.
    pub async fn mget<T: DeserializeOwned>(&self, keys: &[&str]) -> Result<Vec<Option<T>>> {
        if keys.is_empty() {
            return Ok(vec![]);
        }
        let mut conn = self.conn.clone();
        let raws: Vec<Option<String>> = redis::cmd("MGET")
            .arg(keys)
            .query_async(&mut conn)
            .await?;
        let mut result = Vec::with_capacity(raws.len());
        for raw in raws {
            match raw {
                None => result.push(None),
                Some(s) => {
                    let v: T = serde_json::from_str(&s)
                        .or_else(|_| serde_json::from_value(serde_json::Value::String(s)))?;
                    result.push(Some(v));
                }
            }
        }
        Ok(result)
    }

    // ── Distributed lock ──────────────────────────────────────────────────────

    /// Acquire a distributed lock using Redis `SET NX PX` (Redlock-lite pattern).
    ///
    /// Returns `Ok(Some(LockHandle))` on success, `Ok(None)` when all attempts
    /// are exhausted. The lock auto-expires after `ttl_ms` to prevent deadlocks.
    ///
    /// - `ttl_ms`: safety TTL in milliseconds
    /// - `retries`: max acquire attempts (default: 3)
    /// - `retry_delay_ms`: milliseconds between retries (default: 50)
    pub async fn lock(
        &self,
        key: &str,
        ttl_ms: u64,
        retries: Option<u32>,
        retry_delay_ms: Option<u64>,
    ) -> Result<Option<LockHandle>> {
        let retries = retries.unwrap_or(3);
        let retry_delay = retry_delay_ms.unwrap_or(50);
        let lock_key = format!("cachly:lock:{key}");
        let token = Uuid::new_v4().to_string();

        for attempt in 0..=retries {
            let mut conn = self.conn.clone();
            let result: Option<String> = redis::cmd("SET")
                .arg(&lock_key)
                .arg(&token)
                .arg("NX")
                .arg("PX")
                .arg(ttl_ms)
                .query_async(&mut conn)
                .await?;
            if result.as_deref() == Some("OK") {
                return Ok(Some(LockHandle { conn: self.conn.clone(), lock_key, token }));
            }
            if attempt < retries {
                tokio::time::sleep(std::time::Duration::from_millis(retry_delay)).await;
            }
        }
        Ok(None)
    }

    // ── Batch API ─────────────────────────────────────────────────────────────

    /// Execute multiple cache operations in a **single round-trip**.
    ///
    /// When `batch_url` is set (via [`with_batch_url`](CachlyClient::with_batch_url)),
    /// all ops are sent in one `POST {batch_url}/batch` HTTP request.
    /// Otherwise they fall back to a Redis pipeline.
    ///
    /// ```rust,no_run
    /// # use cachly::{CachlyClient, BatchOp};
    /// # #[tokio::main]
    /// # async fn main() -> cachly::Result<()> {
    /// let cache = CachlyClient::new("redis://:pw@host:6379").await?;
    /// let results = cache.batch(vec![
    ///     BatchOp::get("user:1"),
    ///     BatchOp::get("config:app"),
    ///     BatchOp::set("visits", "42", Some(86400)),
    /// ]).await?;
    /// let user_val = &results[0].value;   // Option<String>
    /// let ok       =  results[2].ok;      // bool
    /// # Ok(())
    /// # }
    /// ```
    pub async fn batch(&self, ops: Vec<BatchOp>) -> Result<Vec<BatchOpResult>> {
        if ops.is_empty() {
            return Ok(vec![]);
        }
        if let Some(ref base_url) = self.batch_url {
            self.batch_via_http(ops, base_url).await
        } else {
            self.batch_via_pipeline(ops).await
        }
    }

    #[cfg(feature = "reqwest")]
    async fn batch_via_http(&self, ops: Vec<BatchOp>, base_url: &str) -> Result<Vec<BatchOpResult>> {
        use serde_json::Value;
        let url = format!("{base_url}/batch");
        let body = serde_json::json!({ "ops": ops });
        let resp: Value = reqwest::Client::new()
            .post(&url)
            .json(&body)
            .send()
            .await
            .map_err(|e| CachlyError::Http(e.to_string()))?
            .json()
            .await
            .map_err(|e| CachlyError::Http(e.to_string()))?;
        let results_raw = resp["results"].as_array().cloned().unwrap_or_default();
        let mut out = Vec::with_capacity(ops.len());
        for (i, r) in results_raw.iter().enumerate() {
            if i >= ops.len() { break; }
            let mut res = BatchOpResult { value: None, found: false, ok: false, exists: false, ttl_seconds: -2, error: None };
            if let Some(e) = r["error"].as_str().filter(|s| !s.is_empty()) {
                res.error = Some(e.to_string());
                out.push(res);
                continue;
            }
            match ops[i].op.as_str() {
                "get" => {
                    res.found = r["found"].as_bool().unwrap_or(false);
                    res.value = r["value"].as_str().map(str::to_string);
                }
                "set"    => res.ok = r["ok"].as_bool().unwrap_or(false),
                "del"    => res.ok = r["deleted"].as_i64().unwrap_or(0) > 0,
                "exists" => res.exists = r["exists"].as_bool().unwrap_or(false),
                "ttl"    => res.ttl_seconds = r["ttl_seconds"].as_i64().unwrap_or(-2),
                _        => {}
            }
            out.push(res);
        }
        Ok(out)
    }

    #[cfg(not(feature = "reqwest"))]
    async fn batch_via_http(&self, ops: Vec<BatchOp>, _base_url: &str) -> Result<Vec<BatchOpResult>> {
        // No HTTP client available without the `reqwest` feature – fall back to pipeline.
        self.batch_via_pipeline(ops).await
    }

    async fn batch_via_pipeline(&self, ops: Vec<BatchOp>) -> Result<Vec<BatchOpResult>> {
        use redis::pipe;
        let mut pipeline = pipe();
        for op in &ops {
            match op.op.as_str() {
                "get"    => { pipeline.get(&op.key); }
                "set"    => {
                    let val = op.value.as_deref().unwrap_or("");
                    if let Some(secs) = op.ttl {
                        pipeline.cmd("SETEX").arg(&op.key).arg(secs).arg(val);
                    } else {
                        pipeline.set(&op.key, val);
                    }
                }
                "del"    => { pipeline.del(&op.key); }
                "exists" => { pipeline.exists(&op.key); }
                "ttl"    => { pipeline.ttl(&op.key); }
                _        => {}
            }
        }
        let mut conn = self.conn.clone();
        let raw: Vec<redis::Value> = pipeline.query_async(&mut conn).await?;
        let mut out = Vec::with_capacity(ops.len());
        for (i, val) in raw.iter().enumerate() {
            if i >= ops.len() { break; }
            let mut res = BatchOpResult { value: None, found: false, ok: false, exists: false, ttl_seconds: -2, error: None };
            match ops[i].op.as_str() {
                "get" => {
                    if let redis::Value::Data(b) = val {
                        res.value = String::from_utf8(b.clone()).ok();
                        res.found = true;
                    }
                }
                "set"    => res.ok = matches!(val, redis::Value::Okay | redis::Value::Status(_)),
                "del"    => res.ok = matches!(val, redis::Value::Int(n) if *n > 0),
                "exists" => res.exists = matches!(val, redis::Value::Int(n) if *n > 0),
                "ttl"    => res.ttl_seconds = if let redis::Value::Int(n) = val { *n } else { -2 },
                _ => {}
            }
            out.push(res);
        }
        Ok(out)
    }

    // ── Streaming cache ───────────────────────────────────────────────────────

    /// Cache a streaming response chunk-by-chunk via Redis `RPUSH`.
    /// Replay with [`stream_get`](CachlyClient::stream_get).
    pub async fn stream_set<I>(&self, key: &str, chunks: I, ttl: Option<u64>) -> Result<()>
    where
        I: IntoIterator<Item = String>,
    {
        let list_key = format!("cachly:stream:{key}");
        let mut conn = self.conn.clone();
        let _: i64 = conn.del(&list_key).await?;
        for chunk in chunks {
            let _: i64 = redis::cmd("RPUSH")
                .arg(&list_key)
                .arg(&chunk)
                .query_async::<_, i64>(&mut conn)
                .await?;
        }
        if let Some(secs) = ttl {
            let _: bool = conn.expire(&list_key, secs as i64).await?;
        }
        Ok(())
    }

    /// Retrieve a cached stream as `Vec<String>` (all chunks via `LRANGE`).
    /// Returns `Ok(None)` on cache miss.
    pub async fn stream_get(&self, key: &str) -> Result<Option<Vec<String>>> {
        let list_key = format!("cachly:stream:{key}");
        let mut conn = self.conn.clone();
        let len: i64 = conn.llen(&list_key).await?;
        if len == 0 {
            return Ok(None);
        }
        let chunks: Vec<String> = redis::cmd("LRANGE")
            .arg(&list_key)
            .arg(0i64)
            .arg(-1i64)
            .query_async(&mut conn)
            .await?;
        Ok(Some(chunks))
    }

    /// Return a [`SemanticCache`] backed by the given embedding function.
    /// Uses linear SCAN. For HNSW, use [`semantic_with_vector`].
    pub fn semantic<EF, EFut>(&self, embed_fn: EF) -> SemanticCache<EF>
    where
        EF: Fn(String) -> EFut,
        EFut: Future<Output = Result<Vec<f64>>>,
    {
        SemanticCache { conn: self.conn.clone(), embed_fn }
    }

    /// Return a [`SemanticCache`] with pgvector API support (HNSW O(log n)).
    ///
    /// Pass `vector_url = None` to disable API mode; the cache will use SCAN.
    /// Requires the `vector-api` feature to be enabled in `Cargo.toml`.
    pub fn semantic_with_vector<EF, EFut>(
        &self,
        embed_fn: EF,
        _vector_url: Option<String>,
    ) -> SemanticCache<EF>
    where
        EF: Fn(String) -> EFut,
        EFut: Future<Output = Result<Vec<f64>>>,
    {
        // vector_url is passed per-call via SemanticOptions.vector_url.
        // This method exists for API symmetry with other SDKs.
        SemanticCache { conn: self.conn.clone(), embed_fn }
    }

    // ── Tag-based invalidation ─────────────────────────────────────────────────

    /// Associate a cache key with tags. `POST /v1/cache/:token/tags`
    #[cfg(feature = "vector-api")]
    pub async fn set_tags(&self, key: &str, tags: Vec<&str>) -> Result<TagsResult> {
        let base = self.batch_url.as_deref()
            .ok_or_else(|| CachlyError::Http("batch_url required for set_tags".into()))?;
        let resp = http_request(reqwest::Method::POST, &format!("{base}/tags"),
            Some(serde_json::json!({ "key": key, "tags": tags }))).await?;
        Ok(serde_json::from_value(resp).map_err(|e| CachlyError::Serde(e))?)
    }

    /// Delete all cache keys associated with a tag. `POST /v1/cache/:token/invalidate`
    #[cfg(feature = "vector-api")]
    pub async fn invalidate_tag(&self, tag: &str) -> Result<InvalidateTagResult> {
        let base = self.batch_url.as_deref()
            .ok_or_else(|| CachlyError::Http("batch_url required for invalidate_tag".into()))?;
        let resp = http_request(reqwest::Method::POST, &format!("{base}/invalidate"),
            Some(serde_json::json!({ "tag": tag }))).await?;
        Ok(serde_json::from_value(resp).map_err(|e| CachlyError::Serde(e))?)
    }

    /// Get all tags for a cache key. `GET /v1/cache/:token/tags/:key`
    #[cfg(feature = "vector-api")]
    pub async fn get_tags(&self, key: &str) -> Result<TagsResult> {
        let base = self.batch_url.as_deref()
            .ok_or_else(|| CachlyError::Http("batch_url required for get_tags".into()))?;
        let resp = http_request(reqwest::Method::GET, &format!("{base}/tags/{key}"), None).await?;
        Ok(serde_json::from_value(resp).map_err(|e| CachlyError::Serde(e))?)
    }

    /// Remove all tag associations for a key. `DELETE /v1/cache/:token/tags/:key`
    #[cfg(feature = "vector-api")]
    pub async fn delete_tags(&self, key: &str) -> Result<serde_json::Value> {
        let base = self.batch_url.as_deref()
            .ok_or_else(|| CachlyError::Http("batch_url required for delete_tags".into()))?;
        http_request(reqwest::Method::DELETE, &format!("{base}/tags/{key}"), None).await
    }

    // ── Stale-While-Revalidate (SWR) ──────────────────────────────────────────

    /// Register a key for Stale-While-Revalidate. `POST /v1/cache/:token/swr/register`
    #[cfg(feature = "vector-api")]
    pub async fn swr_register(
        &self,
        key: &str,
        ttl_seconds: u64,
        stale_window_seconds: u64,
        fetcher_hint: Option<&str>,
    ) -> Result<serde_json::Value> {
        let base = self.batch_url.as_deref()
            .ok_or_else(|| CachlyError::Http("batch_url required for swr_register".into()))?;
        let mut body = serde_json::json!({
            "key": key, "ttl_seconds": ttl_seconds, "stale_window_seconds": stale_window_seconds,
        });
        if let Some(hint) = fetcher_hint {
            body["fetcher_hint"] = serde_json::Value::String(hint.to_string());
        }
        http_request(reqwest::Method::POST, &format!("{base}/swr/register"), Some(body)).await
    }

    /// Query the list of stale keys. `POST /v1/cache/:token/swr/check`
    #[cfg(feature = "vector-api")]
    pub async fn swr_check(&self) -> Result<SwrCheckResult> {
        let base = self.batch_url.as_deref()
            .ok_or_else(|| CachlyError::Http("batch_url required for swr_check".into()))?;
        let resp = http_request(reqwest::Method::POST, &format!("{base}/swr/check"),
            Some(serde_json::json!({}))).await?;
        Ok(serde_json::from_value(resp).map_err(|e| CachlyError::Serde(e))?)
    }

    /// Remove a key from the SWR registry. `DELETE /v1/cache/:token/swr/:key`
    #[cfg(feature = "vector-api")]
    pub async fn swr_remove(&self, key: &str) -> Result<serde_json::Value> {
        let base = self.batch_url.as_deref()
            .ok_or_else(|| CachlyError::Http("batch_url required for swr_remove".into()))?;
        http_request(reqwest::Method::DELETE, &format!("{base}/swr/{key}"), None).await
    }

    // ── Bulk Warmup ───────────────────────────────────────────────────────────

    /// Bulk-warm the KV cache via a server-side pipeline. `POST /v1/cache/:token/warm`
    #[cfg(feature = "vector-api")]
    pub async fn bulk_warmup(&self, entries: Vec<BulkWarmupEntry>) -> Result<BulkWarmupResult> {
        let base = self.batch_url.as_deref()
            .ok_or_else(|| CachlyError::Http("batch_url required for bulk_warmup".into()))?;
        let resp = http_request(reqwest::Method::POST, &format!("{base}/warm"),
            Some(serde_json::json!({ "entries": entries }))).await?;
        Ok(serde_json::from_value(resp).map_err(|e| CachlyError::Serde(e))?)
    }

    // ── LLM Proxy Stats ───────────────────────────────────────────────────────

    /// Return LLM proxy statistics. `GET /v1/llm-proxy/:token/stats`
    #[cfg(feature = "vector-api")]
    pub async fn llm_proxy_stats(&self) -> Result<LlmProxyStatsResult> {
        let base = self.llm_proxy_url.as_deref()
            .ok_or_else(|| CachlyError::Http("llm_proxy_url required for llm_proxy_stats".into()))?;
        let resp = http_request(reqwest::Method::GET, &format!("{base}/stats"), None).await?;
        Ok(serde_json::from_value(resp).unwrap_or_default())
    }

    // ── Sub-client accessors ──────────────────────────────────────────────────

    /// Return a `PubSubClient` if `pubsub_url` is configured.
    #[cfg(feature = "vector-api")]
    pub fn pub_sub(&self) -> Option<PubSubClient> {
        self.pubsub_url.as_deref().map(|url| PubSubClient { pubsub_url: url.to_owned() })
    }

    /// Return a `WorkflowClient` if `workflow_url` is configured.
    #[cfg(feature = "vector-api")]
    pub fn workflow(&self) -> Option<WorkflowClient> {
        self.workflow_url.as_deref().map(|url| WorkflowClient { workflow_url: url.to_owned() })
    }
}

// ── PubSubClient ──────────────────────────────────────────────────────────────

/// Pub/Sub client. Obtain via [`CachlyClient::pub_sub`].
#[cfg(feature = "vector-api")]
pub struct PubSubClient {
    pubsub_url: String,
}

#[cfg(feature = "vector-api")]
impl PubSubClient {
    /// Publish a message to a channel. `POST /v1/pubsub/:token/publish`
    pub async fn publish(&self, channel: &str, message: &str) -> Result<serde_json::Value> {
        http_request(
            reqwest::Method::POST,
            &format!("{}/publish", self.pubsub_url),
            Some(serde_json::json!({ "channel": channel, "message": message })),
        )
        .await
    }

    /// Subscribe to channels and collect SSE messages. `POST /v1/pubsub/:token/subscribe`
    ///
    /// Collects all messages until the stream closes. Returns a `Vec<PubSubMessage>`.
    pub async fn subscribe(&self, channels: &[&str]) -> Result<Vec<PubSubMessage>> {
        let client = reqwest::Client::builder()
            .timeout(std::time::Duration::from_secs(300))
            .build()
            .map_err(|e| CachlyError::Http(e.to_string()))?;
        let text = client
            .post(&format!("{}/subscribe", self.pubsub_url))
            .header("Accept", "text/event-stream")
            .json(&serde_json::json!({ "channels": channels }))
            .send()
            .await
            .map_err(|e| CachlyError::Http(e.to_string()))?
            .text()
            .await
            .map_err(|e| CachlyError::Http(e.to_string()))?;
        let mut messages = Vec::new();
        for line in text.lines() {
            if let Some(data) = line.strip_prefix("data:") {
                let data = data.trim();
                if data.is_empty() || data == "{}" { continue; }
                if let Ok(msg) = serde_json::from_str::<PubSubMessage>(data) {
                    messages.push(msg);
                }
            }
        }
        Ok(messages)
    }

    /// List active channels. `GET /v1/pubsub/:token/channels`
    pub async fn channels(&self) -> Result<serde_json::Value> {
        http_request(reqwest::Method::GET, &format!("{}/channels", self.pubsub_url), None).await
    }

    /// Pub/Sub statistics. `GET /v1/pubsub/:token/stats`
    pub async fn stats(&self) -> Result<serde_json::Value> {
        http_request(reqwest::Method::GET, &format!("{}/stats", self.pubsub_url), None).await
    }
}

// ── WorkflowClient ────────────────────────────────────────────────────────────

/// Workflow checkpoint client. Obtain via [`CachlyClient::workflow`].
#[cfg(feature = "vector-api")]
pub struct WorkflowClient {
    workflow_url: String,
}

#[cfg(feature = "vector-api")]
impl WorkflowClient {
    /// Save a workflow checkpoint. `POST /v1/workflow/:token/checkpoints`
    pub async fn save_checkpoint(
        &self,
        run_id: &str,
        step_index: i32,
        step_name: &str,
        agent_name: &str,
        status: &str,
        state: Option<&str>,
        output: Option<&str>,
        duration_ms: Option<u64>,
    ) -> Result<WorkflowCheckpoint> {
        let mut body = serde_json::json!({
            "run_id": run_id, "step_index": step_index, "step_name": step_name,
            "agent_name": agent_name, "status": status,
        });
        if let Some(s) = state   { body["state"] = s.into(); }
        if let Some(o) = output  { body["output"] = o.into(); }
        if let Some(d) = duration_ms { body["duration_ms"] = d.into(); }
        let resp = http_request(reqwest::Method::POST,
            &format!("{}/checkpoints", self.workflow_url), Some(body)).await?;
        Ok(serde_json::from_value(resp).unwrap_or_default())
    }

    /// List all workflow runs. `GET /v1/workflow/:token/runs`
    pub async fn list_runs(&self) -> Result<Vec<WorkflowRun>> {
        let resp = http_request(reqwest::Method::GET,
            &format!("{}/runs", self.workflow_url), None).await?;
        Ok(resp.get("runs")
            .and_then(|v| serde_json::from_value(v.clone()).ok())
            .unwrap_or_default())
    }

    /// Get checkpoints for a run. `GET /v1/workflow/:token/runs/:runId`
    pub async fn get_run(&self, run_id: &str) -> Result<WorkflowRun> {
        let resp = http_request(reqwest::Method::GET,
            &format!("{}/runs/{run_id}", self.workflow_url), None).await?;
        Ok(serde_json::from_value(resp).map_err(|e| CachlyError::Serde(e))?)
    }

    /// Get the latest checkpoint. `GET /v1/workflow/:token/runs/:runId/latest`
    pub async fn latest_checkpoint(&self, run_id: &str) -> Result<WorkflowCheckpoint> {
        let resp = http_request(reqwest::Method::GET,
            &format!("{}/runs/{run_id}/latest", self.workflow_url), None).await?;
        Ok(serde_json::from_value(resp).unwrap_or_default())
    }

    /// Delete a workflow run. `DELETE /v1/workflow/:token/runs/:runId`
    pub async fn delete_run(&self, run_id: &str) -> Result<serde_json::Value> {
        http_request(reqwest::Method::DELETE,
            &format!("{}/runs/{run_id}", self.workflow_url), None).await
    }
}

// ── Helpers ───────────────────────────────────────────────────────────────────

fn cosine_similarity(a: &[f64], b: &[f64]) -> f64 {
    let len = a.len().min(b.len());
    let (mut dot, mut norm_a, mut norm_b) = (0.0_f64, 0.0_f64, 0.0_f64);
    for i in 0..len {
        dot += a[i] * b[i];
        norm_a += a[i] * a[i];
        norm_b += b[i] * b[i];
    }
    let denom = norm_a.sqrt() * norm_b.sqrt();
    if denom == 0.0 { 0.0 } else { dot / denom }
}

/// Format an RFC3339 expiry timestamp `ttl` seconds from now.
#[cfg(feature = "vector-api")]
fn chrono_expires_at(ttl_secs: u64) -> String {
    use std::time::{SystemTime, UNIX_EPOCH};
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs();
    let exp = now + ttl_secs;
    // Format as RFC3339: YYYY-MM-DDTHH:MM:SSZ (seconds precision, UTC)
    let secs = exp % 60;
    let min  = (exp / 60) % 60;
    let hr   = (exp / 3600) % 24;
    let days = exp / 86400;
    // Approximate date arithmetic – good enough for expiry hints.
    let epoch_day = days as i64;
    let (y, mo, d) = epoch_day_to_ymd(epoch_day);
    format!("{y:04}-{mo:02}-{d:02}T{hr:02}:{min:02}:{secs:02}Z")
}

#[cfg(feature = "vector-api")]
fn epoch_day_to_ymd(mut z: i64) -> (i32, u32, u32) {
    z += 719_468;
    let era = z.div_euclid(146_097);
    let doe = z.rem_euclid(146_097);
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp  = (5 * doy + 2) / 153;
    let d   = doy - (153 * mp + 2) / 5 + 1;
    let m   = if mp < 10 { mp + 3 } else { mp - 9 };
    let y   = if m <= 2 { y + 1 } else { y };
    (y as i32, m as u32, d as u32)
}

// ── Edge Cache (Feature #5) ──────────────────────────────────────────────────

/// Edge cache configuration for an instance.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct EdgeCacheConfig {
    pub id: String,
    pub instance_id: String,
    pub enabled: bool,
    pub edge_ttl: i32,
    pub worker_url: String,
    pub cloudflare_zone_id: String,
    pub purge_on_write: bool,
    pub cache_search_results: bool,
    pub total_hits: i64,
    pub total_misses: i64,
    pub hit_rate: f64,
}

/// Fields to update in edge cache configuration.
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct EdgeCacheConfigUpdate {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub enabled: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub edge_ttl: Option<i32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub worker_url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cloudflare_zone_id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub purge_on_write: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cache_search_results: Option<bool>,
}

/// Specifies what to purge from the edge cache.
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct EdgePurgeOptions {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub namespaces: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub urls: Option<Vec<String>>,
}

/// Result of an edge cache purge operation.
#[derive(Debug, Clone, serde::Deserialize)]
pub struct EdgePurgeResult {
    pub purged: i32,
    pub urls: Vec<String>,
}

/// Edge cache hit/miss statistics.
#[derive(Debug, Clone, serde::Deserialize)]
pub struct EdgeCacheStats {
    pub enabled: bool,
    pub worker_url: String,
    pub edge_ttl: i32,
    pub total_hits: i64,
    pub total_misses: i64,
    pub hit_rate: f64,
}

/// Client for managing Cloudflare Edge Cache.
/// Obtain via [`CachlyClient::edge`].
#[cfg(feature = "vector-api")]
pub struct EdgeCacheClient {
    base_url: String,
}

#[cfg(feature = "vector-api")]
impl EdgeCacheClient {
    /// Get the current edge cache configuration.
    pub async fn get_config(&self) -> Result<EdgeCacheConfig> {
        let resp = http_request(
            reqwest::Method::GET,
            &format!("{}/config", self.base_url),
            None,
        ).await?;
        Ok(serde_json::from_value(resp).map_err(CachlyError::Serde)?)
    }

    /// Update the edge cache configuration.
    pub async fn set_config(&self, update: EdgeCacheConfigUpdate) -> Result<EdgeCacheConfig> {
        let resp = http_request(
            reqwest::Method::PUT,
            &format!("{}/config", self.base_url),
            Some(serde_json::to_value(&update).map_err(CachlyError::Serde)?),
        ).await?;
        Ok(serde_json::from_value(resp).map_err(CachlyError::Serde)?)
    }

    /// Disable and remove the edge cache configuration.
    pub async fn delete_config(&self) -> Result<()> {
        http_request(
            reqwest::Method::DELETE,
            &format!("{}/config", self.base_url),
            None,
        ).await?;
        Ok(())
    }

    /// Purge cached entries from the Cloudflare CDN.
    /// Pass `None` to purge all entries, or specify namespaces/URLs to purge selectively.
    pub async fn purge(&self, options: Option<EdgePurgeOptions>) -> Result<EdgePurgeResult> {
        let body = options.map(|o| serde_json::to_value(&o).ok()).flatten();
        let resp = http_request(
            reqwest::Method::POST,
            &format!("{}/purge", self.base_url),
            body,
        ).await?;
        Ok(serde_json::from_value(resp).map_err(CachlyError::Serde)?)
    }

    /// Get edge cache hit/miss statistics.
    pub async fn stats(&self) -> Result<EdgeCacheStats> {
        let resp = http_request(
            reqwest::Method::GET,
            &format!("{}/stats", self.base_url),
            None,
        ).await?;
        Ok(serde_json::from_value(resp).map_err(CachlyError::Serde)?)
    }
}

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

#[cfg(test)]
mod tests {
    use super::{
        cosine_similarity, normalize_prompt, detect_namespace, confidence_band,
        quantize_embedding, val_key, extract_id, extract_namespace,
        SemanticConfidence,
    };

    // ── cosine_similarity ─────────────────────────────────────────────────────

    #[test]
    fn test_cosine_identical() {
        let v = vec![1.0, 2.0, 3.0];
        assert!((cosine_similarity(&v, &v) - 1.0).abs() < 1e-9);
    }

    #[test]
    fn test_cosine_orthogonal() {
        let a = vec![1.0, 0.0];
        let b = vec![0.0, 1.0];
        assert!((cosine_similarity(&a, &b)).abs() < 1e-9);
    }

    #[test]
    fn test_cosine_zero_vector() {
        let a = vec![0.0, 0.0];
        let b = vec![1.0, 2.0];
        assert_eq!(cosine_similarity(&a, &b), 0.0);
    }

    #[test]
    fn test_cosine_opposite_vectors() {
        let a = vec![1.0, 0.0];
        let b = vec![-1.0, 0.0];
        assert!((cosine_similarity(&a, &b) - (-1.0)).abs() < 1e-9);
    }

    #[test]
    fn test_cosine_similar_vectors_high_score() {
        let a = vec![0.9, 0.1, 0.0];
        let b = vec![0.85, 0.15, 0.0];
        assert!(cosine_similarity(&a, &b) > 0.99);
    }

    // ── key helpers ───────────────────────────────────────────────────────────

    #[test]
    fn test_val_key() {
        assert_eq!(
            val_key("cachly:sem:emb:abc-123"),
            "cachly:sem:val:abc-123"
        );
    }

    #[test]
    fn test_extract_id() {
        assert_eq!(extract_id("cachly:sem:emb:abc-123"), "abc-123");
    }

    #[test]
    fn test_extract_namespace() {
        assert_eq!(extract_namespace("cachly:sem:emb:abc-123"), "cachly:sem");
    }

    // ── normalize_prompt ──────────────────────────────────────────────────────

    #[test]
    fn test_normalize_removes_filler_please() {
        let result = normalize_prompt("please tell me about Redis", None);
        assert!(!result.contains("please"), "filler 'please' should be removed, got: {result}");
        assert!(result.contains("tell me about redis"));
    }

    #[test]
    fn test_normalize_removes_filler_bitte() {
        let result = normalize_prompt("bitte erkläre mir Redis", None);
        assert!(!result.contains("bitte"), "filler 'bitte' should be removed, got: {result}");
    }

    #[test]
    fn test_normalize_lowercases() {
        let result = normalize_prompt("WHAT IS REDIS?", None);
        assert_eq!(result, "what is redis?");
    }

    #[test]
    fn test_normalize_collapses_whitespace() {
        let result = normalize_prompt("what   is   redis?", None);
        assert!(!result.contains("  "), "extra spaces should collapse, got: {result}");
    }

    #[test]
    fn test_normalize_trailing_exclamation_becomes_question() {
        let result = normalize_prompt("what is redis!", None);
        assert!(result.ends_with('?'), "trailing ! should become ?, got: {result}");
    }

    #[test]
    fn test_normalize_empty_string() {
        assert_eq!(normalize_prompt("", None), "");
    }

    #[test]
    fn test_normalize_custom_filler_words() {
        let fillers: &[&str] = &["yo"];
        let result = normalize_prompt("yo what is redis?", Some(fillers));
        assert!(!result.starts_with("yo"), "custom filler 'yo' should be removed, got: {result}");
    }

    // ── §4 detect_namespace ───────────────────────────────────────────────────

    #[test]
    fn test_namespace_code_python_def() {
        assert_eq!(detect_namespace("def process(data): return data * 2"), "cachly:sem:code");
    }

    #[test]
    fn test_namespace_code_const() {
        assert_eq!(detect_namespace("const x = () => 42"), "cachly:sem:code");
    }

    #[test]
    fn test_namespace_code_class() {
        assert_eq!(detect_namespace("class MyService { constructor() {} }"), "cachly:sem:code");
    }

    #[test]
    fn test_namespace_code_import() {
        assert_eq!(detect_namespace("import React from \"react\""), "cachly:sem:code");
    }

    #[test]
    fn test_namespace_code_shebang() {
        assert_eq!(detect_namespace("#!/usr/bin/env python3"), "cachly:sem:code");
    }

    #[test]
    fn test_namespace_code_go_func() {
        assert_eq!(detect_namespace("func main() { fmt.Println(\"hi\") }"), "cachly:sem:code");
    }

    #[test]
    fn test_namespace_translation_translate() {
        assert_eq!(detect_namespace("translate this text to english"), "cachly:sem:translation");
    }

    #[test]
    fn test_namespace_translation_uebersetze() {
        assert_eq!(detect_namespace("übersetze diesen Text auf englisch"), "cachly:sem:translation");
    }

    #[test]
    fn test_namespace_translation_traduis() {
        assert_eq!(detect_namespace("traduis ce texte en anglais"), "cachly:sem:translation");
    }

    #[test]
    fn test_namespace_summary_summarize() {
        assert_eq!(detect_namespace("summarize this article for me"), "cachly:sem:summary");
    }

    #[test]
    fn test_namespace_summary_tldr() {
        assert_eq!(detect_namespace("tl;dr of this document?"), "cachly:sem:summary");
    }

    #[test]
    fn test_namespace_summary_key_points() {
        assert_eq!(detect_namespace("give me the key points of this paper"), "cachly:sem:summary");
    }

    #[test]
    fn test_namespace_summary_nutshell() {
        assert_eq!(detect_namespace("explain in a nutshell"), "cachly:sem:summary");
    }

    #[test]
    fn test_namespace_qa_what_is() {
        assert_eq!(detect_namespace("what is the capital of France?"), "cachly:sem:qa");
    }

    #[test]
    fn test_namespace_qa_how_does() {
        assert_eq!(detect_namespace("how does photosynthesis work?"), "cachly:sem:qa");
    }

    #[test]
    fn test_namespace_qa_trailing_question_mark() {
        assert_eq!(detect_namespace("Is Redis faster than Memcached?"), "cachly:sem:qa");
    }

    #[test]
    fn test_namespace_qa_german_wer() {
        assert_eq!(detect_namespace("wer ist der aktuelle Bundeskanzler?"), "cachly:sem:qa");
    }

    #[test]
    fn test_namespace_creative_poem() {
        assert_eq!(detect_namespace("Write a poem about autumn"), "cachly:sem:creative");
    }

    #[test]
    fn test_namespace_creative_story() {
        assert_eq!(detect_namespace("Tell me a story about dragons"), "cachly:sem:creative");
    }

    #[test]
    fn test_namespace_code_priority_over_translation() {
        assert_eq!(
            detect_namespace("translate this function def foo(): pass"),
            "cachly:sem:code"
        );
    }

    #[test]
    fn test_namespace_case_insensitive() {
        assert_eq!(detect_namespace("CONST X = 1"), "cachly:sem:code");
        assert_eq!(detect_namespace("TRANSLATE TO GERMAN"), "cachly:sem:translation");
        assert_eq!(detect_namespace("SUMMARIZE THIS"), "cachly:sem:summary");
        assert_eq!(detect_namespace("WHAT IS GRAVITY?"), "cachly:sem:qa");
    }

    #[test]
    fn test_namespace_trims_whitespace() {
        assert_eq!(detect_namespace("   what is dark matter?   "), "cachly:sem:qa");
        assert_eq!(detect_namespace("  const x = 1  "), "cachly:sem:code");
    }

    // ── confidence_band ───────────────────────────────────────────────────────

    #[test]
    fn test_confidence_high() {
        assert_eq!(confidence_band(0.98, 0.85, 0.97), SemanticConfidence::High);
    }

    #[test]
    fn test_confidence_medium() {
        assert_eq!(confidence_band(0.90, 0.85, 0.97), SemanticConfidence::Medium);
    }

    #[test]
    fn test_confidence_uncertain() {
        assert_eq!(confidence_band(0.80, 0.85, 0.97), SemanticConfidence::Uncertain);
    }

    #[test]
    fn test_confidence_exactly_at_high_threshold() {
        assert_eq!(confidence_band(0.97, 0.85, 0.97), SemanticConfidence::High);
    }

    #[test]
    fn test_confidence_exactly_at_threshold() {
        assert_eq!(confidence_band(0.85, 0.85, 0.97), SemanticConfidence::Medium);
    }

    // ── §7 quantize_embedding ─────────────────────────────────────────────────

    #[test]
    fn test_quantize_empty_vector() {
        let q = quantize_embedding(&[]);
        assert!(q.values.is_empty());
        assert_eq!(q.min, 0.0);
        assert_eq!(q.max, 0.0);
    }

    #[test]
    fn test_quantize_constant_vector_all_zeros() {
        let q = quantize_embedding(&[0.5, 0.5, 0.5]);
        assert!(q.values.iter().all(|&v| v == 0));
        assert!((q.min - 0.5).abs() < 1e-9);
        assert!((q.max - 0.5).abs() < 1e-9);
    }

    #[test]
    fn test_quantize_min_is_neg128_max_is_127() {
        let q = quantize_embedding(&[0.0, 1.0]);
        assert_eq!(q.values[0], -128);
        assert_eq!(q.values[1], 127);
    }

    #[test]
    fn test_quantize_all_values_in_int8_range() {
        let vec: Vec<f64> = (0..16).map(|i| (i as f64).sin()).collect();
        let q = quantize_embedding(&vec);
        for v in &q.values {
            assert!(*v >= -128 && *v <= 127, "value {v} is outside int8 range");
        }
    }

    #[test]
    fn test_quantize_preserves_min_max() {
        let q = quantize_embedding(&[0.1, 0.4, 0.9]);
        assert!((q.min - 0.1).abs() < 1e-9);
        assert!((q.max - 0.9).abs() < 1e-9);
    }
}
