package executor

import (
	"context"
	"reflect"
	"strings"
	"testing"

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

func TestIsToJSONExprAliases(t *testing.T) {
	cases := []struct {
		name string
		expr parser.Expression
	}{
		{
			name: "to_json",
			expr: &parser.FunctionCall{
				Name: "to_json",
				Args: []parser.Expression{&parser.ColumnExpr{Name: "value"}},
			},
		},
		{
			name: "tojson",
			expr: &parser.FunctionCall{
				Name: "tojson",
				Args: []parser.Expression{&parser.ColumnExpr{Name: "value"}},
			},
		},
		{
			name: "aliased_tojson",
			expr: &parser.AliasedExpr{
				Alias: "val",
				Expr: &parser.FunctionCall{
					Name: "ToJson",
					Args: []parser.Expression{&parser.ColumnExpr{Name: "value"}},
				},
			},
		},
	}

	for _, tc := range cases {
		if !isToJSONExpr(tc.expr) {
			t.Fatalf("expected %s to be recognized as to_json", tc.name)
		}
	}
}

func TestTranslateSelectExprToJSONAlias(t *testing.T) {
	exec := &Executor{translator: translator.NewTranslator()}
	expr := &parser.FunctionCall{
		Name: "tojson",
		Args: []parser.Expression{&parser.ColumnExpr{Name: "value"}},
	}

	sql, spec, _, err := exec.translateSelectExpr(expr, map[string]columnTypeInfo{})
	if err != nil {
		t.Fatalf("translateSelectExpr failed: %v", err)
	}
	if spec == nil {
		t.Fatal("expected to_json spec for tojson")
	}
	expected := "`value` AS `system.tojson(value)`"
	if sql != expected {
		t.Fatalf("expected sql %q, got %q", expected, sql)
	}
	if !strings.Contains(spec.name, "system.tojson") {
		t.Fatalf("expected spec name to include system.tojson, got %q", spec.name)
	}
}

func TestSelectPlanLimitLiteralMin(t *testing.T) {
	exec := &Executor{translator: translator.NewTranslator()}
	stmt, err := parser.NewParser("SELECT a FROM ks.tbl LIMIT 7 PER PARTITION LIMIT 5").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	selectStmt, ok := stmt.(*parser.SelectStatement)
	if !ok {
		t.Fatalf("expected select statement, got %T", stmt)
	}
	plan, err := exec.buildSelectPlan(context.Background(), "", "", selectStmt, false, false)
	if err != nil {
		t.Fatalf("buildSelectPlan failed: %v", err)
	}
	if plan.limitPlan != nil {
		t.Fatalf("expected no limit plan for literal limits")
	}
	if !strings.Contains(plan.sql, "LIMIT 5") {
		t.Fatalf("expected computed limit in SQL: %s", plan.sql)
	}
}

func TestSelectPlanLimitBindOmit(t *testing.T) {
	exec := &Executor{translator: translator.NewTranslator()}
	stmt, err := parser.NewParser("SELECT a FROM ks.tbl WHERE b = ? LIMIT ? PER PARTITION LIMIT ?").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	selectStmt, ok := stmt.(*parser.SelectStatement)
	if !ok {
		t.Fatalf("expected select statement, got %T", stmt)
	}
	plan, err := exec.buildSelectPlan(context.Background(), "", "", selectStmt, true, false)
	if err != nil {
		t.Fatalf("buildSelectPlan failed: %v", err)
	}
	if plan.limitPlan == nil {
		t.Fatalf("expected limit plan for bind markers")
	}
	if strings.Contains(plan.sql, "LIMIT") {
		t.Fatalf("expected LIMIT to be omitted in SQL: %s", plan.sql)
	}
	expectedDrops := []int{1, 2}
	if !reflect.DeepEqual(plan.limitPlan.dropIndexes, expectedDrops) {
		t.Fatalf("unexpected drop indexes: %#v", plan.limitPlan.dropIndexes)
	}
	limit, err := plan.limitPlan.effectiveLimit([]interface{}{int32(9), int32(7), int32(5)})
	if err != nil {
		t.Fatalf("effectiveLimit failed: %v", err)
	}
	if limit != 5 {
		t.Fatalf("expected effective limit 5, got %d", limit)
	}
	args, err := filterLimitParams([]interface{}{int32(9), int32(7), int32(5)}, plan.limitPlan.dropIndexes)
	if err != nil {
		t.Fatalf("filterLimitParams failed: %v", err)
	}
	if len(args) != 1 || args[0].(int32) != 9 {
		t.Fatalf("unexpected args: %#v", args)
	}
}

func TestSelectMetadataQueryBindsPlaceholdersAndForcesZeroRows(t *testing.T) {
	sql := "SELECT * FROM `ks`.`tbl` WHERE `user_id` = ? ORDER BY `ts` DESC LIMIT ?"
	metaSQL, args := selectMetadataQuery(sql)

	expectedSQL := "SELECT * FROM (" + sql + ") AS cql_metadata LIMIT 0"
	if metaSQL != expectedSQL {
		t.Fatalf("unexpected metadata SQL:\n got: %s\nwant: %s", metaSQL, expectedSQL)
	}
	if !reflect.DeepEqual(args, []interface{}{0, 0}) {
		t.Fatalf("unexpected metadata args: %#v", args)
	}
}

func TestSelectMetadataQueryCountsFinalSQLPlaceholdersOnly(t *testing.T) {
	cases := []struct {
		name string
		sql  string
		args []interface{}
	}{
		{
			name: "single quoted literal",
			sql:  "SELECT * FROM `ks`.`tbl` WHERE `user_id` = ? AND `body` = 'literal ?' ORDER BY `ts` DESC",
			args: []interface{}{0},
		},
		{
			name: "backtick identifier",
			sql:  "SELECT `what?` FROM `ks`.`tbl` WHERE `user_id` = ?",
			args: []interface{}{0},
		},
		{
			name: "doubled single quote",
			sql:  "SELECT * FROM `ks`.`tbl` WHERE `user_id` = ? AND `body` = 'it''s ?'",
			args: []interface{}{0},
		},
		{
			name: "backslash escaped single quote",
			sql:  "SELECT * FROM `ks`.`tbl` WHERE `user_id` = ? AND `body` = 'it\\'s ?'",
			args: []interface{}{0},
		},
		{
			name: "doubled backtick identifier",
			sql:  "SELECT `odd``?` FROM `ks`.`tbl` WHERE `user_id` = ?",
			args: []interface{}{0},
		},
	}

	for _, tc := range cases {
		metaSQL, args := selectMetadataQuery(tc.sql)

		expectedSQL := "SELECT * FROM (" + tc.sql + ") AS cql_metadata LIMIT 0"
		if metaSQL != expectedSQL {
			t.Fatalf("%s: unexpected metadata SQL:\n got: %s\nwant: %s", tc.name, metaSQL, expectedSQL)
		}
		if !reflect.DeepEqual(args, tc.args) {
			t.Fatalf("%s: unexpected metadata args: %#v", tc.name, args)
		}
	}
}

func TestSelectPlanLimitProxyFilterSuppressed(t *testing.T) {
	exec := &Executor{translator: translator.NewTranslator()}
	stmt, err := parser.NewParser("SELECT a FROM ks.tbl LIMIT 5").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	selectStmt, ok := stmt.(*parser.SelectStatement)
	if !ok {
		t.Fatalf("expected select statement, got %T", stmt)
	}
	plan, err := exec.buildSelectPlan(context.Background(), "", "", selectStmt, false, true)
	if err != nil {
		t.Fatalf("buildSelectPlan failed: %v", err)
	}
	if plan.limitPlan == nil {
		t.Fatalf("expected limit plan when proxy filtering is active")
	}
	if !plan.limitPlan.applyInMemory {
		t.Fatalf("expected limit plan to apply in memory")
	}
	if strings.Contains(plan.sql, "LIMIT") {
		t.Fatalf("expected LIMIT to be omitted in SQL: %s", plan.sql)
	}
	limit, err := plan.limitPlan.effectiveLimit(nil)
	if err != nil {
		t.Fatalf("effectiveLimit failed: %v", err)
	}
	if limit != 5 {
		t.Fatalf("expected limit 5, got %d", limit)
	}
}