use std::sync::Arc;
use std::time::Duration;

use reqwest::header::{HeaderMap, HeaderValue, AUTHORIZATION, CONTENT_TYPE};
use reqwest::Url;
use tokio::task::JoinHandle;
use tracing::debug;
use uuid::Uuid;

use crate::buffer::{BufferHandle, BufferWorker, FlushFailureHandler};
use crate::config::{BufferConfig, ClientConfig, RetryConfig};
use crate::error::MonkError;
use crate::models::*;
use crate::retry::RetryPolicy;
use crate::validation;

// ── Builder ─────────────────────────────────────────────────────────────────

/// Fluent builder for constructing a [`MonkClient`].
pub struct MonkClientBuilder {
    base_url: Option<String>,
    api_key: Option<String>,
    timeout: Duration,
    retry: RetryConfig,
    buffer: BufferConfig,
    on_flush_failure: Option<FlushFailureHandler>,
}

impl Default for MonkClientBuilder {
    fn default() -> Self {
        Self {
            base_url: None,
            api_key: None,
            timeout: Duration::from_secs(30),
            retry: RetryConfig::default(),
            buffer: BufferConfig::default(),
            on_flush_failure: None,
        }
    }
}

impl MonkClientBuilder {
    /// Base URL of the Monk Events API (e.g. `https://events.monk.dev`).
    pub fn base_url(mut self, url: impl Into<String>) -> Self {
        self.base_url = Some(url.into());
        self
    }

    /// Bearer API key for authentication.
    pub fn api_key(mut self, key: impl Into<String>) -> Self {
        self.api_key = Some(key.into());
        self
    }

    /// Per-request HTTP timeout (default 30 s).
    pub fn timeout(mut self, dur: Duration) -> Self {
        self.timeout = dur;
        self
    }

    /// Override default retry configuration.
    pub fn retry(mut self, config: RetryConfig) -> Self {
        self.retry = config;
        self
    }

    /// Override default buffer configuration.
    pub fn buffer(mut self, config: BufferConfig) -> Self {
        self.buffer = config;
        self
    }

    /// Register a callback invoked when a buffered batch flush fails after all
    /// retries. Receives the number of events that were dropped and the error.
    ///
    /// Use this for alerting, metrics, or dead-letter persistence.
    pub fn on_flush_failure(
        mut self,
        handler: impl Fn(usize, &MonkError) + Send + Sync + 'static,
    ) -> Self {
        self.on_flush_failure = Some(Arc::new(handler));
        self
    }

    /// Build the client. Returns an error if required fields are missing.
    pub fn build(self) -> crate::error::Result<MonkClient> {
        let base_url = self
            .base_url
            .ok_or_else(|| MonkError::Validation("base_url is required".into()))?
            .trim_end_matches('/')
            .to_string();
        Url::parse(&base_url)
            .map_err(|e| MonkError::Validation(format!("base_url is invalid: {e}")))?;

        let api_key = self
            .api_key
            .ok_or_else(|| MonkError::Validation("api_key is required".into()))?;

        let mut default_headers = HeaderMap::new();
        default_headers.insert(
            AUTHORIZATION,
            HeaderValue::from_str(&format!("Bearer {api_key}"))
                .map_err(|_| MonkError::Validation("api_key contains invalid characters".into()))?,
        );
        default_headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));

        let http = reqwest::Client::builder()
            .timeout(self.timeout)
            .default_headers(default_headers)
            .pool_max_idle_per_host(20)
            .build()
            .map_err(MonkError::Network)?;

        let retry_policy = Arc::new(RetryPolicy::new(self.retry));
        let inner = Arc::new(ClientInner {
            http,
            base_url,
            retry: Arc::clone(&retry_policy),
        });

        let inner_clone = Arc::clone(&inner);
        let (buffer_handle, worker_join) = BufferWorker::spawn(
            self.buffer,
            move |events| {
                let inner = Arc::clone(&inner_clone);
                async move { inner.do_ingest_batch(events).await.map(|_| ()) }
            },
            self.on_flush_failure,
        );

        Ok(MonkClient {
            inner,
            buffer: buffer_handle,
            _worker: Arc::new(worker_join),
        })
    }
}

// ── Client ──────────────────────────────────────────────────────────────────

/// Production-grade client for the Monk Events API.
///
/// Created via [`MonkClient::builder()`]. Cheaply cloneable (all state is `Arc`-wrapped).
#[derive(Clone)]
pub struct MonkClient {
    inner: Arc<ClientInner>,
    buffer: BufferHandle,
    _worker: Arc<JoinHandle<()>>,
}

struct ClientInner {
    http: reqwest::Client,
    base_url: String,
    retry: Arc<RetryPolicy>,
}

impl MonkClient {
    /// Start building a new client.
    pub fn builder() -> MonkClientBuilder {
        MonkClientBuilder::default()
    }

    /// Build a client from a top-level configuration struct.
    pub fn from_config(config: ClientConfig) -> crate::error::Result<Self> {
        MonkClient::builder()
            .base_url(config.base_url)
            .api_key(config.api_key)
            .timeout(config.timeout)
            .retry(config.retry)
            .buffer(config.buffer)
            .build()
    }

    // ── Ingest (single) ─────────────────────────────────────────────────

    /// Ingest a single event with automatic retries.
    ///
    /// Validates the event client-side, then sends `POST /v1/events`.
    pub async fn ingest(&self, mut event: Event) -> crate::error::Result<IngestResponse> {
        assign_idempotency_key_if_missing(&mut event);
        validation::validate_event(&event)?;
        self.inner.do_ingest(event).await
    }

    // ── Ingest (batch) ──────────────────────────────────────────────────

    /// Ingest a batch of events with automatic retries.
    ///
    /// Validates all events client-side, then sends `POST /v1/events/batch`.
    /// Batch size must be <= 10,000.
    pub async fn ingest_batch(
        &self,
        mut events: Vec<Event>,
    ) -> crate::error::Result<BatchIngestResponse> {
        for event in &mut events {
            assign_idempotency_key_if_missing(event);
        }
        validation::validate_batch(&events)?;
        self.inner.do_ingest_batch(events).await
    }

    // ── Buffered Ingest ─────────────────────────────────────────────────

    /// Enqueue an event into the in-memory buffer for automatic batching and flushing.
    ///
    /// Events are accumulated and flushed via `POST /v1/events/batch` either when
    /// the batch size threshold is reached or the flush interval elapses.
    ///
    /// Behaviour when the buffer is full depends on
    /// [`crate::config::BackpressureStrategy`]:
    /// - `FailFast` (default): returns [`MonkError::BufferFull`] immediately.
    /// - `Block`: suspends the caller until space is available.
    pub async fn ingest_buffered(&self, mut event: Event) -> crate::error::Result<()> {
        assign_idempotency_key_if_missing(&mut event);
        validation::validate_event(&event)?;
        self.buffer.send(event).await
    }

    /// Manually trigger a flush of the event buffer.
    pub fn flush(&self) {
        self.buffer.flush_signal().notify_one();
    }

    /// Gracefully shut down: flush remaining buffered events and release resources.
    ///
    /// After calling `close`, further calls to `ingest_buffered` will fail.
    pub fn close(&self) {
        self.buffer.close();
        self.buffer.flush_signal().notify_one();
    }

    // ── List Events ─────────────────────────────────────────────────────

    /// Begin building a list-events query (`GET /v1/events`).
    pub fn list_events(&self) -> ListEventsRequest {
        ListEventsRequest {
            inner: Arc::clone(&self.inner),
            event_name: None,
            customer_id: None,
            start: None,
            end: None,
            limit: None,
            offset: None,
        }
    }

    // ── Usage ───────────────────────────────────────────────────────────

    /// Begin building a usage query (`GET /v1/usage`).
    ///
    /// `start` and `end` must be in `YYYY-MM-DD` format.
    pub fn query_usage(&self, start: impl Into<String>, end: impl Into<String>) -> UsageRequest {
        UsageRequest {
            inner: Arc::clone(&self.inner),
            start: start.into(),
            end: end.into(),
            event_name: None,
        }
    }

    // ── Health ───────────────────────────────────────────────────────────

    /// Liveness probe (`GET /health`). Does not require authentication.
    pub async fn health(&self) -> crate::error::Result<HealthResponse> {
        let url = format!("{}/health", self.inner.base_url);
        let resp = self.inner.http.get(&url).send().await?;
        parse_response(resp).await
    }

    /// Readiness probe (`GET /ready`). Does not require authentication.
    pub async fn ready(&self) -> crate::error::Result<ReadyResponse> {
        let url = format!("{}/ready", self.inner.base_url);
        let resp = self.inner.http.get(&url).send().await?;
        parse_response(resp).await
    }
}

// ── Internal helpers on ClientInner ─────────────────────────────────────────

impl ClientInner {
    async fn do_ingest(&self, event: Event) -> crate::error::Result<IngestResponse> {
        let url = format!("{}/v1/events", self.base_url);
        let body = serde_json::to_value(&event)?;

        self.retry
            .execute(|attempt| {
                let url = url.clone();
                let body = body.clone();
                let http = self.http.clone();
                async move {
                    let request_id = Uuid::new_v4().to_string();
                    debug!(attempt, request_id, "POST /v1/events");

                    let resp = http
                        .post(&url)
                        .header("X-Request-ID", &request_id)
                        .json(&body)
                        .send()
                        .await
                        .map_err(|e| {
                            if e.is_timeout() {
                                MonkError::Timeout
                            } else {
                                MonkError::Network(e)
                            }
                        })?;

                    parse_response(resp).await
                }
            })
            .await
    }

    async fn do_ingest_batch(
        &self,
        events: Vec<Event>,
    ) -> crate::error::Result<BatchIngestResponse> {
        let url = format!("{}/v1/events/batch", self.base_url);
        let body = serde_json::to_value(&BatchRequest { events })?;

        self.retry
            .execute(|attempt| {
                let url = url.clone();
                let body = body.clone();
                let http = self.http.clone();
                async move {
                    let request_id = Uuid::new_v4().to_string();
                    debug!(attempt, request_id, "POST /v1/events/batch");

                    let resp = http
                        .post(&url)
                        .header("X-Request-ID", &request_id)
                        .json(&body)
                        .send()
                        .await
                        .map_err(|e| {
                            if e.is_timeout() {
                                MonkError::Timeout
                            } else {
                                MonkError::Network(e)
                            }
                        })?;

                    parse_response(resp).await
                }
            })
            .await
    }
}

// ── Request builders for query endpoints ────────────────────────────────────

/// Builder for `GET /v1/events` queries.
pub struct ListEventsRequest {
    inner: Arc<ClientInner>,
    event_name: Option<String>,
    customer_id: Option<String>,
    start: Option<String>,
    end: Option<String>,
    limit: Option<usize>,
    offset: Option<usize>,
}

impl ListEventsRequest {
    pub fn event_name(mut self, name: impl Into<String>) -> Self {
        self.event_name = Some(name.into());
        self
    }
    pub fn customer_id(mut self, id: impl Into<String>) -> Self {
        self.customer_id = Some(id.into());
        self
    }
    pub fn start(mut self, date: impl Into<String>) -> Self {
        self.start = Some(date.into());
        self
    }
    pub fn end(mut self, date: impl Into<String>) -> Self {
        self.end = Some(date.into());
        self
    }
    pub fn limit(mut self, n: usize) -> Self {
        self.limit = Some(n);
        self
    }
    pub fn offset(mut self, n: usize) -> Self {
        self.offset = Some(n);
        self
    }

    /// Execute the query.
    pub async fn send(self) -> crate::error::Result<EventListResponse> {
        let url = format!("{}/v1/events", self.inner.base_url);
        let event_name = self.event_name;
        let customer_id = self.customer_id;
        let start = self.start;
        let end = self.end;
        let limit = self.limit;
        let offset = self.offset;

        self.inner
            .retry
            .execute(|attempt| {
                let url = url.clone();
                let http = self.inner.http.clone();
                let event_name = event_name.clone();
                let customer_id = customer_id.clone();
                let start = start.clone();
                let end = end.clone();
                async move {
                    let request_id = Uuid::new_v4().to_string();
                    debug!(attempt, request_id, "GET /v1/events");

                    let mut req = http.get(&url).header("X-Request-ID", &request_id);

                    let mut params: Vec<(&str, String)> = Vec::new();
                    if let Some(ref v) = event_name {
                        params.push(("eventName", v.clone()));
                    }
                    if let Some(ref v) = customer_id {
                        params.push(("customerId", v.clone()));
                    }
                    if let Some(ref v) = start {
                        params.push(("start", v.clone()));
                    }
                    if let Some(ref v) = end {
                        params.push(("end", v.clone()));
                    }
                    if let Some(v) = limit {
                        params.push(("limit", v.to_string()));
                    }
                    if let Some(v) = offset {
                        params.push(("offset", v.to_string()));
                    }

                    req = req.query(&params);
                    let resp = req.send().await.map_err(|e| {
                        if e.is_timeout() {
                            MonkError::Timeout
                        } else {
                            MonkError::Network(e)
                        }
                    })?;

                    parse_response(resp).await
                }
            })
            .await
    }
}

/// Builder for `GET /v1/usage` queries.
pub struct UsageRequest {
    inner: Arc<ClientInner>,
    start: String,
    end: String,
    event_name: Option<String>,
}

impl UsageRequest {
    pub fn event_name(mut self, name: impl Into<String>) -> Self {
        self.event_name = Some(name.into());
        self
    }

    /// Execute the query.
    pub async fn send(self) -> crate::error::Result<UsageResponse> {
        let url = format!("{}/v1/usage", self.inner.base_url);
        let start = self.start;
        let end = self.end;
        let event_name = self.event_name;

        self.inner
            .retry
            .execute(|attempt| {
                let url = url.clone();
                let http = self.inner.http.clone();
                let start = start.clone();
                let end = end.clone();
                let event_name = event_name.clone();
                async move {
                    let request_id = Uuid::new_v4().to_string();
                    debug!(attempt, request_id, "GET /v1/usage");

                    let mut params = vec![("start", start), ("end", end)];
                    if let Some(ref v) = event_name {
                        params.push(("eventName", v.clone()));
                    }

                    let resp = http
                        .get(&url)
                        .header("X-Request-ID", &request_id)
                        .query(&params)
                        .send()
                        .await
                        .map_err(|e| {
                            if e.is_timeout() {
                                MonkError::Timeout
                            } else {
                                MonkError::Network(e)
                            }
                        })?;

                    parse_response(resp).await
                }
            })
            .await
    }
}

// ── Response parsing ────────────────────────────────────────────────────────

async fn parse_response<T: serde::de::DeserializeOwned>(
    resp: reqwest::Response,
) -> crate::error::Result<T> {
    let status = resp.status().as_u16();

    if (200..300).contains(&status) {
        let body = resp.json::<T>().await.map_err(|e| {
            MonkError::Validation(format!("failed to decode success response: {e}"))
        })?;
        return Ok(body);
    }

    let raw = resp.text().await.unwrap_or_default();
    match serde_json::from_str::<ApiErrorBody>(&raw) {
        Ok(api_err) => Err(MonkError::from_api(status, api_err)),
        Err(_) => Err(MonkError::from_status(status, raw)),
    }
}

fn assign_idempotency_key_if_missing(event: &mut Event) {
    if event.idempotency_key.is_none() {
        event.idempotency_key = Some(Uuid::new_v4().to_string());
    }
}
