//! RPC service and server definitions for the Typesense client.
#![cfg(feature = "rpc")]
use anyhow::{Context, Result};
use irpc::channel::oneshot;
use irpc::rpc_requests;
use irpc::{Client, WithChannels};
use n0_future::task::AbortOnDropHandle;
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
use tracing::instrument;

use crate::client::api_responses::Health;
use crate::{SearchClient, SearchParametersWrapper, SearchResponse};

/// Protocol for in-server search RPC.
#[rpc_requests(message = SearchMessage)]
#[derive(Debug, Serialize, Deserialize)]
pub enum SearchProtocol {
    /// Query Typesense for health.
    #[rpc(tx=oneshot::Sender<Health>)]
    #[wrap(GetHealth)]
    Health,

    /// Submit a search query, return the search response.
    #[rpc(tx=oneshot::Sender<SearchResponse>)]
    #[wrap(Search)]
    Search {
        /// Search parameters.
        params: SearchParametersWrapper,
    },
}

/// Actor for search RPC.
#[derive(Debug, Clone)]
struct SearchActor {
    /// REST Typesense client.
    client: SearchClient,
}

impl SearchActor {
    /// Initializes the actor with the given search client.
    #[must_use]
    fn new(client: SearchClient) -> Self {
        Self { client }
    }

    #[instrument(skip(self, rx))]
    async fn run(mut self, mut rx: mpsc::Receiver<SearchMessage>) {
        while let Some(msg) = rx.recv().await {
            if let Err(err) = self.handle(msg).await {
                tracing::error!("Error: {err}");
            }
        }
    }

    #[instrument(skip(self))]
    async fn handle(&mut self, msg: SearchMessage) -> Result<()> {
        match msg {
            SearchMessage::Health(health) => {
                tracing::trace!("health {health:#?}");
                let WithChannels { tx, .. } = health;
                let resp = self
                    .client
                    .health()
                    .await
                    .context("Error sending health request")?;
                Ok(tx.send(resp).await?)
            }
            SearchMessage::Search(search) => {
                tracing::trace!("search {search:#?}");
                let WithChannels { tx, inner, .. } = search;
                let resp = self
                    .client
                    .search(&inner.params.into())
                    .await
                    .context("Error sending search request")?;
                Ok(tx.send(resp).await?)
            }
        }
    }
}

/// API for interacting with the search service.
///
/// Adapted from
/// <https://github.com/n0-computer/irpc/blob/3bfe01cdba985545c6572af69cf60404124fbc68/irpc-iroh/examples/remote-and-local.rs>
#[derive(Debug)]
pub struct SearchApi {
    client: Client<SearchProtocol>,
    _actor_task: Option<AbortOnDropHandle<()>>,
}

impl SearchApi {
    /// Creates a new instance of search service API to communicate
    /// with the [`SearchClient`].
    #[instrument(skip(search_client))]
    pub fn spawn(search_client: SearchClient) -> Self {
        const CHANNEL_CAP: usize = 100;

        let (tx, rx) = mpsc::channel(CHANNEL_CAP);
        let actor = SearchActor::new(search_client);
        let actor_task = AbortOnDropHandle::new(n0_future::task::spawn(actor.run(rx)));
        let client = Client::local(tx);
        Self {
            client,
            _actor_task: Some(actor_task),
        }
    }

    /// Gets Typesense server health.
    #[instrument(skip(self))]
    #[allow(clippy::missing_errors_doc)]
    pub async fn health(&self) -> irpc::Result<Health> {
        self.client.rpc(GetHealth).await
    }

    /// Submits a search query.
    #[instrument(skip(self))]
    #[allow(clippy::missing_errors_doc)]
    pub async fn search(&self, params: SearchParametersWrapper) -> irpc::Result<SearchResponse> {
        let msg = Search { params };
        self.client.rpc(msg).await
    }
}
