//! ctl HTTP client: signs every business request with the four X-RE-* headers.

use std::time::{SystemTime, UNIX_EPOCH};

use anyhow::{anyhow, Result};
use base64::{
    engine::general_purpose::{STANDARD as B64, URL_SAFE_NO_PAD as B64URL},
    Engine,
};
use rand::{rngs::OsRng, RngCore};
use reqwest::Method;
use serde::{de::DeserializeOwned, Serialize};
use serde_json::Value;

use dragoon_proto::{
    canonical::canonical_string,
    constants,
    models::{AuthChallenge, AuthLoginRequest, AuthLoginResponse, Task, TaskKind, TaskLimits, TaskSubmit},
    signer::Signer,
};

pub struct CtlClient {
    base: String,
    http: reqwest::Client,
    signer: Box<dyn Signer + Send + Sync>,
    session_token: Option<String>,
}

impl CtlClient {
    pub fn new(
        base_url: impl Into<String>,
        signer: Box<dyn Signer + Send + Sync>,
        verify_tls: bool,
        session_token: Option<String>,
    ) -> Result<Self> {
        let mut builder = reqwest::Client::builder().timeout(std::time::Duration::from_secs(60));
        if !verify_tls {
            builder = builder.danger_accept_invalid_certs(true);
        }
        Ok(Self {
            base: base_url.into().trim_end_matches('/').to_string(),
            http: builder.build()?,
            signer,
            session_token,
        })
    }

    pub fn session_token(&self) -> Option<&str> {
        self.session_token.as_deref()
    }

    pub async fn login(
        &mut self,
        username: &str,
        password: &str,
        totp_code: &str,
    ) -> Result<AuthLoginResponse> {
        let r = self
            .http
            .post(format!("{}/v1/auth/challenge", self.base))
            .json(&serde_json::json!({}))
            .send()
            .await?;
        if !r.status().is_success() {
            return Err(anyhow!("challenge failed: {}", r.status()));
        }
        let ch: AuthChallenge = r.json().await?;
        let sig_wire = self
            .signer
            .sign(ch.challenge.as_bytes())
            .map_err(|e| anyhow!("sign challenge: {e}"))?;
        let body = AuthLoginRequest {
            username: username.into(),
            password: password.into(),
            totp_code: totp_code.into(),
            challenge: ch.challenge,
            challenge_signature_b64: B64.encode(&sig_wire),
            key_fingerprint: self.signer.fingerprint().to_string(),
        };
        let r = self
            .http
            .post(format!("{}/v1/auth/login", self.base))
            .json(&body)
            .send()
            .await?;
        if !r.status().is_success() {
            return Err(anyhow!("login failed: {} {}", r.status(), r.text().await.unwrap_or_default()));
        }
        let resp: AuthLoginResponse = r.json().await?;
        self.session_token = Some(resp.session_token.clone());
        Ok(resp)
    }

    pub async fn logout(&mut self) -> Result<()> {
        let _: Value = self.signed("POST", "/v1/auth/logout", Some(&serde_json::json!({}))).await?;
        self.session_token = None;
        Ok(())
    }

    pub async fn list_workers(&self) -> Result<Vec<Value>> {
        self.signed("GET", "/v1/workers", None::<&Value>).await
    }

    pub async fn submit(
        &self,
        worker_name: &str,
        kind: TaskKind,
        payload: &str,
        collect: Vec<String>,
        limits: TaskLimits,
        fetch_path: Option<String>,
    ) -> Result<Task> {
        let body = TaskSubmit {
            worker_name: worker_name.into(),
            kind,
            payload: payload.into(),
            collect,
            limits,
            fetch_path,
        };
        self.signed("POST", "/v1/tasks", Some(&body)).await
    }

    pub async fn get_task(&self, task_id: &str) -> Result<Task> {
        self.signed("GET", &format!("/v1/tasks/{task_id}"), None::<&Value>).await
    }

    pub async fn cancel_task(&self, task_id: &str) -> Result<Task> {
        self.signed(
            "POST",
            &format!("/v1/tasks/{task_id}/cancel"),
            Some(&serde_json::json!({})),
        )
        .await
    }

    pub async fn tail_log(&self, task_id: &str, since_seq: u64, stream: &str) -> Result<Value> {
        self.signed(
            "GET",
            &format!("/v1/tasks/{task_id}/log?since_seq={since_seq}&stream={stream}"),
            None::<&Value>,
        )
        .await
    }

    pub async fn download_artifact(&self, task_id: &str, name: &str) -> Result<bytes::Bytes> {
        let encoded = url::form_urlencoded::byte_serialize(name.as_bytes())
            .collect::<String>();
        let resp = self
            .signed_request(
                "GET",
                &format!("/v1/tasks/{task_id}/artifact?name={encoded}"),
                None::<&Value>,
            )
            .await?;
        Ok(resp.bytes().await?)
    }

    pub async fn fetch(&self, worker: &str, path: &str, glob: &str) -> Result<Task> {
        self.signed(
            "POST",
            &format!("/v1/workers/{worker}/fetch"),
            Some(&serde_json::json!({"path": path, "glob": glob})),
        )
        .await
    }

    pub async fn fetch_messages(
        &self,
        since_id: i64,
        include_read: bool,
        long_poll: bool,
    ) -> Result<Value> {
        self.signed(
            "GET",
            &format!(
                "/v1/messages?since_id={since_id}&include_read={include_read}&long_poll={long_poll}"
            ),
            None::<&Value>,
        )
        .await
    }

    pub async fn ack_messages(&self, ids: Vec<i64>) -> Result<i64> {
        let v: Value = self
            .signed("POST", "/v1/messages/ack", Some(&serde_json::json!({"ids": ids})))
            .await?;
        Ok(v.get("acked").and_then(Value::as_i64).unwrap_or(0))
    }

    // --------------------------------------------------------------

    async fn signed<T: DeserializeOwned, B: Serialize>(
        &self,
        method: &str,
        path: &str,
        body: Option<&B>,
    ) -> Result<T> {
        let resp = self.signed_request(method, path, body).await?;
        Ok(resp.json().await?)
    }

    async fn signed_request<B: Serialize>(
        &self,
        method: &str,
        path: &str,
        body: Option<&B>,
    ) -> Result<reqwest::Response> {
        let token = self
            .session_token
            .as_ref()
            .ok_or_else(|| anyhow!("not logged in"))?;
        let body_bytes: Vec<u8> = match body {
            None => Vec::new(),
            Some(b) => serde_json::to_vec(b)?,
        };
        let ts = i64::try_from(SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs())?;
        let mut nonce_bytes = [0u8; 12];
        OsRng.fill_bytes(&mut nonce_bytes);
        let nonce = B64URL.encode(nonce_bytes);
        let canonical = canonical_string(method, path, ts, &nonce, &body_bytes);
        let sig_wire = self
            .signer
            .sign(&canonical)
            .map_err(|e| anyhow!("signer: {e}"))?;
        let sig_b64 = B64.encode(&sig_wire);

        let m = Method::from_bytes(method.as_bytes())?;
        let mut req = self
            .http
            .request(m, format!("{}{}", self.base, path))
            .header("Authorization", format!("Bearer {token}"))
            .header(constants::HDR_TIMESTAMP, ts.to_string())
            .header(constants::HDR_NONCE, &nonce)
            .header(constants::HDR_KEY_FP, self.signer.fingerprint())
            .header(constants::HDR_SIG, &sig_b64);
        if !body_bytes.is_empty() {
            req = req
                .header("Content-Type", "application/json")
                .body(body_bytes);
        }
        let resp = req.send().await?;
        if !resp.status().is_success() {
            return Err(anyhow!(
                "{method} {path}: {} {}",
                resp.status(),
                resp.text().await.unwrap_or_default()
            ));
        }
        Ok(resp)
    }
}
