// Copyright 2023 PingCAP, Inc.
// TODO(iosmanthus): introduce TLS support for the whole integration test.

use std::{
    convert::TryFrom,
    error::Error as StdError,
    fmt, fs,
    future::Future,
    io,
    pin::Pin,
    result::Result as StdResult,
    sync::{
        Arc, Mutex,
        atomic::{AtomicUsize, Ordering},
    },
    task::{Context, Poll},
    time::{Duration, SystemTime},
};

use bstr::ByteSlice;
use bytes::Bytes;
use futures_util::{StreamExt, ready};
use http::{Method, Request, StatusCode, Version};
use http_body::Frame;
use http_body_util::{BodyStream, StreamBody};
use hyper::{Body, Uri, client::HttpConnector, server::conn::AddrIncoming};
use hyper_rustls::HttpsConnector;
use hyper_util::{client::legacy::Client as Hyper1Client, rt::TokioExecutor};
use rustls::{
    ServerConnection, SignatureScheme,
    client::ResolvesClientCert,
    crypto::ring::sign::any_supported_type,
    pki_types::{CertificateDer, PrivateKeyDer},
    server::{ClientHello, ResolvesServerCert, WebPkiClientVerifier},
    sign::CertifiedKey,
};
use tikv_util::{Either, box_err, debug, error, time::Instant};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};

use crate::{SecurityConfig, SecurityManager, metrics::HYPER_RELOAD_CERT_COUNTER};

type Hyper1HttpsConnector = HttpsConnector<hyper_util::client::legacy::connect::HttpConnector>;
type Hyper1Body = StreamBody<futures_util::stream::BoxStream<'static, Result<Frame<Bytes>>>>;
type Hyper1HttpsClient = Hyper1Client<Hyper1HttpsConnector, Hyper1Body>;
type ResponseFuture = Pin<Box<dyn Future<Output = Result<hyper::Response<Body>>> + Send>>;

#[derive(Clone, Copy, Default)]
/// Options supported by both the hyper 0.14 HTTP client path and the hyper 1
/// HTTPS client path.
pub struct HttpClientOptions {
    pool_max_idle_per_host: Option<usize>,
}

impl HttpClientOptions {
    /// Sets the maximum idle connections per host kept in the pool.
    pub fn pool_max_idle_per_host(mut self, max_idle: usize) -> Self {
        self.pool_max_idle_per_host = Some(max_idle);
        self
    }

    fn build_http_client(&self) -> hyper::Client<HttpConnector> {
        let mut builder = hyper::Client::builder();
        if let Some(max_idle) = self.pool_max_idle_per_host {
            builder.pool_max_idle_per_host(max_idle);
        }
        builder.build_http()
    }

    fn build_hyper1_client(&self, connector: Hyper1HttpsConnector) -> Hyper1HttpsClient {
        let mut builder = Hyper1Client::builder(TokioExecutor::new());
        if let Some(max_idle) = self.pool_max_idle_per_host {
            builder.pool_max_idle_per_host(max_idle);
        }
        builder.build(connector)
    }
}

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("HTTP error: {0}:{1}")]
    Http(StatusCode, String),
    #[error("request timeout: {0}")]
    Timeout(String),
    #[error(transparent)]
    Hyper(#[from] hyper::Error),
    #[error("HTTP client error: {0}")]
    HyperUtil(#[from] hyper_util::client::legacy::Error),
    #[error("HTTP compatibility error: {0}")]
    Compatibility(String),
    #[error(transparent)]
    InvalidUri(#[from] http::uri::InvalidUri),
    #[error(transparent)]
    Tls(#[from] rustls::Error),
    #[error(transparent)]
    Io(#[from] io::Error),
    #[error(transparent)]
    Json(#[from] serde_json::Error),
    #[error(transparent)]
    Other(#[from] Box<dyn StdError + Sync + Send>),
}

pub type Result<T> = std::result::Result<T, Error>;

impl SecurityManager {
    pub fn acceptor(&self, incoming: AddrIncoming) -> Result<Either<AddrIncoming, TlsAcceptor>> {
        if self.cfg.ca_path.is_empty() {
            Ok(Either::Left(incoming))
        } else {
            Ok(Either::Right(self.tls_acceptor(incoming)?))
        }
    }

    pub fn build_uri<T: AsRef<str>>(&self, rest: T) -> Result<Uri> {
        let scheme = if self.cfg.ca_path.is_empty() {
            "http"
        } else {
            "https"
        };
        Ok(Uri::try_from(format!("{}://{}", scheme, rest.as_ref()))?)
    }

    fn tls_acceptor(&self, incoming: AddrIncoming) -> Result<TlsAcceptor> {
        let ca = load_root_store(&self.cfg.ca_path)?;
        let cert_resolver = ReloadWhenChanged::new(self.cfg.clone())?;

        let provider = rustls_crypto_provider();
        let client_verifier =
            WebPkiClientVerifier::builder_with_provider(Arc::new(ca), provider.clone())
                .allow_unauthenticated()
                .build()
                .map_err(|e| {
                    Error::Other(box_err!("failed to build client cert verifier: {}", e))
                })?;
        let tls_config = rustls::ServerConfig::builder_with_provider(provider)
            .with_safe_default_protocol_versions()?
            .with_client_cert_verifier(client_verifier)
            .with_cert_resolver(cert_resolver);

        let mut tls_config = tls_config;
        tls_config.alpn_protocols = vec![b"http/1.1".to_vec(), b"h2".to_vec()];
        let acceptor = TlsAcceptor::new(incoming, tls_config);

        Ok(acceptor)
    }

    pub fn http_client(&self, client_builder: hyper::client::Builder) -> Result<HttpClient> {
        if self.cfg.ca_path.is_empty() {
            Ok(HttpClient::Http(client_builder.build_http()))
        } else {
            self.http_client_with_options(HttpClientOptions::default())
        }
    }

    /// Builds an HTTP client with options that are preserved across HTTP and
    /// HTTPS modes.
    pub fn http_client_with_options(&self, options: HttpClientOptions) -> Result<HttpClient> {
        if self.cfg.ca_path.is_empty() {
            Ok(HttpClient::Http(options.build_http_client()))
        } else {
            let ca = load_root_store(&self.cfg.ca_path)?;
            let cert_resolver = ReloadWhenChanged::new(self.cfg.clone())?;
            let tls_config = rustls::ClientConfig::builder_with_provider(rustls_crypto_provider())
                .with_safe_default_protocol_versions()?
                .with_root_certificates(ca)
                .with_client_cert_resolver(cert_resolver);

            let connector = hyper_rustls::HttpsConnectorBuilder::new()
                .with_tls_config(tls_config)
                .https_only()
                .enable_http1()
                .build();
            let client = options.build_hyper1_client(connector);
            Ok(HttpClient::Https(client))
        }
    }
}

pub struct TlsAcceptor {
    incoming: AddrIncoming,
    config: Arc<rustls::ServerConfig>,
}

impl TlsAcceptor {
    fn new(incoming: AddrIncoming, config: rustls::ServerConfig) -> Self {
        Self {
            incoming,
            config: Arc::new(config),
        }
    }
}

impl hyper::server::accept::Accept for TlsAcceptor {
    type Conn = TlsStream<hyper::server::conn::AddrStream>;
    type Error = io::Error;

    fn poll_accept(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<std::result::Result<Self::Conn, Self::Error>>> {
        match Pin::new(&mut self.incoming).poll_accept(cx) {
            Poll::Ready(Some(Ok(stream))) => {
                Poll::Ready(Some(Ok(TlsStream::new(stream, self.config.clone()))))
            }
            Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(err))),
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}

pub struct TlsStream<S> {
    state: TlsStreamState<S>,
}

enum TlsStreamState<S> {
    Handshaking(tokio_rustls::Accept<S>),
    Streaming(tokio_rustls::server::TlsStream<S>),
}

impl<S> TlsStream<S>
where
    S: AsyncRead + AsyncWrite + Unpin,
{
    fn new(stream: S, config: Arc<rustls::ServerConfig>) -> Self {
        let accept = tokio_rustls::TlsAcceptor::from(config).accept(stream);
        Self {
            state: TlsStreamState::Handshaking(accept),
        }
    }

    /// Returns the underlying TCP stream when it is still available.
    pub fn io(&self) -> Option<&S> {
        match &self.state {
            TlsStreamState::Handshaking(accept) => accept.get_ref(),
            TlsStreamState::Streaming(stream) => Some(stream.get_ref().0),
        }
    }

    /// Returns the negotiated TLS connection after the handshake has completed.
    pub fn connection(&self) -> Option<&ServerConnection> {
        match &self.state {
            TlsStreamState::Handshaking(_) => None,
            TlsStreamState::Streaming(stream) => Some(stream.get_ref().1),
        }
    }
}

impl<S> AsyncRead for TlsStream<S>
where
    S: AsyncRead + AsyncWrite + Unpin,
{
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        let this = self.get_mut();
        let accept = match &mut this.state {
            TlsStreamState::Handshaking(accept) => accept,
            TlsStreamState::Streaming(stream) => return Pin::new(stream).poll_read(cx, buf),
        };

        let mut stream = ready!(Pin::new(accept).poll(cx))?;
        let result = Pin::new(&mut stream).poll_read(cx, buf);
        this.state = TlsStreamState::Streaming(stream);
        result
    }
}

impl<S> AsyncWrite for TlsStream<S>
where
    S: AsyncRead + AsyncWrite + Unpin,
{
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        let this = self.get_mut();
        let accept = match &mut this.state {
            TlsStreamState::Handshaking(accept) => accept,
            TlsStreamState::Streaming(stream) => return Pin::new(stream).poll_write(cx, buf),
        };

        let mut stream = ready!(Pin::new(accept).poll(cx))?;
        let result = Pin::new(&mut stream).poll_write(cx, buf);
        this.state = TlsStreamState::Streaming(stream);
        result
    }

    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        match &mut self.state {
            TlsStreamState::Handshaking(_) => Poll::Ready(Ok(())),
            TlsStreamState::Streaming(stream) => Pin::new(stream).poll_flush(cx),
        }
    }

    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        match &mut self.state {
            TlsStreamState::Handshaking(_) => Poll::Ready(Ok(())),
            TlsStreamState::Streaming(stream) => Pin::new(stream).poll_shutdown(cx),
        }
    }
}

struct ReloadWhenChanged {
    cfg: Arc<SecurityConfig>,
    latest: Mutex<CertState>,
}

impl fmt::Debug for ReloadWhenChanged {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ReloadWhenChanged").finish_non_exhaustive()
    }
}

struct CertState {
    key: Arc<CertifiedKey>,
    atime: SystemTime,
    mtime: SystemTime,
}

#[derive(Debug)]
enum CertType {
    Server,
    Client,
}

impl ReloadWhenChanged {
    fn new(cfg: Arc<SecurityConfig>) -> StdResult<Arc<Self>, Box<dyn StdError + Sync + Send>> {
        let mtime = fs::metadata(&cfg.cert_path)?.modified()?;
        Ok(Arc::new(Self {
            cfg: cfg.clone(),
            latest: Mutex::new(CertState {
                key: Self::load_certified_key(&cfg)?,
                atime: SystemTime::now(),
                mtime,
            }),
        }))
    }

    fn load_certified_key(
        cfg: &SecurityConfig,
    ) -> std::result::Result<Arc<CertifiedKey>, Box<dyn StdError + Sync + Send>> {
        let cert = load_certs(&cfg.cert_path)?;
        let key = load_key(&cfg.key_path)?;
        Ok(Arc::new(CertifiedKey::new(cert, any_supported_type(&key)?)))
    }

    fn resolve_impl(&self, cert_type: CertType) -> Arc<CertifiedKey> {
        let mut latest = self.latest.lock().unwrap();
        let old_key = latest.key.clone();
        if SystemTime::now()
            .duration_since(latest.atime)
            .unwrap_or_default()
            < self.cfg.cert_reload_interval.0
        {
            return old_key;
        }

        // Update the access time of the cert state.
        latest.atime = SystemTime::now();
        let mut mtime = Some(latest.mtime);
        match self.cfg.is_modified(&mut mtime) {
            Ok(false) => {
                return old_key;
            }
            Err(e) => {
                HYPER_RELOAD_CERT_COUNTER
                    .with_label_values(&["fail", &format!("{:?}", cert_type)])
                    .inc();
                error!( "fail to check mtime of certificates"; "err"=>?e, "cert_type"=>?cert_type);
                return old_key;
            }
            _ => {}
        }

        match Self::load_certified_key(&self.cfg) {
            Ok(new_key) => {
                latest.key = new_key;
                latest.mtime = mtime.unwrap();
                HYPER_RELOAD_CERT_COUNTER
                    .with_label_values(&["success", &format!("{:?}", cert_type)])
                    .inc();
                latest.key.clone()
            }
            Err(e) => {
                HYPER_RELOAD_CERT_COUNTER
                    .with_label_values(&["fail", &format!("{:?}", cert_type)])
                    .inc();
                error!("fail to load certificates";"err"=>?e,"cert_type"=>?cert_type);
                old_key
            }
        }
    }
}

impl ResolvesServerCert for ReloadWhenChanged {
    fn resolve(&self, _: ClientHello<'_>) -> Option<Arc<CertifiedKey>> {
        Some(self.resolve_impl(CertType::Server))
    }
}

impl ResolvesClientCert for ReloadWhenChanged {
    fn resolve(
        &self,
        _acceptable_issuers: &[&[u8]],
        _sigschemes: &[SignatureScheme],
    ) -> Option<Arc<CertifiedKey>> {
        Some(self.resolve_impl(CertType::Client))
    }

    fn has_certs(&self) -> bool {
        true
    }
}

fn error<E>(err: E) -> io::Error
where
    E: Into<Box<dyn StdError + Send + Sync>>,
{
    io::Error::other(err)
}

fn load_certs(filename: &str) -> io::Result<Vec<CertificateDer<'static>>> {
    let certfile = fs::File::open(filename)?;
    let mut reader = io::BufReader::new(certfile);
    rustls_pemfile::certs(&mut reader).collect()
}

fn load_key(filename: &str) -> io::Result<PrivateKeyDer<'static>> {
    let keyfile = fs::File::open(filename)?;
    let mut reader = io::BufReader::new(keyfile);
    rustls_pemfile::private_key(&mut reader)?.ok_or_else(|| error("no private keys found"))
}

fn load_root_store(filename: &str) -> io::Result<rustls::RootCertStore> {
    let ca_certs = load_certs(filename)?;
    let mut store = rustls::RootCertStore::empty();
    for cert in ca_certs {
        store.add(cert).map_err(error)?;
    }
    Ok(store)
}

fn rustls_crypto_provider() -> Arc<rustls::crypto::CryptoProvider> {
    Arc::new(rustls::crypto::ring::default_provider())
}

fn map_http1_version(version: Version) -> http1::Version {
    match version {
        Version::HTTP_09 => http1::Version::HTTP_09,
        Version::HTTP_10 => http1::Version::HTTP_10,
        Version::HTTP_11 => http1::Version::HTTP_11,
        Version::HTTP_2 => http1::Version::HTTP_2,
        Version::HTTP_3 => http1::Version::HTTP_3,
        _ => http1::Version::HTTP_11,
    }
}

fn map_hyper_version(version: http1::Version) -> Version {
    match version {
        http1::Version::HTTP_09 => Version::HTTP_09,
        http1::Version::HTTP_10 => Version::HTTP_10,
        http1::Version::HTTP_11 => Version::HTTP_11,
        http1::Version::HTTP_2 => Version::HTTP_2,
        http1::Version::HTTP_3 => Version::HTTP_3,
        _ => Version::HTTP_11,
    }
}

fn convert_request(req: hyper::Request<Body>) -> Result<http1::Request<Hyper1Body>> {
    let (parts, body) = req.into_parts();
    let method = http1::Method::from_bytes(parts.method.as_str().as_bytes())
        .map_err(|err| Error::Compatibility(format!("invalid method: {err}")))?;
    let uri = parts
        .uri
        .to_string()
        .parse::<http1::Uri>()
        .map_err(|err| Error::Compatibility(format!("invalid URI: {err}")))?;

    let mut builder = http1::Request::builder()
        .method(method)
        .uri(uri)
        .version(map_http1_version(parts.version));
    for (name, value) in parts.headers.iter() {
        let name = http1::header::HeaderName::from_bytes(name.as_str().as_bytes())
            .map_err(|err| Error::Compatibility(format!("invalid header name: {err}")))?;
        let value = http1::HeaderValue::from_bytes(value.as_bytes())
            .map_err(|err| Error::Compatibility(format!("invalid header value: {err}")))?;
        builder = builder.header(name, value);
    }

    let body = StreamBody::new(
        body.map(|chunk| chunk.map(Frame::data).map_err(Error::Hyper))
            .boxed(),
    );
    builder
        .body(body)
        .map_err(|err| Error::Compatibility(format!("failed to build request: {err}")))
}

fn convert_headers(headers: &http1::HeaderMap, target: &mut http::HeaderMap) -> Result<()> {
    for (name, value) in headers.iter() {
        let name = http::header::HeaderName::from_bytes(name.as_str().as_bytes())
            .map_err(|err| Error::Compatibility(format!("invalid response header name: {err}")))?;
        let value = http::HeaderValue::from_bytes(value.as_bytes())
            .map_err(|err| Error::Compatibility(format!("invalid response header value: {err}")))?;
        target.append(name, value);
    }
    Ok(())
}

fn convert_response(
    resp: http1::Response<hyper1::body::Incoming>,
) -> Result<hyper::Response<Body>> {
    let (parts, body) = resp.into_parts();
    let status = StatusCode::from_u16(parts.status.as_u16())
        .map_err(|err| Error::Compatibility(format!("invalid response status: {err}")))?;
    let hyper_body = convert_body(body);

    let mut builder = hyper::Response::builder()
        .status(status)
        .version(map_hyper_version(parts.version));
    let headers = builder
        .headers_mut()
        .ok_or_else(|| Error::Compatibility("missing response headers".to_owned()))?;
    convert_headers(&parts.headers, headers)?;
    builder
        .body(hyper_body)
        .map_err(|err| Error::Compatibility(format!("failed to build response: {err}")))
}

fn convert_body<B>(body: B) -> Body
where
    B: http_body::Body<Data = Bytes, Error = hyper1::Error> + Send + Unpin + 'static,
{
    // Lazily bridge the hyper 1.x body into a hyper 0.14 Body, mirroring
    // convert_request. No detached task, so dropping the consumer drops the
    // upstream body and releases the connection.
    Body::wrap_stream(
        BodyStream::new(body)
            .filter_map(|frame| async move {
                match frame {
                    Ok(frame) => frame.into_data().ok().map(Ok),
                    Err(e) => Some(Err(e)),
                }
            })
            .boxed(),
    )
}

#[derive(Clone)]
pub enum HttpClient {
    Http(hyper::Client<HttpConnector>),
    Https(Hyper1HttpsClient),
}

impl HttpClient {
    pub fn get(&self, uri: Uri) -> ResponseFuture {
        match self {
            HttpClient::Http(client) => {
                let client = client.clone();
                Box::pin(async move { Ok(client.get(uri).await?) })
            }
            HttpClient::Https(_) => {
                let req = hyper::Request::builder()
                    .method(Method::GET)
                    .uri(uri)
                    .body(Body::empty())
                    .unwrap();
                self.request(req)
            }
        }
    }

    pub fn request(&self, req: hyper::http::Request<hyper::Body>) -> ResponseFuture {
        match self {
            HttpClient::Http(client) => {
                let client = client.clone();
                Box::pin(async move { Ok(client.request(req).await?) })
            }
            HttpClient::Https(client) => {
                let client = client.clone();
                Box::pin(async move {
                    let req = convert_request(req)?;
                    let resp = client.request(req).await?;
                    convert_response(resp)
                })
            }
        }
    }
}

pub struct RestfulClient {
    tag: String,
    security_mgr: Arc<SecurityManager>,
    endpoints: Vec<String>,
    last_endpoint_idx: Option<AtomicUsize>,
    req_timeout: Duration,
    retry_timeout: Duration,
    retry_on_server_error: bool,
}

impl Clone for RestfulClient {
    fn clone(&self) -> Self {
        Self {
            tag: self.tag.clone(),
            security_mgr: self.security_mgr.clone(),
            endpoints: self.endpoints.clone(),
            last_endpoint_idx: self
                .last_endpoint_idx
                .as_ref()
                .map(|idx| AtomicUsize::new(idx.load(Ordering::Relaxed))),
            req_timeout: self.req_timeout,
            retry_timeout: self.retry_timeout,
            retry_on_server_error: self.retry_on_server_error,
        }
    }
}

impl RestfulClient {
    pub fn new(
        tag: impl AsRef<str>,
        endpoints: Vec<String>,
        security_mgr: Arc<SecurityManager>,
    ) -> Result<Self> {
        let endpoints = Self::trim_schema(endpoints);
        if endpoints.is_empty() {
            return Err(box_err!("{}: no endpoints configured", tag.as_ref()));
        }
        let endpoint_idx = (endpoints.len() > 1).then(|| AtomicUsize::new(0));
        Ok(Self {
            tag: tag.as_ref().to_owned(),
            endpoints,
            security_mgr,
            last_endpoint_idx: endpoint_idx,
            req_timeout: Duration::from_secs(3),
            retry_timeout: Duration::from_secs(6),
            retry_on_server_error: true,
        })
    }

    pub fn set_timeout(&mut self, req_timeout: Duration, retry_timeout: Duration) {
        self.req_timeout = req_timeout;
        self.retry_timeout = retry_timeout;
    }

    pub fn set_retry_on_server_error(&mut self, retry: bool) {
        self.retry_on_server_error = retry;
    }

    pub async fn request(
        &self,
        path: impl AsRef<str>,
        method: Method,
        body_data: Option<Bytes>,
    ) -> Result<Bytes> {
        let path = path.as_ref();
        let client = self.security_mgr.http_client(hyper::Client::builder())?;
        let mut err = None;
        let last_ep_idx = self.last_endpoint_idx();
        let mut idx = 0;
        let start_time = Instant::now_coarse();
        while start_time.saturating_elapsed() < self.retry_timeout {
            let current_ep_idx = (last_ep_idx + idx) % self.endpoints.len();
            idx += 1;
            let endpoint = &self.endpoints[current_ep_idx];

            let uri = self.security_mgr.build_uri(format!("{endpoint}/{path}"))?;
            let tag = format!("{}:{}:{}", self.tag, method.as_str(), uri);
            let req = Request::builder()
                .method(method.clone())
                .uri(uri)
                .body(match body_data {
                    Some(ref data) => Body::from(data.clone()),
                    None => Body::empty(),
                })
                .unwrap();
            let resp = tokio::time::timeout(self.req_timeout, client.request(req)).await;
            match resp {
                Err(_elapsed) => err = Some(Error::Timeout(format!("{}: request timeout", tag))),
                Ok(Err(e)) => err = Some(e),
                Ok(Ok(resp)) => {
                    let status = resp.status();
                    let body_res = tokio::time::timeout(
                        self.req_timeout,
                        hyper::body::to_bytes(resp.into_body()),
                    )
                    .await;
                    if status.is_success() {
                        let Ok(body) = body_res else {
                            err = Some(Error::Timeout(format!("{}: read body timeout", tag)));
                            continue;
                        };
                        let body = match body {
                            Ok(body) => body,
                            Err(e) => {
                                err = Some(Error::Hyper(e));
                                continue;
                            }
                        };
                        debug!("{}: success: {}", tag, body.to_str_lossy());
                        if current_ep_idx != last_ep_idx {
                            self.set_last_endpoint_idx(current_ep_idx);
                        }
                        return Ok(body);
                    } else {
                        // Body is only used for the error message, so a read
                        // failure falls back to an empty message.
                        let body = body_res.ok().and_then(|body| body.ok()).unwrap_or_default();
                        let msg = body.to_str_lossy();
                        error!("{}: http client error: {}: {}", tag, status, msg.as_ref());
                        let this_err = Error::Http(status, msg.to_string());
                        if status.is_client_error() || !self.retry_on_server_error {
                            return Err(this_err);
                        } else {
                            err = Some(this_err);
                        }
                    }
                }
            }
        }
        Err(err.unwrap_or_else(|| {
            box_err!(
                "{}:{}:{}: request failed within retry timeout {:?}",
                self.tag,
                method.as_str(),
                path,
                self.retry_timeout
            )
        }))
    }

    pub async fn request_with_status(
        &self,
        path: impl AsRef<str>,
        method: Method,
        body_data: Option<Bytes>,
    ) -> Result<(StatusCode, Bytes)> {
        let path = path.as_ref();
        let client = self.security_mgr.http_client(hyper::Client::builder())?;
        let mut err = None;
        let last_ep_idx = self.last_endpoint_idx();
        let mut idx = 0;
        let start_time = Instant::now_coarse();
        while start_time.saturating_elapsed() < self.retry_timeout {
            let current_ep_idx = (last_ep_idx + idx) % self.endpoints.len();
            idx += 1;
            let endpoint = &self.endpoints[current_ep_idx];

            let uri = self.security_mgr.build_uri(format!("{endpoint}/{path}"))?;
            let tag = format!("{}:{}:{}", self.tag, method.as_str(), uri);
            let req = Request::builder()
                .method(method.clone())
                .uri(uri)
                .body(match body_data {
                    Some(ref data) => Body::from(data.clone()),
                    None => Body::empty(),
                })
                .unwrap();
            let resp = tokio::time::timeout(self.req_timeout, client.request(req)).await;
            match resp {
                Err(_elapsed) => err = Some(Error::Timeout(format!("{}: request timeout", tag))),
                Ok(Err(e)) => err = Some(box_err!("{tag}: return error: {}", e.to_string())),
                Ok(Ok(resp)) => {
                    let status = resp.status();
                    let body_res = tokio::time::timeout(
                        self.req_timeout,
                        hyper::body::to_bytes(resp.into_body()),
                    )
                    .await;
                    if status.is_success() {
                        // A success needs the full body, so a read failure is
                        // retryable.
                        let Ok(body) = body_res else {
                            err = Some(Error::Timeout(format!("{}: read body timeout", tag)));
                            continue;
                        };
                        let body = match body {
                            Ok(body) => body,
                            Err(e) => {
                                err = Some(Error::Hyper(e));
                                continue;
                            }
                        };
                        debug!("{}: success: {}", tag, body.to_str_lossy());
                        if current_ep_idx != last_ep_idx {
                            self.set_last_endpoint_idx(current_ep_idx);
                        }
                        return Ok((status, body));
                    } else if status.is_client_error() {
                        // The status is authoritative, so a body-read failure
                        // falls back to an empty body.
                        let body = body_res.ok().and_then(|body| body.ok()).unwrap_or_default();
                        debug!("{}: client error: {}", tag, body.to_str_lossy());
                        return Ok((status, body));
                    } else {
                        let body = body_res.ok().and_then(|body| body.ok()).unwrap_or_default();
                        err = Some(box_err!(
                            "{tag}: return error: {status}: {}",
                            body.to_str_lossy()
                        ));
                    }
                }
            }
        }
        Err(err.unwrap_or_else(|| {
            box_err!(
                "{}:{}:{}: request_with_status failed within retry timeout {:?}",
                self.tag,
                method.as_str(),
                path,
                self.retry_timeout
            )
        }))
    }

    pub async fn get_with_status<T>(&self, path: impl AsRef<str>) -> Result<(StatusCode, Option<T>)>
    where
        T: std::fmt::Debug + for<'a> serde::de::Deserialize<'a>,
    {
        let path = path.as_ref();
        let (status, resp) = self.request_with_status(path, Method::GET, None).await?;
        if status.is_success() {
            let t: T = serde_json::from_slice(&resp)?;
            debug!("{}: get", self.tag; "path" => path, "resp" => ?t);
            Ok((status, Some(t)))
        } else {
            Ok((status, None))
        }
    }

    pub async fn get<T>(&self, path: impl AsRef<str>) -> Result<T>
    where
        T: std::fmt::Debug + for<'a> serde::de::Deserialize<'a>,
    {
        let path = path.as_ref();
        match self.request(path, Method::GET, None).await {
            Ok(resp) => {
                let t: T = serde_json::from_slice(&resp)?;
                debug!("{}: get", self.tag; "path" => path, "resp" => ?t);
                Ok(t)
            }
            Err(err) => Err(err),
        }
    }

    pub async fn post<Req, Resp>(&self, path: impl AsRef<str>, data: &Req) -> Result<Resp>
    where
        Req: serde::ser::Serialize,
        Resp: std::fmt::Debug + for<'a> serde::de::Deserialize<'a>,
    {
        self.post_or_put(Method::POST, path, data).await
    }

    pub async fn put<Req, Resp>(&self, path: impl AsRef<str>, data: &Req) -> Result<Resp>
    where
        Req: serde::ser::Serialize,
        Resp: std::fmt::Debug + for<'a> serde::de::Deserialize<'a>,
    {
        self.post_or_put(Method::PUT, path, data).await
    }

    async fn post_or_put<Req, Resp>(
        &self,
        method: Method,
        path: impl AsRef<str>,
        data: &Req,
    ) -> Result<Resp>
    where
        Req: serde::ser::Serialize,
        Resp: std::fmt::Debug + for<'a> serde::de::Deserialize<'a>,
    {
        let path = path.as_ref();
        let body_data = Bytes::from(serde_json::to_vec(data)?);
        match self.request(path, method, Some(body_data)).await {
            Ok(resp) => {
                let t: Resp = serde_json::from_slice(&resp)?;
                debug!("{}: post", self.tag; "path" => path, "resp" => ?t);
                Ok(t)
            }
            Err(err) => Err(err),
        }
    }

    pub async fn delete(&self, path: impl AsRef<str>) -> Result<Bytes> {
        let path = path.as_ref();
        match self.request(path, Method::DELETE, None).await {
            Ok(resp) => {
                debug!("{}: delete", self.tag; "path" => path, "resp" => resp.to_str_lossy().as_ref());
                Ok(resp)
            }
            Err(err) => Err(err),
        }
    }

    fn trim_schema(mut endpoints: Vec<String>) -> Vec<String> {
        endpoints.iter_mut().for_each(|endpoint| {
            for prefix in &["http://", "https://"] {
                if endpoint.starts_with(prefix) {
                    endpoint.replace_range(0..prefix.len(), "");
                    break;
                }
            }
        });
        endpoints
    }

    fn last_endpoint_idx(&self) -> usize {
        self.last_endpoint_idx
            .as_ref()
            .map_or(0, |idx| idx.load(Ordering::Relaxed))
    }

    fn set_last_endpoint_idx(&self, ep_idx: usize) {
        if let Some(idx) = self.last_endpoint_idx.as_ref() {
            idx.store(ep_idx, Ordering::Relaxed);
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{
        io::{Read, Write},
        net::{SocketAddr, TcpListener, TcpStream},
        sync::atomic::AtomicBool,
        thread,
    };

    use super::*;

    // A hyper 1.x body that never yields a frame and flips a flag when dropped.
    struct StallBody(Arc<AtomicBool>);

    impl Drop for StallBody {
        fn drop(&mut self) {
            self.0.store(true, Ordering::SeqCst);
        }
    }

    impl http_body::Body for StallBody {
        type Data = Bytes;
        type Error = hyper1::Error;

        fn poll_frame(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
        ) -> Poll<Option<StdResult<Frame<Bytes>, hyper1::Error>>> {
            Poll::Pending
        }
    }

    // Dropping the converted body must drop the upstream body. Otherwise a
    // stalled HTTPS response leaks the bridge that owns the upstream connection.
    #[tokio::test]
    async fn test_convert_body_releases_upstream_when_consumer_dropped() {
        let dropped = Arc::new(AtomicBool::new(false));
        let hyper_body = convert_body(StallBody(dropped.clone()));
        drop(hyper_body);
        tokio::task::yield_now().await;
        assert!(
            dropped.load(Ordering::SeqCst),
            "converting the body must not leak a task that owns the upstream body"
        );
    }

    #[test]
    fn test_restful_client() {
        let endpoints = vec![
            "127.0.0.1".to_string(),
            "http://127.0.0.2:4379".to_string(),
            "https://127.0.0.3:4380".to_string(),
        ];
        let endpoints = RestfulClient::trim_schema(endpoints);
        assert_eq!(
            endpoints,
            vec![
                "127.0.0.1".to_string(),
                "127.0.0.2:4379".to_string(),
                "127.0.0.3:4380".to_string(),
            ]
        );
    }

    // Runs `respond` for each accepted connection and returns the server
    // address. Body-read tests use it to send crafted or stalled responses.
    fn spawn_server(respond: impl Fn(&mut TcpStream) + Send + 'static) -> SocketAddr {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        thread::spawn(move || {
            for stream in listener.incoming() {
                let Ok(mut stream) = stream else { break };
                let mut buf = [0u8; 1024];
                let _ = stream.read(&mut buf);
                respond(&mut stream);
            }
        });
        addr
    }

    fn test_client(
        addr: SocketAddr,
        req_timeout: Duration,
        retry_timeout: Duration,
    ) -> RestfulClient {
        let mgr = Arc::new(SecurityManager::new(&SecurityConfig::default()).unwrap());
        let mut client = RestfulClient::new("test", vec![addr.to_string()], mgr).unwrap();
        client.set_timeout(req_timeout, retry_timeout);
        client
    }

    // Promise 100 body bytes but write 5 then close, so the body read fails with
    // an incomplete-message error.
    const TRUNCATED_200: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\nshort";

    // A truncated response body must surface as Err, not panic the caller.
    // ServiceLimitFetcher runs request_with_status on a production background
    // thread, so a body-read failure there previously crashed that thread.
    #[tokio::test]
    async fn test_request_with_status_truncated_body_returns_err_not_panic() {
        let addr = spawn_server(|s| {
            let _ = s.write_all(TRUNCATED_200);
        });
        let client = test_client(addr, Duration::from_secs(1), Duration::from_millis(200));
        let res = client.request_with_status("path", Method::GET, None).await;
        assert!(
            matches!(res, Err(Error::Hyper(_))),
            "truncated body should return a Hyper body error, got {:?}",
            res
        );
    }

    // request() shares the same body-read path and must also return Err.
    #[tokio::test]
    async fn test_request_truncated_body_returns_err_not_panic() {
        let addr = spawn_server(|s| {
            let _ = s.write_all(TRUNCATED_200);
        });
        let client = test_client(addr, Duration::from_secs(1), Duration::from_millis(200));
        let res = client.request("path", Method::GET, None).await;
        assert!(
            matches!(res, Err(Error::Hyper(_))),
            "truncated body should return a Hyper body error, got {:?}",
            res
        );
    }

    // A non-success status whose body then fails must not panic while building
    // the error message from that body.
    #[tokio::test]
    async fn test_request_error_status_truncated_body_returns_err_not_panic() {
        let addr = spawn_server(|s| {
            let _ = s
                .write_all(b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 100\r\n\r\nshort");
        });
        let client = test_client(addr, Duration::from_secs(1), Duration::from_millis(200));
        let res = client.request("path", Method::GET, None).await;
        assert!(
            matches!(res, Err(Error::Http(status, _)) if status == StatusCode::SERVICE_UNAVAILABLE),
            "5xx with truncated body should return the HTTP status error, got {:?}",
            res
        );
    }

    // A body that never arrives must time out, not hang the caller.
    #[tokio::test]
    async fn test_request_with_status_stalled_body_times_out() {
        // Send headers promising a body, then hold the connection without
        // sending it; without the read timeout the caller would hang.
        let addr = spawn_server(|s| {
            let _ = s.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n");
            thread::sleep(Duration::from_secs(3));
        });
        let client = test_client(addr, Duration::from_millis(300), Duration::from_millis(50));
        let start = Instant::now_coarse();
        let res = client.request_with_status("path", Method::GET, None).await;
        assert!(
            matches!(res, Err(Error::Timeout(_))),
            "stalled body should time out, got {:?}",
            res
        );
        assert!(
            start.saturating_elapsed() < Duration::from_secs(2),
            "body read must time out instead of hanging"
        );
    }

    // A peer that accepts the request but never sends response headers must time
    // out at the request phase, not hang the caller before the body is reached.
    #[tokio::test]
    async fn test_request_with_status_no_response_headers_times_out() {
        let addr = spawn_server(|_s| {
            // Read the request but send nothing back, holding the connection so
            // the client waits for response headers that never arrive.
            thread::sleep(Duration::from_secs(3));
        });
        let client = test_client(addr, Duration::from_millis(300), Duration::from_millis(50));
        let start = Instant::now_coarse();
        let res = client.request_with_status("path", Method::GET, None).await;
        assert!(
            matches!(res, Err(Error::Timeout(_))),
            "missing response headers should time out, got {:?}",
            res
        );
        assert!(
            start.saturating_elapsed() < Duration::from_secs(2),
            "request must time out instead of hanging"
        );
    }

    // A client-error status must be preserved even if its body read fails, since
    // get_with_status treats a status like 404 as data, not an error to retry.
    #[tokio::test]
    async fn test_request_with_status_client_error_truncated_body_preserves_status() {
        let addr = spawn_server(|s| {
            let _ = s.write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 100\r\n\r\nshort");
        });
        let client = test_client(addr, Duration::from_secs(1), Duration::from_millis(200));
        let res = client.request_with_status("path", Method::GET, None).await;
        assert!(
            matches!(res, Ok((status, _)) if status == StatusCode::NOT_FOUND),
            "client-error status must be preserved on body failure, got {:?}",
            res
        );
    }
}