//! This module provides APIs to "bridge" flora-cli with the flora engine server via [`flora-service`]
//! (using tarpc).
use anyhow::{Context, Result};
pub use omry_service::archiving::{Record, RecordMeta, RecordParams};
use omry_service::iroh::endpoint::presets;
use omry_service::iroh::{Endpoint, EndpointAddr};
use omry_service::irpc::channel::mpsc;
use omry_service::search::SearchParameters;
use omry_service::{MetaHit, OmryApi, PONG};
use semver::Version;
use tokio::task::JoinSet;
use tracing::instrument;

use crate::common::RecordId;

/// Trait to bridge the user-facing application with the flora backend
/// over RPC.
#[allow(async_fn_in_trait)]
pub trait ServiceBridge {
    /// Returns a reference to the underlying tarpc client used by this bridge.
    fn api(&self) -> &OmryApi;

    /// Pings the flora server.
    ///
    /// Returns the expected [`PONG`] response on success.
    #[instrument(skip(self))]
    async fn ping(&self) -> Result<String> {
        let rx = self
            .api()
            .ping()
            .await
            .context("error sending ping request")?;
        let resp = rx.await?;
        if resp == PONG {
            Ok(resp)
        } else {
            anyhow::bail!("Unexpected ping response: {resp}")
        }
    }

    /// Adds a single record to the flora archive.
    #[instrument(skip(self))]
    async fn add_record(&self, record: RecordParams) -> Result<RecordId> {
        let (tx, mut rx) = self.api().add().await.context("error calling add")?;
        tx.send(record).await.context("Failed to send record")?;
        drop(tx);

        let id = rx
            .recv()
            .await?
            .ok_or_else(|| anyhow::anyhow!("got empty response"))?;
        Ok(id)
    }

    /// Queries the backend for the number of records it currently stores.
    #[instrument(skip(self))]
    async fn count(&self) -> Result<i64> {
        let rx = self.api().count().await.context("failed calling count")?;
        Ok(rx.await?)
    }

    /// Queries the backend for records with the given IDs.
    #[instrument(skip(self))]
    async fn get_records(
        &self,
        record_ids: Vec<RecordId>,
    ) -> Result<mpsc::Receiver<Option<Record>>> {
        let (tx, rx) = self
            .api()
            .record()
            .await
            .context("failed calling get record")?;

        let mut set = JoinSet::new();
        for id in &record_ids {
            let tx = tx.clone();
            let id = *id;
            set.spawn(async move { tx.send(id).await });
        }

        drop(tx);
        set.join_all().await;
        Ok(rx)
    }

    /// Gets metadata for all records stored in the Omry database.
    #[instrument(skip(self))]
    async fn get_all_records_meta(
        &self,
        start_offset: Option<usize>,
    ) -> Result<mpsc::Receiver<Option<RecordMeta>>> {
        let rx = self
            .api()
            .all_meta(start_offset)
            .await
            .context("Failed to invoke 'all_meta' RPC")?;

        Ok(rx)
    }

    /// Gets metadata for all records stored in Omry,
    /// in reverse chronological order (i.e. most recently added first).
    async fn most_recent_meta(
        &self,
        start_offset: Option<usize>,
    ) -> Result<mpsc::Receiver<Option<RecordMeta>>> {
        let rx = self
            .api()
            .most_recent_meta(start_offset)
            .await
            .context("Failed to invoke 'most_recent_meta'")?;
        Ok(rx)
    }

    /// Sends a search request to the flora backend.
    #[instrument(skip(self))]
    async fn search(&self, search_parameters: SearchParameters) -> Result<mpsc::Receiver<MetaHit>> {
        self.api()
            .search(search_parameters.into())
            .await
            .context("failed while calling search")
    }

    /// Sends a requests for the Omry API to delete record(s) with
    /// the given IDs.
    #[instrument(skip(self))]
    async fn delete(&self, ids: Vec<RecordId>) -> Result<usize> {
        tracing::debug!(
            "Will make {} RPCs to delete the given IDs (one per each)",
            ids.len()
        );

        let (tx, mut rx) = self.api().delete_record().await?;
        let mut set = JoinSet::new();

        for id in ids {
            let tx = tx.clone();
            set.spawn(async move { tx.send(id).await });
        }

        drop(tx);

        set.join_all().await;

        let mut n_deleted = 0usize;
        let mut n_errors = 0usize;
        loop {
            match rx.recv().await {
                Ok(Some(num_affected)) => n_deleted += num_affected,
                Ok(None) => break,
                Err(e) => {
                    tracing::error!("{e}");
                    n_errors += 1;
                }
            }
        }

        if n_errors != 0 {
            anyhow::bail!("{n_errors} records couldn't be scheduled for deletion due to errors.");
        }

        Ok(n_deleted)
    }

    /// Returns the semantic version of the backend Omry service.
    #[instrument(skip(self))]
    async fn version(&self) -> Result<Version> {
        let rx = self
            .api()
            .version()
            .await
            .context("Failed to invoke 'version' RPC")?;
        Ok(rx.await?.parse()?)
    }
}

/// Gets the iroh endpoint address for the given server,
/// via its web API.
#[allow(clippy::missing_errors_doc)]
async fn endpoint_addr(server_url: &str) -> Result<EndpointAddr> {
    const ENDPOINT_SUBPATH: &str = "/endpoint-addr";
    let url = format!("{server_url}{ENDPOINT_SUBPATH}");
    let bytes = reqwest::get(&url)
        .await?
        .error_for_status()
        .context("Couldn't get the endpoint_address")?
        .bytes()
        .await
        .context("Couldn't read the bytes")?;
    let endpoint_addr =
        postcard::from_bytes(&bytes).context("Error deserializing the endpoint address")?;
    Ok(endpoint_addr)
}

/// Creates a client for the given Omry web server endpoint address.
#[instrument]
async fn connect(server_addr: EndpointAddr) -> Result<(OmryApi, Endpoint)> {
    tracing::trace!("connecting API");
    let endpoint = Endpoint::bind(presets::N0).await?;
    let api = OmryApi::connect(endpoint.clone(), server_addr);
    Ok((api, endpoint))
}

/// Spawns a new instance of the flora RPC bridge, using the given config.
///
/// Returns an endpoint, which must be closed manually with [`Endpoint::close`].
///
/// ## Errors
/// Returns [`anyhow::Error`] if the config specifies RPC host and port, and
/// we can't create a client with those.
#[instrument(skip(auth_token))]
pub async fn connect_and_auth(server_url: &str, auth_token: String) -> Result<(OmryApi, Endpoint)> {
    let endpoint_addr = endpoint_addr(server_url).await?;
    let (api, endpoint) = connect(endpoint_addr)
        .await
        .context("Couldn't connect to the API")?;
    api.auth(auth_token).await.context("Couldn't do auth")?;
    Ok((api, endpoint))
}
