// Copyright 2026 TiKV Project Authors. Licensed under Apache-2.0.

use std::collections::HashSet;

use anyhow::{Result, bail};
use tantivy::schema::{Field, FieldType, IndexRecordOption, Schema, Type};

pub const SINGLE_TEXT_FIELD: Field = Field::from_field_id(0);

/// Tantivy value type selected for one physical index field.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
pub enum PhysicalFieldType {
    Text,
    I64,
    U64,
    F64,
    Bytes,
}

impl PhysicalFieldType {
    fn tantivy_value_type(self) -> Type {
        match self {
            Self::Text => Type::Str,
            Self::I64 => Type::I64,
            Self::U64 => Type::U64,
            Self::F64 => Type::F64,
            Self::Bytes => Type::Bytes,
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Hash)]
pub struct SingleFieldLayout {
    field: Field,
    field_name: String,
    /// Whether the physical field stores field norms required for BM25.
    /// Query scoring/filter behavior remains a query-AST choice.
    enable_bm25: bool,
    field_type: PhysicalFieldType,
}

impl Default for SingleFieldLayout {
    fn default() -> Self {
        Self::legacy_body()
    }
}

impl SingleFieldLayout {
    pub fn new(field_name: impl Into<String>) -> Self {
        Self::new_with_field(SINGLE_TEXT_FIELD, field_name)
    }

    pub fn new_with_field(field: Field, field_name: impl Into<String>) -> Self {
        Self::new_with_field_and_bm25(field, field_name, true)
    }

    pub fn new_with_field_and_bm25(
        field: Field,
        field_name: impl Into<String>,
        enable_bm25: bool,
    ) -> Self {
        Self::new_with_field_type_and_bm25(field, field_name, PhysicalFieldType::Text, enable_bm25)
    }

    pub fn new_with_field_type_and_bm25(
        field: Field,
        field_name: impl Into<String>,
        field_type: PhysicalFieldType,
        enable_bm25: bool,
    ) -> Self {
        let field_name = field_name.into();
        assert!(!field_name.is_empty());
        Self {
            field,
            field_name,
            enable_bm25,
            field_type,
        }
    }

    pub fn legacy_body() -> Self {
        Self::new("body")
    }

    #[inline]
    pub fn field(&self) -> Field {
        self.field
    }

    #[inline]
    pub fn field_name(&self) -> &str {
        &self.field_name
    }

    #[inline]
    pub fn enable_bm25(&self) -> bool {
        self.enable_bm25
    }

    #[inline]
    pub fn field_type(&self) -> PhysicalFieldType {
        self.field_type
    }

    pub fn validate_schema(&self, schema: &Schema) -> Result<()> {
        let field = schema.get_field(self.field_name())?;
        if field != self.field() {
            bail!(
                "unexpected field id for field layout {:?}, expected {}, got {}",
                self.field_name(),
                self.field().field_id(),
                field.field_id()
            );
        }
        let actual = schema.get_field_entry(field).field_type();
        if actual.value_type() != self.field_type.tantivy_value_type() {
            bail!(
                "unexpected physical type for field {:?}, expected {:?}, got {:?}",
                self.field_name(),
                self.field_type,
                actual
            );
        }
        if !actual.is_indexed() {
            bail!("FTS field {:?} is not indexed", self.field_name());
        }
        if actual.has_fieldnorms() != self.enable_bm25 {
            bail!(
                "unexpected fieldnorm option for field {:?}, expected bm25={}, got fieldnorms={}",
                self.field_name(),
                self.enable_bm25,
                actual.has_fieldnorms(),
            );
        }
        if self.field_type == PhysicalFieldType::Text {
            let FieldType::Str(text_options) = actual else {
                unreachable!("text physical type was checked above")
            };
            let Some(indexing_options) = text_options.get_indexing_options() else {
                bail!("FTS text field {:?} is not indexed", self.field_name());
            };
            let tokenizer_name = indexing_options.tokenizer();
            // The tokenizer name is part of the Tantivy schema. Exact text
            // fields only emit one whole-value token, so Basic postings are
            // sufficient even when field norms are enabled for BM25.
            let expected_record = if tokenizer_name == "EXACT_VALUE_V1" || !self.enable_bm25 {
                IndexRecordOption::Basic
            } else {
                IndexRecordOption::WithFreqsAndPositions
            };
            if actual.get_index_record_option() != Some(expected_record) {
                bail!(
                    "unexpected text index record option for field {:?}, expected {:?}, got {:?}",
                    self.field_name(),
                    expected_record,
                    actual.get_index_record_option(),
                );
            }
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct IndexLayout {
    fields: Vec<SingleFieldLayout>,
}

impl Default for IndexLayout {
    fn default() -> Self {
        Self::legacy_body()
    }
}

impl IndexLayout {
    pub fn legacy_body() -> Self {
        Self::from_field_names(["body"])
    }

    /// Builds an index layout where iterator order defines Tantivy field ids.
    ///
    /// Callers must pass field names in the exact physical schema order used
    /// when the index is written.
    pub fn from_field_names<I, S>(field_names: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        let fields = field_names
            .into_iter()
            .enumerate()
            .map(|(field_id, field_name)| {
                SingleFieldLayout::new_with_field(Field::from_field_id(field_id as u32), field_name)
            })
            .collect::<Vec<_>>();
        Self::new(fields)
    }

    pub fn new(fields: Vec<SingleFieldLayout>) -> Self {
        assert!(!fields.is_empty());
        let mut field_names = HashSet::with_capacity(fields.len());
        for (field_id, field) in fields.iter().enumerate() {
            assert_eq!(field.field(), Field::from_field_id(field_id as u32));
            assert!(field_names.insert(field.field_name().to_string()));
        }
        Self { fields }
    }

    #[inline]
    pub fn fields(&self) -> &[SingleFieldLayout] {
        &self.fields
    }

    #[inline]
    pub fn first_field(&self) -> &SingleFieldLayout {
        &self.fields[0]
    }

    pub fn field_by_name(&self, field_name: &str) -> Option<SingleFieldLayout> {
        self.fields
            .iter()
            .find(|field| field.field_name() == field_name)
            .cloned()
    }

    pub fn field_by_id(&self, field_id: usize) -> Option<SingleFieldLayout> {
        self.fields.get(field_id).cloned()
    }

    pub fn validate_schema(&self, schema: &Schema) -> Result<()> {
        if schema.num_fields() != self.fields.len() {
            bail!(
                "unexpected number of fields in tantivy index schema, expected {}, got {}",
                self.fields.len(),
                schema.num_fields()
            );
        }
        for field in &self.fields {
            field.validate_schema(schema)?;
        }
        Ok(())
    }
}

pub fn validate_single_text_schema(schema: &Schema) -> Result<()> {
    IndexLayout::legacy_body().validate_schema(schema)
}