use crate::sync_http::{MAX_BODY_BYTES, MAX_HEADER_BYTES};
use crate::{
    DeckVisibility, SyncAuthSession, SyncBundlePayload, SyncDeckSummary, SyncDeleteError,
    SyncDownload, SyncDownloadError, SyncDownloadKind, SyncOAuthStart, SyncQuotaStatus,
    SyncRemoteDeckStatus, SyncSharedDeckStatus, SyncUploadError, SyncUploadOutcome,
    SyncUploadRequest, SyncUuid, SyncValidationError,
};
use std::collections::BTreeMap;
use std::fmt;
use std::io::{self, Read, Write};
use std::net::TcpStream;
use std::time::Duration;

const DEFAULT_HTTP_PORT: u16 = 80;
const DEFAULT_HTTPS_PORT: u16 = 443;
const CLIENT_METADATA_TIMEOUT: Duration = Duration::from_secs(30);
const CLIENT_TRANSFER_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
const CLIENT_TRANSFER_TOTAL_TIMEOUT: Duration = Duration::from_secs(90);
const TRANSFER_PROGRESS_CHUNK_BYTES: usize = 64 * 1024;

type TransferProgressCallback = Box<dyn FnMut(u64, Option<u64>) + Send>;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct SyncHttpClientTimeouts {
    total: Duration,
    connect: Duration,
}

fn sync_http_client_timeouts(transfer_request: bool) -> SyncHttpClientTimeouts {
    if transfer_request {
        SyncHttpClientTimeouts {
            total: CLIENT_TRANSFER_TOTAL_TIMEOUT,
            connect: CLIENT_TRANSFER_CONNECT_TIMEOUT,
        }
    } else {
        SyncHttpClientTimeouts {
            total: CLIENT_METADATA_TIMEOUT,
            connect: CLIENT_METADATA_TIMEOUT,
        }
    }
}

#[cfg(test)]
struct TestServerRequest<'a> {
    state: &'a crate::sync_http::ServerState,
    method: &'a str,
    path: &'a str,
    headers: &'a [(&'a str, &'a str)],
    body: &'a [u8],
    upload_progress: Option<TransferProgressCallback>,
    download_progress: Option<TransferProgressCallback>,
}

#[derive(Debug, Clone)]
pub struct HttpSyncClient {
    endpoint: HttpSyncEndpoint,
    bearer_token: Option<String>,
    #[cfg(test)]
    test_server: Option<std::sync::Arc<crate::sync_http::ServerState>>,
}

impl HttpSyncClient {
    pub fn new(server_url: &str) -> Result<Self, HttpSyncClientError> {
        Ok(Self {
            endpoint: HttpSyncEndpoint::parse(server_url)?,
            bearer_token: None,
            #[cfg(test)]
            test_server: None,
        })
    }

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

    #[cfg(test)]
    fn for_test_server(
        server_url: &str,
        state: std::sync::Arc<crate::sync_http::ServerState>,
    ) -> Result<Self, HttpSyncClientError> {
        Ok(Self {
            endpoint: HttpSyncEndpoint::parse_for_test_server(server_url)?,
            bearer_token: None,
            test_server: Some(state),
        })
    }

    pub fn health(&self) -> Result<String, HttpSyncClientError> {
        let response = self.send("GET", "/api/v1/health", &[], &[])?;
        expect_success(response, &[200]).and_then(response_text)
    }

    pub fn list_public_decks(&self) -> Result<Vec<SyncSharedDeckStatus>, HttpSyncClientError> {
        let response = self.send("GET", "/api/v1/decks", &[], &[])?;
        let text = response_text(expect_success(response, &[200])?)?;
        parse_shared_status_list(&text)
    }

    pub fn quota_status(&self) -> Result<SyncQuotaStatus, HttpSyncClientError> {
        let authorization;
        let headers = match self.bearer_token.as_deref() {
            Some(token) => {
                authorization = bearer_authorization_header(token);
                vec![("Authorization", authorization.as_str())]
            }
            None => Vec::new(),
        };
        let response = self.send("GET", "/api/v1/quota", &headers, &[])?;
        let text = response_text(expect_success(response, &[200])?)?;
        parse_quota_status(&text)
    }

    pub fn start_fedi_browser_login(&self) -> Result<SyncOAuthStart, HttpSyncClientError> {
        let response = self.send("POST", "/api/v1/auth/fedi/browser-start", &[], &[])?;
        let text = response_text(expect_success(response, &[201])?)?;
        let login_token = required_json_string_field(&text, "login_token")?;
        let registration_path = required_json_string_field(&text, "registration_url")?;
        validate_sync_registration_path(&registration_path)?;
        let registration_url = self.endpoint.absolute_url(&registration_path);
        SyncOAuthStart::from_parts_with_registration_url(
            &login_token,
            &registration_url,
            Some(&registration_url),
        )
        .map_err(HttpSyncClientError::from)
    }

    pub fn start_fedi_browser_device_login(&self) -> Result<SyncOAuthStart, HttpSyncClientError> {
        let response = self.send("POST", "/api/v1/auth/fedi/device-browser-start", &[], &[])?;
        let text = response_text(expect_success(response, &[201])?)?;
        let login_token = required_json_string_field(&text, "login_token")?;
        let authorization_path = required_json_string_field(&text, "registration_url")?;
        validate_sync_registration_path(&authorization_path)?;
        let authorization_url = self.endpoint.absolute_url(&authorization_path);
        SyncOAuthStart::from_parts_with_registration_url(
            &login_token,
            &authorization_url,
            Some(&authorization_url),
        )
        .map_err(HttpSyncClientError::from)
    }

    pub fn account_deletion_url(&self) -> String {
        self.endpoint.absolute_url("/auth/fedi/delete")
    }

    pub fn poll_fedi_browser_login(
        &self,
        login_token: &str,
    ) -> Result<Option<SyncAuthSession>, HttpSyncClientError> {
        let response = self.send(
            "POST",
            "/api/v1/auth/fedi/poll",
            &[("X-Cosimo-Login-Token", login_token)],
            &[],
        )?;
        match response.status {
            202 => Ok(None),
            201 => {
                let text = response_text(response)?;
                parse_auth_session_json(&text).map(Some)
            }
            status => {
                let _ = status;
                Err(unexpected_status(response))
            }
        }
    }

    pub fn revoke_auth_session(&self) -> Result<(), HttpSyncClientError> {
        let authorization;
        let headers = match self.bearer_token.as_deref() {
            Some(token) => {
                authorization = bearer_authorization_header(token);
                vec![("Authorization", authorization.as_str())]
            }
            None => Vec::new(),
        };
        let response = self.send("POST", "/api/v1/auth/logout", &headers, &[])?;
        expect_success(response, &[200]).map(|_| ())
    }

    pub fn revoke_all_auth_sessions(&self) -> Result<(), HttpSyncClientError> {
        let authorization;
        let headers = match self.bearer_token.as_deref() {
            Some(token) => {
                authorization = bearer_authorization_header(token);
                vec![("Authorization", authorization.as_str())]
            }
            None => Vec::new(),
        };
        let response = self.send("POST", "/api/v1/auth/logout-all", &headers, &[])?;
        expect_success(response, &[200]).map(|_| ())
    }

    pub fn list_owner_decks(
        &self,
        _requester_owner_identity: &str,
    ) -> Result<Vec<SyncRemoteDeckStatus>, HttpSyncClientError> {
        let authorization;
        let headers = match self.bearer_token.as_deref() {
            Some(token) => {
                authorization = bearer_authorization_header(token);
                vec![("Authorization", authorization.as_str())]
            }
            None => Vec::new(),
        };
        let response = self.send("GET", "/api/v1/me/decks", &headers, &[])?;
        let text = response_text(expect_success(response, &[200])?)?;
        parse_owner_status_list(&text)
    }

    pub fn inspect_shared_deck(
        &self,
        deck_uuid: &SyncUuid,
    ) -> Result<SyncSharedDeckStatus, HttpSyncClientError> {
        let path = format!("/api/v1/decks/{}", deck_uuid.as_str());
        let response = self.send("GET", &path, &[], &[])?;
        let response = match response.status {
            200 => response,
            403 => return Err(SyncDownloadError::Forbidden.into()),
            404 if is_remote_deck_not_found_response(&response) => {
                return Err(SyncDownloadError::NotFound.into());
            }
            _ => return Err(unexpected_status(response)),
        };
        parse_shared_status(&response_text(response)?)
    }

    pub fn download(
        &self,
        deck_uuid: &SyncUuid,
        kind: SyncDownloadKind,
        _requester_owner_identity: Option<&str>,
    ) -> Result<SyncDownload, HttpSyncClientError> {
        self.download_with_progress(deck_uuid, kind, _requester_owner_identity, |_, _| {})
    }

    pub fn download_with_progress<P>(
        &self,
        deck_uuid: &SyncUuid,
        kind: SyncDownloadKind,
        _requester_owner_identity: Option<&str>,
        progress: P,
    ) -> Result<SyncDownload, HttpSyncClientError>
    where
        P: FnMut(u64, Option<u64>) + Send + 'static,
    {
        let kind_query = match kind {
            SyncDownloadKind::SharedContent => "shared",
            SyncDownloadKind::OwnerStudyHistory => "owner-study-history",
        };
        let path = format!(
            "/api/v1/decks/{}/download?kind={kind_query}",
            deck_uuid.as_str()
        );
        let mut headers = Vec::new();
        let authorization;
        if let Some(token) = self.bearer_token.as_deref() {
            authorization = bearer_authorization_header(token);
            headers.push(("Authorization", authorization.as_str()));
        }
        let response =
            self.send_with_progress("GET", &path, &headers, &[], None, Some(Box::new(progress)))?;
        let response = match response.status {
            200 => response,
            403 => return Err(SyncDownloadError::Forbidden.into()),
            404 if is_remote_deck_not_found_response(&response) => {
                return Err(SyncDownloadError::NotFound.into());
            }
            _ => return Err(unexpected_status(response)),
        };

        let revision = required_response_header(&response, "x-cosimo-revision")?;
        let visibility =
            parse_visibility(required_response_header(&response, "x-cosimo-visibility")?)?;
        let bundle_hash = required_response_header(&response, "x-cosimo-bundle-hash")?;
        let includes_review_database = parse_bool_header(
            "x-cosimo-includes-review-database",
            required_response_header(&response, "x-cosimo-includes-review-database")?,
        )?;
        let device_name = response
            .headers
            .get("x-cosimo-last-upload-device-name")
            .map(String::as_str);
        let bundle =
            SyncBundlePayload::from_parts(response.body, &bundle_hash, includes_review_database)?;
        SyncDownload::from_parts(&revision, visibility, bundle, device_name)
            .map_err(HttpSyncClientError::from)
    }

    pub fn upload(
        &self,
        request: SyncUploadRequest,
    ) -> Result<SyncUploadOutcome, HttpSyncClientError> {
        self.upload_with_progress(request, |_, _| {})
    }

    pub fn upload_with_progress<P>(
        &self,
        request: SyncUploadRequest,
        progress: P,
    ) -> Result<SyncUploadOutcome, HttpSyncClientError>
    where
        P: FnMut(u64, Option<u64>) + Send + 'static,
    {
        let authorization;
        let mut headers = vec![("X-Cosimo-Visibility", request.visibility().code())];
        if let Some(token) = self.bearer_token.as_deref() {
            authorization = bearer_authorization_header(token);
            headers.push(("Authorization", authorization.as_str()));
        }
        let content_length;
        if let Some(content) = request.content_bundle() {
            headers.push(("X-Cosimo-Content-Hash", content.bundle_hash()));
            content_length = content.bytes().len().to_string();
            headers.push(("X-Cosimo-Content-Length", &content_length));
        }
        let history_length;
        if let Some(history) = request.study_history_bundle() {
            history_length = history.bytes().len().to_string();
            headers.push(("X-Cosimo-Study-History-Hash", history.bundle_hash()));
            headers.push(("X-Cosimo-Study-History-Length", &history_length));
        }
        if let Some(expected_revision) = request.expected_revision() {
            headers.push(("X-Cosimo-Expected-Revision", expected_revision));
        }
        if let Some(device_name) = request.device_identity().device_name() {
            headers.push(("X-Cosimo-Device-Name", device_name));
        }

        let mut body = Vec::with_capacity(
            request
                .content_bundle()
                .map(|bundle| bundle.bytes().len())
                .unwrap_or(0)
                + request
                    .study_history_bundle()
                    .map(|bundle| bundle.bytes().len())
                    .unwrap_or(0),
        );
        if let Some(content) = request.content_bundle() {
            body.extend_from_slice(content.bytes());
        }
        if let Some(history) = request.study_history_bundle() {
            body.extend_from_slice(history.bytes());
        }

        let path = format!("/api/v1/decks/{}/upload", request.deck_uuid().as_str());
        let response = self.send_with_progress(
            "POST",
            &path,
            &headers,
            &body,
            Some(Box::new(progress)),
            None,
        )?;
        let response = match response.status {
            201 => response,
            403 => return Err(upload_forbidden_error(&response).into()),
            409 => return Err(parse_upload_conflict(&response_text(response)?)?.into()),
            _ => return Err(unexpected_status(response)),
        };
        let text = response_text(response)?;
        SyncUploadOutcome::from_parts(
            &required_json_string_field(&text, "revision")?,
            &required_json_string_field(&text, "bundle_hash")?,
        )
        .map_err(HttpSyncClientError::from)
    }

    pub fn delete_deck(&self, deck_uuid: &SyncUuid) -> Result<(), HttpSyncClientError> {
        let path = format!("/api/v1/decks/{}", deck_uuid.as_str());
        let authorization;
        let headers = match self.bearer_token.as_deref() {
            Some(token) => {
                authorization = bearer_authorization_header(token);
                vec![("Authorization", authorization.as_str())]
            }
            None => Vec::new(),
        };
        let response = self.send("DELETE", &path, &headers, &[])?;
        match response.status {
            200 => Ok(()),
            403 => Err(SyncDeleteError::Forbidden.into()),
            404 => Err(SyncDeleteError::NotFound.into()),
            _ => Err(unexpected_status(response)),
        }
    }

    fn send(
        &self,
        method: &str,
        path: &str,
        headers: &[(&str, &str)],
        body: &[u8],
    ) -> Result<HttpResponse, HttpSyncClientError> {
        self.send_with_progress(method, path, headers, body, None, None)
    }

    fn send_with_progress(
        &self,
        method: &str,
        path: &str,
        headers: &[(&str, &str)],
        body: &[u8],
        upload_progress: Option<TransferProgressCallback>,
        download_progress: Option<TransferProgressCallback>,
    ) -> Result<HttpResponse, HttpSyncClientError> {
        #[cfg(test)]
        if let Some(state) = &self.test_server {
            return self.send_to_test_server(TestServerRequest {
                state,
                method,
                path,
                headers,
                body,
                upload_progress,
                download_progress,
            });
        }

        match self.endpoint.scheme {
            HttpSyncScheme::Http => self.send_plain_http_with_progress(
                method,
                path,
                headers,
                body,
                upload_progress,
                download_progress,
            ),
            HttpSyncScheme::Https => self.send_reqwest_with_progress(
                method,
                path,
                headers,
                body,
                upload_progress,
                download_progress,
            ),
        }
    }

    fn send_plain_http_with_progress(
        &self,
        method: &str,
        path: &str,
        headers: &[(&str, &str)],
        body: &[u8],
        upload_progress: Option<TransferProgressCallback>,
        download_progress: Option<TransferProgressCallback>,
    ) -> Result<HttpResponse, HttpSyncClientError> {
        let mut stream = TcpStream::connect(self.endpoint.socket_address())?;
        stream.set_read_timeout(Some(CLIENT_TRANSFER_CONNECT_TIMEOUT))?;
        stream.set_write_timeout(Some(CLIENT_TRANSFER_CONNECT_TIMEOUT))?;

        let request_head = self.request_head_bytes(method, path, headers, body.len())?;
        stream.write_all(&request_head)?;
        write_body_with_progress(&mut stream, body, upload_progress)?;
        stream.flush()?;

        read_http_response_with_progress(&mut stream, download_progress)
    }

    fn send_reqwest_with_progress(
        &self,
        method: &str,
        path: &str,
        headers: &[(&str, &str)],
        body: &[u8],
        upload_progress: Option<TransferProgressCallback>,
        mut download_progress: Option<TransferProgressCallback>,
    ) -> Result<HttpResponse, HttpSyncClientError> {
        for (_, value) in headers {
            validate_http_header_value(value)?;
        }
        let method = reqwest::Method::from_bytes(method.as_bytes()).map_err(|_| {
            HttpSyncClientError::Protocol(format!("unsupported sync HTTP method {method}"))
        })?;
        let timeouts =
            sync_http_client_timeouts(upload_progress.is_some() || download_progress.is_some());
        let client = reqwest::blocking::Client::builder()
            .timeout(timeouts.total)
            .connect_timeout(timeouts.connect)
            .redirect(reqwest::redirect::Policy::none())
            .build()?;
        let mut request = client.request(method, self.endpoint.absolute_url(path));
        for (name, value) in headers {
            request = request.header(*name, *value);
        }
        let request_body = reqwest_body_with_progress(body.to_vec(), upload_progress);
        let mut response = request.body(request_body).send()?;
        let status = response.status().as_u16();
        let headers = response_headers(response.headers())?;
        let progress_total = response_body_progress_total(response.content_length(), &headers)?;
        if let Some(length) = progress_total
            && length > MAX_BODY_BYTES as u64
        {
            return Err(HttpSyncClientError::Protocol(
                "server response body is too large".to_string(),
            ));
        }
        let body = read_body_stream_with_progress(
            &mut response,
            progress_total,
            download_progress.as_mut(),
        )?;
        if body.len() > MAX_BODY_BYTES {
            return Err(HttpSyncClientError::Protocol(
                "server response body is too large".to_string(),
            ));
        }
        Ok(HttpResponse {
            status,
            headers,
            body,
        })
    }

    #[cfg(test)]
    fn request_bytes(
        &self,
        method: &str,
        path: &str,
        headers: &[(&str, &str)],
        body: &[u8],
    ) -> Result<Vec<u8>, HttpSyncClientError> {
        let mut bytes = self.request_head_bytes(method, path, headers, body.len())?;
        bytes.extend_from_slice(body);
        Ok(bytes)
    }

    fn request_head_bytes(
        &self,
        method: &str,
        path: &str,
        headers: &[(&str, &str)],
        body_len: usize,
    ) -> Result<Vec<u8>, HttpSyncClientError> {
        let target = self.endpoint.request_target(path);
        let mut request = format!(
            "{method} {target} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\nContent-Length: {}\r\n",
            self.endpoint.host_header, body_len
        );
        for (name, value) in headers {
            validate_http_header_value(value)?;
            request.push_str(name);
            request.push_str(": ");
            request.push_str(value);
            request.push_str("\r\n");
        }
        request.push_str("\r\n");
        Ok(request.into_bytes())
    }

    #[cfg(test)]
    fn send_to_test_server(
        &self,
        mut request: TestServerRequest<'_>,
    ) -> Result<HttpResponse, HttpSyncClientError> {
        use crate::sync_http::{handle_request, http_error_response, read_http_request};
        use std::io::Cursor;

        let request_bytes =
            self.request_bytes(request.method, request.path, request.headers, request.body)?;
        report_transfer_progress(
            request.upload_progress.as_mut(),
            request.body.len() as u64,
            Some(request.body.len() as u64),
        );
        let mut input = Cursor::new(request_bytes);
        let response = match read_http_request(&mut input) {
            Ok(http_request) => handle_request(request.state, http_request),
            Err(error) => http_error_response(error),
        };
        let mut output = Cursor::new(response.to_http_bytes());
        read_http_response_with_progress(&mut output, request.download_progress)
    }
}

fn reqwest_body_with_progress(
    body: Vec<u8>,
    progress: Option<TransferProgressCallback>,
) -> reqwest::blocking::Body {
    let len = body.len() as u64;
    reqwest::blocking::Body::sized(ProgressRead::new(body, len, progress), len)
}

struct ProgressRead {
    inner: std::io::Cursor<Vec<u8>>,
    total: u64,
    transferred: u64,
    progress: Option<TransferProgressCallback>,
}

impl ProgressRead {
    fn new(body: Vec<u8>, total: u64, progress: Option<TransferProgressCallback>) -> Self {
        Self {
            inner: std::io::Cursor::new(body),
            total,
            transferred: 0,
            progress,
        }
    }
}

impl Read for ProgressRead {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let read = self.inner.read(buf)?;
        if read > 0 {
            self.transferred = self.transferred.saturating_add(read as u64);
            report_transfer_progress(self.progress.as_mut(), self.transferred, Some(self.total));
        } else if self.total == 0 {
            report_transfer_progress(self.progress.as_mut(), 0, Some(0));
        }
        Ok(read)
    }
}

fn write_body_with_progress(
    writer: &mut impl Write,
    body: &[u8],
    mut progress: Option<TransferProgressCallback>,
) -> Result<(), HttpSyncClientError> {
    if body.is_empty() {
        report_transfer_progress(progress.as_mut(), 0, Some(0));
        return Ok(());
    }

    let total = body.len() as u64;
    let mut transferred = 0_u64;
    for chunk in body.chunks(TRANSFER_PROGRESS_CHUNK_BYTES) {
        writer.write_all(chunk)?;
        transferred = transferred.saturating_add(chunk.len() as u64);
        report_transfer_progress(progress.as_mut(), transferred, Some(total));
    }
    Ok(())
}

fn read_http_response_with_progress(
    reader: &mut impl Read,
    mut progress: Option<TransferProgressCallback>,
) -> Result<HttpResponse, HttpSyncClientError> {
    let mut bytes = Vec::new();
    let mut header_info = None;
    let mut buffer = [0_u8; TRANSFER_PROGRESS_CHUNK_BYTES];
    loop {
        let read = reader.read(&mut buffer)?;
        if read == 0 {
            break;
        }
        bytes.extend_from_slice(&buffer[..read]);
        if header_info.is_none() {
            header_info = response_header_info(&bytes)?;
        }
        if let Some((body_start, total)) = header_info {
            let transferred = bytes.len().saturating_sub(body_start);
            if transferred > MAX_BODY_BYTES {
                return Err(HttpSyncClientError::Protocol(
                    "server response body is too large".to_string(),
                ));
            }
            report_transfer_progress(progress.as_mut(), transferred as u64, total);
        } else if bytes.len() > MAX_HEADER_BYTES + 4 {
            return Err(HttpSyncClientError::Protocol(
                "server response headers are too large".to_string(),
            ));
        }
    }

    parse_http_response_bytes(bytes)
}

fn read_body_stream_with_progress(
    reader: &mut impl Read,
    expected_length: Option<u64>,
    mut progress: Option<&mut TransferProgressCallback>,
) -> Result<Vec<u8>, HttpSyncClientError> {
    let mut body = Vec::new();
    let mut buffer = [0_u8; TRANSFER_PROGRESS_CHUNK_BYTES];
    loop {
        let read = reader.read(&mut buffer)?;
        if read == 0 {
            break;
        }
        body.extend_from_slice(&buffer[..read]);
        if body.len() > MAX_BODY_BYTES {
            return Err(HttpSyncClientError::Protocol(
                "server response body is too large".to_string(),
            ));
        }
        report_transfer_progress(progress.as_deref_mut(), body.len() as u64, expected_length);
    }
    Ok(body)
}

fn report_transfer_progress(
    progress: Option<&mut TransferProgressCallback>,
    transferred: u64,
    total: Option<u64>,
) {
    if let Some(progress) = progress {
        progress(transferred, total);
    }
}

fn response_header_info(bytes: &[u8]) -> Result<Option<(usize, Option<u64>)>, HttpSyncClientError> {
    let Some(header_end) = find_header_end(bytes) else {
        return Ok(None);
    };
    if header_end > MAX_HEADER_BYTES {
        return Err(HttpSyncClientError::Protocol(
            "server response headers are too large".to_string(),
        ));
    }
    let header_text = std::str::from_utf8(&bytes[..header_end]).map_err(|_| {
        HttpSyncClientError::Protocol("server response headers are not UTF-8".to_string())
    })?;
    let (_, headers) = parse_response_headers(header_text)?;
    let total = response_body_progress_total(None, &headers)?;
    Ok(Some((header_end + 4, total)))
}

fn response_body_progress_total(
    transport_content_length: Option<u64>,
    headers: &BTreeMap<String, String>,
) -> Result<Option<u64>, HttpSyncClientError> {
    if let Some(length) = transport_content_length {
        return Ok(Some(length));
    }
    if let Some(value) = headers.get("content-length") {
        return value
            .parse::<u64>()
            .map(Some)
            .map_err(|_| invalid_response_length_header("Content-Length"));
    }
    if let Some(value) = headers.get("x-cosimo-bundle-length") {
        return value
            .parse::<u64>()
            .map(Some)
            .map_err(|_| invalid_response_length_header("X-Cosimo-Bundle-Length"));
    }
    Ok(None)
}

fn invalid_response_length_header(header_name: &str) -> HttpSyncClientError {
    HttpSyncClientError::Protocol(format!("server response has invalid {header_name}"))
}

fn validate_http_header_value(value: &str) -> Result<(), HttpSyncClientError> {
    if value
        .chars()
        .any(|character| character == '\r' || character == '\n')
    {
        return Err(HttpSyncClientError::Protocol(
            "sync HTTP header values must not contain line breaks".to_string(),
        ));
    }
    Ok(())
}

fn response_headers(
    headers: &reqwest::header::HeaderMap,
) -> Result<BTreeMap<String, String>, HttpSyncClientError> {
    let mut parsed = BTreeMap::new();
    for (name, value) in headers {
        let value = value.to_str().map_err(|_| {
            HttpSyncClientError::Protocol("server response header is not UTF-8".to_string())
        })?;
        parsed.insert(name.as_str().to_ascii_lowercase(), value.trim().to_string());
    }
    Ok(parsed)
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum HttpSyncScheme {
    Http,
    Https,
}

impl HttpSyncScheme {
    fn code(self) -> &'static str {
        match self {
            Self::Http => "http",
            Self::Https => "https",
        }
    }

    fn default_port(self) -> u16 {
        match self {
            Self::Http => DEFAULT_HTTP_PORT,
            Self::Https => DEFAULT_HTTPS_PORT,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct HttpSyncEndpoint {
    scheme: HttpSyncScheme,
    host: String,
    port: u16,
    host_header: String,
    base_path: String,
}

impl HttpSyncEndpoint {
    fn parse(server_url: &str) -> Result<Self, HttpSyncClientError> {
        Self::parse_with_http_policy(server_url, HttpSyncPolicy::Deployment)
    }

    #[cfg(test)]
    fn parse_for_test_server(server_url: &str) -> Result<Self, HttpSyncClientError> {
        Self::parse_with_http_policy(server_url, HttpSyncPolicy::InProcessTest)
    }

    fn parse_with_http_policy(
        server_url: &str,
        policy: HttpSyncPolicy,
    ) -> Result<Self, HttpSyncClientError> {
        let server_url = server_url.trim();
        if server_url
            .chars()
            .any(|character| character.is_control() || character.is_whitespace())
        {
            return Err(HttpSyncClientError::InvalidServerUrl(
                "sync server URL must not contain whitespace or control characters".to_string(),
            ));
        }
        let (scheme, rest) = if let Some(rest) = server_url.strip_prefix("http://") {
            (HttpSyncScheme::Http, rest)
        } else if let Some(rest) = server_url.strip_prefix("https://") {
            (HttpSyncScheme::Https, rest)
        } else {
            return Err(HttpSyncClientError::InvalidServerUrl(
                "sync server URL must start with http:// or https://".to_string(),
            ));
        };
        if rest.is_empty() {
            return Err(HttpSyncClientError::InvalidServerUrl(
                "sync server URL is missing a host".to_string(),
            ));
        }
        if rest.contains('@') {
            return Err(HttpSyncClientError::InvalidServerUrl(
                "sync server URL must not contain user information".to_string(),
            ));
        }
        let (authority, path) = rest.split_once('/').unwrap_or((rest, ""));
        let (host, port) = parse_authority(authority, scheme.default_port())?;
        let host_header = if port == scheme.default_port() {
            host.clone()
        } else {
            format!("{host}:{port}")
        };
        let base_path = normalize_base_path(path)?;
        if scheme == HttpSyncScheme::Http
            && policy == HttpSyncPolicy::Deployment
            && !is_loopback_http_host(&host)
        {
            return Err(HttpSyncClientError::InvalidServerUrl(
                "plain http sync URLs are only allowed for local loopback development; use https:// for deployed sync servers".to_string(),
            ));
        }
        Ok(Self {
            scheme,
            host,
            port,
            host_header,
            base_path,
        })
    }

    fn socket_address(&self) -> String {
        format!("{}:{}", self.host, self.port)
    }

    fn request_target(&self, path: &str) -> String {
        let path = path.trim_start_matches('/');
        if self.base_path.is_empty() {
            format!("/{path}")
        } else {
            format!("{}/{path}", self.base_path)
        }
    }

    fn absolute_url(&self, path: &str) -> String {
        format!(
            "{}://{}{}",
            self.scheme.code(),
            self.host_header,
            self.request_target(path)
        )
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum HttpSyncPolicy {
    Deployment,
    #[cfg(test)]
    InProcessTest,
}

fn is_loopback_http_host(host: &str) -> bool {
    let host = host
        .strip_prefix('[')
        .and_then(|host| host.strip_suffix(']'))
        .unwrap_or(host);
    if host.eq_ignore_ascii_case("localhost") {
        return true;
    }
    host.parse::<std::net::IpAddr>()
        .map(|address| address.is_loopback())
        .unwrap_or(false)
}

fn parse_authority(
    authority: &str,
    default_port: u16,
) -> Result<(String, u16), HttpSyncClientError> {
    if authority.is_empty() {
        return Err(HttpSyncClientError::InvalidServerUrl(
            "sync server URL is missing a host".to_string(),
        ));
    }
    if authority.starts_with('[') {
        let Some(close) = authority.find(']') else {
            return Err(HttpSyncClientError::InvalidServerUrl(
                "IPv6 host is missing a closing bracket".to_string(),
            ));
        };
        let host = &authority[..=close];
        let port = match authority[close + 1..].strip_prefix(':') {
            Some(port) => parse_port(port)?,
            None if authority.len() == close + 1 => default_port,
            None => {
                return Err(HttpSyncClientError::InvalidServerUrl(
                    "invalid IPv6 host and port".to_string(),
                ));
            }
        };
        return Ok((host.to_string(), port));
    }

    let (host, port) = match authority.rsplit_once(':') {
        Some((host, port)) if !host.is_empty() && !port.is_empty() => (host, parse_port(port)?),
        Some(_) => {
            return Err(HttpSyncClientError::InvalidServerUrl(
                "invalid host and port".to_string(),
            ));
        }
        None => (authority, default_port),
    };
    if host.is_empty() {
        return Err(HttpSyncClientError::InvalidServerUrl(
            "sync server URL is missing a host".to_string(),
        ));
    }
    Ok((host.to_string(), port))
}

fn parse_port(port: &str) -> Result<u16, HttpSyncClientError> {
    port.parse::<u16>().map_err(|_| {
        HttpSyncClientError::InvalidServerUrl("sync server URL has an invalid port".to_string())
    })
}

fn normalize_base_path(path: &str) -> Result<String, HttpSyncClientError> {
    if path.contains('?') || path.contains('#') {
        return Err(HttpSyncClientError::InvalidServerUrl(
            "sync server URL must not contain a query string or fragment".to_string(),
        ));
    }
    let trimmed = path.trim_matches('/');
    if trimmed.is_empty() {
        Ok(String::new())
    } else {
        Ok(format!("/{trimmed}"))
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct HttpResponse {
    status: u16,
    headers: BTreeMap<String, String>,
    body: Vec<u8>,
}

fn parse_http_response_bytes(bytes: Vec<u8>) -> Result<HttpResponse, HttpSyncClientError> {
    let header_end = find_header_end(&bytes).ok_or_else(|| {
        HttpSyncClientError::Protocol("server response did not contain HTTP headers".to_string())
    })?;
    if header_end > MAX_HEADER_BYTES {
        return Err(HttpSyncClientError::Protocol(
            "server response headers are too large".to_string(),
        ));
    }

    let header_text = std::str::from_utf8(&bytes[..header_end]).map_err(|_| {
        HttpSyncClientError::Protocol("server response headers are not UTF-8".to_string())
    })?;
    let (status, headers) = parse_response_headers(header_text)?;
    let body_start = header_end + 4;
    let body = bytes[body_start..].to_vec();
    if body.len() > MAX_BODY_BYTES {
        return Err(HttpSyncClientError::Protocol(
            "server response body is too large".to_string(),
        ));
    }
    if let Some(content_length) = headers.get("content-length") {
        let declared = content_length.parse::<usize>().map_err(|_| {
            HttpSyncClientError::Protocol("server response has invalid Content-Length".to_string())
        })?;
        if declared != body.len() {
            return Err(HttpSyncClientError::Protocol(
                "server response body length does not match Content-Length".to_string(),
            ));
        }
    }

    Ok(HttpResponse {
        status,
        headers,
        body,
    })
}

fn parse_response_headers(
    header_text: &str,
) -> Result<(u16, BTreeMap<String, String>), HttpSyncClientError> {
    let mut lines = header_text.split("\r\n");
    let status_line = lines
        .next()
        .ok_or_else(|| HttpSyncClientError::Protocol("missing HTTP status line".to_string()))?;
    let mut parts = status_line.split_whitespace();
    let version = parts
        .next()
        .ok_or_else(|| HttpSyncClientError::Protocol("missing HTTP version".to_string()))?;
    let status = parts
        .next()
        .ok_or_else(|| HttpSyncClientError::Protocol("missing HTTP status code".to_string()))?;
    if !version.starts_with("HTTP/") {
        return Err(HttpSyncClientError::Protocol(
            "invalid HTTP status line".to_string(),
        ));
    }
    let status = status
        .parse::<u16>()
        .map_err(|_| HttpSyncClientError::Protocol("invalid HTTP status code".to_string()))?;

    let mut headers = BTreeMap::new();
    for line in lines {
        if line.is_empty() {
            continue;
        }
        let Some((name, value)) = line.split_once(':') else {
            return Err(HttpSyncClientError::Protocol(
                "invalid HTTP response header".to_string(),
            ));
        };
        let name = name.trim().to_ascii_lowercase();
        let value = value.trim();
        if name.is_empty() {
            return Err(HttpSyncClientError::Protocol(
                "empty HTTP response header name".to_string(),
            ));
        }
        headers.insert(name, value.to_string());
    }

    Ok((status, headers))
}

fn find_header_end(bytes: &[u8]) -> Option<usize> {
    bytes.windows(4).position(|window| window == b"\r\n\r\n")
}

fn expect_success(
    response: HttpResponse,
    statuses: &[u16],
) -> Result<HttpResponse, HttpSyncClientError> {
    if statuses.contains(&response.status) {
        Ok(response)
    } else {
        Err(unexpected_status(response))
    }
}

fn unexpected_status(response: HttpResponse) -> HttpSyncClientError {
    let status = response.status;
    let body = response_error_summary(&response);
    match status {
        400 => HttpSyncClientError::Protocol(body),
        401 => HttpSyncClientError::Authentication(body),
        503 => HttpSyncClientError::ServerBusy {
            message: body,
            retry_after_seconds: response
                .headers
                .get("retry-after")
                .and_then(|value| value.parse::<u64>().ok()),
        },
        500..=599 => HttpSyncClientError::Server(body),
        _ => HttpSyncClientError::Protocol(format!("unexpected HTTP status {status}: {body}")),
    }
}

fn is_remote_deck_not_found_response(response: &HttpResponse) -> bool {
    response_text_lossy(&response.body) == "remote deck was not found"
}

fn upload_forbidden_error(response: &HttpResponse) -> SyncUploadError {
    match response_text_lossy(&response.body).as_str() {
        "sync upload is not allowed because this deck UUID is reserved for another owner" => {
            SyncUploadError::ReservedDeckIdentity
        }
        "sync upload is not allowed because this deck UUID has been retired" => {
            SyncUploadError::RetiredDeckIdentity
        }
        _ => SyncUploadError::Forbidden,
    }
}

fn response_error_summary(response: &HttpResponse) -> String {
    let body = response_text_lossy(&response.body);
    if response
        .headers
        .get("content-type")
        .is_some_and(|content_type| content_type.starts_with("text/html"))
    {
        return html_title(&body)
            .map(|title| format!("HTML response from server: {title}"))
            .unwrap_or_else(|| "HTML response from server".to_string());
    }
    const MAX_ERROR_BODY_CHARS: usize = 240;
    if body.chars().count() <= MAX_ERROR_BODY_CHARS {
        return body;
    }
    let mut summary = body.chars().take(MAX_ERROR_BODY_CHARS).collect::<String>();
    summary.push_str("...");
    summary
}

fn html_title(text: &str) -> Option<String> {
    let lower = text.to_ascii_lowercase();
    let start = lower.find("<title>")? + "<title>".len();
    let end = lower[start..].find("</title>")? + start;
    Some(text[start..end].trim().to_string()).filter(|title| !title.is_empty())
}

fn response_text(response: HttpResponse) -> Result<String, HttpSyncClientError> {
    String::from_utf8(response.body)
        .map_err(|_| HttpSyncClientError::Protocol("server response is not UTF-8".to_string()))
}

fn response_text_lossy(bytes: &[u8]) -> String {
    String::from_utf8_lossy(bytes).trim().to_string()
}

fn required_response_header(
    response: &HttpResponse,
    name: &'static str,
) -> Result<String, HttpSyncClientError> {
    response.headers.get(name).cloned().ok_or_else(|| {
        HttpSyncClientError::Protocol(format!("server response is missing {name} header"))
    })
}

fn parse_visibility(value: String) -> Result<DeckVisibility, HttpSyncClientError> {
    DeckVisibility::from_code(&value).ok_or_else(|| {
        HttpSyncClientError::Protocol(format!("server response has unknown visibility {value}"))
    })
}

fn parse_bool_header(header: &'static str, value: String) -> Result<bool, HttpSyncClientError> {
    match value.as_str() {
        "true" => Ok(true),
        "false" => Ok(false),
        _ => Err(HttpSyncClientError::Protocol(format!(
            "server response has invalid {header}"
        ))),
    }
}

fn bearer_authorization_header(token: &str) -> String {
    format!("Bearer {token}")
}

fn validate_sync_registration_path(path: &str) -> Result<(), HttpSyncClientError> {
    if path
        .chars()
        .any(|character| character.is_control() || character.is_whitespace())
    {
        return Err(HttpSyncClientError::Protocol(
            "sync server returned unsafe registration URL path".to_string(),
        ));
    }
    if !(path.starts_with("/auth/fedi/register?") || path.starts_with("/auth/fedi/authorize?"))
        || path.contains("://")
    {
        return Err(HttpSyncClientError::Protocol(
            "sync server returned unexpected registration URL path".to_string(),
        ));
    }
    Ok(())
}

fn parse_auth_session_json(text: &str) -> Result<SyncAuthSession, HttpSyncClientError> {
    let token_type = required_json_string_field(text, "token_type")?;
    if token_type != "Bearer" {
        return Err(HttpSyncClientError::Protocol(format!(
            "server returned unsupported sync auth token type {token_type}"
        )));
    }
    SyncAuthSession::from_parts(
        &required_json_string_field(text, "owner_identity")?,
        &required_json_string_field(text, "access_token")?,
    )
    .map_err(|error| HttpSyncClientError::Protocol(error.to_string()))
}

fn parse_upload_conflict(body: &str) -> Result<SyncUploadError, HttpSyncClientError> {
    if let Ok(conflict) = parse_upload_conflict_json(body) {
        return Ok(conflict);
    }

    parse_upload_conflict_text(body)
}

fn parse_upload_conflict_json(body: &str) -> Result<SyncUploadError, HttpSyncClientError> {
    let error = required_json_string_field(body, "error")?;
    if error != "revision_conflict" {
        return Err(HttpSyncClientError::Protocol(body.to_string()));
    }

    Ok(SyncUploadError::Conflict {
        expected: optional_json_string_field(body, "expected_revision")?,
        actual: optional_json_string_field(body, "actual_revision")?,
    })
}

fn parse_upload_conflict_text(body: &str) -> Result<SyncUploadError, HttpSyncClientError> {
    let Some(rest) = body.strip_prefix("remote revision conflict: expected ") else {
        return Err(HttpSyncClientError::Protocol(body.to_string()));
    };
    let Some((expected, actual)) = rest.split_once(", found ") else {
        return Err(HttpSyncClientError::Protocol(body.to_string()));
    };
    Ok(SyncUploadError::Conflict {
        expected: conflict_revision(expected),
        actual: conflict_revision(actual),
    })
}

fn conflict_revision(value: &str) -> Option<String> {
    let value = value.trim();
    (value != "none" && !value.is_empty()).then(|| value.to_string())
}

fn parse_shared_status_list(text: &str) -> Result<Vec<SyncSharedDeckStatus>, HttpSyncClientError> {
    json_array_entries(text)?
        .iter()
        .map(|entry| parse_shared_status(entry))
        .collect()
}

fn parse_owner_status_list(text: &str) -> Result<Vec<SyncRemoteDeckStatus>, HttpSyncClientError> {
    json_array_entries(text)?
        .iter()
        .map(|entry| parse_owner_status(entry))
        .collect()
}

fn parse_shared_status(text: &str) -> Result<SyncSharedDeckStatus, HttpSyncClientError> {
    SyncSharedDeckStatus::from_parts_with_summary(
        required_json_string_field(text, "deck_uuid")?
            .parse()
            .map_err(|error| {
                HttpSyncClientError::Protocol(format!(
                    "server response contains invalid deck UUID: {error}"
                ))
            })?,
        &required_json_string_field(text, "revision")?,
        parse_visibility(required_json_string_field(text, "visibility")?)?,
        &required_json_string_field(text, "owner_identity")?,
        &required_json_string_field(text, "content_bundle_hash")?,
        parse_deck_summary(text)?,
    )
    .map_err(HttpSyncClientError::from)
}

fn parse_owner_status(text: &str) -> Result<SyncRemoteDeckStatus, HttpSyncClientError> {
    let shared = parse_shared_status(text)?;
    SyncRemoteDeckStatus::from_shared_status(
        shared,
        optional_json_string_field(text, "study_history_bundle_hash")?.as_deref(),
        optional_json_string_field(text, "last_upload_device_name")?.as_deref(),
    )
    .map_err(HttpSyncClientError::from)
}

fn parse_deck_summary(text: &str) -> Result<SyncDeckSummary, HttpSyncClientError> {
    let value = serde_json::from_str::<serde_json::Value>(text).map_err(|error| {
        HttpSyncClientError::Protocol(format!("invalid remote deck status JSON: {error}"))
    })?;
    let object = value.as_object().ok_or_else(|| {
        HttpSyncClientError::Protocol("remote deck status is not a JSON object".to_string())
    })?;
    let title = optional_json_object_string_field(object, "title")?;
    let deck_name = optional_json_object_string_field(object, "deck_name")?;
    SyncDeckSummary::from_parts(
        title.as_deref(),
        deck_name.as_deref(),
        optional_json_object_usize_field(object, "card_count")?,
        optional_json_object_usize_field(object, "audio_file_count")?,
        optional_json_object_usize_field(object, "managed_audio_file_count")?,
        optional_json_object_string_array_field(object, "audio_formats")?,
    )
    .map_err(HttpSyncClientError::from)
}

fn parse_quota_status(text: &str) -> Result<SyncQuotaStatus, HttpSyncClientError> {
    let value = serde_json::from_str::<serde_json::Value>(text)
        .map_err(|error| HttpSyncClientError::Protocol(format!("invalid quota JSON: {error}")))?;
    let object = value.as_object().ok_or_else(|| {
        HttpSyncClientError::Protocol("quota response is not a JSON object".to_string())
    })?;
    Ok(SyncQuotaStatus::from_parts(
        optional_quota_usize_field(object, "max_decks_per_owner")?,
        optional_quota_u64_field(object, "max_storage_bytes_per_owner")?,
        optional_quota_u64_field(object, "max_storage_bytes_total")?,
        optional_quota_usize_field(object, "owner_deck_count")?,
        optional_quota_u64_field(object, "owner_storage_bytes")?,
    ))
}

fn optional_quota_usize_field(
    object: &serde_json::Map<String, serde_json::Value>,
    field: &'static str,
) -> Result<Option<usize>, HttpSyncClientError> {
    optional_quota_u64_field(object, field)?
        .map(|value| {
            usize::try_from(value).map_err(|_| {
                HttpSyncClientError::Protocol(format!(
                    "quota response contains out-of-range {field}"
                ))
            })
        })
        .transpose()
}

fn optional_quota_u64_field(
    object: &serde_json::Map<String, serde_json::Value>,
    field: &'static str,
) -> Result<Option<u64>, HttpSyncClientError> {
    match object.get(field) {
        None | Some(serde_json::Value::Null) => Ok(None),
        Some(serde_json::Value::Number(number)) => number
            .as_u64()
            .ok_or_else(|| {
                HttpSyncClientError::Protocol(format!(
                    "quota response contains invalid unsigned integer {field}"
                ))
            })
            .map(Some),
        Some(_) => Err(HttpSyncClientError::Protocol(format!(
            "quota response contains non-numeric {field}"
        ))),
    }
}

fn optional_json_object_string_field(
    object: &serde_json::Map<String, serde_json::Value>,
    field: &'static str,
) -> Result<Option<String>, HttpSyncClientError> {
    match object.get(field) {
        None | Some(serde_json::Value::Null) => Ok(None),
        Some(serde_json::Value::String(value)) => Ok(Some(value.clone())),
        Some(_) => Err(HttpSyncClientError::Protocol(format!(
            "remote deck status contains non-string {field}"
        ))),
    }
}

fn optional_json_object_usize_field(
    object: &serde_json::Map<String, serde_json::Value>,
    field: &'static str,
) -> Result<Option<usize>, HttpSyncClientError> {
    match object.get(field) {
        None | Some(serde_json::Value::Null) => Ok(None),
        Some(serde_json::Value::Number(number)) => number
            .as_u64()
            .ok_or_else(|| {
                HttpSyncClientError::Protocol(format!(
                    "remote deck status contains invalid unsigned integer {field}"
                ))
            })
            .and_then(|value| {
                usize::try_from(value).map_err(|_| {
                    HttpSyncClientError::Protocol(format!(
                        "remote deck status contains out-of-range {field}"
                    ))
                })
            })
            .map(Some),
        Some(_) => Err(HttpSyncClientError::Protocol(format!(
            "remote deck status contains non-numeric {field}"
        ))),
    }
}

fn optional_json_object_string_array_field(
    object: &serde_json::Map<String, serde_json::Value>,
    field: &'static str,
) -> Result<Vec<String>, HttpSyncClientError> {
    match object.get(field) {
        None | Some(serde_json::Value::Null) => Ok(Vec::new()),
        Some(serde_json::Value::Array(values)) => values
            .iter()
            .map(|value| match value {
                serde_json::Value::String(value) => Ok(value.clone()),
                _ => Err(HttpSyncClientError::Protocol(format!(
                    "remote deck status contains non-string {field} entry"
                ))),
            })
            .collect(),
        Some(_) => Err(HttpSyncClientError::Protocol(format!(
            "remote deck status contains non-array {field}"
        ))),
    }
}

fn json_array_entries(text: &str) -> Result<Vec<&str>, HttpSyncClientError> {
    let text = text.trim();
    if !text.starts_with('[') || !text.ends_with(']') {
        return Err(HttpSyncClientError::Protocol(
            "server response is not a JSON array".to_string(),
        ));
    }
    let inner = &text[1..text.len() - 1];
    let mut entries = Vec::new();
    let mut object_start = None;
    let mut depth = 0usize;
    let mut in_string = false;
    let mut escaped = false;
    for (index, character) in inner.char_indices() {
        if in_string {
            if escaped {
                escaped = false;
            } else if character == '\\' {
                escaped = true;
            } else if character == '"' {
                in_string = false;
            }
            continue;
        }

        match character {
            '"' => in_string = true,
            '{' => {
                if depth == 0 {
                    object_start = Some(index);
                }
                depth += 1;
            }
            '}' => {
                depth = depth.checked_sub(1).ok_or_else(|| {
                    HttpSyncClientError::Protocol("invalid JSON object nesting".to_string())
                })?;
                if depth == 0 {
                    let start = object_start.take().ok_or_else(|| {
                        HttpSyncClientError::Protocol("invalid JSON object".to_string())
                    })?;
                    entries.push(&inner[start..=index]);
                }
            }
            _ => {}
        }
    }
    if depth != 0 || in_string {
        return Err(HttpSyncClientError::Protocol(
            "unterminated JSON in server response".to_string(),
        ));
    }
    Ok(entries)
}

fn required_json_string_field(
    text: &str,
    field: &'static str,
) -> Result<String, HttpSyncClientError> {
    optional_json_string_field(text, field)?
        .ok_or_else(|| HttpSyncClientError::Protocol(format!("server response is missing {field}")))
}

fn optional_json_string_field(
    text: &str,
    field: &'static str,
) -> Result<Option<String>, HttpSyncClientError> {
    let marker = format!("\"{field}\":");
    let Some(start) = text.find(&marker) else {
        return Ok(None);
    };
    let mut index = start + marker.len();
    skip_json_whitespace(text, &mut index);
    if text[index..].starts_with("null") {
        return Ok(None);
    }
    parse_json_string_at(text, &mut index).map(Some)
}

fn skip_json_whitespace(text: &str, index: &mut usize) {
    while let Some(character) = text[*index..].chars().next() {
        if !character.is_whitespace() {
            break;
        }
        *index += character.len_utf8();
    }
}

fn parse_json_string_at(text: &str, index: &mut usize) -> Result<String, HttpSyncClientError> {
    if !text[*index..].starts_with('"') {
        return Err(HttpSyncClientError::Protocol(
            "server response contains a non-string JSON field".to_string(),
        ));
    }
    *index += 1;
    let mut value = String::new();
    while *index < text.len() {
        let character = text[*index..]
            .chars()
            .next()
            .ok_or_else(|| HttpSyncClientError::Protocol("unterminated JSON string".to_string()))?;
        *index += character.len_utf8();
        match character {
            '"' => return Ok(value),
            '\\' => value.push(parse_json_escape(text, index)?),
            character => value.push(character),
        }
    }
    Err(HttpSyncClientError::Protocol(
        "unterminated JSON string".to_string(),
    ))
}

fn parse_json_escape(text: &str, index: &mut usize) -> Result<char, HttpSyncClientError> {
    let character = text[*index..]
        .chars()
        .next()
        .ok_or_else(|| HttpSyncClientError::Protocol("unterminated JSON escape".to_string()))?;
    *index += character.len_utf8();
    match character {
        '"' => Ok('"'),
        '\\' => Ok('\\'),
        '/' => Ok('/'),
        'b' => Ok('\u{0008}'),
        'f' => Ok('\u{000c}'),
        'n' => Ok('\n'),
        'r' => Ok('\r'),
        't' => Ok('\t'),
        'u' => {
            let end = *index + 4;
            let hex = text.get(*index..end).ok_or_else(|| {
                HttpSyncClientError::Protocol("short JSON unicode escape".to_string())
            })?;
            *index = end;
            let code = u32::from_str_radix(hex, 16).map_err(|_| {
                HttpSyncClientError::Protocol("invalid JSON unicode escape".to_string())
            })?;
            char::from_u32(code).ok_or_else(|| {
                HttpSyncClientError::Protocol("invalid JSON unicode scalar".to_string())
            })
        }
        _ => Err(HttpSyncClientError::Protocol(
            "invalid JSON escape".to_string(),
        )),
    }
}

#[derive(Debug)]
pub enum HttpSyncClientError {
    InvalidServerUrl(String),
    UnsupportedScheme(String),
    Io(io::Error),
    Transport(reqwest::Error),
    Authentication(String),
    Protocol(String),
    Server(String),
    ServerBusy {
        message: String,
        retry_after_seconds: Option<u64>,
    },
    Upload(SyncUploadError),
    Download(SyncDownloadError),
    Delete(SyncDeleteError),
    Validation(SyncValidationError),
}

impl fmt::Display for HttpSyncClientError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidServerUrl(message) => write!(f, "invalid sync server URL: {message}"),
            Self::UnsupportedScheme(scheme) => {
                write!(f, "{scheme} sync URLs are not supported by this build")
            }
            Self::Io(error) => write!(f, "sync HTTP connection failed: {error}"),
            Self::Transport(error) => write!(f, "sync HTTP request failed: {error}"),
            Self::Authentication(message) => write!(f, "sync authentication failed: {message}"),
            Self::Protocol(message) => write!(f, "sync HTTP protocol error: {message}"),
            Self::Server(message) => write!(f, "sync server error: {message}"),
            Self::ServerBusy {
                message,
                retry_after_seconds,
            } => match retry_after_seconds {
                Some(seconds) => write!(
                    f,
                    "sync server is temporarily busy: {message}. Try again in about {seconds} seconds."
                ),
                None => write!(
                    f,
                    "sync server is temporarily busy: {message}. Try again later."
                ),
            },
            Self::Upload(error) => error.fmt(f),
            Self::Download(error) => error.fmt(f),
            Self::Delete(error) => error.fmt(f),
            Self::Validation(error) => error.fmt(f),
        }
    }
}

impl std::error::Error for HttpSyncClientError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io(error) => Some(error),
            Self::Transport(error) => Some(error),
            Self::Upload(error) => Some(error),
            Self::Download(error) => Some(error),
            Self::Delete(error) => Some(error),
            Self::Validation(error) => Some(error),
            Self::InvalidServerUrl(_)
            | Self::UnsupportedScheme(_)
            | Self::Authentication(_)
            | Self::Protocol(_)
            | Self::Server(_)
            | Self::ServerBusy { .. } => None,
        }
    }
}

impl From<io::Error> for HttpSyncClientError {
    fn from(error: io::Error) -> Self {
        Self::Io(error)
    }
}

impl From<reqwest::Error> for HttpSyncClientError {
    fn from(error: reqwest::Error) -> Self {
        Self::Transport(error)
    }
}

impl From<SyncUploadError> for HttpSyncClientError {
    fn from(error: SyncUploadError) -> Self {
        Self::Upload(error)
    }
}

impl From<SyncDownloadError> for HttpSyncClientError {
    fn from(error: SyncDownloadError) -> Self {
        Self::Download(error)
    }
}

impl From<SyncDeleteError> for HttpSyncClientError {
    fn from(error: SyncDeleteError) -> Self {
        Self::Delete(error)
    }
}

impl From<SyncValidationError> for HttpSyncClientError {
    fn from(error: SyncValidationError) -> Self {
        Self::Validation(error)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sync_http::ServerState;
    use crate::sync_oauth::{FediOAuthApplication, FediOAuthError, SyncOAuthProvider};
    use crate::{
        LocalDeviceIdentity, SyncBundlePayload, SyncOAuthPendingLogin, SyncServerLimits,
        sync_bundle_hash,
    };
    use std::sync::{Arc, Mutex};

    const DECK_UUID: &str = "550e8400-e29b-41d4-a716-446655440000";
    fn authenticated_test_client() -> HttpSyncClient {
        let client = test_oauth_client();
        let start = start_and_accept_test_login(&client);
        let session = client
            .poll_fedi_browser_login(start.login_token())
            .unwrap()
            .unwrap();
        client.with_bearer_token(session.bearer_token().to_string())
    }

    fn start_and_accept_test_login(client: &HttpSyncClient) -> SyncOAuthStart {
        let start = client.start_fedi_browser_login().unwrap();
        assert!(
            start
                .registration_url()
                .is_some_and(|url| url.contains("/auth/fedi/register?login_token="))
        );
        let body = format!(
            "login_token={}&fedi_instance=example.invalid&privacy_acceptance=yes",
            start.login_token()
        );
        let response = client
            .send(
                "POST",
                "/auth/fedi/register",
                &[("Content-Type", "application/x-www-form-urlencoded")],
                body.as_bytes(),
            )
            .unwrap();
        expect_success(response, &[303]).unwrap();
        let callback = client
            .send(
                "GET",
                &format!(
                    "/auth/fedi/callback?state={}&code=code",
                    start.login_token()
                ),
                &[],
                &[],
            )
            .unwrap();
        expect_success(callback, &[200]).unwrap();
        start
    }

    fn test_oauth_client() -> HttpSyncClient {
        HttpSyncClient::for_test_server(
            "http://sync.example.invalid",
            Arc::new(ServerState::memory_with_oauth_provider(Arc::new(
                TestOAuthProvider,
            ))),
        )
        .unwrap()
    }

    fn test_oauth_client_with_limits(limits: SyncServerLimits) -> HttpSyncClient {
        HttpSyncClient::for_test_server(
            "http://sync.example.invalid",
            Arc::new(ServerState::memory_with_oauth_provider_and_limits(
                Arc::new(TestOAuthProvider),
                limits,
            )),
        )
        .unwrap()
    }

    #[derive(Debug)]
    struct TestOAuthProvider;

    impl SyncOAuthProvider for TestOAuthProvider {
        fn register_application_with_redirect_uri(
            &self,
            instance_url: &str,
            redirect_uri: &str,
        ) -> Result<FediOAuthApplication, FediOAuthError> {
            FediOAuthApplication::from_parts_with_redirect_uri(
                instance_url,
                "client-id",
                "client-secret",
                redirect_uri,
            )
        }

        fn exchange_code_for_owner_identity(
            &self,
            pending: &SyncOAuthPendingLogin,
            authorization_code: &str,
        ) -> Result<String, FediOAuthError> {
            assert_eq!(pending.client_id(), "client-id");
            assert_eq!(authorization_code, "code");
            Ok("acct:modulus@example.invalid".to_string())
        }
    }

    #[test]
    fn response_progress_total_uses_bundle_length_when_content_length_is_hidden() {
        let mut headers = BTreeMap::new();
        headers.insert("x-cosimo-bundle-length".to_string(), "12345".to_string());

        assert_eq!(
            response_body_progress_total(None, &headers).unwrap(),
            Some(12345)
        );
    }

    #[test]
    fn plain_http_response_header_info_uses_bundle_length_fallback() {
        let bytes = b"HTTP/1.1 200 OK\r\nx-cosimo-bundle-length: 20\r\n\r\npartial";
        let body_start = bytes
            .windows(4)
            .position(|window| window == b"\r\n\r\n")
            .unwrap()
            + 4;

        assert_eq!(
            response_header_info(bytes).unwrap(),
            Some((body_start, Some(20)))
        );
    }

    fn upload_request(expected_revision: Option<&str>) -> SyncUploadRequest {
        upload_request_for_owner("acct:modulus@example.invalid", expected_revision)
    }

    fn upload_request_for_owner(
        owner_identity: &str,
        expected_revision: Option<&str>,
    ) -> SyncUploadRequest {
        SyncUploadRequest::from_parts(
            DECK_UUID.parse().unwrap(),
            owner_identity,
            DeckVisibility::Public,
            expected_revision,
            LocalDeviceIdentity::new(Some("laptop".to_string())),
            Some(payload(b"content bundle", false)),
            Some(payload(b"history bundle", true)),
        )
        .unwrap()
    }

    fn payload(bytes: &[u8], includes_review_database: bool) -> SyncBundlePayload {
        let bytes = bytes.to_vec();
        let bundle_hash = sync_bundle_hash(&bytes);
        SyncBundlePayload::from_parts(bytes, &bundle_hash, includes_review_database).unwrap()
    }

    #[test]
    fn endpoint_parsing_supports_base_paths() {
        let endpoint = HttpSyncEndpoint::parse("http://127.0.0.1:44791/cosimo-sync.cgi").unwrap();

        assert_eq!(endpoint.scheme, HttpSyncScheme::Http);
        assert_eq!(endpoint.host, "127.0.0.1");
        assert_eq!(endpoint.port, 44791);
        assert_eq!(endpoint.host_header, "127.0.0.1:44791");
        assert_eq!(
            endpoint.request_target("/api/v1/health"),
            "/cosimo-sync.cgi/api/v1/health"
        );
    }

    #[test]
    fn account_deletion_url_uses_sync_server_base_path() {
        let client = HttpSyncClient::new("https://sync.example.invalid/sync.cgi").unwrap();

        assert_eq!(
            client.account_deletion_url(),
            "https://sync.example.invalid/sync.cgi/auth/fedi/delete"
        );
    }

    #[test]
    fn endpoint_parsing_supports_https() {
        let endpoint = HttpSyncEndpoint::parse("https://example.invalid/sync").unwrap();

        assert_eq!(endpoint.scheme, HttpSyncScheme::Https);
        assert_eq!(endpoint.host, "example.invalid");
        assert_eq!(endpoint.port, 443);
        assert_eq!(endpoint.host_header, "example.invalid");
        assert_eq!(
            endpoint.absolute_url("/api/v1/health"),
            "https://example.invalid/sync/api/v1/health"
        );
    }

    #[test]
    fn endpoint_parsing_keeps_explicit_https_port() {
        let endpoint = HttpSyncEndpoint::parse("https://example.invalid:8443/sync").unwrap();

        assert_eq!(endpoint.scheme, HttpSyncScheme::Https);
        assert_eq!(endpoint.port, 8443);
        assert_eq!(endpoint.host_header, "example.invalid:8443");
        assert_eq!(
            endpoint.absolute_url("/api/v1/health"),
            "https://example.invalid:8443/sync/api/v1/health"
        );
    }

    #[test]
    fn endpoint_parsing_rejects_non_loopback_plain_http() {
        let error = HttpSyncEndpoint::parse("http://example.invalid/sync").unwrap_err();

        assert!(matches!(error, HttpSyncClientError::InvalidServerUrl(_)));
    }

    #[test]
    fn endpoint_parsing_allows_loopback_plain_http_for_development() {
        let localhost = HttpSyncEndpoint::parse("http://localhost:44791/sync").unwrap();
        let ipv6 = HttpSyncEndpoint::parse("http://[::1]:44791/sync").unwrap();

        assert_eq!(localhost.scheme, HttpSyncScheme::Http);
        assert_eq!(localhost.host_header, "localhost:44791");
        assert_eq!(ipv6.host_header, "[::1]:44791");
    }

    #[test]
    fn endpoint_parsing_rejects_request_line_injection() {
        let error = HttpSyncEndpoint::parse("http://example.invalid/\r\nX-Bad: yes").unwrap_err();

        assert!(matches!(error, HttpSyncClientError::InvalidServerUrl(_)));
    }

    #[test]
    fn transfer_requests_use_longer_total_timeout_but_keep_short_connect_timeout() {
        assert_eq!(
            sync_http_client_timeouts(false),
            SyncHttpClientTimeouts {
                total: Duration::from_secs(30),
                connect: Duration::from_secs(30),
            }
        );
        assert_eq!(
            sync_http_client_timeouts(true),
            SyncHttpClientTimeouts {
                total: Duration::from_secs(90),
                connect: Duration::from_secs(30),
            }
        );
    }

    #[test]
    fn http_client_uploads_and_downloads_owner_history() {
        let client = authenticated_test_client();

        let upload = client.upload(upload_request(None)).unwrap();
        assert_eq!(upload.revision(), "1");
        assert_eq!(upload.bundle_hash(), sync_bundle_hash(b"history bundle"));

        let deck_uuid: SyncUuid = DECK_UUID.parse().unwrap();
        let download = client
            .download(
                &deck_uuid,
                SyncDownloadKind::OwnerStudyHistory,
                Some("acct:modulus@example.invalid"),
            )
            .unwrap();

        assert_eq!(download.revision(), "1");
        assert_eq!(download.bundle().bytes(), b"history bundle");
        assert!(download.bundle().includes_review_database());
        assert_eq!(download.last_upload_device_name(), Some("laptop"));
    }

    #[test]
    fn http_client_reports_upload_and_download_transfer_progress() {
        let client = authenticated_test_client();
        let upload_progress = Arc::new(Mutex::new(Vec::new()));
        let upload_progress_writer = Arc::clone(&upload_progress);
        let uploaded = client
            .upload_with_progress(upload_request(None), move |transferred, total| {
                upload_progress_writer
                    .lock()
                    .unwrap()
                    .push((transferred, total));
            })
            .unwrap();
        let upload_bytes = b"content bundle".len() + b"history bundle".len();

        assert_eq!(uploaded.revision(), "1");
        assert!(
            upload_progress
                .lock()
                .unwrap()
                .contains(&(upload_bytes as u64, Some(upload_bytes as u64)))
        );

        let deck_uuid: SyncUuid = DECK_UUID.parse().unwrap();
        let download_progress = Arc::new(Mutex::new(Vec::new()));
        let download_progress_writer = Arc::clone(&download_progress);
        let download = client
            .download_with_progress(
                &deck_uuid,
                SyncDownloadKind::OwnerStudyHistory,
                Some("acct:modulus@example.invalid"),
                move |transferred, total| {
                    download_progress_writer
                        .lock()
                        .unwrap()
                        .push((transferred, total));
                },
            )
            .unwrap();
        let download_bytes = b"history bundle".len();

        assert_eq!(download.bundle().bytes(), b"history bundle");
        assert!(
            download_progress
                .lock()
                .unwrap()
                .contains(&(download_bytes as u64, Some(download_bytes as u64)))
        );
    }

    #[test]
    fn http_client_lists_public_and_owner_decks() {
        let client = authenticated_test_client();
        client.upload(upload_request(None)).unwrap();

        let public = client.list_public_decks().unwrap();
        let owner = client
            .list_owner_decks("acct:modulus@example.invalid")
            .unwrap();

        assert_eq!(public.len(), 1);
        assert_eq!(public[0].deck_uuid().as_str(), DECK_UUID);
        assert_eq!(owner.len(), 1);
        let history_hash = sync_bundle_hash(b"history bundle");
        assert_eq!(
            owner[0].study_history_bundle_hash(),
            Some(history_hash.as_str())
        );
    }

    #[test]
    fn remote_status_parser_reads_deck_summary_fields() {
        let status = parse_shared_status(
            r#"{"deck_uuid":"550e8400-e29b-41d4-a716-446655440000","revision":"3","visibility":"public","owner_identity":"acct:modulus@example.invalid","content_bundle_hash":"sha256:content","title":"Indonesian","deck_name":"deck","card_count":284,"audio_file_count":568,"managed_audio_file_count":400,"audio_formats":["opus","wav"]}"#,
        )
        .unwrap();

        assert_eq!(status.summary().title(), Some("Indonesian"));
        assert_eq!(status.summary().deck_name(), Some("deck"));
        assert_eq!(status.summary().card_count(), Some(284));
        assert_eq!(status.summary().audio_file_count(), Some(568));
        assert_eq!(status.summary().managed_audio_file_count(), Some(400));
        assert_eq!(status.summary().manual_audio_file_count(), Some(168));
        assert_eq!(
            status.summary().audio_formats(),
            ["opus".to_string(), "wav".to_string()]
        );
    }

    #[test]
    fn http_client_reads_quota_status() {
        let client = test_oauth_client_with_limits(SyncServerLimits {
            max_decks_per_owner: Some(7),
            max_storage_bytes_per_owner: Some(16_384),
            max_storage_bytes_total: Some(65_536),
            ..SyncServerLimits::default()
        });
        let start = start_and_accept_test_login(&client);
        let session = client
            .poll_fedi_browser_login(start.login_token())
            .unwrap()
            .unwrap();
        let client = client.with_bearer_token(session.bearer_token().to_string());
        client.upload(upload_request(None)).unwrap();

        let status = client.quota_status().unwrap();

        assert_eq!(status.max_decks_per_owner(), Some(7));
        assert_eq!(status.max_storage_bytes_per_owner(), Some(16_384));
        assert_eq!(status.max_storage_bytes_total(), Some(65_536));
        assert_eq!(status.owner_deck_count(), Some(1));
        assert_eq!(status.owner_storage_bytes(), Some(14));
    }

    #[test]
    fn http_client_reads_public_quota_without_login() {
        let client = test_oauth_client_with_limits(SyncServerLimits {
            max_decks_per_owner: Some(7),
            max_storage_bytes_per_owner: Some(16_384),
            max_storage_bytes_total: Some(65_536),
            ..SyncServerLimits::default()
        });

        let status = client.quota_status().unwrap();

        assert_eq!(status.max_decks_per_owner(), Some(7));
        assert_eq!(status.max_storage_bytes_per_owner(), Some(16_384));
        assert_eq!(status.max_storage_bytes_total(), Some(65_536));
        assert_eq!(status.owner_deck_count(), None);
        assert_eq!(status.owner_storage_bytes(), None);
    }

    #[test]
    fn quota_parser_uses_json_structure_not_raw_text_search() {
        let status = parse_quota_status(
            r#"{"note":"\"owner_storage_bytes\":999","owner_storage_bytes":14}"#,
        )
        .unwrap();
        let missing = parse_quota_status(r#"{"note":"\"owner_storage_bytes\":999"}"#).unwrap();

        assert_eq!(status.owner_storage_bytes(), Some(14));
        assert_eq!(missing.owner_storage_bytes(), None);
    }

    #[test]
    fn quota_parser_rejects_invalid_numeric_shapes() {
        for text in [
            r#"[]"#,
            r#"{"owner_storage_bytes":"14"}"#,
            r#"{"owner_storage_bytes":-1}"#,
            r#"{"owner_storage_bytes":1.5}"#,
            r#"{"owner_storage_bytes":true}"#,
            r#"{"owner_storage_bytes":14x}"#,
            r#"{"owner_storage_bytes":18446744073709551616}"#,
        ] {
            let error = parse_quota_status(text).unwrap_err();

            assert!(matches!(error, HttpSyncClientError::Protocol(_)));
        }
    }

    #[test]
    fn client_distinguishes_cosimo_deck_404_from_server_route_404() {
        let cosimo_response = HttpResponse {
            status: 404,
            headers: BTreeMap::from([(
                "content-type".to_string(),
                "text/plain; charset=utf-8".to_string(),
            )]),
            body: b"remote deck was not found".to_vec(),
        };
        let apache_response = HttpResponse {
            status: 404,
            headers: BTreeMap::from([(
                "content-type".to_string(),
                "text/html; charset=iso-8859-1".to_string(),
            )]),
            body: b"<!doctype html><title>404 Not Found</title><h1>Not Found</h1>".to_vec(),
        };

        assert!(is_remote_deck_not_found_response(&cosimo_response));
        assert!(!is_remote_deck_not_found_response(&apache_response));
        assert_eq!(
            unexpected_status(apache_response).to_string(),
            "sync HTTP protocol error: unexpected HTTP status 404: HTML response from server: 404 Not Found"
        );
    }

    #[test]
    fn upload_forbidden_response_preserves_specific_identity_errors() {
        let reserved_response = HttpResponse {
            status: 403,
            headers: BTreeMap::new(),
            body:
                b"sync upload is not allowed because this deck UUID is reserved for another owner"
                    .to_vec(),
        };
        let retired_response = HttpResponse {
            status: 403,
            headers: BTreeMap::new(),
            body: b"sync upload is not allowed because this deck UUID has been retired".to_vec(),
        };
        let generic_response = HttpResponse {
            status: 403,
            headers: BTreeMap::new(),
            body: b"sync upload is not allowed for this owner".to_vec(),
        };

        assert_eq!(
            upload_forbidden_error(&reserved_response),
            SyncUploadError::ReservedDeckIdentity
        );
        assert_eq!(
            upload_forbidden_error(&retired_response),
            SyncUploadError::RetiredDeckIdentity
        );
        assert_eq!(
            upload_forbidden_error(&generic_response),
            SyncUploadError::Forbidden
        );
    }

    #[test]
    fn http_client_deletes_owned_deck() {
        let client = authenticated_test_client();
        client.upload(upload_request(None)).unwrap();
        let deck_uuid: SyncUuid = DECK_UUID.parse().unwrap();

        client.delete_deck(&deck_uuid).unwrap();

        let public = client.list_public_decks().unwrap();
        let owner = client
            .list_owner_decks("acct:modulus@example.invalid")
            .unwrap();
        let download_error = client
            .download(&deck_uuid, SyncDownloadKind::OwnerStudyHistory, None)
            .unwrap_err();

        assert!(public.is_empty());
        assert!(owner.is_empty());
        assert!(matches!(
            download_error,
            HttpSyncClientError::Download(SyncDownloadError::NotFound)
        ));
    }

    #[test]
    fn http_client_uses_fedi_bearer_session_for_owner_routes() {
        let client = test_oauth_client();
        let start = start_and_accept_test_login(&client);
        let session = client
            .poll_fedi_browser_login(start.login_token())
            .unwrap()
            .unwrap();
        let client = client.with_bearer_token(session.bearer_token().to_string());

        client
            .upload(upload_request_for_owner("acct:other@example.invalid", None))
            .unwrap();

        let owner = client
            .list_owner_decks("acct:ignored@example.invalid")
            .unwrap();
        let deck_uuid: SyncUuid = DECK_UUID.parse().unwrap();
        let download = client
            .download(&deck_uuid, SyncDownloadKind::OwnerStudyHistory, None)
            .unwrap();

        assert_eq!(session.owner_identity(), "acct:modulus@example.invalid");
        assert_eq!(owner.len(), 1);
        assert_eq!(owner[0].owner_identity(), "acct:modulus@example.invalid");
        assert!(download.bundle().includes_review_database());
    }

    #[test]
    fn http_client_can_complete_fedi_browser_login_and_use_bearer_session() {
        let client = test_oauth_client();
        let start = client.start_fedi_browser_login().unwrap();
        assert!(
            start
                .registration_url()
                .is_some_and(|url| url.contains("/auth/fedi/register?login_token="))
        );
        assert!(
            client
                .poll_fedi_browser_login(start.login_token())
                .unwrap()
                .is_none()
        );

        let body = format!(
            "login_token={}&fedi_instance=example.invalid&privacy_acceptance=yes",
            start.login_token()
        );
        let registration = client
            .send(
                "POST",
                "/auth/fedi/register",
                &[("Content-Type", "application/x-www-form-urlencoded")],
                body.as_bytes(),
            )
            .unwrap();
        let registration = expect_success(registration, &[303]).unwrap();
        let location = registration.headers.get("location").unwrap();
        assert!(location.contains("https://example.invalid/oauth/authorize?"));
        assert!(
            location.contains(
                "redirect_uri=https%3A%2F%2Fsync.example.invalid%2Fauth%2Ffedi%2Fcallback"
            )
        );
        assert!(location.contains(&format!("state={}", start.login_token())));

        let callback = client
            .send(
                "GET",
                &format!(
                    "/auth/fedi/callback?state={}&code=code",
                    start.login_token()
                ),
                &[],
                &[],
            )
            .unwrap();
        expect_success(callback, &[200]).unwrap();

        let session = client
            .poll_fedi_browser_login(start.login_token())
            .unwrap()
            .unwrap();
        let client = client.with_bearer_token(session.bearer_token().to_string());
        client
            .upload(upload_request_for_owner("acct:other@example.invalid", None))
            .unwrap();

        let owner = client
            .list_owner_decks("acct:ignored@example.invalid")
            .unwrap();
        assert_eq!(session.owner_identity(), "acct:modulus@example.invalid");
        assert_eq!(owner.len(), 1);
        assert_eq!(owner[0].owner_identity(), "acct:modulus@example.invalid");
    }

    #[test]
    fn http_client_can_complete_fedi_browser_device_login_for_existing_account() {
        let client = test_oauth_client();
        let registration_start = start_and_accept_test_login(&client);
        client
            .poll_fedi_browser_login(registration_start.login_token())
            .unwrap()
            .unwrap();

        let start = client.start_fedi_browser_device_login().unwrap();
        assert!(
            start
                .registration_url()
                .is_some_and(|url| url.contains("/auth/fedi/authorize?login_token="))
        );
        assert!(
            client
                .poll_fedi_browser_login(start.login_token())
                .unwrap()
                .is_none()
        );

        let body = format!(
            "login_token={}&fedi_instance=%40modulus%40example.invalid",
            start.login_token()
        );
        let authorization = client
            .send(
                "POST",
                "/auth/fedi/authorize",
                &[("Content-Type", "application/x-www-form-urlencoded")],
                body.as_bytes(),
            )
            .unwrap();
        let authorization = expect_success(authorization, &[303]).unwrap();
        let location = authorization.headers.get("location").unwrap();
        assert!(location.contains("https://example.invalid/oauth/authorize?"));
        assert!(location.contains(
            "redirect_uri=https%3A%2F%2Fsync.example.invalid%2Fauth%2Ffedi%2Fauthorize%2Fcallback"
        ));
        assert!(location.contains(&format!("state={}", start.login_token())));

        let callback = client
            .send(
                "GET",
                &format!(
                    "/auth/fedi/authorize/callback?state={}&code=code",
                    start.login_token()
                ),
                &[],
                &[],
            )
            .unwrap();
        expect_success(callback, &[200]).unwrap();

        let session = client
            .poll_fedi_browser_login(start.login_token())
            .unwrap()
            .unwrap();
        assert_eq!(session.owner_identity(), "acct:modulus@example.invalid");
    }

    #[test]
    fn http_client_can_revoke_bearer_session() {
        let client = authenticated_test_client();
        client
            .list_owner_decks("acct:modulus@example.invalid")
            .unwrap();

        client.revoke_auth_session().unwrap();
        let error = client
            .list_owner_decks("acct:modulus@example.invalid")
            .unwrap_err();

        assert!(matches!(error, HttpSyncClientError::Authentication(_)));
        assert!(error.to_string().contains("authentication failed"));
    }

    #[test]
    fn http_client_can_revoke_all_owner_bearer_sessions() {
        let client = test_oauth_client();
        let first_start = start_and_accept_test_login(&client);
        let first_session = client
            .poll_fedi_browser_login(first_start.login_token())
            .unwrap()
            .unwrap();
        let second_start = start_and_accept_test_login(&client);
        let second_session = client
            .poll_fedi_browser_login(second_start.login_token())
            .unwrap()
            .unwrap();
        let first_client = client
            .clone()
            .with_bearer_token(first_session.bearer_token().to_string());
        let second_client = client.with_bearer_token(second_session.bearer_token().to_string());

        first_client.revoke_all_auth_sessions().unwrap();

        for client in [first_client, second_client] {
            let error = client
                .list_owner_decks("acct:modulus@example.invalid")
                .unwrap_err();
            assert!(matches!(error, HttpSyncClientError::Authentication(_)));
            assert!(error.to_string().contains("authentication failed"));
        }
    }

    #[test]
    fn http_client_reports_revision_conflicts() {
        let client = authenticated_test_client();
        client.upload(upload_request(None)).unwrap();

        let error = client.upload(upload_request(None)).unwrap_err();

        assert!(matches!(
            error,
            HttpSyncClientError::Upload(SyncUploadError::Conflict {
                expected: None,
                actual: Some(_)
            })
        ));
    }

    #[test]
    fn conflict_parser_accepts_legacy_text_response() {
        let error =
            parse_upload_conflict("remote revision conflict: expected none, found 42").unwrap();

        assert_eq!(
            error,
            SyncUploadError::Conflict {
                expected: None,
                actual: Some("42".to_string())
            }
        );
    }

    #[test]
    fn json_string_parser_handles_escaped_response_values() {
        let text = r#"{"owner_identity":"acct:\"modulus\"@example.invalid","last_upload_device_name":"line\nfeed"}"#;

        assert_eq!(
            required_json_string_field(text, "owner_identity").unwrap(),
            "acct:\"modulus\"@example.invalid"
        );
        assert_eq!(
            optional_json_string_field(text, "last_upload_device_name").unwrap(),
            Some("line\nfeed".to_string())
        );
    }

    #[test]
    fn header_values_reject_line_breaks() {
        let error =
            validate_http_header_value("acct:owner@example.invalid\r\nX-Bad: yes").unwrap_err();

        assert!(matches!(error, HttpSyncClientError::Protocol(_)));
    }
}
