package executor

import (
	"strings"

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

func (e *Executor) protocolModeNormalized() string {
	if e == nil {
		return config.ExecutorProtocolModeStrictCassandra
	}
	switch strings.ToLower(strings.TrimSpace(e.protocolMode)) {
	case config.ExecutorProtocolModeExtendedSQL, config.ExecutorProtocolModeExtendedTiDB:
		return config.ExecutorProtocolModeExtendedSQL
	default:
		return config.ExecutorProtocolModeStrictCassandra
	}
}

func (e *Executor) extendedProtocolModeEnabled() bool {
	return e.protocolModeNormalized() == config.ExecutorProtocolModeExtendedSQL
}

func (e *Executor) batchUpdateMergeEnabled() bool {
	return e.extendedProtocolModeEnabled()
}

func (e *Executor) shouldInsertDuplicateIncrement(table *parser.TableName, column string) bool {
	if e == nil || !e.extendedProtocolModeEnabled() || table == nil || column == "" {
		return false
	}
	if len(e.insertDuplicateIncrements) == 0 {
		return false
	}
	_, ok := e.insertDuplicateIncrements[tableColumnKey(table.Schema, table.Name, column)]
	return ok
}

func parseInsertDuplicateIncrementColumns(values []string) map[string]struct{} {
	if len(values) == 0 {
		return nil
	}
	out := make(map[string]struct{}, len(values))
	for _, raw := range values {
		key, ok := parseQualifiedColumnKey(raw)
		if !ok {
			continue
		}
		out[key] = struct{}{}
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

func parseQualifiedColumnKey(raw string) (string, bool) {
	parts := strings.Split(strings.TrimSpace(raw), ".")
	if len(parts) != 3 {
		return "", false
	}
	for i := range parts {
		parts[i] = strings.ToLower(strings.TrimSpace(parts[i]))
		if parts[i] == "" {
			return "", false
		}
	}
	return parts[0] + "." + parts[1] + "." + parts[2], true
}

func tableColumnKey(schema, table, column string) string {
	return strings.ToLower(strings.TrimSpace(schema)) + "." +
		strings.ToLower(strings.TrimSpace(table)) + "." +
		strings.ToLower(strings.TrimSpace(column))
}