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/keyspace"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/parser"
)

func (e *Executor) validateKeyspaceName(name string) error {
	if err := keyspace.ValidateIdentifier(name, "keyspace"); err != nil {
		return cqlerrors.Invalid(err.Error())
	}
	return nil
}

func (e *Executor) validateTableName(table *parser.TableName) error {
	if table == nil {
		return cqlerrors.Invalid("table name is required")
	}
	if table.Schema == "" {
		if table.Name == "" {
			return cqlerrors.Invalid("keyspace is required")
		}
		return cqlerrors.Invalid(fmt.Sprintf("keyspace is required for %s", table.Name))
	}
	if err := keyspace.ValidateIdentifier(table.Schema, "keyspace"); err != nil {
		return cqlerrors.Invalid(err.Error())
	}
	if err := keyspace.ValidateIdentifier(table.Name, "table"); err != nil {
		return cqlerrors.Invalid(err.Error())
	}
	return nil
}

func (e *Executor) validateColumnName(name string) error {
	if err := keyspace.ValidateIdentifier(name, "column"); err != nil {
		return cqlerrors.Invalid(err.Error())
	}
	return nil
}

func validateQueryWhereClause(expr parser.Expression) error {
	if !exprContainsNullLiteral(expr) {
		return nil
	}
	return cqlerrors.Invalid("NULL literal is not allowed in WHERE clause")
}

func exprContainsNullLiteral(expr parser.Expression) bool {
	if expr == nil {
		return false
	}
	switch e := expr.(type) {
	case *parser.NullLiteral:
		return true
	case *parser.TypeHintExpr:
		return exprContainsNullLiteral(e.Expr)
	case *parser.CastExpr:
		return exprContainsNullLiteral(e.Expr)
	case *parser.IsNullExpr:
		return exprContainsNullLiteral(e.Expr)
	case *parser.AliasedExpr:
		return exprContainsNullLiteral(e.Expr)
	case *parser.BinaryExpr:
		return exprContainsNullLiteral(e.Left) || exprContainsNullLiteral(e.Right)
	case *parser.UnaryExpr:
		return exprContainsNullLiteral(e.Expr)
	case *parser.FunctionCall:
		for _, arg := range e.Args {
			if exprContainsNullLiteral(arg) {
				return true
			}
		}
		return false
	case *parser.CollectionAccessExpr:
		return exprContainsNullLiteral(e.Target) || exprContainsNullLiteral(e.Start) || exprContainsNullLiteral(e.End)
	case *parser.ListLiteral:
		for _, elem := range e.Elements {
			if exprContainsNullLiteral(elem) {
				return true
			}
		}
		return false
	case *parser.SetLiteral:
		for _, elem := range e.Elements {
			if exprContainsNullLiteral(elem) {
				return true
			}
		}
		return false
	case *parser.TupleLiteral:
		for _, elem := range e.Elements {
			if exprContainsNullLiteral(elem) {
				return true
			}
		}
		return false
	case *parser.VectorLiteral:
		for _, elem := range e.Elements {
			if exprContainsNullLiteral(elem) {
				return true
			}
		}
		return false
	case *parser.MapLiteral:
		for _, entry := range e.Entries {
			if exprContainsNullLiteral(entry.Key) || exprContainsNullLiteral(entry.Value) {
				return true
			}
		}
		return false
	default:
		return false
	}
}

func (e *Executor) validateIndexColumns(columns []string) error {
	if len(columns) == 0 {
		return cqlerrors.Invalid("index column is required")
	}
	seen := make(map[string]struct{}, len(columns))
	for _, col := range columns {
		if err := e.validateColumnName(col); err != nil {
			return err
		}
		key := strings.ToLower(col)
		if _, ok := seen[key]; ok {
			return cqlerrors.Invalid(fmt.Sprintf("duplicate index column %q", col))
		}
		seen[key] = struct{}{}
	}
	return nil
}

func (e *Executor) validateIndexName(name string) error {
	if err := keyspace.ValidateIdentifier(name, "index"); err != nil {
		return cqlerrors.Invalid(err.Error())
	}
	return nil
}

func (e *Executor) validateTriggerName(name string) error {
	if err := keyspace.ValidateIdentifier(name, "trigger"); err != nil {
		return cqlerrors.Invalid(err.Error())
	}
	return nil
}

func (e *Executor) ensureUserKeyspace(schema string) error {
	if keyspace.IsBuiltIn(schema) {
		return cqlerrors.Invalid(fmt.Sprintf("keyspace %q is read-only", schema))
	}
	return nil
}

func (e *Executor) validateUseKeyspace(ctx context.Context, user, password, name string) error {
	if err := e.validateKeyspaceName(name); err != nil {
		return err
	}
	if err := validateBlockedKeyspace(name); err != nil {
		return err
	}
	if keyspace.IsBuiltIn(name) {
		return nil
	}
	return e.ensureSchemaExists(ctx, user, password, name)
}

func (e *Executor) ensureSchemaExists(ctx context.Context, user, password, schema string) error {
	if keyspace.IsInternalSchema(schema) {
		return cqlerrors.Invalid(fmt.Sprintf("keyspace %q is reserved", schema))
	}
	if e.schemaManager == nil {
		return cqlerrors.Invalid("schema manager is not configured")
	}
	exists, err := e.schemaManager.SchemaExists(ctx, user, password, schema)
	if err != nil {
		return err
	}
	if !exists {
		return cqlerrors.Invalid(fmt.Sprintf("keyspace %q does not exist", schema))
	}
	return nil
}

func (e *Executor) ensureTableExists(ctx context.Context, user, password, schema, table string) error {
	if e.schemaManager == nil {
		return cqlerrors.Invalid("schema manager is not configured")
	}
	exists, err := e.schemaManager.TableExists(ctx, user, password, schema, table)
	if err != nil {
		return err
	}
	if !exists {
		return cqlerrors.Invalid(fmt.Sprintf("table %q.%q does not exist", schema, table))
	}
	return nil
}

func (e *Executor) ensureColumnExists(ctx context.Context, user, password, schema, table, column string) (bool, error) {
	if e.schemaManager == nil {
		return false, cqlerrors.Invalid("schema manager is not configured")
	}
	return e.schemaManager.ColumnExists(ctx, user, password, schema, table, column)
}

func (e *Executor) validatePreparedStatement(ctx context.Context, user, password string, stmt parser.Statement) error {
	switch s := stmt.(type) {
	case *parser.UseStatement:
		return e.validateUseKeyspace(ctx, user, password, s.Keyspace)
	case *parser.CreateKeyspaceStatement:
		if err := e.validateKeyspaceName(s.Name); err != nil {
			return err
		}
		if keyspace.IsBuiltIn(s.Name) || keyspace.IsInternalSchema(s.Name) {
			return cqlerrors.Invalid(fmt.Sprintf("keyspace %q is reserved", s.Name))
		}
		return nil
	case *parser.DropKeyspaceStatement:
		if err := e.validateKeyspaceName(s.Name); err != nil {
			return err
		}
		if keyspace.IsBuiltIn(s.Name) || keyspace.IsInternalSchema(s.Name) {
			return cqlerrors.Invalid(fmt.Sprintf("keyspace %q is reserved", s.Name))
		}
		return nil
	case *parser.CreateTableStatement:
		if err := e.validateTableName(s.Name); err != nil {
			return err
		}
		if err := e.ensureUserKeyspace(s.Name.Schema); err != nil {
			return err
		}
		if err := e.ensureSchemaExists(ctx, user, password, s.Name.Schema); err != nil {
			return err
		}
		for _, col := range s.Columns {
			if err := e.validateColumnName(col.Name); err != nil {
				return err
			}
		}
		if err := e.validateCreateTableExtensions(s); err != nil {
			return err
		}
		if err := validateCounterTableDefinition(s); err != nil {
			return cqlerrors.Invalid(err.Error())
		}
		return nil
	case *parser.DropTableStatement:
		if err := e.validateTableName(s.Name); err != nil {
			return err
		}
		if err := e.ensureUserKeyspace(s.Name.Schema); err != nil {
			return err
		}
		return e.ensureSchemaExists(ctx, user, password, s.Name.Schema)
	case *parser.AlterTableStatement:
		if err := e.validateTableName(s.Table); err != nil {
			return err
		}
		if err := e.ensureUserKeyspace(s.Table.Schema); err != nil {
			return err
		}
		if err := e.ensureSchemaExists(ctx, user, password, s.Table.Schema); err != nil {
			return err
		}
		switch s.Operation {
		case "ADD", "DROP":
			if err := e.validateColumnName(s.Column); err != nil {
				return err
			}
		case "RENAME":
			for _, pair := range s.Renames {
				if err := e.validateColumnName(pair.From); err != nil {
					return err
				}
				if err := e.validateColumnName(pair.To); err != nil {
					return err
				}
			}
		}
		return nil
	case *parser.CreateIndexStatement:
		if err := e.validateTableName(s.Table); err != nil {
			return err
		}
		if err := e.ensureUserKeyspace(s.Table.Schema); err != nil {
			return err
		}
		if err := e.ensureSchemaExists(ctx, user, password, s.Table.Schema); err != nil {
			return err
		}
		if len(s.Columns) == 0 {
			return cqlerrors.Invalid("index column is required")
		}
		if err := e.validateIndexColumns(s.Columns); err != nil {
			return err
		}
		if s.Name == "" {
			s.Name = defaultIndexName(s.Table.Name, s.Columns)
		}
		return e.validateIndexName(s.Name)
	case *parser.CreateTypeStatement:
		if s.Name == nil {
			return cqlerrors.Invalid("type name is required")
		}
		if err := e.validateTableName(s.Name); err != nil {
			return err
		}
		if err := e.ensureUserKeyspace(s.Name.Schema); err != nil {
			return err
		}
		return e.ensureSchemaExists(ctx, user, password, s.Name.Schema)
	case *parser.AlterTypeStatement:
		if s.Name == nil {
			return cqlerrors.Invalid("type name is required")
		}
		if err := e.validateTableName(s.Name); err != nil {
			return err
		}
		if err := e.ensureUserKeyspace(s.Name.Schema); err != nil {
			return err
		}
		return e.ensureSchemaExists(ctx, user, password, s.Name.Schema)
	case *parser.DropTypeStatement:
		if s.Name == nil {
			return cqlerrors.Invalid("type name is required")
		}
		if err := e.validateTableName(s.Name); err != nil {
			return err
		}
		if err := e.ensureUserKeyspace(s.Name.Schema); err != nil {
			return err
		}
		return e.ensureSchemaExists(ctx, user, password, s.Name.Schema)
	case *parser.CreateMaterializedViewStatement:
		if s.Name == nil {
			return cqlerrors.Invalid("view name is required")
		}
		if err := e.validateTableName(s.Name); err != nil {
			return err
		}
		if err := e.ensureUserKeyspace(s.Name.Schema); err != nil {
			return err
		}
		if err := e.ensureSchemaExists(ctx, user, password, s.Name.Schema); err != nil {
			return err
		}
		if s.Select == nil || s.Select.From == nil {
			return cqlerrors.Invalid("view select statement is required")
		}
		if err := e.validateTableName(s.Select.From); err != nil {
			return err
		}
		if s.Name.Schema != s.Select.From.Schema {
			return cqlerrors.Invalid("view keyspace must match base table keyspace")
		}
		return e.ensureTableExists(ctx, user, password, s.Select.From.Schema, s.Select.From.Name)
	case *parser.DropMaterializedViewStatement:
		if s.Name == nil {
			return cqlerrors.Invalid("view name is required")
		}
		if err := e.validateTableName(s.Name); err != nil {
			return err
		}
		if err := e.ensureUserKeyspace(s.Name.Schema); err != nil {
			return err
		}
		return e.ensureSchemaExists(ctx, user, password, s.Name.Schema)
	case *parser.CreateTriggerStatement:
		if s.Table == nil {
			return cqlerrors.Invalid("trigger table is required")
		}
		if err := e.validateTriggerName(s.Name); err != nil {
			return err
		}
		if err := e.validateTableName(s.Table); err != nil {
			return err
		}
		if err := e.ensureUserKeyspace(s.Table.Schema); err != nil {
			return err
		}
		return e.ensureTableExists(ctx, user, password, s.Table.Schema, s.Table.Name)
	case *parser.DropTriggerStatement:
		if s.Table == nil {
			return cqlerrors.Invalid("trigger table is required")
		}
		if err := e.validateTriggerName(s.Name); err != nil {
			return err
		}
		if err := e.validateTableName(s.Table); err != nil {
			return err
		}
		if err := e.ensureUserKeyspace(s.Table.Schema); err != nil {
			return err
		}
		return e.ensureTableExists(ctx, user, password, s.Table.Schema, s.Table.Name)
	case *parser.DropIndexStatement:
		if s.Name == "" {
			return cqlerrors.Invalid("index name is required")
		}
		schema, index := splitIndexName(s.Name)
		if schema == "" {
			return cqlerrors.Invalid("keyspace is required for DROP INDEX")
		}
		if err := e.validateKeyspaceName(schema); err != nil {
			return err
		}
		if err := e.validateIndexName(index); err != nil {
			return err
		}
		if err := e.ensureUserKeyspace(schema); err != nil {
			return err
		}
		return e.ensureSchemaExists(ctx, user, password, schema)
	case *parser.BatchStatement:
		for _, entry := range s.Entries {
			if err := validateBatchEntry(entry); err != nil {
				return err
			}
			if err := e.validateDML(ctx, user, password, entry); err != nil {
				return err
			}
		}
		if err := e.ValidateBatchCounterUsage(ctx, user, password, isCounterBatchType(s.BatchType), s.Entries); err != nil {
			return err
		}
		return nil
	case *parser.SelectStatement, *parser.InsertStatement, *parser.UpdateStatement, *parser.DeleteStatement:
		return e.validateDML(ctx, user, password, stmt)
	default:
		return nil
	}
}