package executor

import (
	"context"
	"fmt"
	"strings"

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

const (
	counterPrimaryKeyError = "counter columns cannot be part of the primary key"
	counterMixError        = "counter tables may only contain counter columns"
)

func isCounterTypeString(cqlType string) bool {
	return normalizeCQLTypeString(cqlType) == "counter"
}

func validateCounterTableDefinition(table *parser.CreateTableStatement) error {
	if table == nil || table.Name == nil {
		return nil
	}
	pkCols := primaryKeyColumnsFromDefinition(table.PrimaryKey)
	pkSet := make(map[string]struct{}, len(pkCols))
	for _, name := range pkCols {
		pkSet[strings.ToLower(name)] = struct{}{}
	}

	hasCounter := false
	hasNonCounter := false
	for _, col := range table.Columns {
		if col == nil {
			continue
		}
		colName := strings.ToLower(col.Name)
		typeStr := cqlTypeString(col.Type, table.Name.Schema)
		isCounter := isCounterTypeString(typeStr)
		if isCounter {
			if _, ok := pkSet[colName]; ok {
				return fmt.Errorf("%s: %s", counterPrimaryKeyError, col.Name)
			}
		}
		if _, ok := pkSet[colName]; ok {
			continue
		}
		if isCounter {
			hasCounter = true
		} else {
			hasNonCounter = true
		}
		if hasCounter && hasNonCounter {
			return fmt.Errorf(counterMixError)
		}
	}
	return nil
}

func primaryKeyColumnsFromDefinition(pk *parser.PrimaryKey) []string {
	if pk == nil {
		return nil
	}
	total := len(pk.PartitionKeys) + len(pk.ClusteringColumns)
	if total == 0 {
		return nil
	}
	out := make([]string, 0, total)
	out = append(out, pk.PartitionKeys...)
	out = append(out, pk.ClusteringColumns...)
	return out
}

func validateCounterTableAlterAdd(columnType string, columnTypes map[string]string, pkCols []string) error {
	hasCounter, hasNonCounter := counterColumnState(columnTypes, pkCols)
	if hasCounter && hasNonCounter {
		return fmt.Errorf(counterMixError)
	}
	newIsCounter := isCounterTypeString(columnType)
	switch {
	case newIsCounter && hasNonCounter:
		return fmt.Errorf(counterMixError)
	case !newIsCounter && hasCounter:
		return fmt.Errorf(counterMixError)
	default:
		return nil
	}
}

func counterColumnState(columnTypes map[string]string, pkCols []string) (bool, bool) {
	if len(columnTypes) == 0 {
		return false, false
	}
	pkSet := make(map[string]struct{}, len(pkCols))
	for _, name := range pkCols {
		pkSet[strings.ToLower(name)] = struct{}{}
	}
	hasCounter := false
	hasNonCounter := false
	for name, typ := range columnTypes {
		if _, ok := pkSet[strings.ToLower(name)]; ok {
			continue
		}
		if isCounterTypeString(typ) {
			hasCounter = true
		} else {
			hasNonCounter = true
		}
		if hasCounter && hasNonCounter {
			return true, true
		}
	}
	return hasCounter, hasNonCounter
}

func tableHasCounterColumns(columnTypes map[string]string) bool {
	for _, typ := range columnTypes {
		if isCounterTypeString(typ) {
			return true
		}
	}
	return false
}

func (e *dmlExecutor) rejectCounterInsert(ctx context.Context, user, password string, stmt *parser.InsertStatement) error {
	if stmt == nil {
		return nil
	}
	columnTypes, err := e.lookupColumnTypes(ctx, user, password, stmt.Table)
	if err != nil {
		return err
	}
	if tableHasCounterColumns(columnTypes) {
		return cqlerrors.Invalid("INSERT does not support counters")
	}
	return nil
}