package executor

import (
	"context"
	"encoding/json"
	"strconv"
	"testing"
	"time"

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

func TestBuildInsertPlanUpsert(t *testing.T) {
	stmt, err := parser.NewParser("INSERT INTO ks.tbl (a, b) VALUES (?, ?)").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	insertStmt, ok := stmt.(*parser.InsertStatement)
	if !ok {
		t.Fatalf("expected insert statement, got %T", stmt)
	}
	exec := &Executor{translator: translator.NewTranslator()}
	plan, err := exec.buildInsertPlan(context.Background(), "", "", insertStmt)
	if err != nil {
		t.Fatalf("buildInsertPlan failed: %v", err)
	}
	expectedSQL := "INSERT INTO `ks`.`tbl` (`a`, `b`) VALUES (?, ?) ON DUPLICATE KEY UPDATE `a` = VALUES(`a`), `b` = VALUES(`b`)"
	if plan.sql != expectedSQL {
		t.Fatalf("unexpected SQL: %s", plan.sql)
	}
	if len(plan.paramOrder) != 2 || plan.paramOrder[0] != 0 || plan.paramOrder[1] != 1 {
		t.Fatalf("unexpected param order: %#v", plan.paramOrder)
	}
}

func TestCQLWriteRowsForInsertUsesLogicalRows(t *testing.T) {
	stmt, err := parser.NewParser("INSERT INTO ks.tbl (a, b) VALUES (1, 2), (3, 4)").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	insertStmt, ok := stmt.(*parser.InsertStatement)
	if !ok {
		t.Fatalf("expected insert statement, got %T", stmt)
	}
	exec := &Executor{translator: translator.NewTranslator()}
	plan, err := exec.buildInsertPlan(context.Background(), "", "", insertStmt)
	if err != nil {
		t.Fatalf("buildInsertPlan failed: %v", err)
	}

	if got := cqlWriteRowsForDMLStatement(plan, 4); got != 2 {
		t.Fatalf("expected logical insert rows to ignore MySQL affected rows, got %d", got)
	}
	if got := cqlWriteRowsForDMLStatement(plan, 0); got != 2 {
		t.Fatalf("expected successful idempotent insert to count logical rows, got %d", got)
	}
}

func TestCQLWriteRowsForUpdateUsesRequestedRows(t *testing.T) {
	stmt := mustParseStatement(t, "UPDATE ks.tbl SET v = 'x' WHERE id = 1")
	updateStmt, ok := stmt.(*parser.UpdateStatement)
	if !ok {
		t.Fatalf("expected update statement, got %T", stmt)
	}
	exec := &Executor{translator: translator.NewTranslator()}
	dml := &dmlExecutor{Executor: exec}
	plan, err := dml.buildPlainUpdatePlan(context.Background(), "", "", updateStmt, []equalityCondition{
		{column: "id", value: &parser.IntegerLiteral{Value: 1}, op: "="},
	}, buildBindMarkerIndex(stmt), []string{"id"}, nil)
	if err != nil {
		t.Fatalf("buildPlainUpdatePlan failed: %v", err)
	}
	if got := cqlWriteRowsForDMLStatement(plan, 0); got != 1 {
		t.Fatalf("expected update rows to use requested rows, got %d", got)
	}
}

func TestCQLWriteRowsFallsBackToAffectedRows(t *testing.T) {
	plan := &dmlPlan{}
	if got := cqlWriteRowsForDMLStatement(plan, 3); got != 3 {
		t.Fatalf("expected fallback rows to use affected rows, got %d", got)
	}
	if got := cqlWriteRowsForDMLStatement(plan, 0); got != 0 {
		t.Fatalf("expected zero affected fallback rows to be ignored, got %d", got)
	}
}

func TestBuildInsertPlanUpsertIncrementConfiguredColumns(t *testing.T) {
	stmt, err := parser.NewParser("INSERT INTO ks.tbl (id, unread_count, note) VALUES (?, ?, ?)").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	insertStmt, ok := stmt.(*parser.InsertStatement)
	if !ok {
		t.Fatalf("expected insert statement, got %T", stmt)
	}
	exec := &Executor{
		translator:                translator.NewTranslator(),
		protocolMode:              config.ExecutorProtocolModeExtendedSQL,
		insertDuplicateIncrements: map[string]struct{}{tableColumnKey("ks", "tbl", "unread_count"): {}},
	}
	plan, err := exec.buildInsertPlan(context.Background(), "", "", insertStmt)
	if err != nil {
		t.Fatalf("buildInsertPlan failed: %v", err)
	}
	expectedSQL := "INSERT INTO `ks`.`tbl` (`id`, `unread_count`, `note`) VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE `id` = VALUES(`id`), `unread_count` = `unread_count` + VALUES(`unread_count`), `note` = VALUES(`note`)"
	if plan.sql != expectedSQL {
		t.Fatalf("unexpected SQL: %s", plan.sql)
	}
}

func TestBuildInsertPlanUpsertIncrementHintUsesCoalesce(t *testing.T) {
	stmt, err := parser.NewParser("INSERT INTO ks.tbl (id, unread_count, mention_count) VALUES (?, ?, ?)").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	insertStmt, ok := stmt.(*parser.InsertStatement)
	if !ok {
		t.Fatalf("expected insert statement, got %T", stmt)
	}
	insertStmt.OnDuplicateIncrementColumns = map[string]struct{}{
		"unread_count": {},
	}

	exec := &Executor{translator: translator.NewTranslator()}
	plan, err := exec.buildInsertPlan(context.Background(), "", "", insertStmt)
	if err != nil {
		t.Fatalf("buildInsertPlan failed: %v", err)
	}
	expectedSQL := "INSERT INTO `ks`.`tbl` (`id`, `unread_count`, `mention_count`) VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE `id` = VALUES(`id`), `unread_count` = COALESCE(`unread_count`, 0) + VALUES(`unread_count`), `mention_count` = VALUES(`mention_count`)"
	if plan.sql != expectedSQL {
		t.Fatalf("unexpected SQL: %s", plan.sql)
	}
}

func TestAppliedResultShape(t *testing.T) {
	cols := []ColumnInfo{{Name: "v", Type: "int"}}
	row := &Row{Values: []interface{}{int64(7)}}
	result := appliedResult(false, row, cols)
	if len(result.Columns) != 2 {
		t.Fatalf("expected 2 columns, got %d", len(result.Columns))
	}
	if result.Columns[0].Name != "[applied]" || result.Columns[0].Type != "boolean" {
		t.Fatalf("unexpected applied column: %#v", result.Columns[0])
	}
	if result.Columns[1].Name != "v" {
		t.Fatalf("unexpected column name: %#v", result.Columns[1])
	}
	if len(result.Rows) != 1 || len(result.Rows[0].Values) != 2 {
		t.Fatalf("unexpected row shape: %#v", result.Rows)
	}
	if applied, ok := result.Rows[0].Values[0].(bool); !ok || applied {
		t.Fatalf("expected applied=false, got %#v", result.Rows[0].Values[0])
	}
}

func TestConvertJSONRawValue(t *testing.T) {
	tsInput := time.Date(2024, 6, 1, 12, 34, 56, 120*int(time.Millisecond), time.UTC)
	ms := tsInput.UnixNano() / int64(time.Millisecond)
	tsRaw := json.RawMessage(strconv.FormatInt(ms, 10))
	ts, err := convertJSONRawValue(tsRaw, "datetime", "timestamp")
	if err != nil {
		t.Fatalf("timestamp parse failed: %v", err)
	}
	if got := ts.(time.Time).UTC().Format("2006-01-02 15:04:05.000"); got != "2024-06-01 12:34:56.120" {
		t.Fatalf("unexpected timestamp: %s", got)
	}

	microRaw := json.RawMessage(`"2024-06-01 12:34:56.123456"`)
	microTS, err := convertJSONRawValue(microRaw, "datetime", "timestamp")
	if err != nil {
		t.Fatalf("microsecond timestamp parse failed: %v", err)
	}
	if got := microTS.(time.Time).UTC().Format("2006-01-02 15:04:05.000000"); got != "2024-06-01 12:34:56.123456" {
		t.Fatalf("unexpected microsecond timestamp: %s", got)
	}

	val, err := convertJSONRawValue(json.RawMessage(`"foo"`), "varchar", "text")
	if err != nil || val.(string) != "foo" {
		t.Fatalf("unexpected text value: %#v (%v)", val, err)
	}

	val, err = convertJSONRawValue(json.RawMessage("42"), "int", "int")
	if err != nil || val.(int64) != 42 {
		t.Fatalf("unexpected int value: %#v (%v)", val, err)
	}

	val, err = convertJSONRawValue(json.RawMessage(`"0x0a"`), "blob", "blob")
	if err != nil {
		t.Fatalf("blob parse failed: %v", err)
	}
	if got := val.([]byte); len(got) != 1 || got[0] != 0x0a {
		t.Fatalf("unexpected blob value: %#v", got)
	}
}

func TestConditionalNullTranslation(t *testing.T) {
	exec := &Executor{translator: translator.NewTranslator()}
	cases := []struct {
		cql      string
		expected string
	}{
		{
			cql:      "UPDATE ks.tbl SET v = 1 WHERE id = 1 IF value = null",
			expected: "`value` IS NULL",
		},
		{
			cql:      "UPDATE ks.tbl SET v = 1 WHERE id = 1 IF value IS NULL",
			expected: "`value` IS NULL",
		},
		{
			cql:      "UPDATE ks.tbl SET v = 1 WHERE id = 1 IF value IS NOT NULL",
			expected: "`value` IS NOT NULL",
		},
	}

	for _, tc := range cases {
		stmt, err := parser.NewParser(tc.cql).Parse()
		if err != nil {
			t.Fatalf("parse failed: %v", err)
		}
		updateStmt, ok := stmt.(*parser.UpdateStatement)
		if !ok {
			t.Fatalf("expected update statement, got %T", stmt)
		}
		conds, err := extractConditionalConditions(updateStmt.IfCondition)
		if err != nil {
			t.Fatalf("extractConditionalConditions failed: %v", err)
		}
		sql, _ := exec.buildWhereClause(conds, buildBindMarkerIndex(stmt), nil)
		if sql != tc.expected {
			t.Fatalf("unexpected SQL: %s", sql)
		}
	}
}

func TestBindMarkerOrderUpdateTimestamp(t *testing.T) {
	stmt := mustParseStatement(t, "UPDATE ks.tbl USING TIMESTAMP ? SET v = ? WHERE id = ?")
	updateStmt, ok := stmt.(*parser.UpdateStatement)
	if !ok {
		t.Fatalf("expected update statement, got %T", stmt)
	}

	markerIndex := buildBindMarkerIndex(updateStmt)
	tsIndex, ok := markerIndex[updateStmt.Timestamp]
	if !ok {
		t.Fatalf("timestamp bind marker missing")
	}
	assignIndex, ok := markerIndex[updateStmt.Assigments[0].Value]
	if !ok {
		t.Fatalf("assignment bind marker missing")
	}
	whereExpr, ok := updateStmt.Where.(*parser.BinaryExpr)
	if !ok {
		t.Fatalf("expected binary where clause, got %T", updateStmt.Where)
	}
	whereMarker, ok := whereExpr.Right.(*parser.BindMarker)
	if !ok {
		t.Fatalf("expected bind marker in where clause, got %T", whereExpr.Right)
	}
	whereIndex, ok := markerIndex[whereMarker]
	if !ok {
		t.Fatalf("where bind marker missing")
	}

	if tsIndex != 0 || assignIndex != 1 || whereIndex != 2 {
		t.Fatalf("unexpected bind marker order: ts=%d assign=%d where=%d", tsIndex, assignIndex, whereIndex)
	}
}

func TestTranslateExpressionWithTimestampOffsetLiteral(t *testing.T) {
	exec := &Executor{translator: translator.NewTranslator()}
	dml := &dmlExecutor{Executor: exec}
	cases := []struct {
		input    string
		expected string
	}{
		{
			input:    "1970-01-01 00:00:00+0000",
			expected: "'1970-01-01 00:00:00.000'",
		},
		{
			input:    "2024-06-01 12:34:56.123456+0100",
			expected: "'2024-06-01 11:34:56.123456'",
		},
	}
	for _, tc := range cases {
		got := dml.translateExpressionWithType(&parser.StringLiteral{Value: tc.input}, "timestamp")
		if got != tc.expected {
			t.Fatalf("unexpected timestamp translation for %q: %s", tc.input, got)
		}
	}
}

func TestBuildWhereClauseWithInAndRange(t *testing.T) {
	stmt := mustParseStatement(t, "UPDATE ks.tbl SET v = 1 WHERE user_id = 1 AND session_id IN (3, 4) AND ts > 5 AND score BETWEEN 1 AND 2")
	updateStmt, ok := stmt.(*parser.UpdateStatement)
	if !ok {
		t.Fatalf("expected update statement, got %T", stmt)
	}
	conds, err := extractEqualityConditions(updateStmt.Where)
	if err != nil {
		t.Fatalf("extractEqualityConditions failed: %v", err)
	}
	exec := &Executor{translator: translator.NewTranslator()}
	dml := &dmlExecutor{Executor: exec}
	sql, _ := dml.buildWhereClause(conds, buildBindMarkerIndex(stmt), nil)
	expected := "`user_id` = 1 AND `session_id` IN (3, 4) AND `ts` > 5 AND `score` BETWEEN 1 AND 2"
	if sql != expected {
		t.Fatalf("unexpected where SQL: %s", sql)
	}
}

func TestInSingleElementCollapseToEquality(t *testing.T) {
	stmt := mustParseStatement(t, "DELETE FROM ks.tbl WHERE user_id = 1 AND session_id IN (3)")
	deleteStmt, ok := stmt.(*parser.DeleteStatement)
	if !ok {
		t.Fatalf("expected delete statement, got %T", stmt)
	}
	conds, err := extractEqualityConditions(deleteStmt.Where)
	if err != nil {
		t.Fatalf("extractEqualityConditions failed: %v", err)
	}
	exec := &Executor{translator: translator.NewTranslator()}
	dml := &dmlExecutor{Executor: exec}
	sql, _ := dml.buildWhereClause(conds, buildBindMarkerIndex(stmt), nil)
	expected := "`user_id` = 1 AND `session_id` = 3"
	if sql != expected {
		t.Fatalf("unexpected where SQL: %s", sql)
	}
}

func TestHasAllPrimaryKeyExactWithIn(t *testing.T) {
	pkCols := []string{"user_id", "session_id"}
	conds := []equalityCondition{
		{column: "user_id", op: "="},
		{column: "session_id", op: "IN"},
	}
	if !hasAllPrimaryKeyExact(pkCols, conds) {
		t.Fatalf("expected primary key exact match with IN")
	}
	if hasAllPrimaryKeyEquality(pkCols, conds) {
		t.Fatalf("expected primary key equality to reject IN")
	}
}