avassa_client/

client.rs

use crate::volga;
use crate::{Error, Result};
use bytes::Bytes;
use serde_json::json;

/// Cert and private key in PEM format.
#[derive(Clone, serde::Deserialize)]
#[serde(rename_all = "kebab-case")]
#[non_exhaustive]
pub struct X509SVID {
    pub cert: String,
    pub private_key: String,
}

#[derive(Clone, serde::Deserialize)]
#[serde(rename_all = "kebab-case")]
struct LoginToken {
    token: String,
    expires_in: i64,
    expires: chrono::DateTime<chrono::Utc>,
    creation_time: chrono::DateTime<chrono::Utc>,
    jwt_svid: Option<String>,
    x509_svid: Option<X509SVID>,
}

impl LoginToken {
    #[expect(clippy::integer_division, reason = "Good enough precision")]
    #[expect(
        clippy::integer_division_remainder_used,
        reason = "Good enough precision"
    )]
    fn renew_at(&self) -> chrono::DateTime<chrono::Utc> {
        let quarter_expiry = self.expires_in / 4;
        self.expires
            .checked_sub_signed(chrono::Duration::seconds(quarter_expiry))
            .unwrap_or_else(chrono::Utc::now)
    }
}

impl core::fmt::Debug for LoginToken {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("LoginToken")
            .field("creation_time", &self.creation_time)
            .field("expires_in", &self.expires_in)
            .field("expires", &self.expires)
            .field("jwt-svid", &self.jwt_svid.is_some())
            .field("x509-svid", &self.x509_svid.is_some())
            .finish_non_exhaustive()
    }
}

#[derive(Debug)]
struct ClientState {
    login_token: LoginToken,
}

/// Builder for an Avassa [`Client`].
#[derive(Clone)]
#[expect(clippy::module_name_repetitions, reason = "Backwards compat")]
pub struct ClientBuilder {
    reqwest_ca: Vec<reqwest::Certificate>,
    tls_ca: tokio_rustls::rustls::RootCertStore,
    disable_cert_verification: bool,
    connection_verbose: bool,
    auto_renew_token: bool,
    timeout: Option<core::time::Duration>,
    connect_timeout: Option<core::time::Duration>,
}

impl ClientBuilder {
    /// Create a new builder instance.
    #[must_use]
    pub(crate) fn new() -> Self {
        let tls_ca = webpki_roots::TLS_SERVER_ROOTS.iter().cloned().collect();
        Self {
            reqwest_ca: Vec::new(),
            tls_ca,
            disable_cert_verification: false,
            connection_verbose: false,
            auto_renew_token: true,
            timeout: None,
            connect_timeout: None,
        }
    }

    /// Enables a request timeout.
    #[must_use]
    #[inline]
    pub fn timeout(self, timeout: core::time::Duration) -> Self {
        Self {
            timeout: Some(timeout),
            ..self
        }
    }

    /// Set a timeout for only the connect phase of a Client.
    #[must_use]
    #[inline]
    pub fn connection_timeout(self, timeout: core::time::Duration) -> Self {
        Self {
            connect_timeout: Some(timeout),
            ..self
        }
    }

    /// Add a root certificate for API certificate verification.
    #[inline]
    #[expect(clippy::missing_errors_doc, reason = "Convered in the Error type")]
    pub fn add_root_certificate(mut self, cert: &[u8]) -> Result<Self> {
        use core::iter;
        let r_ca = reqwest::Certificate::from_pem(cert)?;
        let mut ca_reader = std::io::BufReader::new(cert);
        for item in iter::from_fn(|| rustls_pemfile::read_one(&mut ca_reader).transpose()) {
            if let rustls_pemfile::Item::X509Certificate(ca_cert) = item? {
                self.tls_ca.add(ca_cert)?;
            }
        }
        self.reqwest_ca.push(r_ca);
        Ok(self)
    }

    /// Disable certificate verification.
    #[must_use]
    #[inline]
    pub fn danger_disable_cert_verification(self) -> Self {
        Self {
            disable_cert_verification: true,
            ..self
        }
    }

    /// Enabling this option will emit log messages at the TRACE level for read and write operations
    /// on the https client.
    #[must_use]
    #[inline]
    pub fn enable_verbose_connection(self) -> Self {
        Self {
            connection_verbose: true,
            ..self
        }
    }

    /// Disable auto renewal of authentication token.
    #[must_use]
    #[inline]
    pub fn disable_token_auto_renewal(self) -> Self {
        Self {
            auto_renew_token: false,
            ..self
        }
    }

    /// Login the application from secret set in the environment
    /// `approle_id` can optionally be provided
    /// This assumes the environment variable `APPROLE_SECRET_ID` is set by the system.
    #[deprecated]
    #[inline]
    #[expect(clippy::missing_errors_doc, reason = "Convered in the Error type")]
    pub async fn application_login(&self, host: &str, approle_id: Option<&str>) -> Result<Client> {
        let secret_id = std::env::var("APPROLE_SECRET_ID")
            .map_err(|_| Error::LoginFailureMissingEnv(String::from("APPROLE_SECRET_ID")))?;

        // If no app role is provided, we can try to use the secret id as app role.
        let role_id = approle_id.unwrap_or(&secret_id);

        let base_url = url::Url::parse(host)?;
        let url = base_url.join("v1/approle-login")?;
        let data = json!({
            "role-id": role_id,
            "secret-id": secret_id,
        });
        Client::do_login(self, base_url, url, data).await
    }

    /// Login using approle.
    #[expect(clippy::missing_errors_doc, reason = "Convered in the Error type")]
    pub async fn approle_login(
        &self,
        host: &str,
        secret_id: &str,
        role_id: Option<&str>,
    ) -> Result<Client> {
        // If no role id is provided, use the secret id;
        let role_id = role_id.unwrap_or(secret_id);

        let base_url = url::Url::parse(host)?;
        let url = base_url.join("v1/approle-login")?;
        let data = json!({
            "role-id": role_id,
            "secret-id": secret_id,
        });
        Client::do_login(self, base_url, url, data).await
    }

    /// Login to an avassa Control Tower or Edge Enforcer instance. If possible,
    /// please use the `application_login` as no credentials needs to be distributed.
    #[tracing::instrument(skip(self, password))]
    #[expect(clippy::missing_errors_doc, reason = "Convered in the Error type")]
    pub async fn login(&self, host: &str, username: &str, password: &str) -> Result<Client> {
        let base_url = url::Url::parse(host)?;
        let url = base_url.join("v1/login")?;

        // If we have a tenant, send it.
        let data = json!({
            "username":username,
            "password":password
        });
        Client::do_login(self, base_url, url, data).await
    }

    /// Login using an existing bearer token.
    #[tracing::instrument(skip(self, token))]
    #[expect(clippy::missing_errors_doc, reason = "Convered in the Error type")]
    #[expect(clippy::arithmetic_side_effects, reason = "Ok to add one second")]
    pub fn token_login(&self, host: &str, token: &str) -> Result<Client> {
        let base_url = url::Url::parse(host)?;
        let client = Client::reqwest_client(self)?;
        let creation_time = chrono::Local::now().into();
        let expires = creation_time + chrono::Duration::seconds(1);

        let login_token = LoginToken {
            token: token.to_owned(),
            expires_in: 1,
            creation_time,
            expires,
            jwt_svid: None,
            x509_svid: None,
        };

        Client::new(self, client, base_url, login_token)
    }

    /// Login using JWT issued by external entity, e.g. MS Entra.
    #[expect(clippy::missing_errors_doc, reason = "Convered in the Error type")]
    pub async fn jwt_login(
        &self,
        host: &str,
        tenant_name: &str,
        jwt_auth: &str,
        role: &str,
        jwt: &str,
    ) -> Result<Client> {
        let base_url = url::Url::parse(host)?;
        let url = base_url.join("v1/jwt-login")?;

        let data = json!({
            "tenant": tenant_name,
            "jwt-auth": jwt_auth,
            "role": role,
            "jwt": jwt
        });
        Client::do_login(self, base_url, url, data).await
    }
}

impl Default for ClientBuilder {
    fn default() -> Self {
        Self::new()
    }
}

/// The `Client` is used for all interaction with Control Tower or Edge Enforcer instances.
/// Use one of the login functions to create an instance.
#[derive(Clone)]
pub struct Client {
    base_url: url::Url,
    websocket_url: url::Url,
    state: std::sync::Arc<tokio::sync::Mutex<ClientState>>,
    #[expect(clippy::struct_field_names, reason = "easier to read")]
    http_client: reqwest::Client,
    tls_ca: tokio_rustls::rustls::RootCertStore,
    disable_cert_verification: bool,
}

impl core::fmt::Debug for Client {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("Client")
            .field("base_url", &self.base_url)
            .field("websocket_url", &self.websocket_url)
            .field("state", &self.state)
            .field("http_client", &self.http_client)
            .field("disable_cert_verification", &self.disable_cert_verification)
            .finish_non_exhaustive()
    }
}

impl Client {
    /// Create a Client builder.
    #[must_use]
    pub fn builder() -> ClientBuilder {
        ClientBuilder::new()
    }

    /// Returns true if the token has expired.
    pub async fn token_is_expired(&self) -> bool {
        let state = self.state.lock().await;
        chrono::Local::now() > state.login_token.expires
    }

    pub async fn token_expires(&self) -> chrono::DateTime<chrono::Utc> {
        let state = self.state.lock().await;
        state.login_token.expires
    }

    /// Returns the JWT SVID if exists.
    pub async fn jwt_svid(&self) -> Option<String> {
        let state = self.state.lock().await;
        state.login_token.jwt_svid.clone()
    }

    pub async fn x509_svid(&self) -> Option<X509SVID> {
        let state = self.state.lock().await;
        state.login_token.x509_svid.clone()
    }

    async fn do_login(
        builder: &ClientBuilder,
        base_url: url::Url,
        url: url::Url,
        payload: serde_json::Value,
    ) -> Result<Self> {
        let json = serde_json::to_string(&payload)?;
        let client = Self::reqwest_client(builder)?;
        let result = client
            .post(url)
            .header("content-type", "application/json")
            .body(json)
            .send()
            .await?;

        if result.status().is_success() {
            let login_token = result.json().await?;
            Self::new(builder, client, base_url, login_token)
        } else {
            let text = result.text().await?;
            tracing::debug!("login returned {}", text);
            Err(Error::LoginFailure(text))
        }
    }

    fn reqwest_client(builder: &ClientBuilder) -> Result<reqwest::Client> {
        let reqwest_client_builder = reqwest::Client::builder().use_rustls_tls();

        // Add CA certificates
        let reqwest_client_builder = builder
            .reqwest_ca
            .iter()
            .fold(reqwest_client_builder, |reqwest_client_builder, ca| {
                reqwest_client_builder.add_root_certificate(ca.clone())
            });

        tracing::debug!("Added {} CA certs", builder.reqwest_ca.len());

        let reqwest_client_builder =
            reqwest_client_builder.danger_accept_invalid_certs(builder.disable_cert_verification);

        let reqwest_client_builder =
            reqwest_client_builder.connection_verbose(builder.connection_verbose);

        let reqwest_client_builder = if let Some(duration) = builder.timeout {
            reqwest_client_builder.timeout(duration)
        } else {
            reqwest_client_builder
        };

        let reqwest_client_builder = if let Some(duration) = builder.connect_timeout {
            reqwest_client_builder.connect_timeout(duration)
        } else {
            reqwest_client_builder
        };

        let client = reqwest_client_builder.build()?;
        Ok(client)
    }

    fn new(
        builder: &ClientBuilder,
        client: reqwest::Client,
        base_url: url::Url,
        login_token: LoginToken,
    ) -> Result<Self> {
        let websocket_url = url::Url::parse(&format!("wss://{}/v1/ws/", base_url.host_port()?))?;

        let renew_at = login_token.renew_at();

        let state = std::sync::Arc::new(tokio::sync::Mutex::new(ClientState { login_token }));

        let weak_state = std::sync::Arc::downgrade(&state);
        let refresh_url = base_url.join("/v1/state/strongbox/token/refresh")?;

        if builder.auto_renew_token {
            tokio::spawn(renew_token_task(
                weak_state,
                renew_at,
                client.clone(),
                refresh_url,
            ));
        }

        Ok(Self {
            http_client: client,
            tls_ca: builder.tls_ca.clone(),
            disable_cert_verification: builder.disable_cert_verification,
            base_url,
            websocket_url,
            state,
        })
    }

    /// Returns the login bearer token.
    pub async fn bearer_token(&self) -> String {
        let state = self.state.lock().await;
        state.login_token.token.clone()
    }

    /// GET a json payload from the REST API.
    #[expect(clippy::missing_errors_doc, reason = "Convered in the Error type")]
    pub async fn get_json<T: serde::de::DeserializeOwned>(
        &self,
        path: &str,
        query_params: Option<&[(&str, &str)]>,
    ) -> Result<T> {
        let url = self.base_url.join(path)?;

        let token = self.bearer_token().await;

        let mut builder = self
            .http_client
            .get(url)
            .bearer_auth(&token)
            .header("Accept", "application/json");
        if let Some(qp) = query_params {
            builder = builder.query(qp);
        }

        let result = builder.send().await?;

        if result.status().is_success() {
            let res = result.json().await?;
            Ok(res)
        } else {
            let status = result.status();
            let error_payload = result
                .text()
                .await
                .unwrap_or_else(|_| "No error payload".to_owned());
            Err(Error::WebServer(
                status.as_u16(),
                status.to_string(),
                error_payload,
            ))
        }
    }

    /// GET a bytes payload from the REST API.
    #[expect(clippy::missing_errors_doc, reason = "Convered in the Error type")]
    pub async fn get_bytes(
        &self,
        path: &str,
        query_params: Option<&[(&str, &str)]>,
    ) -> Result<Bytes> {
        let url = self.base_url.join(path)?;

        let token = self.bearer_token().await;

        let mut builder = self.http_client.get(url).bearer_auth(&token);

        if let Some(qp) = query_params {
            builder = builder.query(qp);
        }

        let result = builder.send().await?;

        if result.status().is_success() {
            let res = result.bytes().await?;
            Ok(res)
        } else {
            let status = result.status();
            let error_payload = result
                .text()
                .await
                .unwrap_or_else(|_| "No error payload".to_owned());
            Err(Error::WebServer(
                status.as_u16(),
                status.to_string(),
                error_payload,
            ))
        }
    }

    /// POST arbitrary JSON to a path.
    /// # Panics
    ///   never.
    #[expect(clippy::missing_errors_doc, reason = "Convered in the Error type")]
    #[expect(clippy::expect_used, reason = "Vec len checked")]
    pub async fn post_json(
        &self,
        path: &str,
        data: &serde_json::Value,
    ) -> Result<serde_json::Value> {
        let url = self.base_url.join(path)?;
        let token = self.bearer_token().await;

        tracing::debug!("POST {} {:?}", url, data);

        let result = self
            .http_client
            .post(url)
            .json(&data)
            .bearer_auth(&token)
            .send()
            .await?;

        if result.status().is_success() {
            let resp = result.bytes().await?;

            let decoder = serde_json::Deserializer::from_slice(&resp);

            let responses: core::result::Result<Vec<serde_json::Value>, _> =
                decoder.into_iter().collect();
            let mut responses = responses?;

            match responses.len() {
                0 => Ok(serde_json::Value::Object(serde_json::Map::default())),
                1 => Ok(responses.pop().expect("We know len == 1")),
                _ => {
                    // Convert to a JSON array
                    Ok(serde_json::Value::Array(responses))
                }
            }
        } else {
            tracing::error!("POST call failed");
            let status = result.status();
            let resp = result.json().await;
            match resp {
                Ok(resp) => Err(Error::REST(resp)),
                Err(_) => Err(Error::WebServer(
                    status.as_u16(),
                    status.to_string(),
                    "Failed to get JSON responses".to_owned(),
                )),
            }
        }
    }

    /// PUT arbitrary data to a path.
    #[expect(clippy::missing_errors_doc, reason = "Convered in the Error type")]
    pub async fn put<T: Into<reqwest::Body> + core::fmt::Debug>(
        &self,
        path: &str,
        content_type: &str,
        data: T,
    ) -> Result<()> {
        let url = self.base_url.join(path)?;
        let token = self.state.lock().await.login_token.token.clone();

        tracing::debug!("PUT {} {:?}", url, data);

        let result = self
            .http_client
            .put(url)
            .header(reqwest::header::CONTENT_TYPE, content_type)
            .body(data)
            .bearer_auth(&token)
            .send()
            .await?;

        //#[allow(clippy::redundant_closure_for_method_calls)]
        if result.status().is_success() {
            Ok(())
        } else {
            tracing::error!("PUT call failed");
            let status = result.status();
            let resp = result.json().await;
            match resp {
                Ok(resp) => Err(Error::REST(resp)),
                Err(_) => Err(Error::WebServer(
                    status.as_u16(),
                    status.to_string(),
                    "Failed to get JSON reply".to_owned(),
                )),
            }
        }
    }

    /// PUT arbitrary JSON to a path.
    /// # Errors
    /// `crate::error::Error`
    pub async fn put_json(
        &self,
        path: &str,
        data: &serde_json::Value,
    ) -> Result<serde_json::Value> {
        let url = self.base_url.join(path)?;
        let token = self.state.lock().await.login_token.token.clone();

        tracing::debug!("PUT {} {:?}", url, data);

        let result = self
            .http_client
            .put(url)
            .json(&data)
            .bearer_auth(&token)
            .send()
            .await?;

        // #[allow(clippy::redundant_closure_for_method_calls)]
        if result.status().is_success() {
            use core::error::Error as _;
            let resp = result.json().await.or_else(|err| match err {
                decode_err if err.is_decode() => {
                    match decode_err
                        .source()
                        .and_then(|source_err| source_err.downcast_ref::<serde_json::Error>())
                    {
                        Some(eof_err) if eof_err.is_eof() => {
                            Ok(serde_json::Value::Object(serde_json::Map::new()))
                        }
                        _ => Err(decode_err),
                    }
                }
                other_err => Err(other_err),
            })?;
            Ok(resp)
        } else {
            tracing::error!("PUT call failed");
            let status = result.status();
            let resp = result.json().await;
            match resp {
                Ok(resp) => Err(Error::REST(resp)),
                Err(_) => Err(Error::WebServer(
                    status.as_u16(),
                    status.to_string(),
                    "Failed to get JSON reply".to_owned(),
                )),
            }
        }
    }

    /// Open a volga producer on a topic.
    /// # Errors
    /// `crate::error::Error`
    pub async fn volga_open_producer(
        &self,
        producer_name: &str,
        topic: &str,
        on_no_exists: volga::OnNoExists,
    ) -> Result<volga::producer::Producer> {
        crate::volga::producer::Builder::new(self, producer_name, topic, on_no_exists)?
            .connect()
            .await
    }

    /// Open a volga NAT producer on a topic in a site.
    /// # Errors
    /// `crate::error::Error`
    pub async fn volga_open_child_site_producer(
        &self,
        producer_name: &str,
        topic: &str,
        site: &str,
        on_no_exists: volga::OnNoExists,
    ) -> Result<volga::producer::Producer> {
        crate::volga::producer::Builder::new_child(self, producer_name, topic, site, on_no_exists)?
            .connect()
            .await
    }

    /// Open a volga producer on a topic in the parent site.
    /// # Errors
    /// `crate::error::Error`
    pub async fn volga_open_parent_site_producer(
        &self,
        producer_name: &str,
        topic: &str,
        on_no_exists: volga::OnNoExists,
    ) -> Result<volga::producer::Producer> {
        crate::volga::producer::Builder::new_parent(self, producer_name, topic, on_no_exists)?
            .connect()
            .await
    }

    /// Creates and opens a Volga consumer.
    /// # Errors
    /// `crate::error::Error`
    #[tracing::instrument(skip(self))]
    pub async fn volga_open_consumer(
        &self,
        consumer_name: &str,
        topic: &str,
        options: crate::volga::consumer::Options<'_>,
    ) -> Result<volga::consumer::Consumer> {
        crate::volga::consumer::Builder::new(self, consumer_name, topic)?
            .set_options(options)
            .connect()
            .await
    }

    /// Creates and opens a Volga consumer on a child site.
    /// # Errors
    /// `crate::error::Error`
    pub async fn volga_open_child_site_consumer(
        &self,
        consumer_name: &str,
        topic: &str,
        site: &str,
        options: crate::volga::consumer::Options<'_>,
    ) -> Result<volga::consumer::Consumer> {
        crate::volga::consumer::Builder::new_child(self, consumer_name, topic, site)?
            .set_options(options)
            .connect()
            .await
    }

    /// Creates and opens a Volga consumer on the parent site.
    /// # Errors
    /// `crate::error::Error`
    pub async fn volga_open_parent_site_consumer(
        &self,
        consumer_name: &str,
        topic: &str,
        options: crate::volga::consumer::Options<'_>,
    ) -> Result<volga::consumer::Consumer> {
        crate::volga::consumer::Builder::new_parent(self, consumer_name, topic)?
            .set_options(options)
            .connect()
            .await
    }

    #[tracing::instrument(skip(self))]
    pub(crate) async fn open_tls_stream(
        &self,
    ) -> Result<tokio_rustls::client::TlsStream<tokio::net::TcpStream>> {
        let mut client_config = tokio_rustls::rustls::ClientConfig::builder()
            .with_root_certificates(self.tls_ca.clone())
            .with_no_client_auth();

        if self.disable_cert_verification {
            let mut danger = client_config.dangerous();

            danger.set_certificate_verifier(std::sync::Arc::new(CertificateVerifier));
        }

        let client_config = std::sync::Arc::new(client_config);

        let connector: tokio_rustls::TlsConnector = client_config.into();
        let addrs = self.websocket_url.socket_addrs(|| None)?;
        let stream = tokio::net::TcpStream::connect(&*addrs).await?;

        let server_name = tokio_rustls::rustls::pki_types::ServerName::try_from(
            self.websocket_url
                .host_str()
                .ok_or_else(|| Error::general("No websocket_url host"))?
                .to_owned(),
        )?;
        let stream = connector.connect(server_name, stream).await?;
        Ok(stream)
    }

    /// Opens a query stream.
    /// # Errors
    /// `crate::error::Error`
    pub async fn volga_query_topic(
        &self,
        query: volga::query_topic::Query,
    ) -> Result<volga::query_topic::QueryStream> {
        volga::query_topic::QueryStream::new(self, query).await
    }

    /// Setup a connection to a remote container.
    /// # Errors
    /// `crate::error::Error`
    pub async fn connect(
        &self,
        application: &str,
        service_instance: &str,
        site: &str,
        protocol: crate::app_connect::Protocol,
        port: u16,
        ip_address: Option<core::net::IpAddr>,
    ) -> Result<crate::app_connect::Connection> {
        crate::app_connect::connect(
            self,
            application,
            service_instance,
            site,
            protocol,
            port,
            ip_address,
        )
        .await
    }

    pub(crate) fn get_websocket_url(&self) -> &url::Url {
        &self.websocket_url
    }

    pub(crate) fn get_base_url(&self) -> &url::Url {
        &self.base_url
    }
}

#[derive(Debug)]
struct CertificateVerifier;

#[expect(clippy::missing_trait_methods, reason="Provided by the trait")]
impl tokio_rustls::rustls::client::danger::ServerCertVerifier for CertificateVerifier {
    fn verify_server_cert(
        &self,
        _end_entity: &tokio_rustls::rustls::pki_types::CertificateDer<'_>,
        _intermediates: &[tokio_rustls::rustls::pki_types::CertificateDer<'_>],
        _server_name: &tokio_rustls::rustls::pki_types::ServerName<'_>,
        _ocsp_response: &[u8],
        _now: tokio_rustls::rustls::pki_types::UnixTime,
    ) -> core::result::Result<
        tokio_rustls::rustls::client::danger::ServerCertVerified,
        tokio_rustls::rustls::Error,
    > {
        Ok(tokio_rustls::rustls::client::danger::ServerCertVerified::assertion())
    }

    fn verify_tls12_signature(
        &self,
        _message: &[u8],
        _cert: &tokio_rustls::rustls::pki_types::CertificateDer<'_>,
        _dss: &tokio_rustls::rustls::DigitallySignedStruct,
    ) -> core::result::Result<
        tokio_rustls::rustls::client::danger::HandshakeSignatureValid,
        tokio_rustls::rustls::Error,
    > {
        Ok(tokio_rustls::rustls::client::danger::HandshakeSignatureValid::assertion())
    }

    fn verify_tls13_signature(
        &self,
        _message: &[u8],
        _cert: &tokio_rustls::rustls::pki_types::CertificateDer<'_>,
        _dss: &tokio_rustls::rustls::DigitallySignedStruct,
    ) -> core::result::Result<
        tokio_rustls::rustls::client::danger::HandshakeSignatureValid,
        tokio_rustls::rustls::Error,
    > {
        Ok(tokio_rustls::rustls::client::danger::HandshakeSignatureValid::assertion())
    }

    fn supported_verify_schemes(&self) -> Vec<tokio_rustls::rustls::SignatureScheme> {
        vec![
            tokio_rustls::rustls::SignatureScheme::RSA_PKCS1_SHA1,
            tokio_rustls::rustls::SignatureScheme::ECDSA_SHA1_Legacy,
            tokio_rustls::rustls::SignatureScheme::RSA_PKCS1_SHA256,
            tokio_rustls::rustls::SignatureScheme::ECDSA_NISTP256_SHA256,
            tokio_rustls::rustls::SignatureScheme::RSA_PKCS1_SHA384,
            tokio_rustls::rustls::SignatureScheme::ECDSA_NISTP384_SHA384,
            tokio_rustls::rustls::SignatureScheme::RSA_PKCS1_SHA512,
            tokio_rustls::rustls::SignatureScheme::ECDSA_NISTP521_SHA512,
            tokio_rustls::rustls::SignatureScheme::RSA_PSS_SHA256,
            tokio_rustls::rustls::SignatureScheme::RSA_PSS_SHA384,
            tokio_rustls::rustls::SignatureScheme::RSA_PSS_SHA512,
            tokio_rustls::rustls::SignatureScheme::ED25519,
            tokio_rustls::rustls::SignatureScheme::ED448,
        ]
    }
}

pub(crate) trait URLExt {
    fn host_port(&self) -> core::result::Result<String, url::ParseError>;
}

impl URLExt for url::Url {
    fn host_port(&self) -> core::result::Result<String, url::ParseError> {
        let host = self.host_str().ok_or(url::ParseError::EmptyHost)?;
        Ok(match (host, self.port()) {
            (host, Some(port)) => format!("{host}:{port}"),
            (host, _) => host.to_owned(),
        })
    }
}

#[tracing::instrument(skip(next_renew_at, weak_state, client, refresh_url))]
#[expect(clippy::arithmetic_side_effects, reason="Ok to subtract now")]
#[expect(clippy::single_call_fn, reason="Don't want to inline")]
async fn renew_token_task(
    weak_state: std::sync::Weak<tokio::sync::Mutex<ClientState>>,
    mut next_renew_at: chrono::DateTime<chrono::Utc>,
    client: reqwest::Client,
    refresh_url: url::Url,
) {
    loop {
        let now: chrono::DateTime<_> = chrono::Local::now().into();

        let sleep_time = next_renew_at - now;

        tracing::info!("renew token in {sleep_time} ({})", next_renew_at);

        tokio::time::sleep(
            sleep_time
                .to_std()
                .unwrap_or_else(|_| core::time::Duration::from_secs(0)),
        )
        .await;

        if let Some(state) = weak_state.upgrade() {
            let mut state = state.lock().await;

            if now > state.login_token.expires {
                tracing::error!("Token is expired and we can't renew. Giving up");
                break;
            }
            let response = client
                .post(refresh_url.clone())
                .bearer_auth(&state.login_token.token)
                .send()
                .await;

            let response = match response {
                Ok(resp) => resp,
                Err(err) => {
                    tracing::error!("Failed to renew token: {err}");
                    next_renew_at = chrono::Utc::now() + chrono::Duration::seconds(1);
                    continue;
                }
            };

            let Ok(text) = response.text().await else {
                tracing::error!("Failed to get token renew response text");
                continue;
            };
            let new_login_token = serde_json::from_str::<LoginToken>(&text);

            match new_login_token {
                Ok(new_login_token) => {
                    next_renew_at = new_login_token.renew_at();
                    state.login_token = new_login_token;
                    tracing::info!(
                        "Successfully renewed token, expires: {}",
                        state.login_token.expires
                    );
                }
                Err(err) => {
                    tracing::error!("Failed to parse or get token: {err}");
                    // After failure, we check every second
                    next_renew_at = chrono::Utc::now() + chrono::Duration::seconds(1);
                }
            }
        } else {
            tracing::debug!("renew_token: State lost");
            // If we can't get the state, the client is gone and we should go as well
            break;
        }
    }
}

#[cfg(test)]
mod test {
    #[test]
    #[expect(clippy::shadow_unrelated, reason="Test")]
    fn url_ext() {
        use super::URLExt as _;
        let url = url::Url::parse("https://1.2.3.4:5000/a/b/c").unwrap();
        let host_port = url.host_port().unwrap();
        assert_eq!(&host_port, "1.2.3.4:5000");

        let url = url::Url::parse("https://1.2.3.4/a/b/c").unwrap();
        let host_port = url.host_port().unwrap();
        assert_eq!(&host_port, "1.2.3.4");

        let url = url::Url::parse("https://www.avassa.com/a/b/c").unwrap();
        let host_port = url.host_port().unwrap();
        assert_eq!(&host_port, "www.avassa.com");

        let url = url::Url::parse("https://www.avassa.com:1234/a/b/c").unwrap();
        let host_port = url.host_port().unwrap();
        assert_eq!(&host_port, "www.avassa.com:1234");
    }
}