package translator

import (
	"fmt"
	"strings"
	"testing"

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

func TestTranslateMultiRowInsert(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)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "INSERT INTO `ks`.`tbl` (`a`, `b`) VALUES (1, 2), (3, 4)"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateInsertCollectionLiterals(t *testing.T) {
	stmt, err := parser.NewParser("INSERT INTO ks.tbl (id, tags, roles, attrs) VALUES (1, ['a', 'b'], {'admin', 'user'}, {'region': 'HK', 'lang': 'zh'})").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "INSERT INTO `ks`.`tbl` (`id`, `tags`, `roles`, `attrs`) VALUES (1, JSON_ARRAY('a', 'b'), JSON_ARRAY('admin', 'user'), JSON_OBJECT('region', 'HK', 'lang', 'zh'))"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateInListLiteral(t *testing.T) {
	stmt, err := parser.NewParser("SELECT * FROM ks.tbl WHERE id IN (1, 2, 3)").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "SELECT * FROM `ks`.`tbl` WHERE (`id` IN (1, 2, 3))"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateBetweenTupleLiteral(t *testing.T) {
	stmt, err := parser.NewParser("SELECT * FROM ks.tbl WHERE id BETWEEN 1 AND 3").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "SELECT * FROM `ks`.`tbl` WHERE (`id` BETWEEN 1 AND 3)"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateStringLiteralEscapesBackslash(t *testing.T) {
	stmt, err := parser.NewParser("SELECT * FROM ks.tbl WHERE v = 'Value with \\ backslash'").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "SELECT * FROM `ks`.`tbl` WHERE (`v` = 'Value with \\\\ backslash')"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateFloatLiteralPreservesRaw(t *testing.T) {
	stmt, err := parser.NewParser("SELECT * FROM ks.tbl WHERE v = 1.2300").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "SELECT * FROM `ks`.`tbl` WHERE (`v` = 1.2300)"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateBlobLiteral(t *testing.T) {
	stmt, err := parser.NewParser("INSERT INTO ks.tbl (payload) VALUES (0xdeadbeef)").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "INSERT INTO `ks`.`tbl` (`payload`) VALUES (0xdeadbeef)"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateEmptyBlobLiteral(t *testing.T) {
	stmt, err := parser.NewParser("INSERT INTO ks.tbl (payload) VALUES (0x)").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "INSERT INTO `ks`.`tbl` (`payload`) VALUES (X'')"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateDecimalTypeUsesHighPrecision(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id int, v decimal, PRIMARY KEY (id))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	if !strings.Contains(sql, "DECIMAL(65,30)") {
		t.Fatalf("expected DECIMAL(65,30) in SQL: %s", sql)
	}
}

func TestTranslateTimestampPrecision(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id int, ts timestamp, t time, PRIMARY KEY (id))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	if !strings.Contains(sql, "`ts` DATETIME(6)") {
		t.Fatalf("expected DATETIME(6) for timestamp: %s", sql)
	}
	if !strings.Contains(sql, "`t` TIME(6)") {
		t.Fatalf("expected TIME(6) for time: %s", sql)
	}
}

func TestTranslateCreateKeyspaceCharset(t *testing.T) {
	stmt, err := parser.NewParser("CREATE KEYSPACE ks").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	if !strings.Contains(sql, "DEFAULT CHARACTER SET utf8mb4") {
		t.Fatalf("expected charset in SQL: %s", sql)
	}
	if !strings.Contains(sql, "COLLATE utf8mb4_bin") {
		t.Fatalf("expected collation in SQL: %s", sql)
	}
}

func TestTranslateRequiresSchema(t *testing.T) {
	stmt, err := parser.NewParser("SELECT * FROM tbl").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	_, err = NewTranslator().Translate(stmt)
	if err == nil {
		t.Fatal("expected schema error for unqualified table")
	}
}

func TestTranslateCreateTableOptions(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id int, name text, PRIMARY KEY (id))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	if !strings.Contains(sql, "ENGINE=InnoDB") {
		t.Fatalf("expected engine in SQL: %s", sql)
	}
	if !strings.Contains(sql, "DEFAULT CHARSET=utf8mb4") {
		t.Fatalf("expected charset in SQL: %s", sql)
	}
	if !strings.Contains(sql, "COLLATE utf8mb4_bin") {
		t.Fatalf("expected collation in SQL: %s", sql)
	}
	if !strings.Contains(sql, "PRIMARY KEY (`id`) NONCLUSTERED") {
		t.Fatalf("expected PRIMARY KEY to be NONCLUSTERED by default: %s", sql)
	}
}

func TestTranslateCreateTableTTLRaw(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id int, name text, PRIMARY KEY (id)) TTL = `created_at` + INTERVAL 30 DAY TTL_ENABLE = 'ON'").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	if !strings.Contains(sql, "ttl = `created_at` + interval 30 day ttl_enable = 'ON'") {
		t.Fatalf("expected TTL raw clause in SQL: %s", sql)
	}
}

func TestTranslateAlterTableTTLRaw(t *testing.T) {
	stmt, err := parser.NewParser("ALTER TABLE ks.tbl TTL = `created_at` + INTERVAL 7 DAY TTL_ENABLE = 'ON' TTL_JOB_INTERVAL = '24h'").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "ALTER TABLE `ks`.`tbl` ttl = `created_at` + interval 7 day ttl_enable = 'ON' ttl_job_interval = '24h'"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateAlterTableTTLRawRemove(t *testing.T) {
	stmt, err := parser.NewParser("ALTER TABLE ks.tbl REMOVE TTL").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "ALTER TABLE `ks`.`tbl` remove ttl"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateAlterTableTTLRawEnableOnly(t *testing.T) {
	stmt, err := parser.NewParser("ALTER TABLE ks.tbl TTL_ENABLE = 'OFF'").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "ALTER TABLE `ks`.`tbl` ttl_enable = 'OFF'"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateAlterTableTTLRawJobIntervalOnly(t *testing.T) {
	stmt, err := parser.NewParser("ALTER TABLE ks.tbl TTL_JOB_INTERVAL = '48h'").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "ALTER TABLE `ks`.`tbl` ttl_job_interval = '48h'"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateAlterTableWithUnsupportedTTLRawOption(t *testing.T) {
	stmt, err := parser.NewParser("ALTER TABLE ks.tbl WITH compaction = {'class': 'LeveledCompactionStrategy'}").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	_, err = NewTranslator().Translate(stmt)
	if err == nil {
		t.Fatalf("expected translate error for unsupported ALTER TABLE WITH option without ttl")
	}
}

func TestTranslateCreateTableWithRemovedTTLRawOptionName(t *testing.T) {
	_, err := parser.NewParser("CREATE TABLE ks.tbl (id int, PRIMARY KEY (id)) WITH tidb_ttl_raw = 'TTL = `created_at` + INTERVAL 1 DAY'").Parse()
	if err != nil {
		return
	}
	t.Fatalf("expected parse error for removed option name tidb_ttl_raw")
}

func TestTranslateAlterTableWithRemovedTTLRawOptionName(t *testing.T) {
	_, err := parser.NewParser("ALTER TABLE ks.tbl WITH tidb_ttl_raw = 'REMOVE TTL'").Parse()
	if err != nil {
		return
	}
	t.Fatalf("expected parse error for removed option name tidb_ttl_raw")
}

func TestTranslateCreateTableWithRemovedWrappedTTLOptionName(t *testing.T) {
	_, err := parser.NewParser("CREATE TABLE ks.tbl (id int, PRIMARY KEY (id)) WITH ttl = 'TTL = `created_at` + INTERVAL 1 DAY'").Parse()
	if err != nil {
		return
	}
	t.Fatalf("expected parse error for removed wrapped option name ttl")
}

func TestTranslateAlterTableWithRemovedWrappedTTLOptionName(t *testing.T) {
	_, err := parser.NewParser("ALTER TABLE ks.tbl WITH ttl = 'REMOVE TTL'").Parse()
	if err != nil {
		return
	}
	t.Fatalf("expected parse error for removed wrapped option name ttl")
}

func TestTranslateCreateTableHashPartitionSingleInt(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id INT, name TEXT, PRIMARY KEY (id))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := expectedPartitionSize()
	if !strings.Contains(sql, fmt.Sprintf("PARTITION BY HASH(`id`) PARTITIONS %d", expected)) {
		t.Fatalf("expected HASH partition in SQL: %s", sql)
	}
}

func TestTranslateCreateTableKeyPartitionSingleVarint(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id VARINT, name TEXT, PRIMARY KEY (id))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := expectedPartitionSize()
	if !strings.Contains(sql, fmt.Sprintf("PARTITION BY KEY(`id`) PARTITIONS %d", expected)) {
		t.Fatalf("expected KEY partition in SQL: %s", sql)
	}
}

func TestTranslateCreateTableKeyPartitionComposite(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (a INT, b INT, c INT, PRIMARY KEY ((a, b), c))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := expectedPartitionSize()
	if !strings.Contains(sql, fmt.Sprintf("PARTITION BY KEY(`a`, `b`) PARTITIONS %d", expected)) {
		t.Fatalf("expected composite KEY partition in SQL: %s", sql)
	}
}

func TestTranslateCreateTablePartitionKeyTextAllowedWithoutPartitionClause(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id TEXT, PRIMARY KEY (id))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	if strings.Contains(sql, "PARTITION BY") {
		t.Fatalf("expected PARTITION BY to be skipped for TEXT partition key: %s", sql)
	}
}

func TestTranslateCreateTablePartitionKeyTextRejectedWhenDisabled(t *testing.T) {
	cfg := config.DefaultConfig()
	cfg.MySQL.DisableTextBlobPartitionKey = true
	config.SetConfig(cfg)
	t.Cleanup(func() {
		config.SetConfig(config.DefaultConfig())
	})

	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id TEXT, PRIMARY KEY (id))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	_, err = NewTranslator().Translate(stmt)
	if err == nil {
		t.Fatal("expected TEXT partition key to be rejected when disabled")
	}
}

func TestTranslateCreateTablePrimaryKeyPrefixLeftmostAllocation(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (tenant TEXT, user TEXT, shard INT, PRIMARY KEY (shard, tenant, user))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	if !strings.Contains(sql, "PRIMARY KEY (`shard`, `tenant`(759), `user`(8))") {
		t.Fatalf("expected leftmost prefix allocation in SQL: %s", sql)
	}
}

func TestTranslateCreateTablePrimaryKeyNoPrefixWhenWithinBudget(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (a ASCII, b ASCII, PRIMARY KEY (a, b))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	if !strings.Contains(sql, "PRIMARY KEY (`a`, `b`) NONCLUSTERED") {
		t.Fatalf("expected no prefix lengths in SQL: %s", sql)
	}
}

func TestTranslateCreateTablePrimaryKeyCanEnableClusteredIndex(t *testing.T) {
	cfg := config.DefaultConfig()
	cfg.MySQL.EnableClusteredIndex = true
	config.SetConfig(cfg)
	t.Cleanup(func() {
		config.SetConfig(config.DefaultConfig())
	})

	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id int, name text, PRIMARY KEY (id))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	if !strings.Contains(sql, "PRIMARY KEY (`id`) CLUSTERED") {
		t.Fatalf("expected PRIMARY KEY to be CLUSTERED when enabled: %s", sql)
	}
}

func TestTranslateCreateTablePrimaryKeyPrefixBudgetTooSmall(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (a INT, b TEXT, c TEXT, PRIMARY KEY (a, b, c))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	tr := NewTranslator()
	tr.indexMaxBytes = 64
	tr.minPrefixChars = 8
	tr.minPrefixBytes = 8
	_, err = tr.Translate(stmt)
	if err == nil {
		t.Fatal("expected prefix allocation error")
	}
}

func TestTranslateCreateTablePrimaryKeyConfigOverrides(t *testing.T) {
	cfg := config.DefaultConfig()
	cfg.MySQL.MaxIndexBytes = 128
	cfg.MySQL.MinIndexPrefixChars = 4
	cfg.MySQL.MinIndexPrefixBytes = 4
	config.SetConfig(cfg)
	t.Cleanup(func() {
		config.SetConfig(config.DefaultConfig())
	})

	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id INT, name TEXT, PRIMARY KEY (id, name))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	if !strings.Contains(sql, "PRIMARY KEY (`id`, `name`(31))") {
		t.Fatalf("expected config-based prefix length in SQL: %s", sql)
	}
}

func TestTranslateCreateTablePrimaryKeyCollectionRejected(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id list<text>, PRIMARY KEY (id))").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	_, err = NewTranslator().Translate(stmt)
	if err == nil {
		t.Fatal("expected collection primary key to be rejected")
	}
	if !strings.Contains(err.Error(), "unsupported type") {
		t.Fatalf("unexpected error: %v", err)
	}
}

func TestTranslateCreateMaterializedView(t *testing.T) {
	stmt, err := parser.NewParser("CREATE MATERIALIZED VIEW ks.mv AS SELECT id FROM ks.tbl WHERE id IS NOT NULL PRIMARY KEY (id)").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "CREATE VIEW `ks`.`mv` AS SELECT `id` FROM `ks`.`tbl` WHERE (`id` IS NOT NULL)"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}

func TestTranslateCreateTableColumnDefaultAndOnUpdate(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id int PRIMARY KEY, created_at timestamp DEFAULT current_timestamp, updated_at timestamp DEFAULT now() ON UPDATE current_timestamp)").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	if !strings.Contains(sql, "`created_at` DATETIME(6) DEFAULT CURRENT_TIMESTAMP(6)") {
		t.Fatalf("expected created_at default in SQL: %s", sql)
	}
	if !strings.Contains(sql, "`updated_at` DATETIME(6) DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6)") {
		t.Fatalf("expected updated_at default/on-update in SQL: %s", sql)
	}
}

func TestTranslateCreateTableRejectsUnsupportedDefaultExpression(t *testing.T) {
	stmt, err := parser.NewParser("CREATE TABLE ks.tbl (id int PRIMARY KEY, v int DEFAULT other_col)").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	_, err = NewTranslator().Translate(stmt)
	if err == nil {
		t.Fatalf("expected unsupported default expression error")
	}
}

func expectedPartitionSize() int {
	cfg := config.GetConfig()
	if cfg == nil {
		return 10
	}
	if cfg.MySQL.PartitionDefaultSize <= 0 {
		return 10
	}
	return cfg.MySQL.PartitionDefaultSize
}

func TestTranslateDropMaterializedView(t *testing.T) {
	stmt, err := parser.NewParser("DROP MATERIALIZED VIEW IF EXISTS ks.mv").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	sql, err := NewTranslator().Translate(stmt)
	if err != nil {
		t.Fatalf("translate failed: %v", err)
	}
	expected := "DROP VIEW IF EXISTS `ks`.`mv`"
	if sql != expected {
		t.Fatalf("unexpected SQL: %s", sql)
	}
}