linkrs-query 0.1.0

//! Full-Text Parser
//!
//! This module implements the parser for full-text search SQL statements,
//! including CREATE FULLTEXT INDEX, SEARCH, and related queries.

use crate::parser::ast::fulltext::{
    AlterFulltextIndex, AlterIndexAction, CreateFulltextIndex, DescribeFulltextIndex,
    DropFulltextIndex, FulltextMatchCondition, FulltextQueryExpr, FulltextYieldClause,
    FulltextYieldItem, IndexFieldDef, IndexOptions, LookupFulltext, MatchFulltext, SearchStatement,
    ShowFulltextIndex,
};
use crate::parser::ast::stmt::Stmt;
use crate::parser::ast::stmt::{OrderByClause, OrderByItem};
use crate::parser::ast::types::{LimitClause, OrderDirection, SkipClause};
use crate::parser::parsing::expr_parser::parse_expression_with_context;
use crate::parser::parsing::parse_context::ParseContext;
use crate::parser::TokenKind;
use linkrs_core::types::expr::{create_contextual_expression, Expression};
use linkrs_core::types::FulltextEngineType;
use std::collections::HashMap;

/// Parse full-text search statements from ParseContext
pub fn parse_fulltext(ctx: &mut ParseContext) -> Result<Stmt, crate::parser::ParseError> {
    if ctx.check_keyword("CREATE") {
        return parse_create_fulltext_index(ctx);
    } else if ctx.check_keyword("DROP") {
        return parse_drop_fulltext_index(ctx);
    } else if ctx.check_keyword("ALTER") {
        return parse_alter_fulltext_index(ctx);
    } else if ctx.check_keyword("SHOW") {
        return parse_show_fulltext_index(ctx);
    } else if ctx.check_keyword("DESCRIBE") || ctx.check_keyword("DESC") {
        return parse_describe_fulltext_index(ctx);
    } else if ctx.check_keyword("SEARCH") {
        // Check if it's SEARCH VECTOR or SEARCH INDEX
        if ctx.check_keyword_sequence(&["SEARCH", "VECTOR"]) {
            // Forward to vector parser
            return crate::parser::parsing::vector_parser::parse_vector(ctx);
        }
        // Check if it's SEARCH INDEX - if so, we need to consume INDEX here
        let is_search_index = ctx.check_keyword_sequence(&["SEARCH", "INDEX"]);
        // Consume SEARCH and continue with fulltext parsing
        ctx.consume_keyword("SEARCH")?;
        if is_search_index {
            // Also consume INDEX here before calling parse_search_statement_after_search
            ctx.consume_keyword("INDEX")?;
        }
        return parse_search_statement_after_search(ctx);
    } else if ctx.check_keyword("LOOKUP") {
        // Check if it's LOOKUP VECTOR
        if ctx.check_keyword_sequence(&["LOOKUP", "VECTOR"]) {
            // Forward to vector parser
            return crate::parser::parsing::vector_parser::parse_vector(ctx);
        }
        return parse_lookup_fulltext(ctx);
    } else if ctx.check_keyword("MATCH") {
        return parse_match_fulltext(ctx);
    }

    Err(crate::parser::ParseError::new(
        crate::parser::core::error::ParseErrorKind::SyntaxError,
        "Not a full-text search statement".to_string(),
        ctx.current_position(),
    ))
}

pub fn parse_create_fulltext_index(
    ctx: &mut ParseContext,
) -> Result<Stmt, crate::parser::ParseError> {
    ctx.consume_keyword("CREATE")?;
    parse_create_fulltext_index_after_create(ctx)
}

pub fn parse_create_fulltext_index_after_create(
    ctx: &mut ParseContext,
) -> Result<Stmt, crate::parser::ParseError> {
    ctx.consume_keyword("FULLTEXT")?;
    ctx.consume_keyword("INDEX")?;

    let if_not_exists = if ctx.check_keyword("IF") {
        ctx.consume_keyword("IF")?;
        ctx.consume_keyword("NOT")?;
        ctx.consume_keyword("EXISTS")?;
        true
    } else {
        false
    };

    let index_name = ctx.consume_identifier()?;
    ctx.consume_keyword("ON")?;
    let schema_name = ctx.consume_identifier()?;

    ctx.expect_token(TokenKind::LParen)?;
    let mut fields = Vec::new();

    loop {
        let field_name = ctx.consume_identifier()?;

        let field_def = IndexFieldDef::new(field_name);

        if ctx.check_keyword("ANALYZER") {
            return Err(crate::parser::ParseError::new(
                crate::parser::core::error::ParseErrorKind::SyntaxError,
                "ANALYZER is not supported yet; create the fulltext index without per-field analyzers"
                    .to_string(),
                ctx.current_position(),
            ));
        }

        if ctx.check_keyword("BOOST") {
            return Err(crate::parser::ParseError::new(
                crate::parser::core::error::ParseErrorKind::SyntaxError,
                "BOOST is not supported yet; create the fulltext index without per-field weights"
                    .to_string(),
                ctx.current_position(),
            ));
        }

        fields.push(field_def);

        if !ctx.consume_optional_token(",") {
            break;
        }
    }

    ctx.expect_token(TokenKind::RParen)?;

    ctx.consume_keyword("ENGINE")?;
    if ctx.current_token().kind == TokenKind::Assign {
        ctx.next_token();
    }
    if ctx.check_keyword("BM25") {
        ctx.consume_keyword("BM25")?;
    } else {
        return Err(crate::parser::ParseError::new(
            crate::parser::core::error::ParseErrorKind::SyntaxError,
            "Expected BM25 engine type".to_string(),
            ctx.current_position(),
        ));
    }
    let engine_type = FulltextEngineType::Bm25;

    if ctx.check_keyword("OPTIONS") {
        return Err(crate::parser::ParseError::new(
            crate::parser::core::error::ParseErrorKind::SyntaxError,
            "OPTIONS (...) is not supported yet; create the fulltext index without BM25 option overrides (defaults k1=1.5, b=0.75 apply)"
                .to_string(),
            ctx.current_position(),
        ));
    }

    let options = IndexOptions {
        bm25_config: None,
        common_options: HashMap::new(),
    };

    let mut create = CreateFulltextIndex::new(
        ctx.current_span(),
        index_name,
        schema_name,
        fields,
        engine_type,
    );
    create.if_not_exists = if_not_exists;
    create.options = options;

    Ok(Stmt::CreateFulltextIndex(create))
}

pub fn parse_drop_fulltext_index(
    ctx: &mut ParseContext,
) -> Result<Stmt, crate::parser::ParseError> {
    ctx.consume_keyword("DROP")?;
    parse_drop_fulltext_index_after_drop(ctx)
}

pub fn parse_drop_fulltext_index_after_drop(
    ctx: &mut ParseContext,
) -> Result<Stmt, crate::parser::ParseError> {
    ctx.consume_keyword("FULLTEXT")?;
    ctx.consume_keyword("INDEX")?;

    let if_exists = if ctx.check_keyword("IF") {
        ctx.consume_keyword("IF")?;
        ctx.consume_keyword("EXISTS")?;
        true
    } else {
        false
    };

    let index_name = ctx.consume_identifier()?;

    let drop = DropFulltextIndex {
        span: ctx.current_span(),
        index_name,
        if_exists,
    };

    Ok(Stmt::DropFulltextIndex(drop))
}

pub fn parse_alter_fulltext_index(
    ctx: &mut ParseContext,
) -> Result<Stmt, crate::parser::ParseError> {
    ctx.consume_keyword("ALTER")?;
    parse_alter_fulltext_index_after_alter(ctx)
}

pub fn parse_alter_fulltext_index_after_alter(
    ctx: &mut ParseContext,
) -> Result<Stmt, crate::parser::ParseError> {
    ctx.consume_keyword("FULLTEXT")?;
    ctx.consume_keyword("INDEX")?;

    let index_name = ctx.consume_identifier()?;
    let mut actions = Vec::new();

    loop {
        if ctx.check_keyword("ADD") {
            ctx.consume_keyword("ADD")?;
            ctx.consume_keyword("FIELD")?;

            let field_name = ctx.consume_identifier()?;

            if ctx.check_keyword("ANALYZER") {
                return Err(crate::parser::ParseError::new(
                    crate::parser::core::error::ParseErrorKind::SyntaxError,
                    "ANALYZER is not supported yet; add the field without a per-field analyzer"
                        .to_string(),
                    ctx.current_position(),
                ));
            }

            let field_def = IndexFieldDef::new(field_name);

            actions.push(AlterIndexAction::AddField(field_def));
        } else if ctx.check_keyword("DROP") {
            ctx.consume_keyword("DROP")?;
            ctx.consume_keyword("FIELD")?;
            let field_name = ctx.consume_identifier()?;
            actions.push(AlterIndexAction::DropField(field_name));
        } else if ctx.check_keyword("SET") {
            ctx.consume_keyword("SET")?;
            let key = ctx.consume_identifier()?;
            ctx.expect_token(TokenKind::Eq)?;
            let value = ctx.consume_value()?;
            actions.push(AlterIndexAction::SetOption(key, value));
        } else if ctx.check_keyword("REBUILD") {
            ctx.consume_keyword("REBUILD")?;
            actions.push(AlterIndexAction::Rebuild);
        } else if ctx.check_keyword("OPTIMIZE") {
            ctx.consume_keyword("OPTIMIZE")?;
            actions.push(AlterIndexAction::Optimize);
        } else {
            return Err(crate::parser::ParseError::new(
                crate::parser::core::error::ParseErrorKind::SyntaxError,
                "Expected ALTER INDEX action".to_string(),
                ctx.current_position(),
            ));
        }

        if !ctx.consume_optional_token(",") {
            break;
        }
    }

    let alter = AlterFulltextIndex {
        span: ctx.current_span(),
        index_name,
        actions,
    };

    Ok(Stmt::AlterFulltextIndex(alter))
}

fn parse_show_fulltext_index(ctx: &mut ParseContext) -> Result<Stmt, crate::parser::ParseError> {
    ctx.consume_keyword("SHOW")?;
    ctx.consume_keyword("FULLTEXT")?;
    ctx.consume_keyword("INDEX")?;

    let mut pattern = None;
    let mut from_schema = None;

    if ctx.check_keyword("LIKE") {
        ctx.consume_keyword("LIKE")?;
        pattern = Some(ctx.consume_string()?);
    }

    if ctx.check_keyword("FROM") || ctx.check_keyword("IN") {
        ctx.consume_keyword("FROM")?;
        from_schema = Some(ctx.consume_identifier()?);
    }

    let show = ShowFulltextIndex {
        span: ctx.current_span(),
        pattern,
        from_schema,
    };

    Ok(Stmt::ShowFulltextIndex(show))
}

fn parse_describe_fulltext_index(
    ctx: &mut ParseContext,
) -> Result<Stmt, crate::parser::ParseError> {
    ctx.consume_keyword("DESCRIBE")?;
    ctx.consume_keyword("FULLTEXT")?;
    ctx.consume_keyword("INDEX")?;

    let index_name = ctx.consume_identifier()?;

    let describe = DescribeFulltextIndex {
        span: ctx.current_span(),
        index_name,
    };

    Ok(Stmt::DescribeFulltextIndex(describe))
}

fn parse_search_statement_after_search(
    ctx: &mut ParseContext,
) -> Result<Stmt, crate::parser::ParseError> {
    // INDEX keyword has already been consumed by the caller
    let index_name = ctx.consume_identifier()?;
    ctx.consume_keyword("MATCH")?;

    let query = parse_fulltext_query_expr(ctx)?;

    let mut search = SearchStatement::new(index_name, query);

    if ctx.check_keyword("YIELD") {
        ctx.consume_keyword("YIELD")?;
        let yield_clause = parse_yield_clause(ctx)?;
        search.yield_clause = Some(yield_clause);
    }

    if ctx.check_keyword("WHERE") {
        ctx.consume_keyword("WHERE")?;
        let where_clause = parse_expression_with_context(ctx, ctx.expression_context_clone())?;
        search.where_clause = Some(where_clause);
    }

    if ctx.check_keyword("ORDER") {
        ctx.consume_keyword("ORDER")?;
        ctx.consume_keyword("BY")?;
        let order_clause = parse_order_clause(ctx)?;
        search.order_clause = Some(order_clause);
    }

    if ctx.check_keyword("LIMIT") {
        ctx.consume_keyword("LIMIT")?;
        let count = ctx.consume_int()? as usize;
        search.limit = Some(LimitClause {
            span: ctx.current_span(),
            count,
        });
    }

    if ctx.check_keyword("OFFSET") {
        ctx.consume_keyword("OFFSET")?;
        let count = ctx.consume_int()? as usize;
        search.skip = Some(SkipClause {
            span: ctx.current_span(),
            count,
        });
    }

    Ok(Stmt::Search(search))
}

fn parse_fulltext_query_expr(
    ctx: &mut ParseContext,
) -> Result<FulltextQueryExpr, crate::parser::ParseError> {
    if let Some(text) = ctx.try_consume_string() {
        return Ok(FulltextQueryExpr::Simple(text));
    }

    if ctx.is_identifier_token() {
        let field = ctx.consume_identifier()?;
        if ctx.consume_optional_token(":") {
            let query = ctx.consume_string()?;
            return Ok(FulltextQueryExpr::Field(field, query));
        }
    }

    if let Some(text) = ctx.try_consume_quoted_string() {
        return Ok(FulltextQueryExpr::Phrase(text));
    }

    Err(crate::parser::ParseError::new(
        crate::parser::core::error::ParseErrorKind::SyntaxError,
        "Expected full-text query expression".to_string(),
        ctx.current_position(),
    ))
}

fn parse_yield_clause(
    ctx: &mut ParseContext,
) -> Result<FulltextYieldClause, crate::parser::ParseError> {
    let mut items = Vec::new();

    loop {
        let expr = if ctx.check_keyword("score") && ctx.peek_token().kind != TokenKind::LParen {
            ctx.consume_identifier()?;
            create_contextual_expression(Expression::Function {
                name: "score".to_string(),
                args: vec![],
            })
        } else if ctx.check_keyword("matched_fields") && ctx.peek_token().kind != TokenKind::LParen
        {
            ctx.consume_identifier()?;
            create_contextual_expression(Expression::Function {
                name: "matched_fields".to_string(),
                args: vec![],
            })
        } else if ctx.consume_optional_token("*") {
            create_contextual_expression(Expression::variable("*"))
        } else {
            parse_expression_with_context(ctx, ctx.expression_context_clone())?
        };

        let alias = if ctx.check_keyword("AS") {
            ctx.consume_keyword("AS")?;
            Some(ctx.consume_identifier()?)
        } else {
            None
        };

        items.push(FulltextYieldItem { expr, alias });

        if !ctx.consume_optional_token(",") {
            break;
        }
    }

    Ok(FulltextYieldClause { items })
}

fn parse_order_clause(ctx: &mut ParseContext) -> Result<OrderByClause, crate::parser::ParseError> {
    let span = ctx.current_span();
    let mut items = Vec::new();

    loop {
        let expression = parse_expression_with_context(ctx, ctx.expression_context_clone())?;
        let direction = if ctx.check_keyword("ASC") {
            ctx.consume_keyword("ASC")?;
            OrderDirection::Asc
        } else if ctx.check_keyword("DESC") {
            ctx.consume_keyword("DESC")?;
            OrderDirection::Desc
        } else {
            OrderDirection::Asc
        };

        items.push(OrderByItem {
            expression,
            direction,
        });

        if !ctx.consume_optional_token(",") {
            break;
        }
    }

    Ok(OrderByClause { span, items })
}

fn parse_lookup_fulltext(ctx: &mut ParseContext) -> Result<Stmt, crate::parser::ParseError> {
    ctx.consume_keyword("LOOKUP")?;
    ctx.consume_keyword("ON")?;

    let schema_name = ctx.consume_identifier()?;
    ctx.consume_keyword("INDEX")?;
    let index_name = ctx.consume_identifier()?;

    ctx.consume_keyword("WHERE")?;
    let query = ctx.consume_string()?;

    let mut lookup = LookupFulltext {
        span: ctx.current_span(),
        schema_name,
        index_name,
        query,
        yield_clause: None,
        limit: None,
    };

    if ctx.check_keyword("YIELD") {
        ctx.consume_keyword("YIELD")?;
        lookup.yield_clause = Some(parse_yield_clause(ctx)?);
    }

    if ctx.check_keyword("LIMIT") {
        ctx.consume_keyword("LIMIT")?;
        lookup.limit = Some(ctx.consume_int()? as usize);
    }

    Ok(Stmt::LookupFulltext(lookup))
}

fn parse_match_fulltext(ctx: &mut ParseContext) -> Result<Stmt, crate::parser::ParseError> {
    ctx.consume_keyword("MATCH")?;
    let pattern = ctx.consume_string()?;

    ctx.consume_keyword("WHERE")?;
    ctx.consume_keyword("FULLTEXT_MATCH")?;
    ctx.expect_token(TokenKind::LParen)?;
    let field = ctx.consume_identifier()?;
    ctx.expect_token(TokenKind::Comma)?;
    let query = ctx.consume_string()?;
    ctx.expect_token(TokenKind::RParen)?;

    let condition = FulltextMatchCondition {
        field,
        query,
        index_name: None,
    };

    let mut match_stmt = MatchFulltext {
        span: ctx.current_span(),
        pattern,
        fulltext_condition: condition,
        yield_clause: None,
    };

    if ctx.check_keyword("YIELD") {
        ctx.consume_keyword("YIELD")?;
        match_stmt.yield_clause = Some(parse_yield_clause(ctx)?);
    }

    Ok(Stmt::MatchFulltext(match_stmt))
}

impl IndexFieldDef {
    fn new(field_name: String) -> Self {
        Self {
            field_name,
            analyzer: None,
            boost: None,
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::parser::parsing::parser::Parser;

    #[test]
    fn test_parse_create_fulltext_index() {
        let sql = r#"CREATE FULLTEXT INDEX idx_article_content
                     ON article(title, content)
                     ENGINE BM25"#;

        let mut parser = Parser::new(sql);
        let result = parser.parse();
        assert!(result.is_ok());
    }

    #[test]
    fn test_parse_create_fulltext_index_with_equals_engine() {
        let sql = r#"CREATE FULLTEXT INDEX idx_article_content
                     ON article(content)
                     ENGINE = BM25"#;

        let mut parser = Parser::new(sql);
        let result = parser.parse();
        assert!(result.is_ok());
    }

    #[test]
    fn test_parse_create_fulltext_index_with_analyzer_rejected() {
        let sql = r#"CREATE FULLTEXT INDEX idx_article_content
                     ON article(content ANALYZER 'standard')
                     ENGINE = BM25"#;

        let mut parser = Parser::new(sql);
        let result = parser.parse();
        assert!(result.is_err());
        let collected = format!("{:?}", parser.take_errors());
        assert!(
            collected.contains("ANALYZER"),
            "rejection must name ANALYZER: {}",
            collected
        );
    }

    #[test]
    fn test_parse_create_fulltext_index_with_boost_rejected() {
        let sql = r#"CREATE FULLTEXT INDEX idx_article_content
                     ON article(content BOOST 2.0)
                     ENGINE = BM25"#;

        let mut parser = Parser::new(sql);
        let result = parser.parse();
        assert!(result.is_err());
        let collected = format!("{:?}", parser.take_errors());
        assert!(
            collected.contains("BOOST"),
            "rejection must name BOOST: {}",
            collected
        );
    }

    #[test]
    fn test_parse_create_fulltext_index_with_options_rejected() {
        let sql = r#"CREATE FULLTEXT INDEX idx_article_content
                     ON article(content)
                     ENGINE = BM25 OPTIONS (k1=1.2, b=0.8)"#;

        let mut parser = Parser::new(sql);
        let result = parser.parse();
        assert!(result.is_err());
        let collected = format!("{:?}", parser.take_errors());
        assert!(
            collected.contains("OPTIONS"),
            "rejection must name OPTIONS: {}",
            collected
        );
    }

    #[test]
    fn test_parse_alter_fulltext_add_field_analyzer_rejected() {
        let sql = "ALTER FULLTEXT INDEX idx_article ADD FIELD summary ANALYZER 'standard'";

        let mut parser = Parser::new(sql);
        let result = parser.parse();
        assert!(result.is_err());
        let collected = format!("{:?}", parser.take_errors());
        assert!(
            collected.contains("ANALYZER"),
            "rejection must name ANALYZER: {}",
            collected
        );
    }

    #[test]
    fn test_parse_search_statement() {
        let sql = r#"SEARCH INDEX idx_article MATCH 'database'
                     YIELD doc_id, score() AS s
                     LIMIT 10"#;

        let mut parser = Parser::new(sql);
        let result = parser.parse();
        assert!(result.is_ok());
    }

    #[test]
    fn test_parse_search_statement_where_and_order() {
        let sql = r#"SEARCH INDEX idx_article MATCH 'database'
                     YIELD doc_id, title, score
                     WHERE score > 0.5 AND doc_id != 'art001'
                     ORDER BY score DESC
                     LIMIT 10"#;

        let mut parser = Parser::new(sql);
        let result = parser.parse().expect("SEARCH should parse");
        let stmt = result.ast.stmt();
        let search = match stmt {
            crate::parser::ast::stmt::Stmt::Search(stmt) => stmt,
            _ => panic!("expected SEARCH statement"),
        };

        assert!(search.where_clause.is_some());
        assert!(search.where_clause.as_ref().unwrap().is_binary());
        assert!(search.order_clause.is_some());
        assert_eq!(search.order_clause.as_ref().unwrap().items.len(), 1);

        let yield_items = search.yield_clause.as_ref().expect("YIELD should parse");
        assert_eq!(yield_items.items.len(), 3);
    }

    #[test]
    fn test_parse_search_statement_highlight() {
        let sql = r#"SEARCH INDEX idx_article MATCH 'database'
                     YIELD highlight(title), matched_fields, score"#;

        let mut parser = Parser::new(sql);
        let result = parser.parse().expect("SEARCH should parse");
        let stmt = result.ast.stmt();
        match stmt {
            crate::parser::ast::stmt::Stmt::Search(search) => {
                let items = search
                    .yield_clause
                    .as_ref()
                    .expect("YIELD should parse")
                    .items
                    .clone();
                assert_eq!(items.len(), 3);
                assert!(items[0].expr.is_function());
                assert!(items[1].expr.is_function());
                assert!(items[2].expr.is_function());
            }
            _ => panic!("expected SEARCH statement"),
        }
    }

    #[test]
    fn test_parse_drop_index() {
        let sql = "DROP FULLTEXT INDEX idx_article";

        let mut parser = Parser::new(sql);
        let result = parser.parse();
        assert!(result.is_ok());
    }
}