//! Module for interacting with the Typesense client.
#![cfg(feature = "client")]
use bytes::buf::Buf;
use omry_archiving::{Record, ToRecord};
use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue, InvalidHeaderValue};
use reqwest::{Client, ClientBuilder, Response, StatusCode, Url};
use secrecy::ExposeSecret;
pub use secrecy::SecretBox;
use serde::Deserialize;
use std::fmt;
use std::future::Future;
use std::io::{BufRead, BufReader, Read};
use thiserror::Error;
use tracing::instrument;

/// Type alias for `UUIDv7` IDs that Typesense assigns
/// to new documents.
pub type SearchId = String;

use crate::model::{Collection, SearchRecord, SearchRecordError};
use crate::{SearchParameters, SearchResponse};

/// Module for API responses not directly related to searching documents.
pub mod api_responses {
    #[cfg(feature = "serde")]
    use serde::{Deserialize, Serialize};

    /// Resource error types that Typesense can report via /health.
    #[derive(Debug)]
    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
    #[serde(rename_all = "SCREAMING_SNAKE_CASE")]
    pub enum ResourceError {
        /// Corresponds to `OUT_OF_DISK` in /health response.
        OutOfDisk,

        /// Corresponds to `OUT_OF_MEMORY` in /health response.
        OutOfMemory,
    }

    /// Response from /health endpoint.
    ///
    /// See <https://typesense.org/docs/27.1/api/cluster-operations.html#health>.
    #[derive(Debug)]
    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
    pub struct Health {
        /// Health status.
        pub ok: bool,

        /// Resource error, if there is one.
        pub resource_error: Option<ResourceError>,
    }

    /// Typesense error response with an arbitrary message.
    #[derive(Debug)]
    #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
    pub struct ErrorWithMessage {
        /// The cause of this error.
        pub message: String,
    }

    /// Response for deleting records by query.
    ///
    /// See <https://typesense.org/docs/29.0/api/documents.html#delete-by-query>.
    #[derive(Debug, Deserialize)]
    pub struct DeleteByQuery {
        /// Number of records that were deleted.
        pub num_deleted: u64,
    }
}

/// The header name that corresponds to the API key.
const API_KEY_HEADER_NAME: HeaderName = HeaderName::from_static("x-typesense-api-key");

/// The default URL of the Typesense instance we communicate with.
const DEFAULT_TYPESENSE_URL: &str = "http://localhost:8108/";

/// Type alias for a secret API key.
type ApiKey = SecretBox<str>;

/// Type alias to denote a Typesense collection field name.
pub type FieldName = String;

/// Response given by Typesense import API.
/// See <https://typesense.org/docs/29.0/api/documents.html#index-multiple-documents>.
#[allow(clippy::missing_docs_in_private_items)]
#[derive(Debug, Deserialize)]
struct ImportResponse {
    success: bool,
    error: Option<String>,
    document: Option<String>,
}

impl ImportResponse {
    /// Returns `None` when either the `error` or the `document` are `None`.
    /// Either of those being `None` means this [`ImportResponse`] likely doesn't
    /// represent an error case.
    fn to_error_repr(&self) -> Option<ImportErrorResponse> {
        Some(ImportErrorResponse {
            error_msg: self.error.clone()?,
            bad_document: self.document.clone()?,
        })
    }
}

/// Response describing Typesense import errors.
///
/// See <https://typesense.org/docs/27.1/api/documents.html#index-multiple-documents>.
#[derive(Debug)]
pub struct ImportErrorResponse {
    /// Corresponds to `error` in the Typesense response.
    pub error_msg: String,

    /// Corresponds to `document` in the Typesense response.
    pub bad_document: String,
}

/// Wraps errors that can occur when using flora's Typesense client.
#[derive(Debug, Error)]
#[error(transparent)]
pub struct SearchClientError(#[from] SearchClientErrorRepr);

/// Result type used for private APIs in this module.
type InternalResult<T> = std::result::Result<T, SearchClientErrorRepr>;

/// Error type for failures that can occur when working with the flora Typesense client.
#[derive(Debug, Error)]
pub enum SearchClientErrorRepr {
    /// Error parsing a string into a valid url.
    #[error("Cannot parse URL: {0}")]
    UrlParse(String),

    /// Could not create a valid header value from the given string.
    #[error("{0}")]
    InvalidHeaderValue(#[from] InvalidHeaderValue),

    /// Error interacting with the Typesense REST API.
    #[error("{0}")]
    Typesense(#[from] TypesenseApiError),

    /// Error deserializing a JSON object.
    #[error("{0}")]
    SerdeJson(#[from] serde_json::Error),

    /// Variant that wraps [`Vec<ImportErrorResponse>`] for errors
    /// that occur when importing documents into Typesense.
    #[error("Error importing records: {0:?}")]
    Import(Vec<ImportErrorResponse>),

    /// Error when using the [`Read`] trait on the response from Typesense.
    #[error("Error reading the response: {0}")]
    ReadResponse(#[from] std::io::Error),

    /// Error building the client
    #[error("Error building the reqwest client: {0}")]
    ClientBuilder(String),

    /// A variant for instances of [`reqwest::Error`] not handled otherwise.
    #[error("Error from reqwest: {0}")]
    Reqwest(#[from] reqwest::Error),

    /// Error converting to [`SearchRecord`].
    #[error("{0}")]
    SearchRecord(#[from] SearchRecordError),

    /// All error types not specified otherwise
    #[error("Search client error: {0}")]
    Other(String),
}

impl From<TypesenseRequestError> for SearchClientErrorRepr {
    fn from(error: TypesenseRequestError) -> Self {
        Self::Typesense(error.into())
    }
}

/// Alias for search client results.
pub type Result<T> = std::result::Result<T, SearchClientError>;

/// Result type alias for Typesense-specific APIs.
type TypesenseResult = std::result::Result<Response, TypesenseRequestError>;

/// Wrapper for errors that can occur when interacting with the Typesense API.
#[derive(Debug, Error)]
#[error(transparent)]
pub struct TypesenseApiError(#[from] TypesenseRequestError);

/// Errors that may arise when interacting with Typesense APIs.
/// See <https://typesense.org/docs/29.0/api/api-errors.html>.
#[derive(Debug, Error)]
enum TypesenseRequestError {
    /// 400.
    #[error("Bad request: {0}")]
    BadRequest(TypesenseErrorResponse),

    /// 404.
    #[error("Not found (404); {0}")]
    NotFound(TypesenseErrorResponse),

    /// Another error returned by a Typesense API.
    #[error("error sending request: {0}, status: {1}")]
    Other(reqwest::Error, StatusCode),

    #[error("unknown reqwest error: {0}")]
    Unknown(#[from] reqwest::Error),
}

/// Error responses from Typesense.
#[derive(Debug)]
enum TypesenseErrorResponse {
    /// An API response we could parse, containing the parsed data.
    Parsed(api_responses::ErrorWithMessage),

    /// An API response we couldn't parse, containing the unparsed original string
    /// returned by Typesense.
    ParseFailed(String),
}

impl fmt::Display for TypesenseErrorResponse {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Parsed(parsed_error) => write!(
                f,
                "Error message returned by Typesense: {}",
                parsed_error.message
            ),
            Self::ParseFailed(raw_response) => write!(
                f,
                "Couldn't parse the error returned by Typesense. \
                The raw error response was: {raw_response}"
            ),
        }
    }
}

/// Result alias for fallible operations that try to parse Typesense error responses.
type TypesenseErrorParseResult = reqwest::Result<api_responses::ErrorWithMessage>;

impl From<TypesenseErrorParseResult> for TypesenseErrorResponse {
    fn from(value: TypesenseErrorParseResult) -> Self {
        match value {
            Ok(error_with_message) => Self::Parsed(error_with_message),
            Err(error) => Self::ParseFailed(error.to_string()),
        }
    }
}

/// We have to use this function to convert between results
/// because of the orphan rule.
///
/// Ideally, we'd implement From on [`reqwest::Result`], but this is less upfront work.
#[instrument(skip(result))]
async fn into_request_result(
    result: impl Future<Output = reqwest::Result<Response>>,
) -> TypesenseResult {
    let response = result.await?;

    if let Err(api_error) = response.error_for_status_ref() {
        let message = response
            .json::<api_responses::ErrorWithMessage>()
            .await
            .into();

        let our_error = match api_error.status() {
            Some(StatusCode::BAD_REQUEST) => TypesenseRequestError::BadRequest(message),
            Some(StatusCode::NOT_FOUND) => TypesenseRequestError::NotFound(message),
            Some(status) => TypesenseRequestError::Other(api_error, status),
            None => TypesenseRequestError::Unknown(api_error),
        };
        Err(our_error)
    } else {
        Ok(response)
    }
}

/// Data structure to encapsulate configuration parameters for
/// [`SearchClient`].
#[derive(Debug)]
pub struct SearchClientConfig {
    /// The Typesense URL, including the port.
    pub url: Option<Url>,

    /// Typesense API key.
    pub api_key: ApiKey,
}

/// Internal search client structure.
#[derive(Debug, Clone)]
struct SearchClientInternal {
    /// The HTTP client.
    client: Client,

    /// The Typesense URL this client connects to.
    url: Url,
}

impl SearchClientInternal {
    /// Creates a new internal client to communicate with the Typesense REST APIs.
    #[instrument]
    fn new(config: SearchClientConfig) -> InternalResult<Self> {
        let mut api_key = HeaderValue::from_str(config.api_key.expose_secret())?;
        api_key.set_sensitive(true);

        let header_map = HeaderMap::from_iter([(API_KEY_HEADER_NAME, api_key)]);
        let client = ClientBuilder::new()
            .default_headers(header_map)
            .use_rustls_tls()
            .build()
            .map_err(|error| {
                SearchClientErrorRepr::ClientBuilder(format!(
                    "Couldn't build the reqwest client: {error}"
                ))
            })?;

        // We know the hardcoded default Typesense URL should be parsed,
        // so the `expect` won't panic unless someone broke that hardcoded URL.
        #[allow(clippy::expect_used)]
        let url = config.url.unwrap_or_else(|| {
            DEFAULT_TYPESENSE_URL
                .parse::<Url>()
                .expect("Should parse the default typesense URL")
        });

        Ok(Self { client, url })
    }

    /// Returns the URL for the Typesense instance, with the given suffix.
    fn join_url(&self, input: &str) -> InternalResult<Url> {
        self.url.join(input).map_err(|error| {
            SearchClientErrorRepr::UrlParse(format!("{} + {input}: {error}", self.url))
        })
    }

    /// Returns a URL for /documents under the collection specified by the schema.
    fn documents_url(&self) -> InternalResult<Url> {
        let url = self
            .join_url(format!("collections/{}/documents", SearchRecord::schema().name).as_str())?;

        Ok(url)
    }

    /// Queries Typesense's `/health` endpoint.
    #[instrument]
    async fn health(&self) -> InternalResult<api_responses::Health> {
        let url = self.join_url("health")?;
        let response = into_request_result(self.client.get(url).send())
            .await?
            .json::<_>()
            .await?;
        Ok(response)
    }

    /// Creates a new Typesense collection.
    async fn create_collection(&self) -> InternalResult<Response> {
        let response = self
            .client
            .post(self.join_url("collections")?)
            .json(&SearchRecord::schema())
            .send()
            .await?;

        match response.status() {
            StatusCode::CREATED => Ok(response),
            StatusCode::BAD_REQUEST => Err(TypesenseRequestError::BadRequest(
                response
                    .json::<api_responses::ErrorWithMessage>()
                    .await
                    .into(),
            )
            .into()),

            _ => response.error_for_status().map_err(Into::into),
        }
    }

    /// Gets the Typesense collection matching the given name.
    async fn get_collection(&self, name: &str) -> InternalResult<Collection> {
        let response = into_request_result(
            self.client
                .get(self.join_url(format!("collections/{name}").as_str())?)
                .send(),
        )
        .await?;
        Ok(response.json::<Collection>().await?)
    }

    /// Imports the given records into Typesense.
    #[instrument(skip(self))]
    async fn import_documents(&self, documents: &[SearchRecord]) -> InternalResult<()> {
        //  Consider adding client-side batching
        // https://typesense.org/docs/guide/syncing-data-into-typesense.html#client-side-batch-size-vs-server-side-batching
        let mut documents_jsonl = Vec::with_capacity(documents.len());
        for document in documents {
            documents_jsonl.push(serde_json::to_string(document)?);
        }
        let documents_jsonl = documents_jsonl.join("\n");

        let url = self.join_url(
            format!(
                "collections/{}/documents/import",
                SearchRecord::schema().name
            )
            .as_str(),
        )?;

        // Look into using action=emplace query parameter to update document
        // with the same ID
        // https://typesense.org/docs/27.1/api/documents.html#index-multiple-documents
        let response = into_request_result(
            self.client
                .post(url)
                .header(CONTENT_TYPE, "text/plain")
                .body(documents_jsonl)
                .send(),
        )
        .await
        .inspect_err(|e| tracing::error!("{e}"))?;

        // Limit how much of the response we read so that we don't block forever
        // on a bad response body.
        let mut response_reader = BufReader::new(response.bytes().await?.reader().take(1024));

        let mut response_buf = String::new();
        let mut error_responses = vec![];

        while response_reader.read_line(&mut response_buf)? != 0 {
            let response_record = serde_json::from_str::<ImportResponse>(response_buf.as_str())
                .inspect_err(|error| tracing::error!("{error}\n{response_buf}"))?;

            if !response_record.success {
                if let Some(error_repr) = response_record.to_error_repr() {
                    error_responses.push(error_repr);
                } else {
                    tracing::warn!("Couldn't get an error representation for {response_record:#?}");
                }
            }

            response_buf.clear();
        }

        if error_responses.is_empty() {
            Ok(())
        } else {
            Err(SearchClientErrorRepr::Import(error_responses))
        }
    }

    /// Retrieves the Typesense document matching the given ID.
    async fn get_document(&self, id: &str) -> InternalResult<Option<SearchRecord>> {
        let url = self.join_url(
            format!("collections/{}/documents/{id}", SearchRecord::schema().name).as_str(),
        )?;

        // Intentionally not using `into_search_result` here because we implement custom
        // error handling logic if Typesense couldn't find the document with the given ID.
        let response = self
            .client
            .get(url)
            .send()
            .await?
            .json::<serde_json::Value>()
            .await?;

        if response.get("message").is_some_and(|v| {
            v.as_str()
                .is_some_and(|message| message.starts_with("Could not find a document with id"))
        }) {
            Ok(None)
        } else {
            Ok(Some(serde_json::from_value::<SearchRecord>(response)?))
        }
    }

    /// Gets Typesense documents matching the given SQLite database ID.
    ///
    /// # Returns
    /// A vec of matching documents.
    /// There may be more than one matching Typesense record for a given
    /// SQLite ID.
    async fn get_docs_by_db_id(&self, db_id: i64) -> InternalResult<Vec<SearchRecord>> {
        let ids_query = format!("filter_by=db-id: [{db_id}]");
        let mut url = self
            .documents_url()?
            .join("documents/export")
            .map_err(|e| SearchClientErrorRepr::UrlParse(e.to_string()))?;
        url.set_query(Some(ids_query.as_str()));

        let resp = self
            .client
            .get(url)
            .send()
            .await?
            .error_for_status()?
            .text()
            .await?;

        let mut recs = vec![];
        for line in resp.lines() {
            let rec: SearchRecord = serde_json::from_str(line)?;
            recs.push(rec);
        }

        Ok(recs)
    }

    /// Performs a search using the given query parameters.
    async fn search(&self, search_parameters: &SearchParameters) -> InternalResult<SearchResponse> {
        let url = self.join_url(
            format!(
                "collections/{}/documents/search",
                SearchRecord::schema().name
            )
            .as_str(),
        )?;

        let response = self.client.get(url).query(search_parameters).send();
        Ok(into_request_result(response)
            .await?
            .json::<SearchResponse>()
            .await?)
    }

    /// Deletes the documents matching the given **SQLite** IDs from Typesense.
    async fn delete_by_db_ids(&self, db_ids: Vec<i64>) -> InternalResult<u64> {
        let db_ids = db_ids
            .iter()
            .map(std::string::ToString::to_string)
            .collect::<Vec<_>>();

        let db_ids = db_ids.join(", ");
        let ids_query = format!("filter_by=db-id: [{db_ids}]");
        let mut url = self.documents_url()?;
        url.set_query(Some(ids_query.as_str()));

        let response = self.client.delete(url).send();
        let response = into_request_result(response)
            .await?
            .json::<api_responses::DeleteByQuery>()
            .await?;

        Ok(response.num_deleted)
    }

    /// Deletes the documents with the given **Typesense** IDs.
    async fn delete(&self, ids: &[&str]) -> InternalResult<u64> {
        let ids = ids.join(", ");
        let ids_query = format!("filter_by=id: [{ids}]");

        let mut url = self.documents_url()?;
        url.set_query(Some(ids_query.as_str()));

        let response = self.client.delete(url).send();
        let response = into_request_result(response)
            .await?
            .json::<api_responses::DeleteByQuery>()
            .await?;

        Ok(response.num_deleted)
    }
}

/// Client for flora to interact with the Typesense REST API.
#[derive(Debug, Clone)]
pub struct SearchClient {
    /// Internal search client.
    client: SearchClientInternal,
}

impl SearchClient {
    /// Constructs a new [`SearchClient`] given a [`SearchClientConfig`].
    ///
    /// # Errors
    /// [`SearchClientError`] if the given API key
    /// in the config cannot be used as a header value.
    /// or if we fail to build a client to interact with Typesense.
    fn new(config: SearchClientConfig) -> Result<Self> {
        let client = SearchClientInternal::new(config)?;
        Ok(Self { client })
    }

    /// Returns a new client if the Typesense health endpoint returns an ok status.
    async fn new_if_healthy(config: SearchClientConfig) -> Result<Self> {
        let client = Self::new(config)?;
        client.health().await?;
        Ok(client)
    }

    /// Creates a new [`SearchClient`].
    /// Checks if the Typesense health endpoint returns an ok status before proceeding,
    /// and creates a Typesense collection for `records` if one doesn't exist.
    ///
    /// # Errors
    /// [`SearchClientError`] if Typesense doesn't report ok health after the given
    /// number of retries, or if the request to Typesense fails.
    pub async fn new_with_records_collection(config: SearchClientConfig) -> Result<Self> {
        let search_client = Self::new_if_healthy(config).await?;

        let collection_name = SearchRecord::schema().name;
        if search_client
            .get_collection(&collection_name)
            .await
            .is_err()
        {
            search_client.create_collection().await?;
        }

        Ok(search_client)
    }

    /// Queries Typesense's `/health` endpoint.
    ///
    /// # Errors
    /// [`SearchClientError`] if the request to Typesense fails.
    pub async fn health(&self) -> Result<api_responses::Health> {
        Ok(self.client.health().await?)
    }

    /// Creates a new collection using the value returned by [`SearchRecord::schema`]
    ///
    /// # Errors
    /// [`SearchClientError`] if the request to Typesense fails.
    async fn create_collection(&self) -> Result<Response> {
        Ok(self.client.create_collection().await?)
    }

    /// Gets a Typesense collection with the given name.
    ///
    /// # Errors
    /// [`SearchClientError`] if the request to Typesense fails.
    pub async fn get_collection(&self, name: &str) -> Result<Collection> {
        Ok(self.client.get_collection(name).await?)
    }

    /// Imports the given records into Typesense.
    ///
    /// Returns a `Vec` of IDs (`UUIDv7`, stored as strings)
    /// assigned to the documents in Typesense.
    ///
    /// # Errors
    /// [`SearchClientError`] if the request to Typesense fails.
    #[instrument(skip_all)]
    pub async fn import_records(&self, records: Vec<Record>) -> Result<Vec<SearchId>> {
        let mut search_recs = Vec::with_capacity(records.len());
        let mut ids = Vec::with_capacity(records.len());

        for rec in records {
            let search_rec =
                SearchRecord::try_from(rec).map_err(SearchClientErrorRepr::SearchRecord)?;
            ids.push(search_rec.id().into());
            search_recs.push(search_rec);
        }

        self.client
            .import_documents(&search_recs)
            .await
            .inspect_err(|e| tracing::error!("{e}"))?;
        tracing::info!("imported {} search records", search_recs.len());

        Ok(ids)
    }

    /// Imports the given documents into Typesense.
    ///
    /// Returns a `Vec` of IDs (`UUIDv7`, stored as strings)
    /// assigned to the documents in Typesense.
    ///
    /// # Errors
    /// [`SearchClientError`] if the request to Typesense fails.
    pub async fn import_documents<D>(&self, documents: &[D]) -> Result<Vec<SearchId>>
    where
        D: ToRecord,
    {
        let records = documents
            .iter()
            .map(ToRecord::to_record)
            .collect::<std::result::Result<Vec<_>, _>>()
            .map_err(|e| SearchClientErrorRepr::Other(e.to_string()))?;
        self.import_records(records).await
    }

    /// Gets a document with the given ID from Typesense, or `None` if no document
    /// with that ID exists.
    ///
    /// # Errors
    /// [`SearchClientError`] if the request to Typesense fails.
    pub async fn get_document(&self, id: &str) -> Result<Option<SearchRecord>> {
        Ok(self.client.get_document(id).await?)
    }

    /// Gets Typesense documents matching the given SQLite database ID.
    ///
    /// # Returns
    /// A vec of matching documents.
    /// There may be more than one matching Typesense record for a given
    /// SQLite ID.
    ///
    /// # Errors
    /// [`SearchClientError`] if the request to Typesense fails.
    pub async fn get_docs_by_db_id(&self, db_id: i64) -> Result<Vec<SearchRecord>> {
        Ok(self.client.get_docs_by_db_id(db_id).await?)
    }

    /// Searches Typesense for documents matching the given search parameters.
    ///
    /// # Errors
    /// [`SearchClientError`] if the request to Typesense fails.
    pub async fn search(&self, search_parameters: &SearchParameters) -> Result<SearchResponse> {
        Ok(self.client.search(search_parameters).await?)
    }

    /// Deletes records with the given **SQLite** IDs from Typesense,
    /// and returns how many records were deleted.
    ///
    /// # Errors
    /// Returns [`SearchClientError`] if the request fails.
    pub async fn delete_by_db_ids(&self, db_ids: Vec<i64>) -> Result<u64> {
        Ok(self.client.delete_by_db_ids(db_ids).await?)
    }

    /// Deletes records with the given **Typesense** IDs,
    /// and returns how many records were deleted.
    ///
    /// # Errors
    /// Returns [`SearchClientError`] if the request fails.
    pub async fn delete(&self, ids: &[&str]) -> Result<u64> {
        Ok(self.client.delete(ids).await?)
    }
}
