package executor

import (
	"context"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"

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

func defaultReplication() map[string]interface{} {
	return map[string]interface{}{
		"class":              "SimpleStrategy",
		"replication_factor": "1",
	}
}

func replicationJSON(replication map[string]interface{}) ([]byte, error) {
	if len(replication) == 0 {
		return json.Marshal(defaultReplication())
	}
	out := make(map[string]interface{}, len(replication))
	for key, value := range replication {
		normalized := normalizeOptionKey(key)
		if expr, ok := value.(parser.Expression); ok {
			out[normalized] = exprToJSONValue(expr)
		} else {
			out[normalized] = value
		}
	}
	return json.Marshal(out)
}

func optionsJSON(options map[string]interface{}) ([]byte, error) {
	if len(options) == 0 {
		return []byte("{}"), nil
	}
	out := make(map[string]interface{}, len(options))
	for key, value := range options {
		normalized := normalizeOptionKey(key)
		if expr, ok := value.(parser.Expression); ok {
			out[normalized] = exprToJSONValue(expr)
		} else {
			out[normalized] = value
		}
	}
	return json.Marshal(out)
}

func normalizeOptionKey(key string) string {
	trimmed := strings.TrimSpace(key)
	if len(trimmed) >= 2 && trimmed[0] == '\'' && trimmed[len(trimmed)-1] == '\'' {
		return trimmed[1 : len(trimmed)-1]
	}
	return trimmed
}

func exprToJSONValue(expr parser.Expression) interface{} {
	switch e := expr.(type) {
	case *parser.StringLiteral:
		return e.Value
	case *parser.IntegerLiteral:
		return e.Value
	case *parser.FloatLiteral:
		if e.Raw != "" {
			return e.Raw
		}
		return e.Value
	case *parser.BooleanLiteral:
		return strings.EqualFold(e.Value, "true")
	case *parser.UUIDLiteral:
		return e.Value
	case *parser.BlobLiteral:
		return "0x" + hex.EncodeToString(e.Value)
	case *parser.NullLiteral:
		return nil
	case *parser.ListLiteral:
		values := make([]interface{}, 0, len(e.Elements))
		for _, elem := range e.Elements {
			values = append(values, exprToJSONValue(elem))
		}
		return values
	case *parser.SetLiteral:
		values := make([]interface{}, 0, len(e.Elements))
		for _, elem := range e.Elements {
			values = append(values, exprToJSONValue(elem))
		}
		return values
	case *parser.TupleLiteral:
		values := make([]interface{}, 0, len(e.Elements))
		for _, elem := range e.Elements {
			values = append(values, exprToJSONValue(elem))
		}
		return values
	case *parser.MapLiteral:
		values := make(map[string]interface{}, len(e.Entries))
		for _, entry := range e.Entries {
			key := exprToJSONValue(entry.Key)
			values[fmt.Sprint(key)] = exprToJSONValue(entry.Value)
		}
		return values
	case *parser.TypeHintExpr:
		return exprToJSONValue(e.Expr)
	default:
		return expr.String()
	}
}

func commentFromOptions(options map[string]interface{}) string {
	if len(options) == 0 {
		return ""
	}
	raw, ok := options["comment"]
	if !ok {
		return ""
	}
	if expr, ok := raw.(parser.Expression); ok {
		value := exprToJSONValue(expr)
		if str, ok := value.(string); ok {
			return str
		}
		return fmt.Sprint(value)
	}
	return fmt.Sprint(raw)
}

func (e *ddlExecutor) writeKeyspaceMeta(ctx context.Context, conn *sql.Conn, stmt *parser.CreateKeyspaceStatement) error {
	replication, err := replicationJSON(stmt.Replication)
	if err != nil {
		return err
	}
	_, err = conn.ExecContext(ctx, `
		INSERT INTO cql_meta.keyspaces (keyspace_name, durable_writes, replication_json)
		VALUES (?, ?, ?)
		ON DUPLICATE KEY UPDATE durable_writes=VALUES(durable_writes), replication_json=VALUES(replication_json)
	`, stmt.Name, true, replication)
	return err
}

func (e *ddlExecutor) deleteKeyspaceMeta(ctx context.Context, conn *sql.Conn, keyspace string) error {
	statements := []string{
		"DELETE FROM cql_meta.columns WHERE keyspace_name = ?",
		"DELETE FROM cql_meta.indexes WHERE keyspace_name = ?",
		"DELETE FROM cql_meta.tables WHERE keyspace_name = ?",
		"DELETE FROM cql_meta.types WHERE keyspace_name = ?",
		"DELETE FROM cql_meta.views WHERE keyspace_name = ?",
		"DELETE FROM cql_meta.triggers WHERE keyspace_name = ?",
		"DELETE FROM cql_meta.keyspaces WHERE keyspace_name = ?",
	}
	for _, stmt := range statements {
		if _, err := conn.ExecContext(ctx, stmt, keyspace); err != nil {
			return err
		}
	}
	return nil
}

func (e *ddlExecutor) writeTableMeta(ctx context.Context, conn *sql.Conn, stmt *parser.CreateTableStatement) error {
	options, err := optionsJSON(stmt.Options)
	if err != nil {
		return err
	}
	comment := commentFromOptions(stmt.Options)
	_, err = conn.ExecContext(ctx, `
		INSERT INTO cql_meta.tables (keyspace_name, table_name, options_json, comment)
		VALUES (?, ?, ?, ?)
		ON DUPLICATE KEY UPDATE options_json=VALUES(options_json), comment=VALUES(comment)
	`, stmt.Name.Schema, stmt.Name.Name, options, comment)
	if err != nil {
		return err
	}

	if _, err := conn.ExecContext(ctx, `
		DELETE FROM cql_meta.columns WHERE keyspace_name = ? AND table_name = ?
	`, stmt.Name.Schema, stmt.Name.Name); err != nil {
		return err
	}

	columns := buildColumnMeta(stmt)
	if len(columns) == 0 {
		return nil
	}

	var sqlBuilder strings.Builder
	sqlBuilder.WriteString(`
		INSERT INTO cql_meta.columns
			(keyspace_name, table_name, column_name, cql_type, default_expr, on_update_expr, kind, position, clustering_order)
		VALUES `)

	args := make([]interface{}, 0, len(columns)*9)
	for i, col := range columns {
		if i > 0 {
			sqlBuilder.WriteString(", ")
		}
		sqlBuilder.WriteString("(?, ?, ?, ?, ?, ?, ?, ?, ?)")
		args = append(args, stmt.Name.Schema, stmt.Name.Name, col.name, col.typ, col.defaultExpr, col.onUpdateExpr, col.kind, col.position, col.order)
	}

	_, err = conn.ExecContext(ctx, sqlBuilder.String(), args...)
	return err
}

func (e *ddlExecutor) deleteTableMeta(ctx context.Context, conn *sql.Conn, keyspace, table string) error {
	statements := []string{
		"DELETE FROM cql_meta.columns WHERE keyspace_name = ? AND table_name = ?",
		"DELETE FROM cql_meta.indexes WHERE keyspace_name = ? AND table_name = ?",
		"DELETE FROM cql_meta.triggers WHERE keyspace_name = ? AND table_name = ?",
		"DELETE FROM cql_meta.tables WHERE keyspace_name = ? AND table_name = ?",
	}
	for _, stmt := range statements {
		if _, err := conn.ExecContext(ctx, stmt, keyspace, table); err != nil {
			return err
		}
	}
	return nil
}

func (e *ddlExecutor) alterTableMeta(ctx context.Context, conn *sql.Conn, stmt *parser.AlterTableStatement) error {
	switch stmt.Operation {
	case "ADD":
		cqlType := cqlTypeString(stmt.Type, stmt.Table.Schema)
		_, err := conn.ExecContext(ctx, `
			INSERT INTO cql_meta.columns
				(keyspace_name, table_name, column_name, cql_type, default_expr, on_update_expr, kind, position, clustering_order)
			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
		`, stmt.Table.Schema, stmt.Table.Name, stmt.Column, cqlType, "", "", "regular", -1, "NONE")
		return err
	case "DROP":
		_, err := conn.ExecContext(ctx, `
			DELETE FROM cql_meta.columns
			WHERE keyspace_name = ? AND table_name = ? AND column_name = ?
		`, stmt.Table.Schema, stmt.Table.Name, stmt.Column)
		return err
	case "RENAME":
		for _, pair := range stmt.Renames {
			if _, err := conn.ExecContext(ctx, `
					UPDATE cql_meta.columns
					SET column_name = ?
					WHERE keyspace_name = ? AND table_name = ? AND column_name = ?
				`, pair.To, stmt.Table.Schema, stmt.Table.Name, pair.From); err != nil {
				return err
			}
		}
		return nil
	case "WITH":
		var rawOptions []byte
		var currentComment sql.NullString
		err := conn.QueryRowContext(ctx, `
			SELECT options_json, comment
			FROM cql_meta.tables
			WHERE keyspace_name = ? AND table_name = ?
		`, stmt.Table.Schema, stmt.Table.Name).Scan(&rawOptions, &currentComment)
		if err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return fmt.Errorf("metadata for table %s.%s not found", stmt.Table.Schema, stmt.Table.Name)
			}
			return err
		}

		merged := make(map[string]interface{})
		if len(rawOptions) > 0 {
			if err := json.Unmarshal(rawOptions, &merged); err != nil {
				return fmt.Errorf("failed to decode table options: %w", err)
			}
		}
		for key, value := range stmt.Options {
			normalized := normalizeOptionKey(key)
			if expr, ok := value.(parser.Expression); ok {
				merged[normalized] = exprToJSONValue(expr)
			} else {
				merged[normalized] = value
			}
		}
		updatedOptions, err := json.Marshal(merged)
		if err != nil {
			return err
		}
		comment := ""
		if currentComment.Valid {
			comment = currentComment.String
		}
		if raw, ok := merged["comment"]; ok {
			if raw == nil {
				comment = ""
			} else {
				comment = fmt.Sprint(raw)
			}
		}
		_, err = conn.ExecContext(ctx, `
			UPDATE cql_meta.tables
			SET options_json = ?, comment = ?
			WHERE keyspace_name = ? AND table_name = ?
		`, updatedOptions, comment, stmt.Table.Schema, stmt.Table.Name)
		return err
	default:
		return nil
	}
}

func (e *ddlExecutor) writeIndexMeta(ctx context.Context, conn *sql.Conn, stmt *parser.CreateIndexStatement) error {
	options := map[string]interface{}{
		"target": stmt.Columns,
	}
	if stmt.UsingClass != "" {
		options["class_name"] = stmt.UsingClass
	}
	if len(stmt.Options) > 0 {
		for key, value := range stmt.Options {
			normalized := normalizeOptionKey(key)
			if expr, ok := value.(parser.Expression); ok {
				options[normalized] = exprToJSONValue(expr)
			} else {
				options[normalized] = value
			}
		}
	}
	optionsJSON, err := json.Marshal(options)
	if err != nil {
		return err
	}
	kind := "CUSTOM"
	_, err = conn.ExecContext(ctx, `
		INSERT INTO cql_meta.indexes (keyspace_name, table_name, index_name, kind, options_json)
		VALUES (?, ?, ?, ?, ?)
		ON DUPLICATE KEY UPDATE kind=VALUES(kind), options_json=VALUES(options_json)
	`, stmt.Table.Schema, stmt.Table.Name, stmt.Name, kind, optionsJSON)
	return err
}

func (e *ddlExecutor) deleteIndexMeta(ctx context.Context, conn *sql.Conn, keyspace, table, index string) error {
	_, err := conn.ExecContext(ctx, `
		DELETE FROM cql_meta.indexes
		WHERE keyspace_name = ? AND table_name = ? AND index_name = ?
	`, keyspace, table, index)
	return err
}

func (e *ddlExecutor) writeTypeMeta(ctx context.Context, conn *sql.Conn, stmt *parser.CreateTypeStatement) error {
	if stmt == nil || stmt.Name == nil {
		return fmt.Errorf("type name is required")
	}
	fieldNames := make([]string, len(stmt.Fields))
	fieldTypes := make([]string, len(stmt.Fields))
	for i, field := range stmt.Fields {
		fieldNames[i] = field.Name
		fieldTypes[i] = cqlTypeString(field.Type, stmt.Name.Schema)
	}
	namesJSON, err := json.Marshal(fieldNames)
	if err != nil {
		return err
	}
	typesJSON, err := json.Marshal(fieldTypes)
	if err != nil {
		return err
	}
	_, err = conn.ExecContext(ctx, `
		INSERT INTO cql_meta.types (keyspace_name, type_name, field_names_json, field_types_json)
		VALUES (?, ?, ?, ?)
	`, stmt.Name.Schema, stmt.Name.Name, namesJSON, typesJSON)
	return err
}

func (e *ddlExecutor) updateTypeMeta(ctx context.Context, conn *sql.Conn, keyspace, name string, fieldNames, fieldTypes []string) error {
	namesJSON, err := json.Marshal(fieldNames)
	if err != nil {
		return err
	}
	typesJSON, err := json.Marshal(fieldTypes)
	if err != nil {
		return err
	}
	_, err = conn.ExecContext(ctx, `
		UPDATE cql_meta.types
		SET field_names_json = ?, field_types_json = ?
		WHERE keyspace_name = ? AND type_name = ?
	`, namesJSON, typesJSON, keyspace, name)
	return err
}

func (e *ddlExecutor) deleteTypeMeta(ctx context.Context, conn *sql.Conn, keyspace, name string) error {
	_, err := conn.ExecContext(ctx, `
		DELETE FROM cql_meta.types
		WHERE keyspace_name = ? AND type_name = ?
	`, keyspace, name)
	return err
}

func (e *ddlExecutor) writeViewMeta(ctx context.Context, conn *sql.Conn, stmt *parser.CreateMaterializedViewStatement) error {
	if stmt == nil || stmt.Name == nil || stmt.Select == nil || stmt.Select.From == nil {
		return fmt.Errorf("view metadata requires name and select")
	}
	options, err := optionsJSON(stmt.Options)
	if err != nil {
		return err
	}
	whereClause := ""
	if stmt.Select.Where != nil {
		whereClause = stmt.Select.Where.String()
	}
	_, err = conn.ExecContext(ctx, `
		INSERT INTO cql_meta.views (keyspace_name, view_name, base_table, where_clause, options_json)
		VALUES (?, ?, ?, ?, ?)
	`, stmt.Name.Schema, stmt.Name.Name, stmt.Select.From.Name, whereClause, options)
	return err
}

func (e *ddlExecutor) deleteViewMeta(ctx context.Context, conn *sql.Conn, keyspace, name string) error {
	_, err := conn.ExecContext(ctx, `
		DELETE FROM cql_meta.views
		WHERE keyspace_name = ? AND view_name = ?
	`, keyspace, name)
	return err
}

func (e *ddlExecutor) writeTriggerMeta(ctx context.Context, conn *sql.Conn, stmt *parser.CreateTriggerStatement) error {
	if stmt == nil || stmt.Table == nil {
		return fmt.Errorf("trigger metadata requires table")
	}
	options, err := optionsJSON(map[string]interface{}{
		"class": stmt.Using,
	})
	if err != nil {
		return err
	}
	_, err = conn.ExecContext(ctx, `
		INSERT INTO cql_meta.triggers (keyspace_name, table_name, trigger_name, options_json)
		VALUES (?, ?, ?, ?)
	`, stmt.Table.Schema, stmt.Table.Name, stmt.Name, options)
	return err
}

func (e *ddlExecutor) deleteTriggerMeta(ctx context.Context, conn *sql.Conn, keyspace, table, name string) error {
	_, err := conn.ExecContext(ctx, `
		DELETE FROM cql_meta.triggers
		WHERE keyspace_name = ? AND table_name = ? AND trigger_name = ?
	`, keyspace, table, name)
	return err
}

type columnMeta struct {
	name         string
	typ          string
	defaultExpr  string
	onUpdateExpr string
	kind         string
	order        string
	position     int
}

func buildColumnMeta(stmt *parser.CreateTableStatement) []columnMeta {
	if stmt == nil || len(stmt.Columns) == 0 {
		return nil
	}
	var partitionKeys []string
	var clusteringKeys []string
	if stmt.PrimaryKey != nil {
		partitionKeys = stmt.PrimaryKey.PartitionKeys
		clusteringKeys = stmt.PrimaryKey.ClusteringColumns
	} else {
		for _, col := range stmt.Columns {
			if col != nil && col.Primary {
				partitionKeys = []string{col.Name}
				break
			}
		}
	}

	meta := make([]columnMeta, 0, len(stmt.Columns))
	for _, col := range stmt.Columns {
		kind := "regular"
		order := "NONE"
		position := -1
		switch {
		case col.Static:
			kind = "static"
		case partitionColumnPosition(col.Name, partitionKeys) >= 0:
			kind = "partition_key"
			position = partitionColumnPosition(col.Name, partitionKeys)
		case clusteringColumnPosition(col.Name, clusteringKeys) >= 0:
			kind = "clustering"
			position = clusteringColumnPosition(col.Name, clusteringKeys)
			order = "ASC"
			if stmt.ClusteringOrder != nil {
				if val, ok := stmt.ClusteringOrder[col.Name]; ok && val != "" {
					order = val
				}
			}
		}

		meta = append(meta, columnMeta{
			name:         col.Name,
			typ:          cqlTypeString(col.Type, stmt.Name.Schema),
			defaultExpr:  normalizeColumnExprForMeta(col.DefaultExpr),
			onUpdateExpr: normalizeColumnExprForMeta(col.OnUpdateExpr),
			kind:         kind,
			order:        order,
			position:     position,
		})
	}

	return meta
}

func partitionColumnPosition(name string, partition []string) int {
	for i, key := range partition {
		if key == name {
			return i
		}
	}
	return -1
}

func clusteringColumnPosition(name string, clustering []string) int {
	for i, key := range clustering {
		if key == name {
			return i
		}
	}
	return -1
}

func nextColumnPosition(ctx context.Context, conn *sql.Conn, keyspace, table string) (int, error) {
	var max sql.NullInt64
	if err := conn.QueryRowContext(ctx, `
		SELECT MAX(position)
		FROM cql_meta.columns
		WHERE keyspace_name = ? AND table_name = ?
	`, keyspace, table).Scan(&max); err != nil {
		return 0, err
	}
	if !max.Valid {
		return 0, nil
	}
	return int(max.Int64 + 1), nil
}