package executor

import (
	"bytes"
	"context"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

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

type jsonInsertColumn struct {
	name      string
	cqlType   string
	mysqlType string
}

func (e *dmlExecutor) executeJSONInsert(ctx context.Context, exec sqlExecutor, user, password string, stmt *parser.InsertStatement, params []interface{}) (*QueryResult, error) {
	if stmt == nil || !stmt.JSON {
		return nil, fmt.Errorf("JSON insert statement is required")
	}
	if stmt.TTL != nil {
		return nil, cqlerrors.Invalid("USING TTL is not supported")
	}

	markerIndex := buildBindMarkerIndex(stmt)
	jsonInput, err := resolveJSONInput(stmt.JSONValue, params, markerIndex)
	if err != nil {
		return nil, cqlerrors.Invalid(err.Error())
	}

	columns, err := e.schemaManager.ColumnsForTable(ctx, user, password, stmt.Table.Schema, stmt.Table.Name)
	if err != nil {
		return nil, err
	}
	if len(columns) == 0 {
		return nil, cqlerrors.Invalid("table columns not found")
	}

	rawMap := make(map[string]json.RawMessage)
	if err := json.Unmarshal([]byte(jsonInput), &rawMap); err != nil {
		return nil, cqlerrors.Invalid(fmt.Sprintf("invalid JSON input: %s", err.Error()))
	}

	columnMap := make(map[string]jsonInsertColumn, len(columns))
	for _, col := range columns {
		cqlType := typemap.MySQLTypeToCQLTypeStringWithPrecisionScale(col.DataType, col.NumericPrecision, col.NumericScale)
		columnMap[strings.ToLower(col.Name)] = jsonInsertColumn{
			name:      col.Name,
			cqlType:   cqlType,
			mysqlType: col.DataType,
		}
	}

	jsonValues := make(map[string]json.RawMessage, len(rawMap))
	for key, raw := range rawMap {
		col, ok := columnMap[strings.ToLower(key)]
		if !ok {
			return nil, cqlerrors.Invalid(fmt.Sprintf("unknown column %q in JSON input", key))
		}
		jsonValues[strings.ToLower(col.name)] = raw
	}

	pkCols, err := e.primaryKeyColumns(ctx, user, password, stmt.Table)
	if err != nil {
		return nil, err
	}
	for _, pk := range pkCols {
		raw, ok := jsonValues[strings.ToLower(pk)]
		if !ok || isJSONNull(raw) {
			return nil, cqlerrors.Invalid(fmt.Sprintf("primary key column %q is required", pk))
		}
	}

	values := make([]interface{}, len(columns))
	for i, col := range columns {
		raw, ok := jsonValues[strings.ToLower(col.Name)]
		if !ok {
			values[i] = nil
			continue
		}
		val, err := convertJSONRawValue(raw, col.DataType, typemap.MySQLTypeToCQLTypeStringWithPrecisionScale(col.DataType, col.NumericPrecision, col.NumericScale))
		if err != nil {
			return nil, cqlerrors.Invalid(err.Error())
		}
		values[i] = val
	}

	sqlStmt := buildJSONInsertSQL(stmt.Table, columns, stmt.IfNotExists)
	metrics.RecordRequestSQL(ctx, sqlStmt, values...)
	start := time.Now()
	result, err := exec.ExecContext(ctx, sqlStmt, values...)
	d := time.Since(start)
	metrics.ObserveStageFromContext(ctx, metrics.StageMySQLQuery, d)
	metrics.ObserveMySQLOperation(ctx, metrics.MySQLOpExecWrite, err == nil, d)
	if err != nil {
		return nil, e.mapMySQLErrorWithSQL(ctx, fmt.Errorf("failed to execute JSON insert: %w", err), mysqlOpWrite, sqlStmt)
	}

	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return nil, fmt.Errorf("failed to get rows affected: %w", err)
	}

	if !stmt.IfNotExists {
		return &QueryResult{RowsAffected: rowsAffected, CQLWriteRows: 1}, nil
	}
	if rowsAffected > 0 {
		result := appliedResult(true, nil)
		result.RowsAffected = rowsAffected
		result.CQLWriteRows = 1
		return result, nil
	}

	fetchSQL, fetchArgs := buildFetchByPKValues(stmt.Table, columns, pkCols, values)
	fetchResult, err := e.executeSelect(ctx, exec, fetchSQL, fetchArgs, mysqlOpWrite)
	if err != nil {
		return nil, err
	}
	fetchCols, err := e.columnInfosForTableColumns(ctx, user, password, stmt.Table, columnNames(columns))
	if err != nil {
		return nil, err
	}
	if len(fetchResult.Rows) == 0 {
		return appliedResult(false, &Row{Values: make([]interface{}, len(fetchCols))}, fetchCols), nil
	}
	return appliedResult(false, &fetchResult.Rows[0], fetchCols), nil
}

func resolveJSONInput(expr parser.Expression, params []interface{}, markerIndex map[parser.Expression]int) (string, error) {
	switch e := expr.(type) {
	case *parser.StringLiteral:
		return e.Value, nil
	case *parser.BindMarker, *parser.NamedBindMarker:
		if len(params) == 0 {
			return "", fmt.Errorf("bind markers require prepared parameters")
		}
		idx, ok := markerIndex[expr]
		if !ok || idx >= len(params) {
			return "", fmt.Errorf("JSON bind marker index out of range")
		}
		if isUnsetParam(params[idx]) {
			return "{}", nil
		}
		switch v := params[idx].(type) {
		case string:
			return v, nil
		case []byte:
			return string(v), nil
		default:
			return fmt.Sprint(v), nil
		}
	case *parser.TypeHintExpr:
		return resolveJSONInput(e.Expr, params, markerIndex)
	default:
		return "", fmt.Errorf("JSON value must be a string or bind marker")
	}
}

func buildJSONInsertSQL(table *parser.TableName, columns []mysqlbackend.ColumnInfo, ifNotExists bool) string {
	var sql strings.Builder
	if ifNotExists {
		sql.WriteString("INSERT IGNORE INTO ")
	} else {
		sql.WriteString("INSERT INTO ")
	}
	sql.WriteString(quoteMySQLIdentifier(table.Schema))
	sql.WriteString(".")
	sql.WriteString(quoteMySQLIdentifier(table.Name))

	sql.WriteString(" (")
	for i, col := range columns {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString(quoteMySQLIdentifier(col.Name))
	}
	sql.WriteString(") VALUES (")
	for i := range columns {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString("?")
	}
	sql.WriteString(")")

	if !ifNotExists {
		sql.WriteString(" ON DUPLICATE KEY UPDATE ")
		for i, col := range columns {
			if i > 0 {
				sql.WriteString(", ")
			}
			quoted := quoteMySQLIdentifier(col.Name)
			sql.WriteString(quoted)
			sql.WriteString(" = VALUES(")
			sql.WriteString(quoted)
			sql.WriteString(")")
		}
	}
	return sql.String()
}

func buildFetchByPKValues(table *parser.TableName, columns []mysqlbackend.ColumnInfo, pkCols []string, values []interface{}) (string, []interface{}) {
	indexByColumn := make(map[string]int, len(columns))
	for i, col := range columns {
		indexByColumn[strings.ToLower(col.Name)] = i
	}
	var sql strings.Builder
	sql.WriteString("SELECT ")
	for i, col := range columns {
		if i > 0 {
			sql.WriteString(", ")
		}
		sql.WriteString(quoteMySQLIdentifier(col.Name))
	}
	sql.WriteString(" FROM ")
	sql.WriteString(quoteMySQLIdentifier(table.Schema))
	sql.WriteString(".")
	sql.WriteString(quoteMySQLIdentifier(table.Name))

	args := make([]interface{}, 0, len(pkCols))
	sql.WriteString(" WHERE ")
	for i, pk := range pkCols {
		if i > 0 {
			sql.WriteString(" AND ")
		}
		sql.WriteString(quoteMySQLIdentifier(pk))
		sql.WriteString(" = ?")
		if idx, ok := indexByColumn[strings.ToLower(pk)]; ok && idx < len(values) {
			args = append(args, values[idx])
		} else {
			args = append(args, nil)
		}
	}
	return sql.String(), args
}

func convertJSONRawValue(raw json.RawMessage, mysqlType, cqlType string) (interface{}, error) {
	if isJSONNull(raw) {
		return nil, nil
	}
	if strings.EqualFold(mysqlType, "json") {
		return string(raw), nil
	}
	value, err := decodeJSONScalar(raw)
	if err != nil {
		return nil, err
	}

	switch strings.ToLower(cqlType) {
	case "text", "varchar", "ascii", "uuid", "timeuuid", "inet":
		return stringifyJSONValue(value), nil
	case "boolean":
		return parseJSONBool(value)
	case "int", "smallint", "tinyint":
		return parseJSONInt(value, 32)
	case "bigint", "counter":
		return parseJSONInt(value, 64)
	case "float", "double":
		return parseJSONFloat(value)
	case "decimal", "varint":
		return stringifyJSONValue(value), nil
	case "blob":
		return parseJSONBlob(value)
	case "timestamp":
		return parseJSONTimestamp(value)
	case "date":
		return parseJSONDate(value)
	case "time":
		return stringifyJSONValue(value), nil
	default:
		return stringifyJSONValue(value), nil
	}
}

func decodeJSONScalar(raw json.RawMessage) (interface{}, error) {
	dec := json.NewDecoder(bytes.NewReader(raw))
	dec.UseNumber()
	var value interface{}
	if err := dec.Decode(&value); err != nil {
		return nil, err
	}
	return value, nil
}

func stringifyJSONValue(value interface{}) string {
	switch v := value.(type) {
	case string:
		return v
	case json.Number:
		return v.String()
	case bool:
		if v {
			return "true"
		}
		return "false"
	default:
		return fmt.Sprint(value)
	}
}

func parseJSONBool(value interface{}) (bool, error) {
	switch v := value.(type) {
	case bool:
		return v, nil
	case string:
		return strconv.ParseBool(v)
	case json.Number:
		if v.String() == "1" {
			return true, nil
		}
		if v.String() == "0" {
			return false, nil
		}
	}
	return false, fmt.Errorf("invalid boolean JSON value")
}

func parseJSONInt(value interface{}, bits int) (int64, error) {
	switch v := value.(type) {
	case json.Number:
		return v.Int64()
	case string:
		return strconv.ParseInt(v, 10, bits)
	default:
		return 0, fmt.Errorf("invalid integer JSON value")
	}
}

func parseJSONFloat(value interface{}) (float64, error) {
	switch v := value.(type) {
	case json.Number:
		return v.Float64()
	case string:
		return strconv.ParseFloat(v, 64)
	default:
		return 0, fmt.Errorf("invalid float JSON value")
	}
}

func parseJSONBlob(value interface{}) ([]byte, error) {
	str, ok := value.(string)
	if !ok {
		return nil, fmt.Errorf("invalid blob JSON value")
	}
	if strings.HasPrefix(str, "0x") || strings.HasPrefix(str, "0X") {
		str = str[2:]
	}
	if len(str)%2 != 0 {
		return nil, fmt.Errorf("invalid blob JSON value")
	}
	return hex.DecodeString(str)
}

func parseJSONTimestamp(value interface{}) (time.Time, error) {
	switch v := value.(type) {
	case json.Number:
		ms, err := v.Int64()
		if err != nil {
			return time.Time{}, err
		}
		return time.Unix(0, ms*int64(time.Millisecond)).UTC(), nil
	case string:
		if ts, err := time.ParseInLocation("2006-01-02 15:04:05.000000", v, time.UTC); err == nil {
			return ts, nil
		}
		if ts, err := time.ParseInLocation("2006-01-02 15:04:05.000", v, time.UTC); err == nil {
			return ts, nil
		}
		if ts, err := time.ParseInLocation("2006-01-02 15:04:05", v, time.UTC); err == nil {
			return ts, nil
		}
		return time.Time{}, fmt.Errorf("invalid timestamp JSON value")
	default:
		return time.Time{}, fmt.Errorf("invalid timestamp JSON value")
	}
}

func parseJSONDate(value interface{}) (time.Time, error) {
	str, ok := value.(string)
	if !ok {
		return time.Time{}, fmt.Errorf("invalid date JSON value")
	}
	return time.ParseInLocation("2006-01-02", str, time.UTC)
}

func isJSONNull(raw json.RawMessage) bool {
	return len(bytes.TrimSpace(raw)) == 4 && strings.EqualFold(string(bytes.TrimSpace(raw)), "null")
}

func columnNames(columns []mysqlbackend.ColumnInfo) []string {
	names := make([]string, len(columns))
	for i, col := range columns {
		names[i] = col.Name
	}
	return names
}