use std::marker::PhantomData;

use queue::consumer::SubscribeHandle;
use queue::typed::TypedMessage;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};

use crate::error::{SearchError, SearchResult};
use crate::index::{IndexHandle, IndexSettings};
use crate::metrics;
use crate::query::{SearchRequest, SearchResponse};

pub trait Searchable: Send + Sync + 'static {
    const INDEX_UID: &'static str;
    const PRIMARY_KEY: &'static str;

    fn index_settings() -> IndexSettings {
        IndexSettings::default()
    }

    fn document_id(&self) -> String;
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "op", content = "payload", rename_all = "snake_case")]
pub enum SyncMessage<T> {
    Upsert { document: T },
    Update { document: T },
    Delete { id: String },
    Clear {},
}

#[derive(Debug, Clone, Default)]
pub struct SyncSummary {
    pub index: String,
    pub operation: &'static str,
    pub documents: usize,
}

impl SyncSummary {
    pub fn new(index: &str, operation: &'static str, documents: usize) -> Self {
        Self {
            index: index.to_string(),
            operation,
            documents,
        }
    }
}

pub struct SyncedIndex<T> {
    index: IndexHandle,
    _marker: PhantomData<T>,
}

impl<T> Clone for SyncedIndex<T> {
    fn clone(&self) -> Self {
        Self {
            index: self.index.clone(),
            _marker: PhantomData,
        }
    }
}

impl<T> SyncedIndex<T>
where
    T: Searchable,
{
    pub(crate) fn new(index: IndexHandle) -> Self {
        Self {
            index,
            _marker: PhantomData,
        }
    }

    pub fn uid(&self) -> &str {
        self.index.uid()
    }

    pub fn handle(&self) -> &IndexHandle {
        &self.index
    }

    pub async fn ensure_index(&self) -> SearchResult<()> {
        self.index.ensure_index(&T::index_settings()).await
    }
}

impl<T> SyncedIndex<T>
where
    T: Searchable + Clone + Serialize + DeserializeOwned + Send + Sync,
{
    pub fn bridge<'a>(&self, queue: &'a queue::Queue) -> SyncBridge<'a, T> {
        SyncBridge::new(queue, self.clone())
    }
}

impl<T> SyncedIndex<T>
where
    T: Searchable + Serialize,
{
    pub async fn upsert(&self, documents: &[T]) -> SearchResult<SyncSummary> {
        let _timer = metrics::Timer::start(metrics::OP_DOCS_ADD);
        let batch = self.index.batch_size();
        for chunk in documents.chunks(batch) {
            self.index.add_documents(chunk).await?;
        }
        Ok(SyncSummary::new(
            self.uid(),
            metrics::OP_DOCS_ADD,
            documents.len(),
        ))
    }

    pub async fn update(&self, documents: &[T]) -> SearchResult<SyncSummary> {
        let _timer = metrics::Timer::start(metrics::OP_DOCS_UPDATE);
        let batch = self.index.batch_size();
        for chunk in documents.chunks(batch) {
            self.index.update_documents(chunk).await?;
        }
        Ok(SyncSummary::new(
            self.uid(),
            metrics::OP_DOCS_UPDATE,
            documents.len(),
        ))
    }

    pub async fn delete(&self, ids: &[String]) -> SearchResult<SyncSummary> {
        self.index.delete_documents(ids).await?;
        Ok(SyncSummary::new(
            self.uid(),
            metrics::OP_DOCS_DELETE,
            ids.len(),
        ))
    }

    pub async fn clear(&self) -> SearchResult<SyncSummary> {
        self.index.clear().await?;
        Ok(SyncSummary::new(self.uid(), metrics::OP_DOCS_CLEAR, 0))
    }
}

impl<T> SyncedIndex<T>
where
    T: Searchable + DeserializeOwned,
{
    pub async fn search(&self, request: &SearchRequest) -> SearchResult<SearchResponse<T>> {
        self.index.search(request).await
    }
}

pub struct SyncBridge<'a, T> {
    queue: &'a queue::Queue,
    index: SyncedIndex<T>,
}

impl<'a, T> SyncBridge<'a, T>
where
    T: Searchable + Clone + Serialize + DeserializeOwned + Send + Sync,
{
    pub(crate) fn new(queue: &'a queue::Queue, index: SyncedIndex<T>) -> Self {
        Self { queue, index }
    }

    pub fn index(&self) -> &SyncedIndex<T> {
        &self.index
    }

    pub async fn ensure_index(&self) -> SearchResult<()> {
        self.index.ensure_index().await
    }

    pub async fn publish_upsert(&self, document: &T) -> SearchResult<()> {
        self.publish(&SyncMessage::Upsert {
            document: document.clone(),
        })
        .await
    }

    pub async fn publish_update(&self, document: &T) -> SearchResult<()> {
        self.publish(&SyncMessage::Update {
            document: document.clone(),
        })
        .await
    }

    pub async fn publish_delete(&self, id: impl Into<String>) -> SearchResult<()> {
        self.publish(&SyncMessage::Delete { id: id.into() }).await
    }

    pub async fn publish_clear(&self) -> SearchResult<()> {
        self.publish(&SyncMessage::Clear {}).await
    }

    async fn publish(&self, message: &SyncMessage<T>) -> SearchResult<()> {
        self.queue
            .typed::<SyncMessage<T>>()
            .publish(message)
            .await
            .map_err(SearchError::Queue)
    }

    pub async fn spawn_consumer(
        &self,
        mode: queue::SubscriptionMode,
    ) -> Result<SubscribeHandle, SearchError> {
        self.index.ensure_index().await?;
        let index = self.index.clone();
        let handle = self
            .queue
            .typed::<SyncMessage<T>>()
            .subscribe(mode, move |msg: TypedMessage<SyncMessage<T>>| {
                let index = index.clone();
                async move {
                    let result = match msg.value {
                        SyncMessage::Upsert { document } => index.upsert(&[document]).await,
                        SyncMessage::Update { document } => index.update(&[document]).await,
                        SyncMessage::Delete { id } => index.delete(&[id]).await,
                        SyncMessage::Clear {} => index.clear().await,
                    };
                    match result {
                        Ok(summary) => {
                            tracing::debug!(
                                index = %summary.index,
                                operation = summary.operation,
                                documents = summary.documents,
                                "search sync applied"
                            );
                            Ok(())
                        }
                        Err(e) => {
                            tracing::warn!(
                                index = %index.uid(),
                                topic = %msg.meta.topic,
                                error = %e,
                                "search sync message failed"
                            );
                            Err(queue::QueueError::Handler(e.to_string()))
                        }
                    }
                }
            })
            .map_err(SearchError::Queue)?;
        Ok(handle)
    }
}