//! Thin async HTTP client over the tidalDB surface, plus the corpus seeder.
//!
//! The client returns a [`StatusClass`] per request and nothing else — latency is
//! measured by the caller from the request's *intended* send time so the figure
//! is coordinated-omission-corrected (a slow server inflates the number it would
//! otherwise hide). Response bodies are drained so the keep-alive connection is
//! reused (and so the measured time includes full response transfer).

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

use rand::Rng;
use reqwest::header::{AUTHORIZATION, HeaderValue};
use serde::{Deserialize, Serialize};
use tokio::sync::Semaphore;

use crate::error::{Result, StressError};
use crate::metrics::StatusClass;
use crate::workload::{Body, HttpMethod, ItemMetadata, Plan};

/// `POST /vector_search` request body for the recall probe — serialized straight
/// to the wire (borrows the query vector; no owned copy).
#[derive(Serialize)]
struct VectorQuery<'a> {
    vector: &'a [f32],
    k: usize,
    #[serde(skip_serializing_if = "Option::is_none")]
    ef_search: Option<usize>,
}

/// A `POST /signals` body carrying `user_id`, used to BUILD a user's preference
/// vector (a positive-engagement `like` on an embedded item). The standalone
/// surface reads `user_id`; the regular load `Body::Signal` omits it.
#[derive(Serialize)]
struct PreferenceLike {
    entity_id: u64,
    signal: &'static str,
    weight: f64,
    user_id: u64,
}

/// The parts of the `/vector_search` response the recall oracle reads: the ranked
/// entity IDs (distances are ignored — recall@k is a set-overlap metric).
#[derive(Deserialize)]
struct VectorSearchResponseBody {
    items: Vec<VectorSearchResponseItem>,
}

#[derive(Deserialize)]
struct VectorSearchResponseItem {
    entity_id: u64,
}

/// Async client with optional bearer auth. Cheap to clone (Arc inside reqwest).
#[derive(Clone)]
pub struct HttpClient {
    inner: reqwest::Client,
    /// The `Authorization: Bearer <key>` header value, built ONCE at construction
    /// (the token never changes) so the per-request path is a cheap refcounted
    /// clone instead of `format!` + header re-validation on every send. `None` when
    /// no key is configured. Marked sensitive so it is redacted from any debug output.
    auth_header: Option<HeaderValue>,
    /// `x-tidal-ack` value sent with every write (m11p3: `leader`/`quorum`);
    /// `None` = the deployment's topology default.
    ack: Option<String>,
}

impl HttpClient {
    /// `request_timeout` should sit ABOVE the server's 30s request timeout so the
    /// server's own 408 surfaces as a Timeout class rather than a client abort.
    ///
    /// # Errors
    ///
    /// Returns [`StressError::Client`] if the underlying reqwest client cannot be
    /// built (TLS/builder fault), and [`StressError::Auth`] if `api_key` contains
    /// characters that are not valid in an HTTP `Authorization` header value.
    pub fn new(
        request_timeout: Duration,
        api_key: Option<String>,
        ack: Option<String>,
        ca_cert: Option<String>,
        insecure: bool,
    ) -> Result<Self> {
        let mut builder = reqwest::Client::builder()
            .timeout(request_timeout)
            .connect_timeout(Duration::from_secs(5))
            // Warm-connection pool sized to comfortably saturate the server's
            // 100-concurrency limit per host without exhausting container fds at
            // 4 host-pools (≈4×256 idle ceiling).
            .pool_max_idle_per_host(256)
            .pool_idle_timeout(Duration::from_secs(90))
            .tcp_nodelay(true);
        // m11p7: the cluster's :9500 plane serves TLS from a private CA. Trust it
        // via the mounted ca.crt (verified), or skip verification with --insecure.
        if let Some(path) = ca_cert {
            let pem = std::fs::read(&path).map_err(|source| StressError::CaCert {
                path: path.clone(),
                source,
            })?;
            let cert = reqwest::Certificate::from_pem(&pem).map_err(StressError::Client)?;
            builder = builder.add_root_certificate(cert);
        }
        if insecure {
            builder = builder.danger_accept_invalid_certs(true);
        }
        let inner = builder.build().map_err(StressError::Client)?;
        // Build the Authorization header once. reqwest's `bearer_auth` does a fresh
        // `format!("Bearer {token}")` allocation plus a HeaderValue validation pass
        // on EVERY call, over a token that never changes — pure per-request waste.
        let auth_header = match api_key {
            Some(k) => {
                let mut v = HeaderValue::from_str(&format!("Bearer {k}"))
                    .map_err(|_| StressError::Auth("contains invalid header characters".into()))?;
                v.set_sensitive(true);
                Some(v)
            }
            None => None,
        };
        Ok(Self {
            inner,
            auth_header,
            ack,
        })
    }

    fn apply_auth(&self, rb: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
        // A HeaderValue clone is a cheap refcount bump (or inline copy), not a
        // realloc+revalidate. `cluster_status` deliberately does NOT call this, so
        // its status poll stays unauthenticated as before.
        match &self.auth_header {
            Some(v) => rb.header(AUTHORIZATION, v.clone()),
            None => rb,
        }
    }

    /// Issue one request and classify the result. Never returns an error — a
    /// transport fault is a [`StatusClass::Transport`] outcome, not a fatal.
    pub async fn execute(&self, plan: &Plan) -> StatusClass {
        let rb = match plan.method {
            HttpMethod::Get => self.inner.get(&plan.url),
            HttpMethod::Post => {
                let mut rb = self.inner.post(&plan.url);
                if let Some(ack) = &self.ack {
                    rb = rb.header("x-tidal-ack", ack);
                }
                match &plan.body {
                    Some(b) => rb.json(b),
                    None => rb,
                }
            }
        };
        match self.apply_auth(rb).send().await {
            Ok(resp) => {
                let class = StatusClass::from_status(resp.status().as_u16());
                // Drain the body so the connection returns to the pool.
                let _ = resp.bytes().await;
                class
            }
            Err(_) => StatusClass::Transport,
        }
    }

    /// Issue a `POST {base}/vector_search` recall probe (m12p1) and return both
    /// the capacity [`StatusClass`] and, on a 2xx, the returned entity IDs in
    /// rank order (closest-first). The IDs are what the recall oracle compares
    /// against its brute-force ground truth; a non-2xx or a body it cannot parse
    /// yields `None` for the IDs (the status still classifies the outcome).
    ///
    /// Latency is measured by the caller around this call (from the request's
    /// intended send time), preserving the coordinated-omission correction.
    pub async fn vector_search_ids(
        &self,
        base: &str,
        vector: &[f32],
        k: usize,
        ef_search: Option<usize>,
    ) -> (StatusClass, Option<Vec<u64>>) {
        let url = format!("{base}/vector_search");
        let body = VectorQuery {
            vector,
            k,
            ef_search,
        };
        let rb = self.inner.post(&url).json(&body);
        match self.apply_auth(rb).send().await {
            Ok(resp) => {
                let class = StatusClass::from_status(resp.status().as_u16());
                if class != StatusClass::Ok {
                    // Drain so the connection returns to the pool.
                    let _ = resp.bytes().await;
                    return (class, None);
                }
                // The recall oracle needs the ranked IDs; parse them out. A parse
                // failure on a 2xx is reported as Ok-without-ids (a recall miss is
                // not a transport error) so the latency sample is still honest.
                let ids = resp
                    .json::<VectorSearchResponseBody>()
                    .await
                    .ok()
                    .map(|b| b.items.into_iter().map(|i| i.entity_id).collect());
                (class, ids)
            }
            Err(_) => (StatusClass::Transport, None),
        }
    }

    /// Fetch `GET {base}/cluster/status` (unauthenticated) and return the leader
    /// name and the worst replication lag across regions — used to watch whether
    /// follower lag grows under write load between ramp stages. `None` on any fault.
    pub async fn cluster_status(&self, base: &str) -> Option<(String, i64)> {
        let url = format!("{base}/cluster/status");
        let resp = self.inner.get(&url).send().await.ok()?;
        let v: serde_json::Value = resp.json().await.ok()?;
        let leader = v.get("leader")?.as_str()?.to_string();
        let max_lag = v
            .get("regions")?
            .as_array()?
            .iter()
            .filter_map(|r| r.get("lag_events").and_then(serde_json::Value::as_i64))
            .max()
            .unwrap_or(0);
        Some((leader, max_lag))
    }

    /// POST a preference-building `like` (with `user_id`), retrying 429 like the
    /// corpus seeder. Returns whether it ultimately succeeded.
    async fn post_preference(&self, url: &str, body: &PreferenceLike) -> bool {
        const MAX_ATTEMPTS: u32 = 8;
        for attempt in 0..MAX_ATTEMPTS {
            let rb = self.inner.post(url).json(body);
            match self.apply_auth(rb).send().await {
                Ok(resp) => {
                    let code = resp.status().as_u16();
                    let _ = resp.bytes().await;
                    if (200..=299).contains(&code) {
                        return true;
                    }
                    if code == 429 {
                        tokio::time::sleep(Duration::from_millis(40 + u64::from(attempt) * 20))
                            .await;
                        continue;
                    }
                    if code == 503 {
                        tokio::time::sleep(Duration::from_millis(100)).await;
                        continue;
                    }
                    return false;
                }
                Err(_) => return false,
            }
        }
        false
    }

    /// One-off POST returning the raw status, for the seeder (which needs to know
    /// success vs 429-retry rather than a capacity class).
    async fn post_json(&self, url: &str, body: &Body) -> Option<u16> {
        let rb = self.inner.post(url).json(body);
        match self.apply_auth(rb).send().await {
            Ok(resp) => {
                let code = resp.status().as_u16();
                let _ = resp.bytes().await;
                Some(code)
            }
            Err(_) => None,
        }
    }
}

/// Register `count` items (id 1..=count) plus a `dim`-wide content embedding for
/// each, against `base`, using RANDOM embedding values.
///
/// Use the LEADER url so items broadcast to every region and `/feed` on any
/// region can rank them. Bounded-concurrency, retries 429. Returns the number of
/// items confirmed registered.
///
/// The ramp mode seeds random embeddings because their exact values are
/// irrelevant to the write-path cost it measures. The recall mode instead calls
/// [`seed_corpus_with`] with a DETERMINISTIC id-keyed generator so the
/// generator's brute-force ground truth matches what the engine indexed.
///
/// # Errors
///
/// Returns [`StressError::Seed`] if the bounded-concurrency semaphore is closed
/// while acquiring a permit (the run cannot proceed).
pub async fn seed_corpus(
    client: &HttpClient,
    base: &str,
    count: u64,
    dim: usize,
    concurrency: usize,
) -> Result<u64> {
    seed_corpus_with(client, base, count, concurrency, move |_id| {
        // A fresh ThreadRng per call (it is !Send, so it cannot cross the await
        // inside the task; building the Vec up front keeps it off the await path).
        let mut rng = rand::rng();
        (0..dim).map(|_| rng.random::<f32>() - 0.5).collect()
    })
    .await
}

/// Register `count` items (id 1..=count) plus a content embedding for each, where
/// each item's embedding values come from `embedding(id)`.
///
/// This is the seam the recall harness uses: passing a deterministic id-keyed
/// generator makes the seeded corpus exactly reproducible, so the in-RAM
/// brute-force ground truth the oracle computes is bit-for-bit the corpus the
/// engine indexed. Same bounded-concurrency + 429-retry contract as
/// [`seed_corpus`].
///
/// # Errors
///
/// Returns [`StressError::Seed`] if the bounded-concurrency semaphore is closed
/// while acquiring a permit.
pub async fn seed_corpus_with<F>(
    client: &HttpClient,
    base: &str,
    count: u64,
    concurrency: usize,
    embedding: F,
) -> Result<u64>
where
    F: Fn(u64) -> Vec<f32> + Send + Sync + Clone + 'static,
{
    let sem = Arc::new(Semaphore::new(concurrency.max(1)));
    let done = Arc::new(AtomicU64::new(0));
    let items_url = format!("{base}/items");
    let emb_url = format!("{base}/embeddings");
    let mut handles = Vec::new();

    for id in 1..=count {
        let permit = sem
            .clone()
            .acquire_owned()
            .await
            .map_err(|e| StressError::Seed(format!("semaphore closed: {e}")))?;
        let client = client.clone();
        let items_url = items_url.clone();
        let emb_url = emb_url.clone();
        let done = done.clone();
        let embedding = embedding.clone();
        handles.push(tokio::spawn(async move {
            let _permit = permit;
            let category = SEED_CATEGORIES[(id as usize) % SEED_CATEGORIES.len()];
            let item = Body::Item {
                entity_id: id,
                metadata: ItemMetadata {
                    title: format!("post-{id}"),
                    category,
                },
            };
            let emb = Body::Embedding {
                entity_id: id,
                values: embedding(id),
            };

            // The item and embedding writes are independent — the engine keys the
            // embedding purely on entity id with no item-before-embedding
            // precondition (verified: db/items.rs write_item_embedding) — so issue
            // them concurrently rather than serially, roughly halving each id's
            // seed wall-clock. `done` still requires BOTH to succeed.
            let (item_ok, emb_ok) = tokio::join!(
                retry_write(&client, &items_url, &item),
                retry_write(&client, &emb_url, &emb),
            );
            if item_ok && emb_ok {
                done.fetch_add(1, Ordering::Relaxed);
            }
        }));
    }
    for h in handles {
        let _ = h.await;
    }
    Ok(done.load(Ordering::Relaxed))
}

/// Build a preference vector for each user `1..=users` (m12p2).
///
/// Each user sends one positive `like` signal (carrying `user_id`) on a
/// deterministic embedded item, so the `for_you` profile's ANN candidate
/// generation engages for those users instead of degrading to a scan.
///
/// Each user `u` likes item `((u - 1) % corpus) + 1`, whose embedding becomes (a
/// scaled copy of) the user's preference vector. Bounded concurrency; retries
/// 429. Returns the number of preference likes confirmed.
///
/// # Errors
///
/// Returns [`StressError::Seed`] if the bounded-concurrency semaphore is closed.
pub async fn seed_preferences(
    client: &HttpClient,
    base: &str,
    users: u64,
    corpus: u64,
    concurrency: usize,
) -> Result<u64> {
    let sem = Arc::new(Semaphore::new(concurrency.max(1)));
    let done = Arc::new(AtomicU64::new(0));
    let url = format!("{base}/signals");
    let corpus = corpus.max(1);
    let mut handles = Vec::new();

    for u in 1..=users {
        let permit = sem
            .clone()
            .acquire_owned()
            .await
            .map_err(|e| StressError::Seed(format!("semaphore closed: {e}")))?;
        let client = client.clone();
        let url = url.clone();
        let done = done.clone();
        handles.push(tokio::spawn(async move {
            let _permit = permit;
            let item = ((u - 1) % corpus) + 1;
            let body = PreferenceLike {
                entity_id: item,
                signal: "like",
                weight: 1.0,
                user_id: u,
            };
            if client.post_preference(&url, &body).await {
                done.fetch_add(1, Ordering::Relaxed);
            }
        }));
    }
    for h in handles {
        let _ = h.await;
    }
    Ok(done.load(Ordering::Relaxed))
}

const SEED_CATEGORIES: [&str; 12] = [
    "anime",
    "gaming",
    "fitness",
    "music",
    "cosplay",
    "art",
    "fantasy",
    "scifi",
    "romance",
    "comedy",
    "horror",
    "slice-of-life",
];

/// POST with backpressure-aware retry: a 2xx is success; a 429 backs off and
/// retries (the seed must not give up just because the write pool is busy); any
/// other code / transport fault fails fast (a usage bug, not transient load).
async fn retry_write(client: &HttpClient, url: &str, body: &Body) -> bool {
    // A bulk SEED must persist through sustained write-pool saturation: when one
    // leader owns several shards, a fast multi-writer seed keeps its bounded
    // admission pool full and it sheds 429 for seconds at a stretch. The old
    // 8-attempt / ~900ms budget gave up inside that window and silently DROPPED
    // ~0.4% of the corpus, failing the recall harness's exact-count gate. 429 is
    // backpressure, not failure — retry it persistently (exponential backoff
    // capped at 500ms → ~12s total budget, comfortably past a saturation burst).
    const MAX_ATTEMPTS: u32 = 40;
    for attempt in 0..MAX_ATTEMPTS {
        match client.post_json(url, body).await {
            Some(code) if (200..=299).contains(&code) => return true,
            Some(429) => {
                let backoff = std::cmp::min(40 + u64::from(attempt) * 40, 500);
                tokio::time::sleep(Duration::from_millis(backoff)).await;
            }
            Some(503) => {
                // Leader briefly unreachable (e.g. a roll) — short wait, retry.
                tokio::time::sleep(Duration::from_millis(150)).await;
            }
            _ => return false,
        }
    }
    false
}