package config

import (
	"os"
	"path/filepath"
	"runtime"
	"testing"
	"time"
)

func TestDefaultConfigUnloggedBatchParallelDefaults(t *testing.T) {
	cfg := DefaultConfig()
	if cfg.Executor.UnloggedBatchParallel {
		t.Fatalf("expected unlogged batch parallel to be disabled by default")
	}
	want := runtime.NumCPU() * 4
	if want < 1 {
		want = 1
	}
	if cfg.Executor.UnloggedBatchParallelism != want {
		t.Fatalf("expected unlogged batch parallelism %d, got %d", want, cfg.Executor.UnloggedBatchParallelism)
	}
	if cfg.Executor.UnloggedBatchMaxMySQLConnections != defaultExecutorUnloggedBatchMaxMySQLConnections(cfg.Tenancy.Limits.MaxMySQLConnections) {
		t.Fatalf("expected unlogged batch MySQL connections %d, got %d", defaultExecutorUnloggedBatchMaxMySQLConnections(cfg.Tenancy.Limits.MaxMySQLConnections), cfg.Executor.UnloggedBatchMaxMySQLConnections)
	}
	if cfg.Executor.BatchInsertMergeMaxRows != defaultExecutorBatchInsertMergeMaxRows {
		t.Fatalf("expected batch insert merge max rows %d, got %d", defaultExecutorBatchInsertMergeMaxRows, cfg.Executor.BatchInsertMergeMaxRows)
	}
	if cfg.Executor.BatchInsertMergeMaxBytes != defaultExecutorBatchInsertMergeMaxBytes {
		t.Fatalf("expected batch insert merge max bytes %d, got %d", defaultExecutorBatchInsertMergeMaxBytes, cfg.Executor.BatchInsertMergeMaxBytes)
	}
	if cfg.Executor.BatchGroupByTiDBHashPartition {
		t.Fatalf("expected TiDB HASH partition grouping to be disabled by default")
	}
	if cfg.MySQL.EnableClusteredIndex {
		t.Fatalf("expected clustered index DDL to be disabled by default")
	}
	if !cfg.Executor.UnloggedMergeBeforeParallelEnabled() {
		t.Fatalf("expected unlogged merge before parallel to be enabled by default")
	}
	if cfg.Executor.ProtocolModeNormalized() != ExecutorProtocolModeStrictCassandra {
		t.Fatalf("expected default protocol mode %q, got %q", ExecutorProtocolModeStrictCassandra, cfg.Executor.ProtocolModeNormalized())
	}
	if cfg.Executor.ExtendedProtocolModeEnabled() {
		t.Fatalf("expected extended protocol mode to be disabled by default")
	}
	if cfg.Server.Drain.ReadinessDelay != defaultServerDrainReadinessDelay {
		t.Fatalf("expected drain readiness delay %s, got %s", defaultServerDrainReadinessDelay, cfg.Server.Drain.ReadinessDelay)
	}
	if cfg.Server.Drain.WaitTimeout != defaultServerDrainWaitTimeout {
		t.Fatalf("expected drain wait timeout %s, got %s", defaultServerDrainWaitTimeout, cfg.Server.Drain.WaitTimeout)
	}
}

func TestLoadConfigSetsUnloggedBatchParallelismDefault(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	if err := os.WriteFile(path, []byte{}, 0o644); err != nil {
		t.Fatalf("write config failed: %v", err)
	}

	cfg, err := LoadConfig(path)
	if err != nil {
		t.Fatalf("load config failed: %v", err)
	}
	want := runtime.NumCPU() * 4
	if want < 1 {
		want = 1
	}
	if cfg.Executor.UnloggedBatchParallelism != want {
		t.Fatalf("expected unlogged batch parallelism %d, got %d", want, cfg.Executor.UnloggedBatchParallelism)
	}
	if cfg.Executor.UnloggedBatchMaxMySQLConnections != defaultExecutorUnloggedBatchMaxMySQLConnections(cfg.Tenancy.Limits.MaxMySQLConnections) {
		t.Fatalf("expected unlogged batch MySQL connections %d, got %d", defaultExecutorUnloggedBatchMaxMySQLConnections(cfg.Tenancy.Limits.MaxMySQLConnections), cfg.Executor.UnloggedBatchMaxMySQLConnections)
	}
	if cfg.Executor.BatchInsertMergeMaxRows != defaultExecutorBatchInsertMergeMaxRows {
		t.Fatalf("expected batch insert merge max rows %d, got %d", defaultExecutorBatchInsertMergeMaxRows, cfg.Executor.BatchInsertMergeMaxRows)
	}
	if cfg.Executor.BatchInsertMergeMaxBytes != defaultExecutorBatchInsertMergeMaxBytes {
		t.Fatalf("expected batch insert merge max bytes %d, got %d", defaultExecutorBatchInsertMergeMaxBytes, cfg.Executor.BatchInsertMergeMaxBytes)
	}
	if cfg.Executor.BatchGroupByTiDBHashPartition {
		t.Fatalf("expected TiDB HASH partition grouping to be disabled by default")
	}
	if cfg.MySQL.EnableClusteredIndex {
		t.Fatalf("expected clustered index DDL to be disabled by default")
	}
	if !cfg.Executor.UnloggedMergeBeforeParallelEnabled() {
		t.Fatalf("expected unlogged merge before parallel to default to true")
	}
	if cfg.Executor.ProtocolModeNormalized() != ExecutorProtocolModeStrictCassandra {
		t.Fatalf("expected default protocol mode %q, got %q", ExecutorProtocolModeStrictCassandra, cfg.Executor.ProtocolModeNormalized())
	}
}

func TestLoadConfigPreservesBatchInsertMergeLimits(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	content := []byte(`
executor:
  batch_insert_merge_max_rows: 512
  batch_insert_merge_max_bytes: 524288
`)
	if err := os.WriteFile(path, content, 0o644); err != nil {
		t.Fatalf("write config failed: %v", err)
	}

	cfg, err := LoadConfig(path)
	if err != nil {
		t.Fatalf("load config failed: %v", err)
	}
	if cfg.Executor.BatchInsertMergeMaxRows != 512 {
		t.Fatalf("expected batch insert merge max rows 512, got %d", cfg.Executor.BatchInsertMergeMaxRows)
	}
	if cfg.Executor.BatchInsertMergeMaxBytes != 524288 {
		t.Fatalf("expected batch insert merge max bytes 524288, got %d", cfg.Executor.BatchInsertMergeMaxBytes)
	}
}

func TestLoadConfigPreservesUnloggedBatchMySQLConnectionLimit(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	content := []byte(`
mysql:
  max_open_conns: 20
executor:
  unlogged_batch_max_mysql_connections: 7
`)
	if err := os.WriteFile(path, content, 0o644); err != nil {
		t.Fatalf("write config failed: %v", err)
	}

	cfg, err := LoadConfig(path)
	if err != nil {
		t.Fatalf("load config failed: %v", err)
	}
	if cfg.Executor.UnloggedBatchMaxMySQLConnections != 7 {
		t.Fatalf("expected unlogged batch MySQL connection limit 7, got %d", cfg.Executor.UnloggedBatchMaxMySQLConnections)
	}
}

func TestLoadConfigClampsUnloggedBatchMySQLConnectionsToLeaveHeadroom(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	content := []byte(`
mysql:
  max_open_conns: 10
executor:
  unlogged_batch_max_mysql_connections: 99
`)
	if err := os.WriteFile(path, content, 0o644); err != nil {
		t.Fatalf("write config failed: %v", err)
	}

	cfg, err := LoadConfig(path)
	if err != nil {
		t.Fatalf("load config failed: %v", err)
	}
	if cfg.Executor.UnloggedBatchMaxMySQLConnections != 9 {
		t.Fatalf("expected unlogged batch MySQL connection limit 9, got %d", cfg.Executor.UnloggedBatchMaxMySQLConnections)
	}
}

func TestLoadConfigPreservesBatchHashPartitionGroupingFlag(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	content := []byte(`
executor:
  batch_group_by_tidb_hash_partition: true
`)
	if err := os.WriteFile(path, content, 0o644); err != nil {
		t.Fatalf("write config failed: %v", err)
	}

	cfg, err := LoadConfig(path)
	if err != nil {
		t.Fatalf("load config failed: %v", err)
	}
	if !cfg.Executor.BatchGroupByTiDBHashPartition {
		t.Fatalf("expected TiDB HASH partition grouping flag to be preserved")
	}
}

func TestLoadConfigPreservesMySQLEnableClusteredIndex(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	content := []byte(`
mysql:
  enable_clustered_index: true
`)
	if err := os.WriteFile(path, content, 0o644); err != nil {
		t.Fatalf("write config failed: %v", err)
	}

	cfg, err := LoadConfig(path)
	if err != nil {
		t.Fatalf("load config failed: %v", err)
	}
	if !cfg.MySQL.EnableClusteredIndex {
		t.Fatalf("expected mysql.enable_clustered_index to be preserved")
	}
}

func TestLoadConfigAllowsDisableUnloggedMergeBeforeParallel(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	content := []byte(`
executor:
  unlogged_merge_before_parallel: false
`)
	if err := os.WriteFile(path, content, 0o644); err != nil {
		t.Fatalf("write config failed: %v", err)
	}

	cfg, err := LoadConfig(path)
	if err != nil {
		t.Fatalf("load config failed: %v", err)
	}
	if cfg.Executor.UnloggedMergeBeforeParallelEnabled() {
		t.Fatalf("expected unlogged merge before parallel to be disabled")
	}
}

func TestLoadConfigServerDrainDefaults(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	if err := os.WriteFile(path, []byte{}, 0o644); err != nil {
		t.Fatalf("write config failed: %v", err)
	}

	cfg, err := LoadConfig(path)
	if err != nil {
		t.Fatalf("load config failed: %v", err)
	}
	if cfg.Server.Drain.ReadinessDelay != defaultServerDrainReadinessDelay {
		t.Fatalf("expected drain readiness delay %s, got %s", defaultServerDrainReadinessDelay, cfg.Server.Drain.ReadinessDelay)
	}
	if cfg.Server.Drain.WaitTimeout != defaultServerDrainWaitTimeout {
		t.Fatalf("expected drain wait timeout %s, got %s", defaultServerDrainWaitTimeout, cfg.Server.Drain.WaitTimeout)
	}
}

func TestLoadConfigPreservesServerDrainOverrides(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	content := []byte(`
server:
  drain:
    readiness_delay: 35s
    wait_timeout: 40s
`)
	if err := os.WriteFile(path, content, 0o644); err != nil {
		t.Fatalf("write config failed: %v", err)
	}

	cfg, err := LoadConfig(path)
	if err != nil {
		t.Fatalf("load config failed: %v", err)
	}
	if cfg.Server.Drain.ReadinessDelay != 35*time.Second {
		t.Fatalf("expected drain readiness delay 35s, got %s", cfg.Server.Drain.ReadinessDelay)
	}
	if cfg.Server.Drain.WaitTimeout != 40*time.Second {
		t.Fatalf("expected drain wait timeout 40s, got %s", cfg.Server.Drain.WaitTimeout)
	}
}

func TestLoadConfigNormalizesExecutorProtocolModeAndIncrementColumns(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	content := []byte(`
executor:
  protocol_mode: "EXTENDED_SQL"
  insert_duplicate_increment_columns:
    - " ks.tbl.unread_count "
    - "KS.TBL.UNREAD_COUNT"
`)
	if err := os.WriteFile(path, content, 0o644); err != nil {
		t.Fatalf("write config failed: %v", err)
	}

	cfg, err := LoadConfig(path)
	if err != nil {
		t.Fatalf("load config failed: %v", err)
	}
	if cfg.Executor.ProtocolModeNormalized() != ExecutorProtocolModeExtendedSQL {
		t.Fatalf("expected protocol mode %q, got %q", ExecutorProtocolModeExtendedSQL, cfg.Executor.ProtocolModeNormalized())
	}
	if !cfg.Executor.ExtendedProtocolModeEnabled() {
		t.Fatalf("expected extended protocol mode to be enabled")
	}
	if len(cfg.Executor.InsertDuplicateIncrementColumns) != 1 {
		t.Fatalf("expected deduplicated increment columns, got %#v", cfg.Executor.InsertDuplicateIncrementColumns)
	}
	if cfg.Executor.InsertDuplicateIncrementColumns[0] != "ks.tbl.unread_count" {
		t.Fatalf("expected normalized increment column, got %q", cfg.Executor.InsertDuplicateIncrementColumns[0])
	}
}

func TestLoadConfigLegacyExtendedTiDBAliasStillWorks(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	content := []byte(`
executor:
  protocol_mode: "EXTENDED_TIDB"
`)
	if err := os.WriteFile(path, content, 0o644); err != nil {
		t.Fatalf("write config failed: %v", err)
	}

	cfg, err := LoadConfig(path)
	if err != nil {
		t.Fatalf("load config failed: %v", err)
	}
	if cfg.Executor.ProtocolModeNormalized() != ExecutorProtocolModeExtendedSQL {
		t.Fatalf("expected protocol mode %q, got %q", ExecutorProtocolModeExtendedSQL, cfg.Executor.ProtocolModeNormalized())
	}
}

func TestLoadConfigInvalidProtocolModeFallsBackToStrict(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	content := []byte(`
executor:
  protocol_mode: "unknown_mode"
`)
	if err := os.WriteFile(path, content, 0o644); err != nil {
		t.Fatalf("write config failed: %v", err)
	}

	cfg, err := LoadConfig(path)
	if err != nil {
		t.Fatalf("load config failed: %v", err)
	}
	if cfg.Executor.ProtocolModeNormalized() != ExecutorProtocolModeStrictCassandra {
		t.Fatalf("expected fallback protocol mode %q, got %q", ExecutorProtocolModeStrictCassandra, cfg.Executor.ProtocolModeNormalized())
	}
}