package executor

import (
	"testing"
	"time"

	gocql "github.com/apache/cassandra-gocql-driver/v2"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/parser"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/translator"
)

func TestCastSQLSafeMatrix(t *testing.T) {
	cases := []struct {
		from string
		to   string
		safe bool
	}{
		{"tinyint", "int", true},
		{"int", "bigint", true},
		{"bigint", "int", false},
		{"float", "double", true},
		{"double", "float", false},
		{"text", "text", true},
		{"int", "text", false},
	}
	for _, tc := range cases {
		if got := castSQLSafe(tc.from, tc.to); got != tc.safe {
			t.Fatalf("castSQLSafe(%q,%q) = %v, want %v", tc.from, tc.to, got, tc.safe)
		}
	}
}

func TestCastValueNumericToText(t *testing.T) {
	value, err := castValue(int32(12), "int", "text")
	if err != nil {
		t.Fatalf("castValue failed: %v", err)
	}
	if value != "12" {
		t.Fatalf("expected cast to text \"12\", got %#v", value)
	}
}

func TestCastValueTextToInt(t *testing.T) {
	value, err := castValue("42", "text", "int")
	if err != nil {
		t.Fatalf("castValue failed: %v", err)
	}
	if v, ok := value.(int32); !ok || v != 42 {
		t.Fatalf("expected int32(42), got %#v", value)
	}
}

func TestTranslateSelectExprCast(t *testing.T) {
	exec := &Executor{translator: translator.NewTranslator()}
	typeInfo := map[string]columnTypeInfo{
		"a": {cqlType: "int"},
	}

	unsafeCast := &parser.CastExpr{
		Expr: &parser.ColumnExpr{Name: "a"},
		Type: &parser.SimpleType{Name: "text"},
	}
	sql, _, castSpec, err := exec.translateSelectExpr(unsafeCast, typeInfo)
	if err != nil {
		t.Fatalf("translateSelectExpr failed: %v", err)
	}
	if sql != "`a`" {
		t.Fatalf("expected SQL \"`a`\" for unsafe cast, got %q", sql)
	}
	if castSpec == nil || castSpec.sqlHandled {
		t.Fatalf("expected proxy cast spec for unsafe cast")
	}

	safeCast := &parser.CastExpr{
		Expr: &parser.ColumnExpr{Name: "a"},
		Type: &parser.SimpleType{Name: "bigint"},
	}
	sql, _, castSpec, err = exec.translateSelectExpr(safeCast, typeInfo)
	if err != nil {
		t.Fatalf("translateSelectExpr failed: %v", err)
	}
	if sql != "CAST(`a` AS BIGINT)" {
		t.Fatalf("expected SQL cast, got %q", sql)
	}
	if castSpec == nil || !castSpec.sqlHandled {
		t.Fatalf("expected SQL-handled cast spec")
	}
}

func TestTranslateSelectExprToTimestamp(t *testing.T) {
	exec := &Executor{translator: translator.NewTranslator()}
	typeInfo := map[string]columnTypeInfo{
		"ts": {cqlType: "timeuuid"},
	}

	expr := &parser.FunctionCall{
		Name: "toTimestamp",
		Args: []parser.Expression{&parser.ColumnExpr{Name: "ts"}},
	}
	sql, _, castSpec, err := exec.translateSelectExpr(expr, typeInfo)
	if err != nil {
		t.Fatalf("translateSelectExpr failed: %v", err)
	}
	if sql != "`ts`" {
		t.Fatalf("expected SQL \"`ts`\" for toTimestamp, got %q", sql)
	}
	if castSpec == nil {
		t.Fatalf("expected cast spec for toTimestamp")
	}
	if castSpec.fromType != "timeuuid" || castSpec.toType != "timestamp" {
		t.Fatalf("unexpected cast spec types: %#v", castSpec)
	}
	if castSpec.name != "toTimestamp(ts)" {
		t.Fatalf("unexpected cast spec name: %q", castSpec.name)
	}
}

func TestTranslateSelectExprToTimestampFromTimestamp(t *testing.T) {
	exec := &Executor{translator: translator.NewTranslator()}
	typeInfo := map[string]columnTypeInfo{
		"ts": {cqlType: "timestamp"},
	}

	expr := &parser.FunctionCall{
		Name: "toTimestamp",
		Args: []parser.Expression{&parser.ColumnExpr{Name: "ts"}},
	}
	sql, _, castSpec, err := exec.translateSelectExpr(expr, typeInfo)
	if err != nil {
		t.Fatalf("translateSelectExpr failed: %v", err)
	}
	if sql != "`ts`" {
		t.Fatalf("expected SQL \"`ts`\" for toTimestamp, got %q", sql)
	}
	if castSpec == nil {
		t.Fatalf("expected cast spec for toTimestamp")
	}
	if castSpec.fromType != "timestamp" || castSpec.toType != "timestamp" {
		t.Fatalf("unexpected cast spec types: %#v", castSpec)
	}
	if castSpec.name != "toTimestamp(ts)" {
		t.Fatalf("unexpected cast spec name: %q", castSpec.name)
	}
}

func TestApplySelectTransformToTimestamp(t *testing.T) {
	uuid := gocql.TimeUUID()
	expected := uuid.Time().UTC()
	result := &QueryResult{
		Columns: []ColumnInfo{{Name: "ts", Type: "timeuuid", Nullable: true}},
		Rows:    []Row{{Values: []interface{}{uuid.String()}}},
	}
	spec := castSpec{
		index:      0,
		name:       "toTimestamp(ts)",
		fromType:   "timeuuid",
		toType:     "timestamp",
		sqlHandled: false,
	}
	updated, err := applySelectTransform(result, selectTransform{casts: []castSpec{spec}})
	if err != nil {
		t.Fatalf("applySelectTransform failed: %v", err)
	}
	if updated.Columns[0].Type != "timestamp" {
		t.Fatalf("expected timestamp column type, got %q", updated.Columns[0].Type)
	}
	ts, ok := updated.Rows[0].Values[0].(time.Time)
	if !ok {
		t.Fatalf("expected time.Time value, got %T", updated.Rows[0].Values[0])
	}
	if !ts.UTC().Equal(expected) {
		t.Fatalf("expected %v, got %v", expected, ts.UTC())
	}
}

func TestCastValueFloatOverflowToInt(t *testing.T) {
	const maxInt32 = int32(^uint32(0) >> 1)
	const minInt32 = -maxInt32 - 1

	value, err := castValue(1e20, "double", "int")
	if err != nil {
		t.Fatalf("castValue failed: %v", err)
	}
	if got, ok := value.(int32); !ok || got != maxInt32 {
		t.Fatalf("expected int32 max %d, got %#v", maxInt32, value)
	}

	value, err = castValue(-1e20, "double", "int")
	if err != nil {
		t.Fatalf("castValue failed: %v", err)
	}
	if got, ok := value.(int32); !ok || got != minInt32 {
		t.Fatalf("expected int32 min %d, got %#v", minInt32, value)
	}
}

func TestCastValueTimeuuidValidation(t *testing.T) {
	_, err := castValue("550e8400-e29b-41d4-a716-446655440000", "uuid", "timeuuid")
	if err == nil {
		t.Fatalf("expected timeuuid validation error")
	}
}

func TestTranslateSelectExprTimeFunctions(t *testing.T) {
	exec := &Executor{translator: translator.NewTranslator()}
	typeInfo := map[string]columnTypeInfo{
		"ts":  {cqlType: "timestamp"},
		"day": {cqlType: "date"},
		"uid": {cqlType: "timeuuid"},
	}
	cases := []struct {
		name     string
		expr     parser.Expression
		wantSQL  string
		wantFrom string
		wantTo   string
		wantName string
	}{
		{
			name: "toDate",
			expr: &parser.FunctionCall{
				Name: "toDate",
				Args: []parser.Expression{&parser.ColumnExpr{Name: "ts"}},
			},
			wantSQL:  "`ts`",
			wantFrom: "timestamp",
			wantTo:   "date",
			wantName: "toDate(ts)",
		},
		{
			name: "toUnixTimestamp",
			expr: &parser.FunctionCall{
				Name: "toUnixTimestamp",
				Args: []parser.Expression{&parser.ColumnExpr{Name: "day"}},
			},
			wantSQL:  "`day`",
			wantFrom: "date",
			wantTo:   "bigint",
			wantName: "toUnixTimestamp(day)",
		},
		{
			name: "dateOf",
			expr: &parser.FunctionCall{
				Name: "dateOf",
				Args: []parser.Expression{&parser.ColumnExpr{Name: "uid"}},
			},
			wantSQL:  "`uid`",
			wantFrom: "timeuuid",
			wantTo:   "timestamp",
			wantName: "dateOf(uid)",
		},
		{
			name: "unixTimestampOf",
			expr: &parser.FunctionCall{
				Name: "unixTimestampOf",
				Args: []parser.Expression{&parser.ColumnExpr{Name: "uid"}},
			},
			wantSQL:  "`uid`",
			wantFrom: "timeuuid",
			wantTo:   "bigint",
			wantName: "unixTimestampOf(uid)",
		},
	}

	for _, tc := range cases {
		sql, _, castSpec, err := exec.translateSelectExpr(tc.expr, typeInfo)
		if err != nil {
			t.Fatalf("%s translateSelectExpr failed: %v", tc.name, err)
		}
		if sql != tc.wantSQL {
			t.Fatalf("%s expected SQL %q, got %q", tc.name, tc.wantSQL, sql)
		}
		if castSpec == nil {
			t.Fatalf("%s expected cast spec", tc.name)
		}
		if castSpec.fromType != tc.wantFrom || castSpec.toType != tc.wantTo {
			t.Fatalf("%s unexpected cast spec types: %#v", tc.name, castSpec)
		}
		if castSpec.name != tc.wantName {
			t.Fatalf("%s unexpected cast spec name: %q", tc.name, castSpec.name)
		}
	}
}

func TestApplySelectTransformTimeFunctions(t *testing.T) {
	ts := time.Date(2024, 6, 1, 12, 34, 56, 0, time.UTC)
	specDate, ok := timeFunctionSpecForName("toDate")
	if !ok {
		t.Fatalf("expected toDate spec")
	}
	result := &QueryResult{
		Columns: []ColumnInfo{{Name: "ts", Type: "timestamp", Nullable: true}},
		Rows:    []Row{{Values: []interface{}{ts}}},
	}
	spec := castSpec{
		index:      0,
		name:       "toDate(ts)",
		fromType:   "timestamp",
		toType:     "date",
		sqlHandled: false,
		timeFunc:   &specDate,
	}
	updated, err := applySelectTransform(result, selectTransform{casts: []castSpec{spec}})
	if err != nil {
		t.Fatalf("applySelectTransform failed: %v", err)
	}
	if updated.Columns[0].Type != "date" {
		t.Fatalf("expected date column type, got %q", updated.Columns[0].Type)
	}
	gotDate, ok := updated.Rows[0].Values[0].(time.Time)
	if !ok {
		t.Fatalf("expected time.Time value, got %T", updated.Rows[0].Values[0])
	}
	if !gotDate.Equal(dateFromTime(ts)) {
		t.Fatalf("expected %v, got %v", dateFromTime(ts), gotDate)
	}

	specTS, ok := timeFunctionSpecForName("toTimestamp")
	if !ok {
		t.Fatalf("expected toTimestamp spec")
	}
	result = &QueryResult{
		Columns: []ColumnInfo{{Name: "ts", Type: "timestamp", Nullable: true}},
		Rows:    []Row{{Values: []interface{}{ts}}},
	}
	spec = castSpec{
		index:      0,
		name:       "toTimestamp(ts)",
		fromType:   "timestamp",
		toType:     "timestamp",
		sqlHandled: false,
		timeFunc:   &specTS,
	}
	updated, err = applySelectTransform(result, selectTransform{casts: []castSpec{spec}})
	if err != nil {
		t.Fatalf("applySelectTransform failed: %v", err)
	}
	if updated.Columns[0].Type != "timestamp" {
		t.Fatalf("expected timestamp column type, got %q", updated.Columns[0].Type)
	}
	gotTS, ok := updated.Rows[0].Values[0].(time.Time)
	if !ok {
		t.Fatalf("expected time.Time value, got %T", updated.Rows[0].Values[0])
	}
	if !gotTS.Equal(ts) {
		t.Fatalf("expected %v, got %v", ts, gotTS)
	}

	uuid := gocql.TimeUUID()
	specUnix, ok := timeFunctionSpecForName("unixTimestampOf")
	if !ok {
		t.Fatalf("expected unixTimestampOf spec")
	}
	result = &QueryResult{
		Columns: []ColumnInfo{{Name: "uid", Type: "timeuuid", Nullable: true}},
		Rows:    []Row{{Values: []interface{}{uuid.String()}}},
	}
	spec = castSpec{
		index:      0,
		name:       "unixTimestampOf(uid)",
		fromType:   "timeuuid",
		toType:     "bigint",
		sqlHandled: false,
		timeFunc:   &specUnix,
	}
	updated, err = applySelectTransform(result, selectTransform{casts: []castSpec{spec}})
	if err != nil {
		t.Fatalf("applySelectTransform failed: %v", err)
	}
	if updated.Columns[0].Type != "bigint" {
		t.Fatalf("expected bigint column type, got %q", updated.Columns[0].Type)
	}
	gotUnix, ok := updated.Rows[0].Values[0].(int64)
	if !ok {
		t.Fatalf("expected int64 value, got %T", updated.Rows[0].Values[0])
	}
	expectedUnix := unixMillis(uuid.Time())
	if gotUnix != expectedUnix {
		t.Fatalf("expected %d, got %d", expectedUnix, gotUnix)
	}
}