use anyhow::{Context, Result, anyhow, bail};
use tantivy::{TantivyDocument, schema::Value};
use tici_grpc::ColumnInfo;

use crate::{
    reader::query_info::DocField,
    utils::{
        key::ShardingKey,
        schema::{HANDLE_STORE_NAME, IndexMeta, VERSION_STORE_NAME, *},
    },
    writer::cdc_log::{OperationType, RowDataAccess},
};

#[derive(Clone)]
pub struct TantivyDoc {
    pub sharding_key: ShardingKey,
    pub version: u64,
    pub op_type: OperationType,
    pub doc: TantivyDocument,
    pub doc_bytes: usize,
}

impl TantivyDoc {
    pub fn new<R>(row: &R, index_meta: &IndexMeta) -> Result<TantivyDoc>
    where
        R: RowDataAccess,
    {
        let schema = index_meta.get_tantivy_schema();
        let sharding_key = row.get_sharding_key(index_meta)?;
        let version = row.get_version()?;
        let op_type = row.get_dml_type()?;

        if op_type == OperationType::Delete {
            return Ok(TantivyDoc::delete_doc(sharding_key.clone(), version));
        }

        let get_field = |field_name: &str| {
            schema
                .get_field(field_name)
                .with_context(|| format!("Failed to get field={}, schema={:?}", field_name, schema))
        };
        let mut doc_bytes = 0;
        let mut doc = TantivyDocument::default();

        doc.add_bytes(get_field(HANDLE_STORE_NAME)?, sharding_key.raw());
        doc_bytes += sharding_key.raw().len();
        doc.add_u64(get_field(VERSION_STORE_NAME)?, version);
        doc_bytes += std::mem::size_of_val(&version);

        for col in index_meta.get_column_infos() {
            let field = get_field(&col.store_name())?;
            if col.is_text_type() {
                if let Some(value) = row.get_str(col)? {
                    doc_bytes += value.len();
                    doc.add_text(field, value);
                }
            } else if col.is_signed_int_type() {
                if let Some(value) = row.get_i64(col)? {
                    doc_bytes += std::mem::size_of_val(&value);
                    doc.add_i64(field, value);
                }
            } else if col.is_unsigned_int_type() {
                if let Some(value) = row.get_u64(col)? {
                    doc_bytes += std::mem::size_of_val(&value);
                    doc.add_u64(field, value);
                }
            } else if col.is_float_type() {
                if let Some(value) = row.get_f64(col)? {
                    doc_bytes += std::mem::size_of_val(&value);
                    doc.add_f64(field, value);
                }
            } else if col.is_date_time_type() {
                if let Some(value) = row.get_tidb_packed_time(col)? {
                    doc_bytes += std::mem::size_of_val(&value);
                    doc.add_u64(field, value);
                }
            } else if col.is_decimal_type() {
                if let Some(value) = row.get_decimal_bytes(col)? {
                    doc_bytes += value.len();
                    doc.add_bytes(field, &value);
                }
            } else if col.is_binary_type() {
                if let Some(value) = row.get_binary(col)? {
                    doc_bytes += value.len();
                    doc.add_bytes(field, &value);
                }
            } else {
                bail!("Unsupported column={:?}", col);
            }
        }

        // Build fulltext fields by concatenating text from stored column fields
        doc_bytes += index_meta.build_fulltext_fields(&mut doc)?;

        Ok(TantivyDoc {
            sharding_key: sharding_key.clone(),
            version,
            op_type,
            doc,
            doc_bytes,
        })
    }

    pub fn delete_doc(sharding_key: ShardingKey, version: u64) -> TantivyDoc {
        TantivyDoc {
            sharding_key,
            version,
            op_type: OperationType::Delete,
            doc: TantivyDocument::default(),
            doc_bytes: 0,
        }
    }
}

#[derive(Clone, Debug)]
pub struct RawDoc {
    pub handle: ShardingKey,
    pub version: u64,
    pub op_type: OperationType,
    pub cols: Vec<(i64, Option<String>)>, // col_id, value
    pub is_local_timezone: bool,
}

impl RowDataAccess for RawDoc {
    fn get_sharding_key(&self, _index_meta: &IndexMeta) -> Result<ShardingKey> {
        // TODO: support sharding columns
        Ok(self.handle.clone())
    }

    fn get_version(&self) -> Result<u64> {
        Ok(self.version)
    }

    fn get_dml_type(&self) -> Result<OperationType> {
        Ok(self.op_type)
    }

    fn get_column_value(&self, col: &ColumnInfo) -> Result<Option<&str>> {
        let col_value = self
            .cols
            .iter()
            .find(|(cid, _)| *cid == col.column_id)
            .map(|(_, v)| v);

        if col_value.is_none() {
            bail!(format!("Column '{}' not found", col.column_name));
        }

        match col_value.unwrap() {
            Some(s) => Ok(Some(s.as_str())),
            // For nullable column, return empty string if value is NULL.
            None if col.is_nullable() => Ok(None),
            None => bail!(format!(
                "Column '{:?}' is not nullable, but value is NULL",
                col
            )),
        }
    }

    fn is_local_timezone(&self) -> bool {
        self.is_local_timezone
    }
}

pub trait DocFieldAccess {
    fn get_u64(&self, doc_field: &DocField) -> Result<Option<u64>>;

    fn get_i64(&self, doc_field: &DocField) -> Result<Option<i64>>;

    fn get_str(&self, doc_field: &DocField) -> Result<Option<&str>>;

    fn get_bytes(&self, doc_field: &DocField) -> Result<Option<&[u8]>>;

    fn get_f64(&self, doc_field: &DocField) -> Result<Option<f64>>;
}

impl DocFieldAccess for TantivyDocument {
    fn get_u64(&self, doc_field: &DocField) -> Result<Option<u64>> {
        match self.get_first(doc_field.field) {
            Some(value) => {
                let i = value.as_u64().ok_or_else(|| {
                    anyhow!("Field '{:?}' is not u64. value={:?}", doc_field, value)
                })?;
                Ok(Some(i))
            }
            None => Ok(None),
        }
    }

    fn get_i64(&self, doc_field: &DocField) -> Result<Option<i64>> {
        match self.get_first(doc_field.field) {
            Some(value) => {
                let i = value.as_i64().ok_or_else(|| {
                    anyhow!("Field '{:?}' is not u64. value={:?}", doc_field, value)
                })?;
                Ok(Some(i))
            }
            None => Ok(None),
        }
    }

    fn get_str(&self, doc_field: &DocField) -> Result<Option<&str>> {
        match self.get_first(doc_field.field) {
            Some(value) => {
                let s = value.as_str().ok_or_else(|| {
                    anyhow!("Field '{:?}' is not str. value={:?}", doc_field, value)
                })?;
                Ok(Some(s))
            }
            None => Ok(None),
        }
    }

    fn get_bytes(&self, doc_field: &DocField) -> Result<Option<&[u8]>> {
        match self.get_first(doc_field.field) {
            Some(value) => {
                let b = value.as_bytes().ok_or_else(|| {
                    anyhow!("Field '{:?}' is not bytes. value={:?}", doc_field, value,)
                })?;
                Ok(Some(b))
            }
            None => Ok(None),
        }
    }

    fn get_f64(&self, doc_field: &DocField) -> Result<Option<f64>> {
        match self.get_first(doc_field.field) {
            Some(value) => {
                let f = value.as_f64().ok_or_else(|| {
                    anyhow!("Field '{:?}' is not f64. value={:?}", doc_field, value)
                })?;
                Ok(Some(f))
            }
            None => Ok(None),
        }
    }
}