#[cfg(feature = "_async")]
pub(crate) mod async_transport;
#[cfg(feature = "_blocking")]
pub(crate) mod blocking_transport;

use std::collections::BTreeSet;
use std::io;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

#[cfg(any(feature = "_blocking", test))]
use std::io::Read;

#[cfg(feature = "_async")]
use bytes::Bytes;
#[cfg(feature = "_async")]
use futures_core::Stream;
use http::header::{CONTENT_LENGTH, CONTENT_TYPE, HeaderValue, RETRY_AFTER};
use http::{HeaderMap, StatusCode};
use reqx::advanced::RateLimitPolicy;
use reqx::prelude::{RedirectPolicy, RetryPolicy, StatusPolicy};
use serde::Deserialize;
use serde::Serialize;
use serde::de::DeserializeOwned;
use serde_json::Value;

use crate::Error;
use crate::client::{RateLimitConfig, RequestDefaults, RetryConfig};
use crate::types::common::ResponseParameters;
use crate::types::upload::UploadFile;
use crate::util::{
    body_snippet, build_api_path, redact_token, request_id_from_headers, validate_method_name,
};

static LOCAL_REQUEST_ID_SEQ: AtomicU64 = AtomicU64::new(1);

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum TransportRetryMode {
    Inherit,
    Disabled,
}

#[derive(Clone, Copy, Debug)]
pub(crate) struct TransportRequestConfig<'a> {
    defaults: &'a RequestDefaults,
    retry_mode: TransportRetryMode,
}

impl<'a> TransportRequestConfig<'a> {
    pub(crate) fn new(defaults: &'a RequestDefaults, retry_mode: TransportRetryMode) -> Self {
        Self {
            defaults,
            retry_mode,
        }
    }

    pub(crate) fn defaults(&self) -> &'a RequestDefaults {
        self.defaults
    }

    pub(crate) fn retry_mode(&self) -> TransportRetryMode {
        self.retry_mode
    }
}

fn is_configuration_error_code(code: reqx::ErrorCode) -> bool {
    matches!(
        code,
        reqx::ErrorCode::InvalidUri
            | reqx::ErrorCode::InvalidNoProxyRule
            | reqx::ErrorCode::InvalidProxyConfig
            | reqx::ErrorCode::InvalidAdaptiveConcurrencyPolicy
            | reqx::ErrorCode::RequestBuild
            | reqx::ErrorCode::InvalidHeaderName
            | reqx::ErrorCode::InvalidHeaderValue
            | reqx::ErrorCode::TlsBackendUnavailable
            | reqx::ErrorCode::TlsBackendInit
            | reqx::ErrorCode::TlsConfig
    )
}

#[derive(Debug, Deserialize)]
struct TelegramEnvelope<T> {
    ok: bool,
    result: Option<T>,
    description: Option<String>,
    error_code: Option<i64>,
    parameters: Option<ResponseParameters>,
}

pub(crate) fn parse_telegram_response<T>(
    method: &str,
    status: StatusCode,
    headers: &HeaderMap,
    body: &[u8],
    capture_body_snippet: bool,
    snippet_limit: usize,
) -> Result<T, Error>
where
    T: DeserializeOwned,
{
    let request_id = request_id_from_headers(headers);
    let snippet = capture_body_snippet
        .then(|| body_snippet(body, snippet_limit))
        .flatten();

    let envelope: TelegramEnvelope<T> =
        serde_json::from_slice(body).map_err(|source| Error::DeserializeResponse {
            method: method.to_owned(),
            status: Some(status.as_u16()),
            request_id: request_id.clone().map(Into::into),
            body_snippet: snippet.clone().map(Into::into),
            source,
        })?;

    if envelope.ok {
        if let Some(result) = envelope.result {
            return Ok(result);
        }

        return Err(Error::MissingResult {
            method: method.to_owned(),
            status: Some(status.as_u16()),
            request_id: request_id.map(Into::into),
            body_snippet: snippet.map(Into::into),
        });
    }

    let mut parameters = envelope.parameters;
    if let Some(retry_after) = retry_after_seconds_from_headers(headers) {
        let parameters = parameters.get_or_insert_with(ResponseParameters::default);
        if parameters.retry_after.is_none() {
            parameters.retry_after = Some(retry_after);
        }
    }

    Err(Error::Api {
        method: method.to_owned(),
        status: Some(status.as_u16()),
        request_id: request_id.map(Into::into),
        error_code: envelope.error_code,
        description: envelope
            .description
            .unwrap_or_else(|| "telegram api returned an unknown error".to_owned())
            .into(),
        parameters: parameters.map(Box::new),
        body_snippet: snippet.map(Into::into),
    })
}

fn retry_after_seconds_from_headers(headers: &HeaderMap) -> Option<u64> {
    retry_after_seconds_from_headers_at(headers, SystemTime::now())
}

fn retry_after_seconds_from_headers_at(headers: &HeaderMap, now: SystemTime) -> Option<u64> {
    let value = headers.get(RETRY_AFTER)?.to_str().ok()?.trim();
    if let Ok(seconds) = value.parse::<u64>() {
        return Some(seconds);
    }

    let deadline = httpdate::parse_http_date(value).ok()?;
    let delay = deadline.duration_since(now).unwrap_or(Duration::ZERO);
    Some(delay.as_secs() + u64::from(delay.subsec_nanos() > 0))
}

pub(crate) fn local_transport_request_id(method: &str) -> String {
    let sequence = LOCAL_REQUEST_ID_SEQ.fetch_add(1, Ordering::Relaxed);
    let timestamp = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0_u128, |duration| duration.as_millis());
    format!("tele-{method}-{timestamp:x}-{sequence:x}")
}

pub(crate) fn map_reqx_error(
    method: &str,
    token: &str,
    source: reqx::Error,
    fallback_request_id: Option<String>,
) -> Error {
    let code = source.code();
    if is_configuration_error_code(code) {
        return Error::Configuration {
            reason: format!(
                "while calling `{method}`: {} [{}]",
                redact_token(&source.to_string(), token),
                code.as_str()
            ),
        };
    }

    let request_path = source.request_path().map(|path| redact_token(&path, token));

    Error::Transport {
        method: method.to_owned(),
        status: source.status_code(),
        request_id: source
            .request_id()
            .map(ToOwned::to_owned)
            .or(fallback_request_id)
            .map(Into::into),
        retry_after: source.retry_after(SystemTime::now()),
        request_path: request_path.map(Into::into),
        message: redact_token(&source.to_string(), token).into(),
    }
}

pub(crate) fn map_reqx_builder_error(source: reqx::Error) -> Error {
    Error::Configuration {
        reason: format!(
            "failed to build HTTP client: {} [{}]",
            source,
            source.code().as_str()
        ),
    }
}

pub(crate) fn to_retry_policy(retry: &RetryConfig) -> RetryPolicy {
    RetryPolicy::standard()
        .max_attempts(retry.max_attempts)
        .base_backoff(retry.base_backoff)
        .max_backoff(retry.max_backoff)
        .jitter_ratio(retry.jitter_ratio)
}

pub(crate) fn to_rate_limit_policy(config: &RateLimitConfig) -> RateLimitPolicy {
    RateLimitPolicy::standard()
        .requests_per_second(config.requests_per_second)
        .burst(config.burst)
        .max_throttle_delay(config.max_throttle_delay)
}

trait TransportClientBuilder: Sized {
    type Client;

    fn request_timeout(self, timeout: Duration) -> Self;
    fn connect_timeout(self, timeout: Duration) -> Self;
    fn max_response_body_bytes(self, max_response_body_bytes: usize) -> Self;
    fn default_status_policy(self, default_status_policy: StatusPolicy) -> Self;
    fn redirect_policy(self, redirect_policy: RedirectPolicy) -> Self;
    fn retry_policy(self, retry_policy: RetryPolicy) -> Self;
    fn client_name(self, client_name: &'static str) -> Self;
    fn total_timeout(self, total_timeout: Duration) -> Self;
    fn max_in_flight(self, max_in_flight: usize) -> Self;
    fn max_in_flight_per_host(self, max_in_flight_per_host: usize) -> Self;
    fn global_rate_limit_policy(self, global_rate_limit_policy: RateLimitPolicy) -> Self;
    fn per_host_rate_limit_policy(self, per_host_rate_limit_policy: RateLimitPolicy) -> Self;
    fn allow_non_idempotent_retries(self, enabled: bool) -> Self;
    fn http_proxy(self, proxy_uri: http::Uri) -> Self;
    fn proxy_authorization(self, proxy_authorization: HeaderValue) -> Self;
    fn try_add_no_proxy(self, rule: &str) -> reqx::Result<Self>;
    fn try_default_header(self, name: &str, value: &str) -> reqx::Result<Self>;
    fn build(self) -> reqx::Result<Self::Client>;
}

#[cfg(feature = "_async")]
impl TransportClientBuilder for reqx::ClientBuilder {
    type Client = reqx::Client;

    fn request_timeout(self, timeout: Duration) -> Self {
        reqx::ClientBuilder::request_timeout(self, timeout)
    }

    fn connect_timeout(self, timeout: Duration) -> Self {
        reqx::ClientBuilder::connect_timeout(self, timeout)
    }

    fn max_response_body_bytes(self, max_response_body_bytes: usize) -> Self {
        reqx::ClientBuilder::max_response_body_bytes(self, max_response_body_bytes)
    }

    fn default_status_policy(self, default_status_policy: StatusPolicy) -> Self {
        reqx::ClientBuilder::default_status_policy(self, default_status_policy)
    }

    fn redirect_policy(self, redirect_policy: RedirectPolicy) -> Self {
        reqx::ClientBuilder::redirect_policy(self, redirect_policy)
    }

    fn retry_policy(self, retry_policy: RetryPolicy) -> Self {
        reqx::ClientBuilder::retry_policy(self, retry_policy)
    }

    fn client_name(self, client_name: &'static str) -> Self {
        reqx::ClientBuilder::client_name(self, client_name)
    }

    fn total_timeout(self, total_timeout: Duration) -> Self {
        reqx::ClientBuilder::total_timeout(self, total_timeout)
    }

    fn max_in_flight(self, max_in_flight: usize) -> Self {
        reqx::ClientBuilder::max_in_flight(self, max_in_flight)
    }

    fn max_in_flight_per_host(self, max_in_flight_per_host: usize) -> Self {
        reqx::ClientBuilder::max_in_flight_per_host(self, max_in_flight_per_host)
    }

    fn global_rate_limit_policy(self, global_rate_limit_policy: RateLimitPolicy) -> Self {
        reqx::ClientBuilder::global_rate_limit_policy(self, global_rate_limit_policy)
    }

    fn per_host_rate_limit_policy(self, per_host_rate_limit_policy: RateLimitPolicy) -> Self {
        reqx::ClientBuilder::per_host_rate_limit_policy(self, per_host_rate_limit_policy)
    }

    fn allow_non_idempotent_retries(self, enabled: bool) -> Self {
        reqx::ClientBuilder::allow_non_idempotent_retries(self, enabled)
    }

    fn http_proxy(self, proxy_uri: http::Uri) -> Self {
        reqx::ClientBuilder::http_proxy(self, proxy_uri)
    }

    fn proxy_authorization(self, proxy_authorization: HeaderValue) -> Self {
        reqx::ClientBuilder::proxy_authorization(self, proxy_authorization)
    }

    fn try_add_no_proxy(self, rule: &str) -> reqx::Result<Self> {
        reqx::ClientBuilder::try_add_no_proxy(self, rule)
    }

    fn try_default_header(self, name: &str, value: &str) -> reqx::Result<Self> {
        reqx::ClientBuilder::try_default_header(self, name, value)
    }

    fn build(self) -> reqx::Result<Self::Client> {
        reqx::ClientBuilder::build(self)
    }
}

#[cfg(feature = "_blocking")]
impl TransportClientBuilder for reqx::blocking::ClientBuilder {
    type Client = reqx::blocking::Client;

    fn request_timeout(self, timeout: Duration) -> Self {
        reqx::blocking::ClientBuilder::request_timeout(self, timeout)
    }

    fn connect_timeout(self, timeout: Duration) -> Self {
        reqx::blocking::ClientBuilder::connect_timeout(self, timeout)
    }

    fn max_response_body_bytes(self, max_response_body_bytes: usize) -> Self {
        reqx::blocking::ClientBuilder::max_response_body_bytes(self, max_response_body_bytes)
    }

    fn default_status_policy(self, default_status_policy: StatusPolicy) -> Self {
        reqx::blocking::ClientBuilder::default_status_policy(self, default_status_policy)
    }

    fn redirect_policy(self, redirect_policy: RedirectPolicy) -> Self {
        reqx::blocking::ClientBuilder::redirect_policy(self, redirect_policy)
    }

    fn retry_policy(self, retry_policy: RetryPolicy) -> Self {
        reqx::blocking::ClientBuilder::retry_policy(self, retry_policy)
    }

    fn client_name(self, client_name: &'static str) -> Self {
        reqx::blocking::ClientBuilder::client_name(self, client_name)
    }

    fn total_timeout(self, total_timeout: Duration) -> Self {
        reqx::blocking::ClientBuilder::total_timeout(self, total_timeout)
    }

    fn max_in_flight(self, max_in_flight: usize) -> Self {
        reqx::blocking::ClientBuilder::max_in_flight(self, max_in_flight)
    }

    fn max_in_flight_per_host(self, max_in_flight_per_host: usize) -> Self {
        reqx::blocking::ClientBuilder::max_in_flight_per_host(self, max_in_flight_per_host)
    }

    fn global_rate_limit_policy(self, global_rate_limit_policy: RateLimitPolicy) -> Self {
        reqx::blocking::ClientBuilder::global_rate_limit_policy(self, global_rate_limit_policy)
    }

    fn per_host_rate_limit_policy(self, per_host_rate_limit_policy: RateLimitPolicy) -> Self {
        reqx::blocking::ClientBuilder::per_host_rate_limit_policy(self, per_host_rate_limit_policy)
    }

    fn allow_non_idempotent_retries(self, enabled: bool) -> Self {
        reqx::blocking::ClientBuilder::allow_non_idempotent_retries(self, enabled)
    }

    fn http_proxy(self, proxy_uri: http::Uri) -> Self {
        reqx::blocking::ClientBuilder::http_proxy(self, proxy_uri)
    }

    fn proxy_authorization(self, proxy_authorization: HeaderValue) -> Self {
        reqx::blocking::ClientBuilder::proxy_authorization(self, proxy_authorization)
    }

    fn try_add_no_proxy(self, rule: &str) -> reqx::Result<Self> {
        reqx::blocking::ClientBuilder::try_add_no_proxy(self, rule)
    }

    fn try_default_header(self, name: &str, value: &str) -> reqx::Result<Self> {
        reqx::blocking::ClientBuilder::try_default_header(self, name, value)
    }

    fn build(self) -> reqx::Result<Self::Client> {
        reqx::blocking::ClientBuilder::build(self)
    }
}

fn build_transport_client<B>(
    builder: B,
    defaults: &RequestDefaults,
    default_headers: &[(String, String)],
) -> Result<B::Client, Error>
where
    B: TransportClientBuilder,
{
    let mut builder = builder
        .request_timeout(defaults.request_timeout)
        .connect_timeout(defaults.connect_timeout)
        .max_response_body_bytes(defaults.max_response_body_bytes)
        .default_status_policy(StatusPolicy::Response)
        .redirect_policy(RedirectPolicy::none())
        .retry_policy(to_retry_policy(&defaults.retry))
        .client_name("tele");

    if let Some(total_timeout) = defaults.total_timeout {
        builder = builder.total_timeout(total_timeout);
    }

    if let Some(max_in_flight) = defaults.max_in_flight {
        builder = builder.max_in_flight(max_in_flight);
    }

    if let Some(max_in_flight_per_host) = defaults.max_in_flight_per_host {
        builder = builder.max_in_flight_per_host(max_in_flight_per_host);
    }

    if let Some(global_rate_limit) = defaults.global_rate_limit.as_ref() {
        builder = builder.global_rate_limit_policy(to_rate_limit_policy(global_rate_limit));
    }

    if let Some(per_host_rate_limit) = defaults.per_host_rate_limit.as_ref() {
        builder = builder.per_host_rate_limit_policy(to_rate_limit_policy(per_host_rate_limit));
    }

    if defaults.retry.allow_non_idempotent_retries {
        builder = builder.allow_non_idempotent_retries(true);
    }

    if let Some(http_proxy) = defaults.http_proxy.clone() {
        builder = builder.http_proxy(http_proxy);
    }

    if let Some(proxy_authorization) = defaults.proxy_authorization.clone() {
        builder = builder.proxy_authorization(proxy_authorization);
    }

    for rule in &defaults.no_proxy_rules {
        builder = builder
            .try_add_no_proxy(rule)
            .map_err(map_reqx_builder_error)?;
    }

    for (name, value) in default_headers {
        builder = builder
            .try_default_header(name, value)
            .map_err(map_reqx_builder_error)?;
    }

    builder.build().map_err(map_reqx_builder_error)
}

pub(crate) struct PreparedTelegramCall<'a> {
    method: &'a str,
    token: &'a str,
    path: String,
    fallback_request_id: String,
}

impl<'a> PreparedTelegramCall<'a> {
    pub(crate) fn new(method: &'a str, token: &'a str) -> Result<Self, Error> {
        validate_method_name(method)?;

        Ok(Self {
            method,
            token,
            path: build_api_path(token, method),
            fallback_request_id: local_transport_request_id(method),
        })
    }

    pub(crate) fn path(&self) -> &str {
        &self.path
    }

    pub(crate) fn map_transport_error(&self, source: reqx::Error) -> Error {
        map_reqx_error(
            self.method,
            self.token,
            source,
            Some(self.fallback_request_id.clone()),
        )
    }

    pub(crate) fn parse_response<R>(
        &self,
        response: reqx::Response,
        defaults: &RequestDefaults,
    ) -> Result<R, Error>
    where
        R: DeserializeOwned,
    {
        parse_telegram_response(
            self.method,
            response.status(),
            response.headers(),
            response.body(),
            defaults.capture_body_snippet,
            defaults.body_snippet_limit,
        )
    }
}

pub(crate) fn multipart_header_values(
    payload: &MultipartPayload,
) -> Result<(HeaderValue, HeaderValue), Error> {
    let content_type = HeaderValue::from_str(payload.content_type()).map_err(|source| {
        Error::InvalidHeaderValue {
            name: CONTENT_TYPE.as_str().to_owned(),
            source,
        }
    })?;
    let content_length =
        HeaderValue::from_str(&payload.content_length().to_string()).map_err(|source| {
            Error::InvalidHeaderValue {
                name: CONTENT_LENGTH.as_str().to_owned(),
                source,
            }
        })?;

    Ok((content_type, content_length))
}

pub(crate) fn serialize_multipart_fields<P>(
    payload: &P,
    skip_fields: &[&str],
) -> Result<Vec<(String, String)>, Error>
where
    P: Serialize + ?Sized,
{
    let skip: BTreeSet<&str> = skip_fields.iter().copied().collect();

    let value =
        serde_json::to_value(payload).map_err(|source| Error::SerializeRequest { source })?;
    let object = value.as_object().ok_or_else(|| Error::InvalidRequest {
        reason: "multipart payload must serialize into an object".to_owned(),
    })?;

    let mut fields = Vec::new();
    for (key, value) in object {
        if skip.contains(key.as_str()) {
            continue;
        }

        if let Some(value) = value_to_form_string(value)? {
            fields.push((key.clone(), value));
        }
    }

    Ok(fields)
}

fn value_to_form_string(value: &Value) -> Result<Option<String>, Error> {
    match value {
        Value::Null => Ok(None),
        Value::String(value) => Ok(Some(value.clone())),
        Value::Bool(value) => Ok(Some(value.to_string())),
        Value::Number(value) => Ok(Some(value.to_string())),
        Value::Array(_) | Value::Object(_) => {
            let text = serde_json::to_string(value)
                .map_err(|source| Error::SerializeRequest { source })?;
            Ok(Some(text))
        }
    }
}

enum MultipartChunk {
    Owned(Vec<u8>),
    Shared(Arc<[u8]>),
}

impl MultipartChunk {
    fn len(&self) -> usize {
        match self {
            Self::Owned(bytes) => bytes.len(),
            Self::Shared(bytes) => bytes.len(),
        }
    }

    #[cfg(any(feature = "_blocking", test))]
    fn as_slice(&self) -> &[u8] {
        match self {
            Self::Owned(bytes) => bytes,
            Self::Shared(bytes) => bytes,
        }
    }

    #[cfg(feature = "_async")]
    fn into_bytes(self) -> Bytes {
        match self {
            Self::Owned(bytes) => Bytes::from(bytes),
            Self::Shared(bytes) => Bytes::from_owner(bytes),
        }
    }
}

pub(crate) struct MultipartPayload {
    chunks: Vec<MultipartChunk>,
    content_type: String,
    content_length: usize,
}

impl MultipartPayload {
    pub(crate) fn content_type(&self) -> &str {
        &self.content_type
    }

    pub(crate) fn content_length(&self) -> usize {
        self.content_length
    }

    #[cfg(any(feature = "_blocking", test))]
    pub(crate) fn into_reader(self) -> MultipartBodyReader {
        MultipartBodyReader::new(self.chunks)
    }

    #[cfg(feature = "_async")]
    pub(crate) fn into_stream(self) -> MultipartBodyStream {
        MultipartBodyStream::new(self.chunks)
    }
}

#[cfg(any(feature = "_blocking", test))]
pub(crate) struct MultipartBodyReader {
    chunks: Vec<MultipartChunk>,
    chunk_index: usize,
    chunk_offset: usize,
}

#[cfg(any(feature = "_blocking", test))]
impl MultipartBodyReader {
    fn new(chunks: Vec<MultipartChunk>) -> Self {
        Self {
            chunks,
            chunk_index: 0,
            chunk_offset: 0,
        }
    }
}

#[cfg(any(feature = "_blocking", test))]
impl Read for MultipartBodyReader {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }

        let mut written = 0;

        while written < buf.len() {
            if self.chunk_index >= self.chunks.len() {
                break;
            }

            let chunk = &self.chunks[self.chunk_index];
            let chunk_bytes = chunk.as_slice();
            let remaining = &chunk_bytes[self.chunk_offset..];

            if remaining.is_empty() {
                self.chunk_index += 1;
                self.chunk_offset = 0;
                continue;
            }

            let to_copy = remaining.len().min(buf.len() - written);
            buf[written..written + to_copy].copy_from_slice(&remaining[..to_copy]);
            written += to_copy;
            self.chunk_offset += to_copy;

            if self.chunk_offset >= chunk.len() {
                self.chunk_index += 1;
                self.chunk_offset = 0;
            }
        }

        Ok(written)
    }
}

#[cfg(feature = "_async")]
pub(crate) struct MultipartBodyStream {
    chunks: std::vec::IntoIter<MultipartChunk>,
}

#[cfg(feature = "_async")]
impl MultipartBodyStream {
    fn new(chunks: Vec<MultipartChunk>) -> Self {
        Self {
            chunks: chunks.into_iter(),
        }
    }
}

#[cfg(feature = "_async")]
impl Stream for MultipartBodyStream {
    type Item = Result<Bytes, io::Error>;

    fn poll_next(
        mut self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        match self.chunks.next() {
            Some(chunk) => std::task::Poll::Ready(Some(Ok(chunk.into_bytes()))),
            None => std::task::Poll::Ready(None),
        }
    }
}

fn push_chunk(chunks: &mut Vec<MultipartChunk>, content_length: &mut usize, chunk: MultipartChunk) {
    *content_length += chunk.len();
    chunks.push(chunk);
}

pub(crate) fn build_multipart_payload(
    fields: &[(String, String)],
    file_field_name: &str,
    file: &UploadFile,
) -> Result<MultipartPayload, Error> {
    build_multipart_payload_many(fields, &[(file_field_name, file)])
}

pub(crate) fn build_multipart_payload_many(
    fields: &[(String, String)],
    files: &[(&str, &UploadFile)],
) -> Result<MultipartPayload, Error> {
    if files.is_empty() {
        return Err(Error::InvalidRequest {
            reason: "multipart request requires at least one file part".to_owned(),
        });
    }
    validate_multipart_parts(fields, files)?;

    let boundary = next_multipart_boundary(fields, files)?;

    let mut chunks = Vec::new();
    let mut content_length = 0;

    for (name, value) in fields {
        push_chunk(
            &mut chunks,
            &mut content_length,
            MultipartChunk::Owned(format!("--{boundary}\r\n").into_bytes()),
        );
        push_chunk(
            &mut chunks,
            &mut content_length,
            MultipartChunk::Owned(
                format!(
                    "Content-Disposition: form-data; name=\"{}\"\r\n\r\n",
                    escape_quoted_header_value("multipart field name", name)?
                )
                .into_bytes(),
            ),
        );
        push_chunk(
            &mut chunks,
            &mut content_length,
            MultipartChunk::Owned(value.as_bytes().to_vec()),
        );
        push_chunk(
            &mut chunks,
            &mut content_length,
            MultipartChunk::Owned(b"\r\n".to_vec()),
        );
    }

    for (file_field_name, file) in files {
        push_chunk(
            &mut chunks,
            &mut content_length,
            MultipartChunk::Owned(format!("--{boundary}\r\n").into_bytes()),
        );
        push_chunk(
            &mut chunks,
            &mut content_length,
            MultipartChunk::Owned(
                format!(
                    "Content-Disposition: form-data; name=\"{}\"; filename=\"{}\"\r\n",
                    escape_quoted_header_value("multipart file field name", file_field_name)?,
                    escape_quoted_header_value("multipart file name", file.file_name())?
                )
                .into_bytes(),
            ),
        );
        push_chunk(
            &mut chunks,
            &mut content_length,
            MultipartChunk::Owned(
                format!(
                    "Content-Type: {}\r\n\r\n",
                    file.content_type().unwrap_or("application/octet-stream")
                )
                .into_bytes(),
            ),
        );
        push_chunk(
            &mut chunks,
            &mut content_length,
            MultipartChunk::Shared(file.data_arc()),
        );
        push_chunk(
            &mut chunks,
            &mut content_length,
            MultipartChunk::Owned(b"\r\n".to_vec()),
        );
    }

    push_chunk(
        &mut chunks,
        &mut content_length,
        MultipartChunk::Owned(format!("--{boundary}--\r\n").into_bytes()),
    );

    Ok(MultipartPayload {
        chunks,
        content_type: format!("multipart/form-data; boundary={boundary}"),
        content_length,
    })
}

fn next_multipart_boundary(
    fields: &[(String, String)],
    files: &[(&str, &UploadFile)],
) -> Result<String, Error> {
    let timestamp = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0_u128, |duration| duration.as_nanos());
    let sequence = LOCAL_REQUEST_ID_SEQ.fetch_add(1, Ordering::Relaxed);

    for attempt in 0..64_u8 {
        let boundary = format!("tele-sdk-boundary-{timestamp:x}-{sequence:x}-{attempt:x}");
        if !multipart_boundary_conflicts(&boundary, fields, files) {
            return Ok(boundary);
        }
    }

    Err(Error::InvalidRequest {
        reason: "failed to allocate a multipart boundary that is absent from the payload"
            .to_owned(),
    })
}

fn multipart_boundary_conflicts(
    boundary: &str,
    fields: &[(String, String)],
    files: &[(&str, &UploadFile)],
) -> bool {
    let needle = boundary.as_bytes();

    fields.iter().any(|(name, value)| {
        contains_bytes(name.as_bytes(), needle) || contains_bytes(value.as_bytes(), needle)
    }) || files.iter().any(|(name, file)| {
        contains_bytes(name.as_bytes(), needle)
            || contains_bytes(file.file_name().as_bytes(), needle)
            || file
                .content_type()
                .is_some_and(|content_type| contains_bytes(content_type.as_bytes(), needle))
            || contains_bytes(file.data_arc().as_ref(), needle)
    })
}

fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
    !needle.is_empty()
        && haystack
            .windows(needle.len())
            .any(|window| window == needle)
}

fn validate_multipart_parts(
    fields: &[(String, String)],
    files: &[(&str, &UploadFile)],
) -> Result<(), Error> {
    let mut names = BTreeSet::new();

    for (name, _) in fields {
        validate_multipart_part_name("multipart field name", name)?;
        if !names.insert(name.as_str()) {
            return Err(Error::InvalidRequest {
                reason: format!("duplicate multipart field `{name}`"),
            });
        }
    }

    for (name, _) in files {
        validate_multipart_part_name("multipart file field name", name)?;
        if !names.insert(*name) {
            return Err(Error::InvalidRequest {
                reason: format!("duplicate multipart field `{name}`"),
            });
        }
    }

    Ok(())
}

fn validate_multipart_part_name(label: &str, name: &str) -> Result<(), Error> {
    if name.is_empty() {
        return Err(Error::InvalidRequest {
            reason: format!("{label} cannot be empty"),
        });
    }
    if name.chars().any(char::is_control) {
        return Err(Error::InvalidRequest {
            reason: format!("{label} must not contain control characters"),
        });
    }

    Ok(())
}

fn escape_quoted_header_value(label: &str, value: &str) -> Result<String, Error> {
    if value.chars().any(char::is_control) {
        return Err(Error::InvalidRequest {
            reason: format!("{label} must not contain control characters"),
        });
    }
    Ok(value.replace('\\', "\\\\").replace('"', "\\\""))
}

#[cfg(test)]
mod tests {
    use std::io::Read;
    use std::time::{Duration, SystemTime};

    use super::{
        build_multipart_payload, build_multipart_payload_many, multipart_boundary_conflicts,
        parse_telegram_response, retry_after_seconds_from_headers_at, serialize_multipart_fields,
    };
    use crate::Error;
    use crate::types::upload::UploadFile;
    use http::StatusCode;
    use http::header::{HeaderMap, HeaderValue, RETRY_AFTER};
    use serde_json::Value;

    #[derive(serde::Serialize)]
    struct Payload {
        chat_id: i64,
        text: String,
        file: String,
    }

    #[test]
    fn serializes_multipart_fields_skipping_file_field() {
        let payload = Payload {
            chat_id: 1,
            text: "hello".to_owned(),
            file: "ignored".to_owned(),
        };

        let fields_result = serialize_multipart_fields(&payload, &["file"]);
        assert!(fields_result.is_ok());
        let fields = fields_result.unwrap_or_default();
        assert_eq!(fields.len(), 2);
        assert!(
            fields
                .iter()
                .any(|(key, value)| key == "chat_id" && value == "1")
        );
        assert!(
            fields
                .iter()
                .any(|(key, value)| key == "text" && value == "hello")
        );
    }

    #[test]
    fn api_error_uses_http_retry_after_header_when_parameters_are_absent() {
        let mut headers = HeaderMap::new();
        headers.insert(RETRY_AFTER, HeaderValue::from_static("7"));
        let body = br#"{"ok":false,"error_code":429,"description":"Too Many Requests"}"#;

        let result = parse_telegram_response::<Value>(
            "sendMessage",
            StatusCode::TOO_MANY_REQUESTS,
            &headers,
            body,
            true,
            256,
        );
        assert!(result.is_err());
        let error = match result {
            Err(error) => error,
            Ok(_) => return,
        };

        assert!(matches!(error, Error::Api { .. }));
        assert_eq!(error.retry_after(), Some(Duration::from_secs(7)));
    }

    #[test]
    fn api_error_parameters_retry_after_wins_over_http_header() {
        let mut headers = HeaderMap::new();
        headers.insert(RETRY_AFTER, HeaderValue::from_static("7"));
        let body = br#"{"ok":false,"error_code":429,"description":"Too Many Requests","parameters":{"retry_after":3}}"#;

        let result = parse_telegram_response::<Value>(
            "sendMessage",
            StatusCode::TOO_MANY_REQUESTS,
            &headers,
            body,
            true,
            256,
        );
        assert!(result.is_err());
        let error = match result {
            Err(error) => error,
            Ok(_) => return,
        };

        assert!(matches!(error, Error::Api { .. }));
        assert_eq!(error.retry_after(), Some(Duration::from_secs(3)));
    }

    #[test]
    fn retry_after_header_accepts_http_date() {
        let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
        let retry_at = now + Duration::from_secs(9);
        let retry_after = httpdate::fmt_http_date(retry_at);
        let mut headers = HeaderMap::new();
        let header_value = match HeaderValue::from_str(&retry_after) {
            Ok(value) => value,
            Err(_) => return,
        };
        headers.insert(RETRY_AFTER, header_value);

        assert_eq!(retry_after_seconds_from_headers_at(&headers, now), Some(9));
    }

    #[test]
    fn retry_after_header_past_http_date_retries_immediately() {
        let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
        let retry_at = now - Duration::from_secs(9);
        let retry_after = httpdate::fmt_http_date(retry_at);
        let mut headers = HeaderMap::new();
        let header_value = match HeaderValue::from_str(&retry_after) {
            Ok(value) => value,
            Err(_) => return,
        };
        headers.insert(RETRY_AFTER, header_value);

        assert_eq!(retry_after_seconds_from_headers_at(&headers, now), Some(0));
    }

    #[test]
    fn builds_multipart_body_with_binary_file() {
        let file_result = UploadFile::from_bytes("hello.txt", b"abc123".to_vec());
        assert!(file_result.is_ok());
        let file = match file_result {
            Ok(value) => value,
            Err(_) => return,
        };
        let payload_result =
            build_multipart_payload(&[("chat_id".to_owned(), "1".to_owned())], "photo", &file);
        assert!(payload_result.is_ok());
        let payload = match payload_result {
            Ok(value) => value,
            Err(_) => return,
        };
        let content_length = payload.content_length();
        let content_type = payload.content_type().to_owned();

        let mut reader = payload.into_reader();
        let mut body = Vec::new();
        let read_result = reader.read_to_end(&mut body);
        assert!(read_result.is_ok());

        let text = String::from_utf8_lossy(&body);
        assert!(content_type.starts_with("multipart/form-data; boundary="));
        assert_eq!(content_length, body.len());
        assert!(text.contains("name=\"chat_id\""));
        assert!(text.contains("name=\"photo\"; filename=\"hello.txt\""));
        assert!(text.contains("abc123"));
    }

    #[test]
    fn builds_multipart_body_with_multiple_files() -> Result<(), Box<dyn std::error::Error>> {
        let photo = UploadFile::from_bytes("photo.jpg", b"photo-data".to_vec())?;
        let video = UploadFile::from_bytes("video.mp4", b"video-data".to_vec())?;
        let payload = build_multipart_payload_many(
            &[(
                "media".to_owned(),
                r#"[{"type":"photo","media":"attach://photo0"},{"type":"video","media":"attach://video0"}]"#
                    .to_owned(),
            )],
            &[("photo0", &photo), ("video0", &video)],
        )?;
        let content_length = payload.content_length();

        let mut reader = payload.into_reader();
        let mut body = Vec::new();
        reader.read_to_end(&mut body)?;

        let text = String::from_utf8_lossy(&body);
        assert_eq!(content_length, body.len());
        assert!(text.contains("name=\"media\""));
        assert!(text.contains("attach://photo0"));
        assert!(text.contains("name=\"photo0\"; filename=\"photo.jpg\""));
        assert!(text.contains("photo-data"));
        assert!(text.contains("name=\"video0\"; filename=\"video.mp4\""));
        assert!(text.contains("video-data"));
        Ok(())
    }

    #[test]
    fn rejects_multipart_control_chars_in_header_values() {
        let file_result = UploadFile::from_bytes("hello.txt", b"abc123".to_vec());
        assert!(file_result.is_ok());
        let file = match file_result {
            Ok(value) => value,
            Err(_) => return,
        };

        assert!(
            build_multipart_payload(
                &[("bad\r\nname".to_owned(), "1".to_owned())],
                "photo",
                &file
            )
            .is_err()
        );
        assert!(
            build_multipart_payload(
                &[("chat_id".to_owned(), "1".to_owned())],
                "bad\r\nfile",
                &file
            )
            .is_err()
        );
    }

    #[test]
    fn rejects_ambiguous_multipart_file_field_names() {
        let file_result = UploadFile::from_bytes("hello.txt", b"abc123".to_vec());
        assert!(file_result.is_ok());
        let file = match file_result {
            Ok(value) => value,
            Err(_) => return,
        };

        assert!(
            build_multipart_payload_many(
                &[("chat_id".to_owned(), "1".to_owned())],
                &[("document", &file), ("document", &file)]
            )
            .is_err()
        );
        assert!(
            build_multipart_payload_many(&[("chat_id".to_owned(), "1".to_owned())], &[("", &file)])
                .is_err()
        );
        assert!(
            build_multipart_payload_many(
                &[("document".to_owned(), "attach://document".to_owned())],
                &[("document", &file)]
            )
            .is_err()
        );
        assert!(
            build_multipart_payload_many(
                &[
                    ("chat_id".to_owned(), "1".to_owned()),
                    ("chat_id".to_owned(), "2".to_owned())
                ],
                &[("document", &file)]
            )
            .is_err()
        );
        assert!(
            build_multipart_payload_many(
                &[("".to_owned(), "1".to_owned())],
                &[("document", &file)]
            )
            .is_err()
        );
    }

    #[test]
    fn detects_multipart_boundary_collisions_with_payload() -> Result<(), Box<dyn std::error::Error>>
    {
        let safe_file = UploadFile::from_bytes("hello.txt", b"abc123".to_vec())?;
        assert!(!multipart_boundary_conflicts(
            "needle",
            &[("caption".to_owned(), "safe".to_owned())],
            &[("document", &safe_file)]
        ));
        assert!(multipart_boundary_conflicts(
            "needle",
            &[("caption".to_owned(), "contains needle".to_owned())],
            &[("document", &safe_file)]
        ));

        let colliding_file =
            UploadFile::from_bytes("hello.txt", b"file contains needle bytes".to_vec())?;
        assert!(multipart_boundary_conflicts(
            "needle",
            &[("caption".to_owned(), "safe".to_owned())],
            &[("document", &colliding_file)]
        ));

        Ok(())
    }
}
