invoist 1.0.1

use std::fmt;
use std::sync::Arc;
use std::time::Duration;

use bytes::Bytes;
use reqwest::header::{self, HeaderValue};
use reqwest::{Method, Url};
use serde::Serialize;
use serde::de::DeserializeOwned;

use crate::error::{Error, Result, codes};
use crate::files::{FileDownload, UploadFile};
use crate::models::{
    BatchResult, Category, Document, DocumentDetail, DocumentListItem, ExtractResult, Extraction,
    ExtractionStatus, ReviewQueueItem, ReviewQueueListItem, StatsOverview, Usage,
    WebhookEndpointList, WebhookEndpointWithSecret,
};
use crate::params::{
    CreateWebhookEndpoint, ExtractOptions, ListDocumentsParams, ListReviewQueueParams, Query,
    ReviewResolution, WaitOptions,
};
use crate::retry;

/// The production API.
pub const DEFAULT_BASE_URL: &str = "https://api.invoi.st";
/// How long one attempt may take before it fails with code `timeout`.
pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(60);
/// How many times a retryable failure is retried (so at most three attempts).
pub const DEFAULT_MAX_RETRIES: u32 = 2;

pub(crate) const USER_AGENT: &str = concat!("invoist-rust/", env!("CARGO_PKG_VERSION"), " (rust)");

/// The Invoist API client.
///
/// Cheap to clone (the connection pool is shared) and safe to use from many
/// tasks at once: build one and reuse it. Every method is `async`; to cancel
/// a call, drop its future (for instance with `tokio::time::timeout` or
/// `tokio::select!`). A cancelled call is never retried.
///
/// Business workspaces call the methods on `Client` directly. A household
/// (Invoist Personal) reads its bills through [`Client::bills`] and
/// [`Client::held_emails`].
///
/// ```no_run
/// # async fn run() -> Result<(), invoist::Error> {
/// use invoist::{Client, UploadFile, WaitOptions};
///
/// let client = Client::new(std::env::var("INVOIST_API_KEY").unwrap_or_default())?;
/// let file = UploadFile::from_path("invoice.pdf").expect("readable file");
/// let done = client.extract_and_wait(file, WaitOptions::default()).await?;
/// println!("{:?} {:?}", done.extraction.vendor_name, done.extraction.total_cents);
/// # Ok(())
/// # }
/// ```
#[derive(Clone)]
pub struct Client {
    pub(crate) inner: Arc<Inner>,
}

pub(crate) struct Inner {
    http: reqwest::Client,
    base: Url,
    auth: HeaderValue,
    timeout: Option<Duration>,
    max_retries: u32,
}

impl fmt::Debug for Client {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Client")
            .field("base_url", &self.inner.base.as_str())
            .field("api_key", &"<redacted>")
            .field("timeout", &self.inner.timeout)
            .field("max_retries", &self.inner.max_retries)
            .finish()
    }
}

/// Configures a [`Client`]. Start with [`Client::builder`].
///
/// ```
/// use std::time::Duration;
///
/// let client = invoist::Client::builder()
///     .api_key("inv_sk_test_123")
///     .base_url("http://localhost:8080")
///     .timeout(Duration::from_secs(30))
///     .max_retries(4)
///     .build()?;
/// # Ok::<(), invoist::Error>(())
/// ```
#[derive(Default)]
pub struct ClientBuilder {
    api_key: Option<String>,
    base_url: Option<String>,
    timeout: Option<Option<Duration>>,
    max_retries: Option<u32>,
    http: Option<reqwest::Client>,
}

impl fmt::Debug for ClientBuilder {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ClientBuilder")
            .field("api_key", &self.api_key.as_ref().map(|_| "<redacted>"))
            .field("base_url", &self.base_url)
            .field("timeout", &self.timeout)
            .field("max_retries", &self.max_retries)
            .field("http", &self.http)
            .finish()
    }
}

impl ClientBuilder {
    /// The API key (`inv_sk_...`) or OAuth access token (`inv_at_...`).
    /// Required.
    pub fn api_key(mut self, api_key: impl Into<String>) -> Self {
        self.api_key = Some(api_key.into());
        self
    }

    /// The API's address. Defaults to [`DEFAULT_BASE_URL`]; trailing slashes
    /// are ignored.
    pub fn base_url(mut self, base_url: impl Into<String>) -> Self {
        self.base_url = Some(base_url.into());
        self
    }

    /// How long each attempt may take. Defaults to [`DEFAULT_TIMEOUT`],
    /// generous for large uploads.
    pub fn timeout(mut self, timeout: Duration) -> Self {
        self.timeout = Some(Some(timeout));
        self
    }

    /// No per-attempt timeout: each attempt waits as long as the connection
    /// stays open.
    pub fn no_timeout(mut self) -> Self {
        self.timeout = Some(None);
        self
    }

    /// How many times a retryable failure (no response, 429, 5xx) is
    /// retried. Defaults to [`DEFAULT_MAX_RETRIES`]; `0` disables retrying.
    pub fn max_retries(mut self, max_retries: u32) -> Self {
        self.max_retries = Some(max_retries);
        self
    }

    /// The `reqwest::Client` to send requests with, for proxies, custom
    /// certificates, or tests. The SDK sets its own headers and timeout on
    /// every request, so the client's defaults for those don't matter.
    pub fn http_client(mut self, http: reqwest::Client) -> Self {
        self.http = Some(http);
        self
    }

    /// Builds the client. Fails, before any request, when the API key is
    /// empty or the base URL isn't an `http` or `https` URL.
    pub fn build(self) -> Result<Client> {
        let api_key = self.api_key.unwrap_or_default();
        let api_key = api_key.trim();
        if api_key.is_empty() {
            return Err(Error::invalid_argument(
                "An Invoist API key is required: pass one to Client::new or ClientBuilder::api_key",
            ));
        }
        let mut auth = HeaderValue::try_from(format!("Bearer {api_key}")).map_err(|_| {
            Error::invalid_argument("The API key contains characters a header can't carry")
        })?;
        auth.set_sensitive(true);

        let raw = self
            .base_url
            .unwrap_or_else(|| DEFAULT_BASE_URL.to_string());
        let trimmed = raw.trim().trim_end_matches('/');
        let base = Url::parse(trimmed)
            .ok()
            .filter(|u| matches!(u.scheme(), "http" | "https") && !u.cannot_be_a_base())
            .filter(|u| u.query().is_none() && u.fragment().is_none())
            .ok_or_else(|| {
                Error::invalid_argument(format!(
                    "{raw:?} is not a base URL: use an http or https URL such as {DEFAULT_BASE_URL}"
                ))
            })?;

        let http = match self.http {
            Some(http) => http,
            None => reqwest::Client::builder().build().map_err(|e| {
                Error::invalid_argument(format!("Could not create the HTTP client: {e}"))
                    .with_source(e)
            })?,
        };

        Ok(Client {
            inner: Arc::new(Inner {
                http,
                base,
                auth,
                timeout: self.timeout.unwrap_or(Some(DEFAULT_TIMEOUT)),
                max_retries: self.max_retries.unwrap_or(DEFAULT_MAX_RETRIES),
            }),
        })
    }
}

/// One API call, as data, so every retry can rebuild it exactly.
pub(crate) struct Request {
    method: Method,
    path: Vec<String>,
    query: Query,
    body: Body,
    idempotency_key: Option<HeaderValue>,
    accept: &'static str,
}

enum Body {
    Empty,
    Json(Bytes),
    Multipart(Vec<(&'static str, UploadFile)>),
}

impl Request {
    pub(crate) fn new(method: Method, path: &[&str]) -> Result<Self> {
        if let Some(i) = path.iter().position(|seg| seg.trim().is_empty()) {
            let after = path[..i].join("/");
            return Err(Error::invalid_argument(format!(
                "An ID is required after /{after}/"
            )));
        }
        Ok(Request {
            method,
            path: path.iter().map(|s| s.to_string()).collect(),
            query: Query::default(),
            body: Body::Empty,
            idempotency_key: None,
            accept: "application/json",
        })
    }

    pub(crate) fn get(path: &[&str]) -> Result<Self> {
        Request::new(Method::GET, path)
    }

    pub(crate) fn query(mut self, query: Query) -> Self {
        self.query = query;
        self
    }

    pub(crate) fn json(mut self, body: &impl Serialize) -> Result<Self> {
        let bytes = serde_json::to_vec(body)
            .map_err(|e| Error::invalid_argument(format!("Could not encode the body: {e}")))?;
        self.body = Body::Json(bytes.into());
        Ok(self)
    }

    fn multipart(mut self, files: Vec<(&'static str, UploadFile)>) -> Result<Self> {
        for (_, file) in &files {
            file.content_type()
                .parse::<reqwest::header::HeaderValue>()
                .map_err(|_| {
                    Error::invalid_argument(format!(
                        "{:?} is not a media type",
                        file.content_type()
                    ))
                })?;
        }
        self.body = Body::Multipart(files);
        Ok(self)
    }

    pub(crate) fn binary(mut self) -> Self {
        self.accept = "*/*";
        self
    }
}

/// A successful response, fully read.
pub(crate) struct Response {
    status: u16,
    request_id: Option<String>,
    content_type: Option<String>,
    body: Bytes,
}

impl Client {
    /// A client with the given API key and every other setting at its
    /// default. Fails when the key is empty.
    pub fn new(api_key: impl Into<String>) -> Result<Self> {
        Client::builder().api_key(api_key).build()
    }

    /// A client configured from the environment: `INVOIST_API_KEY` (required)
    /// and `INVOIST_BASE_URL` (optional).
    pub fn from_env() -> Result<Self> {
        let mut builder =
            Client::builder().api_key(std::env::var("INVOIST_API_KEY").unwrap_or_default());
        if let Ok(url) = std::env::var("INVOIST_BASE_URL")
            && !url.trim().is_empty()
        {
            builder = builder.base_url(url);
        }
        builder.build()
    }

    /// Starts configuring a client.
    pub fn builder() -> ClientBuilder {
        ClientBuilder::default()
    }

    /// The base URL requests go to.
    pub fn base_url(&self) -> &str {
        self.inner.base.as_str().trim_end_matches('/')
    }

    fn url(&self, req: &Request) -> Url {
        let mut url = self.inner.base.clone();
        url.path_segments_mut()
            .expect("the base URL was checked to be a base")
            .pop_if_empty()
            .extend(&req.path);
        if !req.query.0.is_empty() {
            url.query_pairs_mut().extend_pairs(&req.query.0);
        }
        url
    }

    fn build(&self, req: &Request) -> Result<reqwest::RequestBuilder> {
        let inner = &self.inner;
        let mut rb = inner
            .http
            .request(req.method.clone(), self.url(req))
            .header(header::AUTHORIZATION, inner.auth.clone())
            .header(header::ACCEPT, req.accept)
            .header(header::USER_AGENT, USER_AGENT);
        if let Some(timeout) = inner.timeout {
            rb = rb.timeout(timeout);
        }
        if let Some(key) = &req.idempotency_key {
            rb = rb.header("Idempotency-Key", key.clone());
        }
        rb = match &req.body {
            Body::Empty => rb,
            Body::Json(bytes) => rb
                .header(header::CONTENT_TYPE, "application/json")
                .body(bytes.clone()),
            Body::Multipart(files) => {
                let mut form = reqwest::multipart::Form::new();
                for (field, file) in files {
                    let data = file.data().clone();
                    let len = data.len() as u64;
                    let part = reqwest::multipart::Part::stream_with_length(data, len)
                        .file_name(file.filename().to_string())
                        .mime_str(file.content_type())
                        .map_err(|e| {
                            Error::invalid_argument(format!("Invalid media type: {e}"))
                                .with_source(e)
                        })?;
                    form = form.part(*field, part);
                }
                rb.multipart(form)
            }
        };
        Ok(rb)
    }

    /// Sends a request, retrying what is safe to retry, and returns the
    /// successful response or the last error.
    pub(crate) async fn execute(&self, req: &Request) -> Result<Response> {
        let mut attempt = 0;
        loop {
            let err = match self.build(req)?.send().await {
                Ok(resp) => {
                    let status = resp.status().as_u16();
                    let headers = resp.headers();
                    let text = |name: header::HeaderName| {
                        headers
                            .get(name)
                            .and_then(|v| v.to_str().ok())
                            .map(str::to_string)
                    };
                    let request_id = text(header::HeaderName::from_static("x-request-id"));
                    let content_type = text(header::CONTENT_TYPE);
                    let retry_after = text(header::RETRY_AFTER)
                        .as_deref()
                        .and_then(retry::parse_retry_after);
                    match resp.bytes().await {
                        Ok(body) if (200..300).contains(&status) => {
                            return Ok(Response {
                                status,
                                request_id,
                                content_type,
                                body,
                            });
                        }
                        Ok(body) => Error::from_response(status, request_id, retry_after, &body),
                        Err(e) => Error::transport(e),
                    }
                }
                Err(e) if e.is_builder() => {
                    return Err(Error::invalid_argument(format!(
                        "The request could not be built: {e}"
                    ))
                    .with_source(e));
                }
                Err(e) => Error::transport(e),
            };
            if !err.is_retryable()
                || attempt >= self.inner.max_retries
                || err
                    .retry_after()
                    .is_some_and(|wait| wait > retry::MAX_RETRY_AFTER)
            {
                return Err(err);
            }
            tokio::time::sleep(retry::delay(attempt, err.retry_after())).await;
            attempt += 1;
        }
    }

    pub(crate) async fn json<T: DeserializeOwned>(&self, req: Request) -> Result<T> {
        let resp = self.execute(&req).await?;
        serde_json::from_slice(&resp.body)
            .map_err(|e| Error::decode(resp.status, resp.request_id, &resp.body, e))
    }

    pub(crate) async fn empty(&self, req: Request) -> Result<()> {
        self.execute(&req).await.map(|_| ())
    }

    pub(crate) async fn download(&self, req: Request) -> Result<FileDownload> {
        let resp = self.execute(&req.binary()).await?;
        Ok(FileDownload {
            bytes: resp.body,
            content_type: resp.content_type,
        })
    }

    // ---- Business workspaces ------------------------------------------------

    /// Submits a document (PDF, PNG or JPEG, up to 20 MB) for extraction.
    ///
    /// Returns as soon as it is accepted, with the extraction still
    /// `pending` (reading happens in the background); use
    /// [`wait_for_extraction`](Client::wait_for_extraction), or the
    /// `extraction.completed` webhook, for the result. Uploading bytes this
    /// workspace has already extracted returns the existing document with
    /// `duplicate: true` and charges nothing, so retrying an upload is safe.
    ///
    /// ```no_run
    /// # async fn run(client: invoist::Client) -> Result<(), invoist::Error> {
    /// use invoist::{ExtractOptions, UploadFile};
    ///
    /// let file = UploadFile::new("invoice.pdf", std::fs::read("invoice.pdf").unwrap());
    /// let options = ExtractOptions::new().idempotency_key("order-1042");
    /// let result = client.extract(file, options).await?;
    /// println!("{} duplicate={}", result.document.id, result.duplicate);
    /// # Ok(())
    /// # }
    /// ```
    pub async fn extract(
        &self,
        file: UploadFile,
        options: ExtractOptions,
    ) -> Result<ExtractResult> {
        let mut req =
            Request::new(Method::POST, &["v1", "extract"])?.multipart(vec![("file", file)])?;
        if let Some(key) = options.idempotency_key {
            let value = HeaderValue::try_from(key).map_err(|_| {
                Error::invalid_argument(
                    "The idempotency key contains characters a header can't carry",
                )
            })?;
            req.idempotency_key = Some(value);
        }
        self.json(req).await
    }

    /// Submits up to ten documents in one request. Each file gets its own
    /// outcome in [`BatchResult::results`], in the order given: one bad file
    /// doesn't fail the others.
    pub async fn extract_batch(
        &self,
        files: impl IntoIterator<Item = UploadFile>,
    ) -> Result<BatchResult> {
        let files: Vec<_> = files.into_iter().map(|f| ("files", f)).collect();
        if files.is_empty() {
            return Err(Error::invalid_argument(
                "extract_batch needs at least one file",
            ));
        }
        let req = Request::new(Method::POST, &["v1", "extract", "batch"])?.multipart(files)?;
        self.json(req).await
    }

    /// Lists documents, newest first.
    pub async fn list_documents(
        &self,
        params: ListDocumentsParams,
    ) -> Result<Vec<DocumentListItem>> {
        #[derive(serde::Deserialize)]
        struct Page {
            documents: Vec<DocumentListItem>,
        }
        let req = Request::get(&["v1", "documents"])?.query(params.query());
        Ok(self.json::<Page>(req).await?.documents)
    }

    /// Fetches a document with its extraction and review items.
    pub async fn get_document(&self, id: &str) -> Result<DocumentDetail> {
        self.json(Request::get(&["v1", "documents", id])?).await
    }

    /// Downloads a document's original file, exactly as uploaded.
    pub async fn get_document_file(&self, id: &str) -> Result<FileDownload> {
        self.download(Request::get(&["v1", "documents", id, "file"])?)
            .await
    }

    /// Polls a document until its extraction finishes, and returns it.
    ///
    /// Returns on `done` *and* on `failed`: a failed reading is an answer
    /// about the document (see [`Extraction::error_message`]), not an error.
    /// Past [`WaitOptions::timeout`] it fails with status 0 and code
    /// [`extraction_timeout`](crate::error::codes::EXTRACTION_TIMEOUT).
    pub async fn wait_for_extraction(
        &self,
        document_id: &str,
        options: WaitOptions,
    ) -> Result<Extraction> {
        let deadline = tokio::time::Instant::now() + options.timeout;
        loop {
            let detail = self.get_document(document_id).await?;
            if let Some(extraction) = detail.extraction
                && extraction.status != ExtractionStatus::Pending
            {
                return Ok(extraction);
            }
            let now = tokio::time::Instant::now();
            if now >= deadline {
                return Err(Error::new(
                    0,
                    codes::EXTRACTION_TIMEOUT,
                    format!(
                        "The extraction of document {document_id} did not finish within {:?}",
                        options.timeout
                    ),
                ));
            }
            tokio::time::sleep(options.poll_interval.min(deadline - now)).await;
        }
    }

    /// Submits a document and waits for its extraction: [`extract`](Client::extract)
    /// followed by [`wait_for_extraction`](Client::wait_for_extraction).
    pub async fn extract_and_wait(
        &self,
        file: UploadFile,
        options: WaitOptions,
    ) -> Result<ExtractedDocument> {
        let result = self.extract(file, ExtractOptions::default()).await?;
        let extraction = match result.extraction {
            Some(e) if e.status != ExtractionStatus::Pending => e,
            _ => {
                self.wait_for_extraction(&result.document.id, options)
                    .await?
            }
        };
        Ok(ExtractedDocument {
            document: result.document,
            extraction,
        })
    }

    /// Lists review-queue items: what the engine flagged for a person to check.
    pub async fn list_review_queue(
        &self,
        params: ListReviewQueueParams,
    ) -> Result<Vec<ReviewQueueListItem>> {
        #[derive(serde::Deserialize)]
        struct Page {
            review_queue: Vec<ReviewQueueListItem>,
        }
        let req = Request::get(&["v1", "review-queue"])?.query(params.query());
        Ok(self.json::<Page>(req).await?.review_queue)
    }

    /// Resolves or dismisses a review item.
    pub async fn resolve_review_item(
        &self,
        id: &str,
        resolution: ReviewResolution,
    ) -> Result<ReviewQueueItem> {
        #[derive(Serialize)]
        struct Body {
            status: ReviewResolution,
        }
        #[derive(serde::Deserialize)]
        struct Answer {
            review_queue_item: ReviewQueueItem,
        }
        let req = Request::new(Method::POST, &["v1", "review-queue", id, "resolve"])?
            .json(&Body { status: resolution })?;
        Ok(self.json::<Answer>(req).await?.review_queue_item)
    }

    /// Lists the workspace's spend categories.
    pub async fn list_categories(&self) -> Result<Vec<Category>> {
        #[derive(serde::Deserialize)]
        struct Page {
            categories: Vec<Category>,
        }
        Ok(self
            .json::<Page>(Request::get(&["v1", "categories"])?)
            .await?
            .categories)
    }

    /// Usage for the current billing period.
    pub async fn get_usage(&self) -> Result<Usage> {
        self.json(Request::get(&["v1", "usage"])?).await
    }

    /// Workspace statistics: document counts, pending review, and plan allowance.
    pub async fn get_stats(&self) -> Result<StatsOverview> {
        self.json(Request::get(&["v1", "stats", "overview"])?).await
    }

    /// Lists webhook endpoints, plus the event types the API knows.
    pub async fn list_webhook_endpoints(&self) -> Result<WebhookEndpointList> {
        self.json(Request::get(&["v1", "webhook-endpoints"])?).await
    }

    /// Registers a webhook endpoint. The answer carries the signing secret,
    /// which is never shown again: store it now.
    pub async fn create_webhook_endpoint(
        &self,
        endpoint: CreateWebhookEndpoint,
    ) -> Result<WebhookEndpointWithSecret> {
        #[derive(serde::Deserialize)]
        struct Answer {
            endpoint: WebhookEndpointWithSecret,
        }
        if endpoint.url.trim().is_empty() {
            return Err(Error::invalid_argument("A webhook endpoint needs a URL"));
        }
        let req = Request::new(Method::POST, &["v1", "webhook-endpoints"])?.json(&endpoint)?;
        Ok(self.json::<Answer>(req).await?.endpoint)
    }

    /// Removes a webhook endpoint. Deliveries to it stop immediately.
    pub async fn delete_webhook_endpoint(&self, id: &str) -> Result<()> {
        self.empty(Request::new(
            Method::DELETE,
            &["v1", "webhook-endpoints", id],
        )?)
        .await
    }
}

/// A document and its finished extraction, from
/// [`Client::extract_and_wait`].
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub struct ExtractedDocument {
    /// The stored document.
    pub document: Document,
    /// Its extraction, `done` or `failed`.
    pub extraction: Extraction,
}