// Copyright 2026 PingCAP, Inc.

use anyhow::{Context, Result, bail};
use kvenginepb::fts::{FullTextAnalyzerType, FullTextIndexFieldDef, FullTextIndexFieldOption};
use tantivy::{
    schema::{IndexRecordOption, TextFieldIndexing},
    tokenizer::TokenizerManager,
};

use crate::{PhysicalFieldType, PhysicalIndexSchema};

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ResolvedFieldKind {
    Text,
    NativeExact(PhysicalFieldType),
}

/// Classifies one analyzer/physical-type pair after schema resolution.
///
/// Keeping this matrix shared makes schema validation and query lowering
/// enforce the same exact-versus-text invariant.
pub(crate) fn resolve_field_kind(
    analyzer_type: FullTextAnalyzerType,
    field_type: PhysicalFieldType,
) -> Result<ResolvedFieldKind> {
    match (analyzer_type, field_type) {
        (FullTextAnalyzerType::FtsAnalyzerExactValueV1, PhysicalFieldType::Text) => {
            Ok(ResolvedFieldKind::Text)
        }
        (FullTextAnalyzerType::FtsAnalyzerExactValueV1, field_type) => {
            Ok(ResolvedFieldKind::NativeExact(field_type))
        }
        (_, PhysicalFieldType::Text) => Ok(ResolvedFieldKind::Text),
        (analyzer_type, field_type) => {
            bail!(
                "FTS analyzer {:?} requires a text physical field, got {:?}",
                analyzer_type,
                field_type,
            )
        }
    }
}

pub(crate) fn build_physical_index_analysis(
    index_fields: &[FullTextIndexFieldDef],
    layout: &crate::IndexLayout,
) -> Result<(PhysicalIndexSchema, TokenizerManager)> {
    validate_index_fields(index_fields, layout)?;
    validate_resolved_field_types(index_fields, layout)?;
    let schema = PhysicalIndexSchema::build_validated(index_fields, layout)?;
    let tokenizers = index_tokenizers_from_index_fields(index_fields, schema.index_layout())?;
    Ok((schema, tokenizers))
}

pub(crate) fn legacy_index_field(tokenizer_name: &str) -> Result<FullTextIndexFieldDef> {
    let mut index_field = FullTextIndexFieldDef::new();
    let mut index_field_option = FullTextIndexFieldOption::new();
    index_field_option.set_analyzer_type(legacy_tokenizer_name_to_analyzer_type(tokenizer_name)?);
    index_field_option.set_enable_bm25(true);
    index_field.set_index_field_option(index_field_option);
    Ok(index_field)
}

fn legacy_tokenizer_name_to_analyzer_type(tokenizer_name: &str) -> Result<FullTextAnalyzerType> {
    match tokenizer_name {
        "EXACT_VALUE_V1" => Ok(FullTextAnalyzerType::FtsAnalyzerExactValueV1),
        "STANDARD_V1" => Ok(FullTextAnalyzerType::FtsAnalyzerStandardV1),
        "MULTILINGUAL_V1" => Ok(FullTextAnalyzerType::FtsAnalyzerMultilingualV1),
        "NGRAM_V1" => Ok(FullTextAnalyzerType::FtsAnalyzerNgramV1),
        "PATH_HIERARCHY_V1" => Ok(FullTextAnalyzerType::FtsAnalyzerPathHierarchyV1),
        _ => bail!("unsupported tokenizer {}", tokenizer_name),
    }
}

/// Returns the tokenizer bound to a text field.
///
/// Call this only after physical schema resolution. `EXACT_VALUE_V1` uses the
/// raw tokenizer when its SQL column resolves to a text physical field.
pub(crate) fn text_tokenizer_name(
    analyzer_type: FullTextAnalyzerType,
    field_idx: usize,
) -> Result<String> {
    match analyzer_type {
        FullTextAnalyzerType::FtsAnalyzerExactValueV1 => Ok("EXACT_VALUE_V1".to_string()),
        FullTextAnalyzerType::FtsAnalyzerStandardV1 => Ok("STANDARD_V1".to_string()),
        FullTextAnalyzerType::FtsAnalyzerMultilingualV1 => Ok("MULTILINGUAL_V1".to_string()),
        FullTextAnalyzerType::FtsAnalyzerNgramV1 => Ok(format!("NGRAM_V1_FIELD_{field_idx}")),
        // The tokenizer name is persisted in the Tantivy schema. A field-
        // local name lets one index register different delimiters without
        // colliding in its per-index TokenizerManager.
        FullTextAnalyzerType::FtsAnalyzerPathHierarchyV1 => {
            Ok(format!("PATH_HIERARCHY_V1_FIELD_{field_idx}"))
        }
        FullTextAnalyzerType::FtsAnalyzerUnspecified => {
            bail!("missing analyzer type for fulltext index field")
        }
    }
}

/// Validates tracked index fields against the physical layout.
pub(crate) fn validate_index_fields(
    index_fields: &[FullTextIndexFieldDef],
    layout: &crate::IndexLayout,
) -> Result<()> {
    if index_fields.len() != layout.fields().len() {
        bail!(
            "index field count {} does not match layout field count {}",
            index_fields.len(),
            layout.fields().len()
        );
    }

    for (field_idx, (field_layout, index_field)) in
        layout.fields().iter().zip(index_fields).enumerate()
    {
        if !index_field.has_index_field_option() {
            bail!(
                "missing index_field_option for fulltext index field with col_id {}",
                index_field.get_col_id()
            );
        }
        let option = index_field.get_index_field_option();
        match option.get_analyzer_type() {
            FullTextAnalyzerType::FtsAnalyzerNgramV1 => {
                ngram_spec(index_field, field_idx)?;
            }
            FullTextAnalyzerType::FtsAnalyzerPathHierarchyV1 => {
                if !option.has_path_hierarchy()
                    || option.get_path_hierarchy().get_delimiter().is_empty()
                {
                    bail!(
                        "path hierarchy field with col_id {} requires a non-empty delimiter",
                        index_field.get_col_id()
                    );
                }
            }
            FullTextAnalyzerType::FtsAnalyzerUnspecified => {
                bail!("missing analyzer type for fulltext index field")
            }
            _ => {}
        }
        if option.get_enable_bm25() != field_layout.enable_bm25() {
            bail!(
                "BM25 option for column {} does not match physical field layout",
                index_field.get_col_id()
            );
        }
    }
    Ok(())
}

pub(crate) fn validate_resolved_field_types(
    index_fields: &[FullTextIndexFieldDef],
    layout: &crate::IndexLayout,
) -> Result<()> {
    for (field_layout, index_field) in layout.fields().iter().zip(index_fields) {
        let analyzer_type = index_field.get_index_field_option().get_analyzer_type();
        resolve_field_kind(analyzer_type, field_layout.field_type()).with_context(|| {
            format!(
                "invalid physical FTS field for column {}",
                index_field.get_col_id()
            )
        })?;
    }
    Ok(())
}

fn ngram_spec(
    index_field: &FullTextIndexFieldDef,
    field_idx: usize,
) -> Result<crate::tokenizer::NgramV1Spec> {
    let option = index_field.get_index_field_option();
    if !option.has_ngram() {
        bail!(
            "Ngram fulltext index field {} with col_id {} requires Ngram options",
            field_idx,
            index_field.get_col_id()
        );
    }
    let ngram = option.get_ngram();
    crate::tokenizer::NgramV1Spec::from_option(ngram).with_context(|| {
        format!(
            "invalid Ngram options for fulltext index field {} with col_id {}: min_gram={}, max_gram={}, granularity={:?}",
            field_idx,
            index_field.get_col_id(),
            ngram.get_min_gram(),
            ngram.get_max_gram(),
            ngram.get_granularity(),
        )
    })
}

pub(crate) fn text_field_indexing(
    index_field_option: &FullTextIndexFieldOption,
    tokenizer_name: &str,
) -> TextFieldIndexing {
    // Exact fields emit one whole-value token, so frequencies and positions
    // carry no additional information even when the field contributes BM25.
    let index_option = if index_field_option.get_analyzer_type()
        == FullTextAnalyzerType::FtsAnalyzerExactValueV1
        || !index_field_option.get_enable_bm25()
    {
        IndexRecordOption::Basic
    } else {
        IndexRecordOption::WithFreqsAndPositions
    };
    TextFieldIndexing::default()
        .set_tokenizer(tokenizer_name)
        .set_fieldnorms(index_field_option.get_enable_bm25())
        .set_index_option(index_option)
}

fn index_tokenizers_from_index_fields(
    index_fields: &[FullTextIndexFieldDef],
    layout: &crate::IndexLayout,
) -> Result<TokenizerManager> {
    let manager = crate::tokenizer::index_tokenizers();
    for (field_idx, (index_field, field_layout)) in
        index_fields.iter().zip(layout.fields()).enumerate()
    {
        if field_layout.field_type() != PhysicalFieldType::Text {
            continue;
        }
        let option = index_field.get_index_field_option();
        let tokenizer_name = text_tokenizer_name(option.get_analyzer_type(), field_idx)?;
        match option.get_analyzer_type() {
            FullTextAnalyzerType::FtsAnalyzerNgramV1 => {
                manager.register(
                    &tokenizer_name,
                    ngram_spec(index_field, field_idx)?.index_analyzer(),
                );
            }
            FullTextAnalyzerType::FtsAnalyzerPathHierarchyV1 => {
                manager.register(
                    &tokenizer_name,
                    crate::tokenizer::PathHierarchyTokenizer::new(
                        option.get_path_hierarchy().get_delimiter(),
                    )?,
                );
            }
            _ => {
                if manager.get(&tokenizer_name).is_none() {
                    bail!(
                        "unsupported tokenizer {:?} for fulltext index field with col_id {}",
                        tokenizer_name,
                        index_field.get_col_id()
                    );
                }
            }
        }
    }
    Ok(manager)
}