package executor

import (
	"encoding/hex"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"
	"unicode/utf8"

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

type jsonValueKind uint8

const (
	jsonKindUnknown jsonValueKind = iota
	jsonKindString
	jsonKindNumber
	jsonKindBoolean
	jsonKindBlob
	jsonKindTimestamp
	jsonKindDate
	jsonKindTime
	jsonKindRaw
)

type jsonTypeInfo struct {
	kind jsonValueKind
}

type jsonColumnSpec struct {
	key []byte
	typ jsonTypeInfo
}

func buildJSONColumnSpecs(cols []ColumnInfo) []jsonColumnSpec {
	specs := make([]jsonColumnSpec, len(cols))
	for i, col := range cols {
		specs[i] = jsonColumnSpec{
			key: buildJSONKey(col.Name),
			typ: jsonTypeInfoFromColumn(col),
		}
	}
	return specs
}

func buildJSONKey(name string) []byte {
	buf := make([]byte, 0, len(name)+4)
	buf = appendJSONString(buf, name)
	buf = append(buf, ':')
	return buf
}

func jsonTypeInfoFromColumn(col ColumnInfo) jsonTypeInfo {
	mysqlType := strings.ToLower(strings.TrimSpace(col.Type))
	if mysqlType == "json" {
		return jsonTypeInfo{kind: jsonKindRaw}
	}
	cqlType := normalizeCQLType(col.Type)
	if !isKnownCQLType(cqlType) && !strings.Contains(cqlType, "<") {
		cqlType = typemap.MySQLTypeToCQLTypeString(cqlType)
	}
	return jsonTypeInfoFromCQL(cqlType)
}

func normalizeCQLType(typ string) string {
	trimmed := strings.TrimSpace(typ)
	if trimmed == "" {
		return ""
	}
	if strings.Contains(trimmed, "'") {
		return trimmed
	}
	lower := strings.ToLower(trimmed)
	if strings.Contains(lower, "<") {
		return lower
	}
	base := lower
	if idx := strings.IndexAny(base, " ("); idx != -1 {
		base = base[:idx]
	}
	return base
}

func jsonTypeInfoFromCQL(cqlType string) jsonTypeInfo {
	if cqlType == "" {
		return jsonTypeInfo{kind: jsonKindUnknown}
	}
	if strings.HasPrefix(cqlType, "'") && strings.HasSuffix(cqlType, "'") {
		return jsonTypeInfo{kind: jsonKindString}
	}
	switch {
	case strings.HasPrefix(cqlType, "list<"),
		strings.HasPrefix(cqlType, "set<"),
		strings.HasPrefix(cqlType, "map<"),
		strings.HasPrefix(cqlType, "tuple<"),
		strings.HasPrefix(cqlType, "frozen<"):
		return jsonTypeInfo{kind: jsonKindRaw}
	}
	switch cqlType {
	case "text", "varchar", "ascii", "uuid", "timeuuid", "inet":
		return jsonTypeInfo{kind: jsonKindString}
	case "boolean":
		return jsonTypeInfo{kind: jsonKindBoolean}
	case "int", "bigint", "smallint", "tinyint", "float", "double", "decimal", "varint", "counter":
		return jsonTypeInfo{kind: jsonKindNumber}
	case "blob":
		return jsonTypeInfo{kind: jsonKindBlob}
	case "timestamp":
		return jsonTypeInfo{kind: jsonKindTimestamp}
	case "date":
		return jsonTypeInfo{kind: jsonKindDate}
	case "time":
		return jsonTypeInfo{kind: jsonKindTime}
	default:
		return jsonTypeInfo{kind: jsonKindString}
	}
}

func isKnownCQLType(cqlType string) bool {
	switch cqlType {
	case "text", "varchar", "ascii",
		"uuid", "timeuuid", "inet",
		"boolean",
		"int", "bigint", "smallint", "tinyint",
		"float", "double", "decimal", "varint", "counter",
		"blob",
		"timestamp", "date", "time", "duration":
		return true
	}
	return strings.HasPrefix(cqlType, "'") && strings.HasSuffix(cqlType, "'")
}

func buildSelectJSONResult(result *QueryResult) (*QueryResult, error) {
	if result == nil {
		return &QueryResult{}, nil
	}

	specs := buildJSONColumnSpecs(result.Columns)
	baseCap := estimateJSONRowSize(specs)
	rows := make([]Row, len(result.Rows))
	buf := make([]byte, 0, baseCap)

	for i, row := range result.Rows {
		buf = buf[:0]
		buf = append(buf, '{')
		for j, spec := range specs {
			if j > 0 {
				buf = append(buf, ',')
			}
			buf = append(buf, spec.key...)
			buf = appendJSONValue(buf, row.Values[j], spec.typ)
		}
		buf = append(buf, '}')
		rows[i] = Row{Values: []interface{}{string(buf)}}
	}

	return &QueryResult{
		Columns: []ColumnInfo{
			{
				Name:     "json",
				Type:     "text",
				Nullable: true,
			},
		},
		Rows: rows,
	}, nil
}

func estimateJSONRowSize(specs []jsonColumnSpec) int {
	total := 2
	for _, spec := range specs {
		total += len(spec.key) + 4
	}
	return total
}

func formatJSONValue(value interface{}, typ jsonTypeInfo) string {
	buf := make([]byte, 0, 32)
	buf = appendJSONValue(buf, value, typ)
	return string(buf)
}

func appendJSONValue(buf []byte, value interface{}, typ jsonTypeInfo) []byte {
	if value == nil {
		return append(buf, "null"...)
	}

	switch typ.kind {
	case jsonKindRaw:
		return appendRawJSON(buf, value)
	case jsonKindString:
		return appendJSONString(buf, stringifyValue(value))
	case jsonKindBoolean:
		return appendJSONBool(buf, value)
	case jsonKindNumber:
		return appendJSONNumber(buf, value)
	case jsonKindBlob:
		return appendJSONBlob(buf, value)
	case jsonKindTimestamp:
		return appendJSONString(buf, formatTimestampValue(value))
	case jsonKindDate:
		return appendJSONString(buf, formatDateValue(value))
	case jsonKindTime:
		return appendJSONString(buf, formatTimeValue(value))
	default:
		return appendJSONString(buf, stringifyValue(value))
	}
}

func appendRawJSON(buf []byte, value interface{}) []byte {
	switch v := value.(type) {
	case string:
		return append(buf, v...)
	case []byte:
		return append(buf, v...)
	case json.RawMessage:
		return append(buf, v...)
	default:
		raw, err := json.Marshal(v)
		if err != nil {
			return append(buf, "null"...)
		}
		return append(buf, raw...)
	}
}

func appendJSONBool(buf []byte, value interface{}) []byte {
	switch v := value.(type) {
	case bool:
		if v {
			return append(buf, "true"...)
		}
		return append(buf, "false"...)
	case string:
		if strings.EqualFold(v, "true") {
			return append(buf, "true"...)
		}
		if strings.EqualFold(v, "false") {
			return append(buf, "false"...)
		}
	}
	return append(buf, "false"...)
}

func appendJSONNumber(buf []byte, value interface{}) []byte {
	switch v := value.(type) {
	case int:
		return strconv.AppendInt(buf, int64(v), 10)
	case int8:
		return strconv.AppendInt(buf, int64(v), 10)
	case int16:
		return strconv.AppendInt(buf, int64(v), 10)
	case int32:
		return strconv.AppendInt(buf, int64(v), 10)
	case int64:
		return strconv.AppendInt(buf, v, 10)
	case uint:
		return strconv.AppendUint(buf, uint64(v), 10)
	case uint8:
		return strconv.AppendUint(buf, uint64(v), 10)
	case uint16:
		return strconv.AppendUint(buf, uint64(v), 10)
	case uint32:
		return strconv.AppendUint(buf, uint64(v), 10)
	case uint64:
		return strconv.AppendUint(buf, v, 10)
	case float32:
		return strconv.AppendFloat(buf, float64(v), 'g', -1, 32)
	case float64:
		return strconv.AppendFloat(buf, v, 'g', -1, 64)
	case string:
		return append(buf, v...)
	default:
		return append(buf, "0"...)
	}
}

func appendJSONBlob(buf []byte, value interface{}) []byte {
	switch v := value.(type) {
	case []byte:
		encoded := make([]byte, 2+hex.EncodedLen(len(v)))
		copy(encoded, "0x")
		hex.Encode(encoded[2:], v)
		return appendJSONString(buf, string(encoded))
	case string:
		return appendJSONString(buf, v)
	default:
		return appendJSONString(buf, stringifyValue(value))
	}
}

func formatTimestampValue(value interface{}) string {
	switch v := value.(type) {
	case time.Time:
		utc := v.UTC()
		if utc.Nanosecond()%int(time.Millisecond) == 0 {
			return utc.Format("2006-01-02 15:04:05.000")
		}
		return utc.Format("2006-01-02 15:04:05.000000")
	case string:
		return v
	default:
		return stringifyValue(value)
	}
}

func formatDateValue(value interface{}) string {
	switch v := value.(type) {
	case time.Time:
		return v.UTC().Format("2006-01-02")
	case string:
		return v
	default:
		return stringifyValue(value)
	}
}

func formatTimeValue(value interface{}) string {
	switch v := value.(type) {
	case time.Time:
		return trimTimeFraction(v.UTC().Format("15:04:05.000000000"))
	case time.Duration:
		return formatDurationAsTime(v)
	case string:
		return v
	default:
		return stringifyValue(value)
	}
}

func formatDurationAsTime(d time.Duration) string {
	if d < 0 {
		d = -d
	}
	h := int(d / time.Hour)
	d -= time.Duration(h) * time.Hour
	m := int(d / time.Minute)
	d -= time.Duration(m) * time.Minute
	s := int(d / time.Second)
	d -= time.Duration(s) * time.Second
	ns := int(d)
	hours := strconv.Itoa(h)
	if h < 10 {
		hours = "0" + hours
	}
	if ns == 0 {
		return hours + ":" + twoDigits(m) + ":" + twoDigits(s)
	}
	frac := strconv.FormatInt(int64(ns), 10)
	frac = strings.Repeat("0", 9-len(frac)) + frac
	return hours + ":" + twoDigits(m) + ":" + twoDigits(s) + "." + strings.TrimRight(frac, "0")
}

func twoDigits(value int) string {
	if value < 10 {
		return "0" + strconv.Itoa(value)
	}
	return strconv.Itoa(value)
}

func trimTimeFraction(value string) string {
	if !strings.Contains(value, ".") {
		return value
	}
	value = strings.TrimRight(value, "0")
	value = strings.TrimRight(value, ".")
	return value
}

func stringifyValue(value interface{}) string {
	switch v := value.(type) {
	case string:
		return v
	case []byte:
		return string(v)
	default:
		return fmt.Sprint(value)
	}
}

func appendJSONString(buf []byte, s string) []byte {
	buf = append(buf, '"')
	start := 0
	for i := 0; i < len(s); {
		b := s[i]
		if b < utf8.RuneSelf {
			switch b {
			case '\\', '"':
				buf = append(buf, s[start:i]...)
				buf = append(buf, '\\', b)
				start = i + 1
			default:
				if b < 0x20 {
					buf = append(buf, s[start:i]...)
					buf = append(buf, '\\', 'u', '0', '0', hexDigits[b>>4], hexDigits[b&0x0f])
					start = i + 1
				}
			}
			i++
			continue
		}
		_, size := utf8.DecodeRuneInString(s[i:])
		if size == 1 {
			buf = append(buf, s[start:i]...)
			buf = append(buf, '\\', 'u', 'f', 'f', 'f', 'd')
			start = i + 1
			i++
			continue
		}
		i += size
	}
	buf = append(buf, s[start:]...)
	buf = append(buf, '"')
	return buf
}

var hexDigits = []byte("0123456789abcdef")