package executor

import (
	"context"
	"fmt"
	"strings"

	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/parser"
)

func (e *Executor) validateSelectGroupBy(ctx context.Context, user, password string, stmt *parser.SelectStatement) error {
	if stmt == nil {
		return nil
	}
	if stmt.Having != nil {
		return fmt.Errorf("HAVING is not supported")
	}
	if len(stmt.GroupBy) == 0 {
		return nil
	}

	pkCols, err := e.primaryKeyColumns(ctx, user, password, stmt.From)
	if err != nil {
		return err
	}
	if err := validateGroupByWithPK(stmt, pkCols); err == nil {
		return nil
	} else if e.groupByMatchesIndex(ctx, user, password, stmt) {
		return nil
	} else {
		return err
	}
}

func validateGroupByWithPK(stmt *parser.SelectStatement, pkCols []string) error {
	if len(pkCols) == 0 {
		return fmt.Errorf("primary key is required for GROUP BY")
	}

	whereConditions, err := extractEqualityConditionsOptional(stmt.Where)
	if err != nil {
		return err
	}
	restricted := make(map[string]struct{}, len(whereConditions))
	for _, cond := range whereConditions {
		if !isColumnInList(cond.column, pkCols) {
			return fmt.Errorf("GROUP BY requires primary key equality restrictions")
		}
		restricted[strings.ToLower(cond.column)] = struct{}{}
	}

	groupCols := make([]string, len(stmt.GroupBy))
	for i, expr := range stmt.GroupBy {
		col, ok := expr.(*parser.ColumnExpr)
		if !ok {
			return fmt.Errorf("GROUP BY supports only column names")
		}
		groupCols[i] = col.Name
	}

	pkIndex := 0
	for _, col := range groupCols {
		for pkIndex < len(pkCols) && !strings.EqualFold(pkCols[pkIndex], col) {
			if _, ok := restricted[strings.ToLower(pkCols[pkIndex])]; !ok {
				return fmt.Errorf("GROUP BY must follow primary key order")
			}
			pkIndex++
		}
		if pkIndex >= len(pkCols) || !strings.EqualFold(pkCols[pkIndex], col) {
			return fmt.Errorf("GROUP BY must follow primary key order")
		}
		pkIndex++
	}

	groupSet := make(map[string]struct{}, len(groupCols))
	for _, col := range groupCols {
		groupSet[strings.ToLower(col)] = struct{}{}
	}

	for _, expr := range stmt.Columns {
		if !isGroupedSelectExpr(expr, groupSet) {
			return fmt.Errorf("SELECT clause must use grouped columns or aggregates")
		}
	}

	return nil
}

func validateOffsetRequiresLimit(stmt *parser.SelectStatement) error {
	if stmt == nil {
		return nil
	}
	if stmt.Options.Offset != nil && stmt.Options.Limit == nil && stmt.Options.PerPartitionLimit == nil {
		return fmt.Errorf("OFFSET requires LIMIT")
	}
	return nil
}

func extractEqualityConditionsOptional(expr parser.Expression) ([]equalityCondition, error) {
	if expr == nil {
		return nil, nil
	}
	return extractEqualityConditions(expr)
}

func isGroupedSelectExpr(expr parser.Expression, groupSet map[string]struct{}) bool {
	if expr == nil {
		return false
	}
	if aliased, ok := expr.(*parser.AliasedExpr); ok {
		expr = aliased.Expr
	}
	switch e := expr.(type) {
	case *parser.ColumnExpr:
		_, ok := groupSet[strings.ToLower(e.Name)]
		return ok
	case *parser.FunctionCall:
		return isAggregateFunction(e)
	default:
		return false
	}
}

func isAggregateFunction(call *parser.FunctionCall) bool {
	switch strings.ToUpper(call.Name) {
	case "COUNT", "MIN", "MAX", "SUM", "AVG":
		return true
	default:
		return false
	}
}

func (e *Executor) groupByMatchesIndex(ctx context.Context, user, password string, stmt *parser.SelectStatement) bool {
	if stmt == nil || stmt.From == nil || stmt.From.Schema == "" || stmt.From.Name == "" {
		return false
	}
	if e.schemaManager == nil {
		return false
	}
	indexes, err := e.schemaManager.IndexesForTable(ctx, user, password, stmt.From.Schema, stmt.From.Name)
	if err != nil || len(indexes) == 0 {
		return false
	}
	if _, _, ok, err := selectBestIndexForOR(stmt, indexes); err == nil && ok {
		return true
	}
	conds, disqualify, orPattern, err := extractIndexConditions(stmt.Where)
	if err != nil || disqualify || hasUnsupportedConditions(conds) {
		return false
	}
	eqCols, rangeCols := extractConditionColumns(conds)
	groupByCols := groupByColumns(stmt)

	metas := buildIndexMetas(indexes)
	if orPattern != nil && len(orPattern.columns) > 0 {
		metas = filterIndexMetasForOrPattern(metas, orPattern.columns)
	}
	eligible := filterIndexMetas(metas, eqCols, rangeCols, groupByCols)
	return len(eligible) > 0
}