package executor

import (
	"context"
	"database/sql"
	"fmt"
	"strings"

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

func (e *ddlExecutor) executeCreateView(ctx context.Context, user, password string, stmt *parser.CreateMaterializedViewStatement) (*QueryResult, error) {
	if stmt == nil || stmt.Name == nil {
		return nil, cqlerrors.Invalid("view name is required")
	}
	if err := e.validateTableName(stmt.Name); err != nil {
		return nil, err
	}
	if err := e.ensureUserKeyspace(stmt.Name.Schema); err != nil {
		return nil, err
	}
	if err := e.ensureSchemaExists(ctx, user, password, stmt.Name.Schema); err != nil {
		return nil, err
	}
	if stmt.Select == nil || stmt.Select.From == nil {
		return nil, cqlerrors.Invalid("view select statement is required")
	}
	if err := e.validateTableName(stmt.Select.From); err != nil {
		return nil, err
	}
	if stmt.Name.Schema != stmt.Select.From.Schema {
		return nil, cqlerrors.Invalid("view keyspace must match base table keyspace")
	}
	if err := e.ensureTableExists(ctx, user, password, stmt.Select.From.Schema, stmt.Select.From.Name); err != nil {
		return nil, err
	}
	if stmt.PrimaryKey == nil {
		return nil, cqlerrors.Invalid("PRIMARY KEY is required for materialized view")
	}
	metadataOnly := false
	if err := validateViewSelect(stmt.Select); err != nil {
		if config.GetConfig().Features.MetadataOnlyViews {
			metadataOnly = true
		} else {
			return nil, cqlerrors.Invalid(err.Error())
		}
	}
	if e.schemaManager == nil {
		return nil, cqlerrors.Invalid("schema manager not initialized")
	}
	if err := e.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		return nil, err
	}
	existing, _, err := e.schemaManager.CQLView(ctx, user, password, stmt.Name.Schema, stmt.Name.Name)
	if err != nil {
		return nil, err
	}
	if existing != nil {
		if stmt.IfNotExists {
			return &QueryResult{}, nil
		}
		return nil, cqlerrors.AlreadyExists(fmt.Sprintf("view %q already exists", stmt.Name.Name))
	}

	ddlSQL := ""
	if !metadataOnly {
		var err error
		ddlSQL, err = e.translator.Translate(stmt)
		if err != nil {
			return nil, cqlerrors.Invalid(err.Error())
		}
	}
	result, err := e.executeDDLWithMeta(ctx, user, password, ddlSQL, func(ctx context.Context, conn *sql.Conn) error {
		return e.writeViewMeta(ctx, conn, stmt)
	})
	if err != nil {
		return nil, err
	}
	e.invalidateSchemaCache(user, password)
	out, err := e.withSchemaChange(ctx, user, password, result, "CREATED", "TABLE", stmt.Name.Schema, stmt.Name.Name)
	if err != nil {
		return nil, err
	}
	if metadataOnly {
		out.Warnings = append(out.Warnings, "view definition stored in metadata only; MySQL view not created")
	}
	return out, nil
}

func (e *ddlExecutor) executeDropView(ctx context.Context, user, password string, stmt *parser.DropMaterializedViewStatement) (*QueryResult, error) {
	if stmt == nil || stmt.Name == nil {
		return nil, cqlerrors.Invalid("view name is required")
	}
	if err := e.validateTableName(stmt.Name); err != nil {
		return nil, err
	}
	if err := e.ensureUserKeyspace(stmt.Name.Schema); err != nil {
		return nil, err
	}
	if err := e.ensureSchemaExists(ctx, user, password, stmt.Name.Schema); err != nil {
		return nil, err
	}
	if e.schemaManager == nil {
		return nil, cqlerrors.Invalid("schema manager not initialized")
	}
	if err := e.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		return nil, err
	}

	existing, _, err := e.schemaManager.CQLView(ctx, user, password, stmt.Name.Schema, stmt.Name.Name)
	if err != nil {
		return nil, err
	}
	if existing == nil {
		if stmt.IfExists {
			return &QueryResult{}, nil
		}
		return nil, cqlerrors.Invalid(fmt.Sprintf("view %q does not exist", stmt.Name.Name))
	}

	useIfExists := stmt.IfExists || config.GetConfig().Features.MetadataOnlyViews
	dropStmt := *stmt
	dropStmt.IfExists = useIfExists
	ddlSQL, err := e.translator.Translate(&dropStmt)
	if err != nil {
		return nil, cqlerrors.Invalid(err.Error())
	}
	result, err := e.executeDDLWithMeta(ctx, user, password, ddlSQL, func(ctx context.Context, conn *sql.Conn) error {
		return e.deleteViewMeta(ctx, conn, stmt.Name.Schema, stmt.Name.Name)
	})
	if err != nil {
		return nil, err
	}
	e.invalidateSchemaCache(user, password)
	return e.withSchemaChange(ctx, user, password, result, "DROPPED", "TABLE", stmt.Name.Schema, stmt.Name.Name)
}

func validateViewSelect(stmt *parser.SelectStatement) error {
	if stmt == nil || stmt.From == nil {
		return fmt.Errorf("view select must reference a base table")
	}
	if stmt.Having != nil {
		return fmt.Errorf("HAVING is not supported in view definitions")
	}
	for _, expr := range stmt.Columns {
		if err := validateViewExpression(expr, true, true); err != nil {
			return err
		}
	}
	if err := validateViewExpression(stmt.Where, false, false); err != nil {
		return err
	}
	return nil
}

func validateViewExpression(expr parser.Expression, allowFunctions, allowStar bool) error {
	switch e := expr.(type) {
	case nil:
		return nil
	case *parser.ColumnExpr, *parser.StringLiteral, *parser.BlobLiteral, *parser.IntegerLiteral,
		*parser.FloatLiteral, *parser.BooleanLiteral, *parser.UUIDLiteral, *parser.NullLiteral:
		return nil
	case *parser.StarExpr:
		if allowStar {
			return nil
		}
		return fmt.Errorf("star expressions are not allowed in view WHERE clause")
	case *parser.BindMarker, *parser.NamedBindMarker:
		return fmt.Errorf("bind markers are not allowed in view definitions")
	case *parser.BinaryExpr:
		if viewUnsupportedOperator(e.Op) {
			return fmt.Errorf("unsupported operator %q in view definition", e.Op)
		}
		if err := validateViewExpression(e.Left, allowFunctions, allowStar); err != nil {
			return err
		}
		return validateViewExpression(e.Right, allowFunctions, allowStar)
	case *parser.UnaryExpr:
		return validateViewExpression(e.Expr, allowFunctions, allowStar)
	case *parser.IsNullExpr:
		return validateViewExpression(e.Expr, allowFunctions, allowStar)
	case *parser.AliasedExpr:
		return validateViewExpression(e.Expr, allowFunctions, allowStar)
	case *parser.CastExpr:
		return validateViewExpression(e.Expr, allowFunctions, allowStar)
	case *parser.TypeHintExpr:
		return validateViewExpression(e.Expr, allowFunctions, allowStar)
	case *parser.FunctionCall:
		if !allowFunctions {
			return fmt.Errorf("functions are not supported in view WHERE clause")
		}
		for _, arg := range e.Args {
			if err := validateViewExpression(arg, allowFunctions, allowStar); err != nil {
				return err
			}
		}
		return nil
	case *parser.ListLiteral, *parser.SetLiteral, *parser.MapLiteral,
		*parser.TupleLiteral, *parser.VectorLiteral, *parser.CollectionAccessExpr:
		return fmt.Errorf("unsupported expression type %T in view definition", expr)
	default:
		return fmt.Errorf("unsupported expression type %T in view definition", expr)
	}
}

func viewUnsupportedOperator(op string) bool {
	switch strings.ToUpper(strings.TrimSpace(op)) {
	case "CONTAINS", "CONTAINS KEY":
		return true
	default:
		return false
	}
}