package config

import (
	"net"
	"os"
	"runtime"
	"strconv"
	"strings"
	"time"

	"gopkg.in/yaml.v3"
)

const (
	defaultPartitionDefaultSize = 10
	envPartitionDefaultSize     = "PARTITION_DEFAULT_SIZE"
	// Default max rows to merge into one INSERT ... VALUES statement in batch optimization.
	defaultExecutorBatchInsertMergeMaxRows = 256
	// Default max estimated SQL bytes for merged INSERT ... VALUES in batch optimization.
	defaultExecutorBatchInsertMergeMaxBytes = 256 * 1024
	// Default delay between flipping readiness=false and closing listeners.
	// Must be >= the time for the load balancer (e.g. AWS NLB) to deregister
	// this pod, so connections broken during drain reconnect to a healthy pod
	// rather than back to this one. Keep readiness_delay + wait_timeout well
	// under terminationGracePeriodSeconds.
	defaultServerDrainReadinessDelay = 40 * time.Second
	// Default time to wait for in-flight connections to drain after listeners
	// close, before force-closing the remainder.
	defaultServerDrainWaitTimeout = 25 * time.Second
	// Executor protocol modes.
	ExecutorProtocolModeStrictCassandra = "strict_cassandra"
	ExecutorProtocolModeExtendedSQL     = "extended_sql"
	// ExecutorProtocolModeExtendedTiDB is kept as a legacy alias for backward compatibility.
	ExecutorProtocolModeExtendedTiDB = "extended_tidb"
)

// Config holds the proxy configuration
type Config struct {
	// MySQL backend configuration
	MySQL MySQLConfig `yaml:"mysql"`

	// Proxy server configuration
	Server ServerConfig `yaml:"server"`

	// Tenancy configuration
	Tenancy TenancyConfig `yaml:"tenancy"`

	// System table configuration for Cassandra compatibility.
	SystemTables SystemTablesConfig `yaml:"system_tables"`

	// Schema sync configuration for multi-instance deployments.
	SchemaSync SchemaSyncConfig `yaml:"schema_sync"`

	// SchemaCacheTTL controls information_schema cache TTL.
	SchemaCacheTTL time.Duration `yaml:"schema_cache_ttl"`
	// CQLMetaCacheTTL controls cql_meta cache TTL.
	CQLMetaCacheTTL time.Duration `yaml:"cql_meta_cache_ttl"`

	// Feature flags
	Features FeatureConfig `yaml:"features"`

	// Logging configuration
	Logging LoggingConfig `yaml:"logging"`

	// Executor configuration
	Executor ExecutorConfig `yaml:"executor"`
}

// MySQLConfig configures the MySQL backend
type MySQLConfig struct {
	Host                 string              `yaml:"host"`
	MaxOpenConns         int                 `yaml:"max_open_conns"`
	MaxIdleConns         int                 `yaml:"max_idle_conns"`
	ConnectTimeout       time.Duration       `yaml:"connect_timeout"`
	ReadTimeout          time.Duration       `yaml:"read_timeout"`
	WriteTimeout         time.Duration       `yaml:"write_timeout"`
	ConnMaxLifetime      time.Duration       `yaml:"conn_max_lifetime"`
	ConnMaxIdleTime      time.Duration       `yaml:"conn_max_idle_time"`
	TLS                  MySQLTLSConfig      `yaml:"tls"`
	DSNParams            map[string]string   `yaml:"dsn_params"`
	Prepared             MySQLPreparedConfig `yaml:"prepared"`
	Retry                MySQLRetryConfig    `yaml:"retry"`
	MaxIndexBytes        int                 `yaml:"max_index_bytes"`
	MinIndexPrefixChars  int                 `yaml:"min_index_prefix_chars"`
	MinIndexPrefixBytes  int                 `yaml:"min_index_prefix_bytes"`
	PartitionDefaultSize int                 `yaml:"partition_default_size"`
	// EnableClusteredIndex emits PRIMARY KEY ... CLUSTERED in CREATE TABLE translations.
	// When false, translator emits NONCLUSTERED explicitly to avoid inheriting TiDB server defaults.
	EnableClusteredIndex bool `yaml:"enable_clustered_index"`
	// DisableTextBlobPartitionKey keeps legacy behavior that rejects TEXT/BLOB partition keys.
	DisableTextBlobPartitionKey bool `yaml:"disable_text_blob_partition_key"`
	// DisableTextBlobSecondaryIndex keeps legacy behavior that rejects secondary indexes on TEXT/BLOB.
	DisableTextBlobSecondaryIndex bool `yaml:"disable_text_blob_secondary_index"`
}

// MySQLPreparedConfig configures MySQL prepared statement caching.
type MySQLPreparedConfig struct {
	Enabled          bool          `yaml:"enabled"`
	MaxStatements    int           `yaml:"max_statements"`
	IdleTTL          time.Duration `yaml:"idle_ttl"`
	CleanupInterval  time.Duration `yaml:"cleanup_interval"`
	PrewarmOnPrepare bool          `yaml:"prewarm_on_prepare"`
}

// MySQLRetryConfig configures transparent server-side retry of backend writes.
type MySQLRetryConfig struct {
	// Deadlock controls retry of writes that failed with a deadlock (MySQL
	// 1213). A deadlock fully rolls the transaction back, so the write is safe
	// to re-run.
	Deadlock DeadlockRetryConfig `yaml:"deadlock"`
}

// DeadlockRetryConfig configures deadlock (MySQL 1213) write retries. Zero
// values fall back to defaults (enabled, 3 attempts, 10ms..100ms backoff).
type DeadlockRetryConfig struct {
	// Enabled toggles deadlock retry; nil (unset) means enabled.
	Enabled *bool `yaml:"enabled"`
	// MaxAttempts is the total number of tries including the first.
	MaxAttempts int `yaml:"max_attempts"`
	// BaseBackoff is the first retry's backoff ceiling; it doubles per retry.
	BaseBackoff time.Duration `yaml:"base_backoff"`
	// MaxBackoff caps the per-retry backoff ceiling.
	MaxBackoff time.Duration `yaml:"max_backoff"`
}

// MySQLTLSConfig configures TLS settings for the MySQL driver.
type MySQLTLSConfig struct {
	// Mode maps to the go-sql-driver/mysql tls parameter (true, skip-verify, preferred, or custom name).
	Mode string `yaml:"mode"`
	// CAFile is an optional CA bundle used to validate the server certificate.
	CAFile string `yaml:"ca_file"`
	// CertFile is an optional client certificate for mTLS.
	CertFile string `yaml:"cert_file"`
	// KeyFile is an optional client key for mTLS.
	KeyFile string `yaml:"key_file"`
	// ServerName overrides the TLS SNI/server name used for verification.
	ServerName string `yaml:"server_name"`
	// InsecureSkipVerify skips server cert verification when true.
	InsecureSkipVerify bool `yaml:"insecure_skip_verify"`
}

// ServerConfig configures the proxy server
type ServerConfig struct {
	MaxConnections    int                    `yaml:"max_connections"`
	MaxInFlightFrames int                    `yaml:"max_in_flight_frames"`
	Admin             AdminConfig            `yaml:"admin"`
	Drain             ServerDrainConfig      `yaml:"drain"`
	Listeners         []ServerListenerConfig `yaml:"listeners"`
}

// AdminConfig configures the admin HTTP server (pprof, metrics).
type AdminConfig struct {
	Address string        `yaml:"address"`
	Metrics MetricsConfig `yaml:"metrics"`
	Pprof   PprofConfig   `yaml:"pprof"`
}

// ServerDrainConfig configures graceful drain behavior during shutdown.
type ServerDrainConfig struct {
	ReadinessDelay time.Duration `yaml:"readiness_delay"`
	WaitTimeout    time.Duration `yaml:"wait_timeout"`
}

// PprofConfig configures the pprof admin endpoints.
type PprofConfig struct {
	Enabled *bool  `yaml:"enabled"`
	Path    string `yaml:"path"`
}

// ServerTLSConfig configures TLS settings for the CQL server.
type ServerTLSConfig struct {
	// Mode selects TLS behavior (disabled, manual, auto).
	Mode string `yaml:"mode"`
	// CertFile is the path to the server certificate PEM.
	CertFile string `yaml:"cert_file"`
	// KeyFile is the path to the server private key PEM.
	KeyFile string `yaml:"key_file"`
	// ClientCAFile is an optional CA bundle for verifying client certificates.
	ClientCAFile string `yaml:"client_ca_file"`
	// ClientAuth controls client certificate behavior (none, request, require, verify, require_and_verify).
	ClientAuth string `yaml:"client_auth"`
	// MinVersion sets the minimum TLS version (e.g. "1.2", "1.3").
	MinVersion string `yaml:"min_version"`
	// MaxVersion sets the maximum TLS version (e.g. "1.2", "1.3").
	MaxVersion string `yaml:"max_version"`
}

// ServerListenerConfig configures a single CQL listener.
type ServerListenerConfig struct {
	Name    string          `yaml:"name"`
	Address string          `yaml:"address"`
	TLS     ServerTLSConfig `yaml:"tls"`
}

// TenancyConfig configures tenant isolation behavior.
type TenancyConfig struct {
	TenantRegex         string             `yaml:"tenant_regex"`
	TenantRegexFallback string             `yaml:"tenant_regex_fallback"`
	PruneInterval       time.Duration      `yaml:"prune_interval"`
	Limits              TenancyLimitConfig `yaml:"limits"`
}

// TenancyLimitConfig sets per-tenant limits.
type TenancyLimitConfig struct {
	MaxProxyConnections int `yaml:"max_proxy_connections"`
	MaxMySQLConnections int `yaml:"max_mysql_connections"`
}

// SystemTablesConfig controls system table values surfaced to clients.
type SystemTablesConfig struct {
	ClusterName           string               `yaml:"cluster_name"`
	Datacenter            string               `yaml:"datacenter"`
	Rack                  string               `yaml:"rack"`
	HostID                string               `yaml:"host_id"`
	AdvertisedAddress     string               `yaml:"advertised_address"`
	NativePort            int                  `yaml:"native_port"`
	StoragePort           int                  `yaml:"storage_port"`
	ReleaseVersion        string               `yaml:"release_version"`
	CQLVersion            string               `yaml:"cql_version"`
	NativeProtocolVersion string               `yaml:"native_protocol_version"`
	Partitioner           string               `yaml:"partitioner"`
	Tokens                []string             `yaml:"tokens"`
	SchemaVersion         SchemaVersionConfig  `yaml:"schema_version"`
	Peers                 []SystemTablePeerCfg `yaml:"peers"`
	CapabilityProfile     string               `yaml:"capability_profile"`
}

// SchemaVersionConfig controls schema_version strategy.
type SchemaVersionConfig struct {
	Strategy string `yaml:"strategy"`
	Fixed    string `yaml:"fixed"`
}

// SchemaSyncConfig controls cross-instance schema metadata sync behavior.
type SchemaSyncConfig struct {
	Enabled       *bool         `yaml:"enabled"`
	CheckInterval time.Duration `yaml:"check_interval"`
}

// SystemTablePeerCfg defines a peer node for system.peers*.
type SystemTablePeerCfg struct {
	AdvertisedAddress string   `yaml:"advertised_address"`
	NativePort        int      `yaml:"native_port"`
	StoragePort       int      `yaml:"storage_port"`
	Tokens            []string `yaml:"tokens"`
}

// FeatureConfig configures optional proxy features.
type FeatureConfig struct {
	MetadataOnlyViews    bool `yaml:"metadata_only_views"`
	MetadataOnlyTriggers bool `yaml:"metadata_only_triggers"`
}

// ExecutorConfig configures executor behavior.
type ExecutorConfig struct {
	SchemaSyncBackground bool `yaml:"schema_sync_background"`
	ExecutePinMySQLConn  bool `yaml:"execute_pin_mysql_conn"`
	CQLMetaPrecreate     bool `yaml:"cql_meta_precreate"`
	// ProtocolMode controls Cassandra-compatibility behavior in the executor.
	// strict_cassandra keeps compatibility constraints; extended_sql enables
	// non-compatible features such as column default expressions.
	ProtocolMode string `yaml:"protocol_mode"`
	// InsertDuplicateIncrementColumns lists fully-qualified columns in
	// keyspace.table.column form that should use additive upsert semantics
	// during INSERT conflict handling: col = col + VALUES(col).
	InsertDuplicateIncrementColumns []string `yaml:"insert_duplicate_increment_columns"`
	// UnloggedBatchParallel enables best-effort parallel execution for UNLOGGED BATCH.
	UnloggedBatchParallel bool `yaml:"unlogged_batch_parallel"`
	// UnloggedMergeBeforeParallel enables INSERT coalescing before UNLOGGED best-effort parallel execution.
	// When enabled, merged statement failures fall back to per-statement retry to preserve partial success.
	// Nil defaults to true.
	UnloggedMergeBeforeParallel *bool `yaml:"unlogged_merge_before_parallel"`
	// UnloggedBatchParallelism controls max goroutines for parallel UNLOGGED BATCH.
	// Values <= 0 default to runtime.NumCPU() * 4.
	UnloggedBatchParallelism int `yaml:"unlogged_batch_parallelism"`
	// UnloggedBatchMaxMySQLConnections caps concurrent MySQL connections consumed by
	// UNLOGGED BATCH worker fan-out so read traffic keeps backend headroom.
	// Values <= 0 default to 75% of the effective MySQL connection budget, leaving
	// at least one connection for non-batch requests when possible.
	UnloggedBatchMaxMySQLConnections int `yaml:"unlogged_batch_max_mysql_connections"`
	// BatchInsertMergeMaxRows controls max rows coalesced into one INSERT ... VALUES statement.
	// Values <= 0 default to 256.
	BatchInsertMergeMaxRows int `yaml:"batch_insert_merge_max_rows"`
	// BatchInsertMergeMaxBytes controls max estimated SQL bytes for one merged INSERT statement.
	// Values <= 0 default to 256KB.
	BatchInsertMergeMaxBytes int `yaml:"batch_insert_merge_max_bytes"`
	// BatchGroupByTiDBHashPartition keeps rows that map to the same TiDB HASH partition
	// in the same merged execute group when partition metadata can be resolved reliably.
	BatchGroupByTiDBHashPartition bool `yaml:"batch_group_by_tidb_hash_partition"`
	// PreparedCacheMaxStatements bounds the per-tenant CQL prepared statement cache
	// (content-addressed by keyspace+query). Values <= 0 default to 10000. Evicted
	// statements are transparently re-prepared by clients via CQL UNPREPARED recovery,
	// so this only caps memory; it never breaks well-behaved drivers.
	PreparedCacheMaxStatements int `yaml:"prepared_cache_max_statements"`
	// PreparedCacheIdleTTL drops cached CQL prepared statements not used within the TTL
	// on their next access. Values <= 0 default to 1h. The LRU size cap bounds memory
	// regardless; this mainly bounds staleness for statements re-used after a long gap.
	PreparedCacheIdleTTL time.Duration `yaml:"prepared_cache_idle_ttl"`
}

// LoggingConfig configures logging behavior
type LoggingConfig struct {
	// Log level: debug, info, warn, error (default: info)
	Level string `yaml:"level"`
	// Output destination: stdout, file (default: stdout)
	Output string `yaml:"output"`
	// File path for log output (required if output=file)
	FilePath string `yaml:"file_path"`
	// Development mode: true for pretty console logging, false for production JSON
	Development bool `yaml:"development"`
	// Enable query redaction to mask sensitive data (passwords, tokens, etc.)
	RedactQueries *bool `yaml:"redact_queries"`
	// Custom redaction patterns (optional, in addition to default patterns)
	RedactPatterns []string `yaml:"redact_patterns"`
	// FrameInfo enables per-frame info logging on write path.
	FrameInfo bool `yaml:"frame_info"`
	// Sampling configures optional zap sampling.
	Sampling LoggingSamplingConfig `yaml:"sampling"`
}

// LoggingSamplingConfig configures zap sampling when enabled.
type LoggingSamplingConfig struct {
	Enabled    bool `yaml:"enabled"`
	Initial    int  `yaml:"initial"`
	Thereafter int  `yaml:"thereafter"`
}

// MetricsConfig configures Prometheus metrics and slow-request logging.
type MetricsConfig struct {
	Enabled              bool          `yaml:"enabled"`
	Path                 string        `yaml:"path"`
	SlowRequestThreshold time.Duration `yaml:"slow_request_threshold"`
}

// PprofEnabled returns true when pprof is enabled or unset.
func (c AdminConfig) PprofEnabled() bool {
	if c.Pprof.Enabled == nil {
		return true
	}
	return *c.Pprof.Enabled
}

// PprofPath returns the base path for pprof endpoints.
func (c AdminConfig) PprofPath() string {
	path := c.Pprof.Path
	if path == "" {
		return "/debug/pprof"
	}
	if !strings.HasPrefix(path, "/") {
		path = "/" + path
	}
	path = strings.TrimSuffix(path, "/")
	if path == "" {
		return "/debug/pprof"
	}
	return path
}

func defaultExecutorUnloggedBatchMaxMySQLConnections(total int) int {
	if total <= 1 {
		return 1
	}
	reserve := total / 4
	if reserve < 1 {
		reserve = 1
	}
	limit := total - reserve
	if limit < 1 {
		limit = 1
	}
	if limit >= total {
		limit = total - 1
	}
	return limit
}

func clampExecutorUnloggedBatchMaxMySQLConnections(limit, total int) int {
	if total <= 0 {
		if limit <= 0 {
			return 1
		}
		return limit
	}
	if limit <= 0 {
		limit = defaultExecutorUnloggedBatchMaxMySQLConnections(total)
	}
	if limit < 1 {
		limit = 1
	}
	if total > 1 && limit >= total {
		limit = total - 1
	}
	if total == 1 {
		return 1
	}
	return limit
}

// LoadConfig loads configuration from a YAML file
func LoadConfig(path string) (*Config, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return nil, err
	}

	var cfg Config
	if err := yaml.Unmarshal(data, &cfg); err != nil {
		return nil, err
	}

	// Set defaults
	if len(cfg.Server.Listeners) == 0 {
		cfg.Server.Listeners = []ServerListenerConfig{
			{Name: "default", Address: "0.0.0.0:9042", TLS: ServerTLSConfig{Mode: "disabled"}},
		}
	}
	if cfg.Server.MaxConnections == 0 {
		cfg.Server.MaxConnections = 10000
	}
	if cfg.Server.MaxInFlightFrames == 0 {
		cfg.Server.MaxInFlightFrames = 64
	}
	if cfg.Server.Drain.ReadinessDelay == 0 {
		cfg.Server.Drain.ReadinessDelay = defaultServerDrainReadinessDelay
	}
	if cfg.Server.Drain.WaitTimeout == 0 {
		cfg.Server.Drain.WaitTimeout = defaultServerDrainWaitTimeout
	}
	if cfg.MySQL.MaxOpenConns == 0 {
		cfg.MySQL.MaxOpenConns = 1000
	}
	if cfg.MySQL.MaxIdleConns == 0 {
		cfg.MySQL.MaxIdleConns = 1000
	}
	if cfg.MySQL.ConnectTimeout == 0 {
		cfg.MySQL.ConnectTimeout = 30 * time.Second
	}
	if cfg.MySQL.ReadTimeout == 0 {
		cfg.MySQL.ReadTimeout = 30 * time.Second
	}
	if cfg.MySQL.WriteTimeout == 0 {
		cfg.MySQL.WriteTimeout = 30 * time.Second
	}
	if cfg.MySQL.ConnMaxLifetime == 0 {
		cfg.MySQL.ConnMaxLifetime = time.Hour
	}
	if cfg.MySQL.ConnMaxIdleTime == 0 {
		cfg.MySQL.ConnMaxIdleTime = 10 * time.Minute
	}
	if cfg.MySQL.MaxIndexBytes == 0 {
		cfg.MySQL.MaxIndexBytes = 3072
	}
	if cfg.MySQL.MinIndexPrefixChars == 0 {
		cfg.MySQL.MinIndexPrefixChars = 8
	}
	if cfg.MySQL.MinIndexPrefixBytes == 0 {
		cfg.MySQL.MinIndexPrefixBytes = 8
	}
	if cfg.MySQL.PartitionDefaultSize == 0 {
		cfg.MySQL.PartitionDefaultSize = defaultPartitionDefaultSize
	}
	applyPartitionDefaultSizeFromEnv(&cfg)
	if cfg.MySQL.Prepared.MaxStatements == 0 {
		cfg.MySQL.Prepared.MaxStatements = 500
	}
	if cfg.MySQL.Prepared.IdleTTL == 0 {
		cfg.MySQL.Prepared.IdleTTL = 10 * time.Minute
	}
	if cfg.MySQL.Prepared.CleanupInterval == 0 {
		cfg.MySQL.Prepared.CleanupInterval = time.Minute
	}
	if cfg.Executor.PreparedCacheMaxStatements == 0 {
		cfg.Executor.PreparedCacheMaxStatements = 10000
	}
	if cfg.Executor.PreparedCacheIdleTTL == 0 {
		cfg.Executor.PreparedCacheIdleTTL = time.Hour
	}
	if cfg.Tenancy.TenantRegex == "" {
		cfg.Tenancy.TenantRegex = "^(?P<tenant>[^:]+):"
	}
	if cfg.Tenancy.TenantRegexFallback == "" {
		cfg.Tenancy.TenantRegexFallback = "username"
	}
	if cfg.Tenancy.PruneInterval == 0 {
		cfg.Tenancy.PruneInterval = 5 * time.Minute
	}
	if cfg.Tenancy.Limits.MaxProxyConnections == 0 {
		cfg.Tenancy.Limits.MaxProxyConnections = cfg.Server.MaxConnections
	}
	if cfg.Tenancy.Limits.MaxMySQLConnections == 0 {
		cfg.Tenancy.Limits.MaxMySQLConnections = cfg.MySQL.MaxOpenConns
	}
	effectiveMySQLBudget := cfg.MySQL.MaxOpenConns
	if cfg.Tenancy.Limits.MaxMySQLConnections > 0 && (effectiveMySQLBudget <= 0 || cfg.Tenancy.Limits.MaxMySQLConnections < effectiveMySQLBudget) {
		effectiveMySQLBudget = cfg.Tenancy.Limits.MaxMySQLConnections
	}
	if cfg.SystemTables.ClusterName == "" {
		cfg.SystemTables.ClusterName = "Cassandra"
	}
	if cfg.SystemTables.Datacenter == "" {
		cfg.SystemTables.Datacenter = "datacenter1"
	}
	if cfg.SystemTables.Rack == "" {
		cfg.SystemTables.Rack = "rack1"
	}
	if cfg.SystemTables.StoragePort == 0 {
		cfg.SystemTables.StoragePort = 7000
	}
	if cfg.SystemTables.ReleaseVersion == "" {
		cfg.SystemTables.ReleaseVersion = "4.0"
	}
	if cfg.SystemTables.CQLVersion == "" {
		cfg.SystemTables.CQLVersion = "3.4.7"
	}
	if cfg.SystemTables.Partitioner == "" {
		cfg.SystemTables.Partitioner = "org.apache.cassandra.dht.Murmur3Partitioner"
	}
	if cfg.SystemTables.SchemaVersion.Strategy == "" {
		cfg.SystemTables.SchemaVersion.Strategy = "cql_meta"
	}
	if cfg.SystemTables.CapabilityProfile == "" {
		cfg.SystemTables.CapabilityProfile = "compat_v4"
	}
	if cfg.SchemaSync.Enabled == nil {
		cfg.SchemaSync.Enabled = boolPtr(true)
	}
	if cfg.SchemaSync.CheckInterval == 0 {
		cfg.SchemaSync.CheckInterval = 5 * time.Second
	}
	if cfg.SchemaCacheTTL == 0 {
		cfg.SchemaCacheTTL = 30 * time.Second
	}
	if cfg.CQLMetaCacheTTL == 0 {
		cfg.CQLMetaCacheTTL = 30 * time.Second
	}
	if cfg.Executor.UnloggedBatchParallelism <= 0 {
		cfg.Executor.UnloggedBatchParallelism = runtime.NumCPU() * 4
		if cfg.Executor.UnloggedBatchParallelism < 1 {
			cfg.Executor.UnloggedBatchParallelism = 1
		}
	}
	cfg.Executor.UnloggedBatchMaxMySQLConnections = clampExecutorUnloggedBatchMaxMySQLConnections(
		cfg.Executor.UnloggedBatchMaxMySQLConnections,
		effectiveMySQLBudget,
	)
	if cfg.Executor.UnloggedMergeBeforeParallel == nil {
		cfg.Executor.UnloggedMergeBeforeParallel = boolPtr(true)
	}
	if cfg.Executor.BatchInsertMergeMaxRows <= 0 {
		cfg.Executor.BatchInsertMergeMaxRows = defaultExecutorBatchInsertMergeMaxRows
	}
	if cfg.Executor.BatchInsertMergeMaxBytes <= 0 {
		cfg.Executor.BatchInsertMergeMaxBytes = defaultExecutorBatchInsertMergeMaxBytes
	}
	cfg.Executor.ProtocolMode = normalizeExecutorProtocolMode(cfg.Executor.ProtocolMode)
	cfg.Executor.InsertDuplicateIncrementColumns = normalizeExecutorQualifiedColumns(cfg.Executor.InsertDuplicateIncrementColumns)
	// Logging defaults
	if cfg.Logging.Level == "" {
		cfg.Logging.Level = "info"
	}
	if cfg.Logging.Output == "" {
		cfg.Logging.Output = "stdout"
	}
	if cfg.Logging.RedactQueries == nil {
		cfg.Logging.RedactQueries = boolPtr(true)
	}
	if cfg.Logging.Sampling.Enabled {
		if cfg.Logging.Sampling.Initial <= 0 {
			cfg.Logging.Sampling.Initial = 100
		}
		if cfg.Logging.Sampling.Thereafter <= 0 {
			cfg.Logging.Sampling.Thereafter = 100
		}
	}
	if cfg.Server.Admin.Pprof.Enabled == nil {
		cfg.Server.Admin.Pprof.Enabled = boolPtr(true)
	}
	if cfg.Server.Admin.Pprof.Path == "" {
		cfg.Server.Admin.Pprof.Path = "/debug/pprof"
	}
	if cfg.Server.Admin.Metrics.Path == "" {
		cfg.Server.Admin.Metrics.Path = "/metrics"
	}

	return &cfg, nil
}

// DefaultConfig returns a default configuration
func DefaultConfig() *Config {
	cfg := &Config{
		Server: ServerConfig{
			MaxConnections:    10000,
			MaxInFlightFrames: 64,
			Admin: AdminConfig{
				Pprof: PprofConfig{
					Enabled: boolPtr(true),
					Path:    "/debug/pprof",
				},
				Metrics: MetricsConfig{
					Enabled:              false,
					Path:                 "/metrics",
					SlowRequestThreshold: 0,
				},
			},
			Drain: ServerDrainConfig{
				ReadinessDelay: defaultServerDrainReadinessDelay,
				WaitTimeout:    defaultServerDrainWaitTimeout,
			},
			Listeners: []ServerListenerConfig{
				{Name: "default", Address: "0.0.0.0:9042", TLS: ServerTLSConfig{Mode: "disabled"}},
			},
		},
		Tenancy: TenancyConfig{
			TenantRegex:         "^(?P<tenant>[^:]+):",
			TenantRegexFallback: "username",
			PruneInterval:       5 * time.Minute,
			Limits: TenancyLimitConfig{
				MaxProxyConnections: 10000,
				MaxMySQLConnections: 1000,
			},
		},
		SystemTables: SystemTablesConfig{
			ClusterName:       "Cassandra",
			Datacenter:        "datacenter1",
			Rack:              "rack1",
			StoragePort:       7000,
			ReleaseVersion:    "4.0",
			CQLVersion:        "3.4.7",
			Partitioner:       "org.apache.cassandra.dht.Murmur3Partitioner",
			SchemaVersion:     SchemaVersionConfig{Strategy: "cql_meta"},
			CapabilityProfile: "compat_v4",
		},
		SchemaSync: SchemaSyncConfig{
			Enabled:       boolPtr(true),
			CheckInterval: 5 * time.Second,
		},
		SchemaCacheTTL:  30 * time.Second,
		CQLMetaCacheTTL: 30 * time.Second,
		MySQL: MySQLConfig{
			Host:            "localhost:3306",
			MaxOpenConns:    1000,
			MaxIdleConns:    1000,
			ConnectTimeout:  30 * time.Second,
			ReadTimeout:     30 * time.Second,
			WriteTimeout:    30 * time.Second,
			ConnMaxLifetime: time.Hour,
			ConnMaxIdleTime: 10 * time.Minute,
			Prepared: MySQLPreparedConfig{
				Enabled:          false,
				MaxStatements:    500,
				IdleTTL:          10 * time.Minute,
				CleanupInterval:  time.Minute,
				PrewarmOnPrepare: false,
			},
			MaxIndexBytes:                 3072,
			MinIndexPrefixChars:           8,
			MinIndexPrefixBytes:           8,
			PartitionDefaultSize:          defaultPartitionDefaultSize,
			EnableClusteredIndex:          false,
			DisableTextBlobPartitionKey:   false,
			DisableTextBlobSecondaryIndex: false,
		},
		Features: FeatureConfig{
			MetadataOnlyViews:    false,
			MetadataOnlyTriggers: false,
		},
		Executor: ExecutorConfig{
			SchemaSyncBackground:             false,
			ExecutePinMySQLConn:              false,
			CQLMetaPrecreate:                 false,
			ProtocolMode:                     ExecutorProtocolModeStrictCassandra,
			InsertDuplicateIncrementColumns:  nil,
			UnloggedBatchParallel:            false,
			UnloggedMergeBeforeParallel:      boolPtr(true),
			UnloggedBatchParallelism:         runtime.NumCPU() * 4,
			UnloggedBatchMaxMySQLConnections: defaultExecutorUnloggedBatchMaxMySQLConnections(1000),
			BatchInsertMergeMaxRows:          defaultExecutorBatchInsertMergeMaxRows,
			BatchInsertMergeMaxBytes:         defaultExecutorBatchInsertMergeMaxBytes,
			BatchGroupByTiDBHashPartition:    false,
			PreparedCacheMaxStatements:       10000,
			PreparedCacheIdleTTL:             time.Hour,
		},
		Logging: LoggingConfig{
			Level:          "info",
			Output:         "stdout",
			Development:    false,
			RedactQueries:  boolPtr(true),
			RedactPatterns: []string{},
			FrameInfo:      false,
			Sampling:       LoggingSamplingConfig{},
		},
	}
	applyPartitionDefaultSizeFromEnv(cfg)
	return cfg
}

// globalConfig holds the active configuration
var globalConfig *Config

// SetConfig sets the global configuration instance
func SetConfig(cfg *Config) {
	globalConfig = cfg
}

// GetConfig returns the global configuration instance
func GetConfig() *Config {
	if globalConfig == nil {
		return DefaultConfig()
	}
	return globalConfig
}

func applyPartitionDefaultSizeFromEnv(cfg *Config) {
	if cfg == nil {
		return
	}
	value, ok := os.LookupEnv(envPartitionDefaultSize)
	if !ok {
		return
	}
	value = strings.TrimSpace(value)
	if value == "" {
		return
	}
	size, err := strconv.Atoi(value)
	if err != nil || size <= 0 {
		return
	}
	cfg.MySQL.PartitionDefaultSize = size
}

// ToVirtualTableRows converts the configuration to virtual table row format
// Returns rows suitable for system_views.settings virtual table
func (c *Config) ToVirtualTableRows() []struct {
	Name  string
	Value string
} {
	listenAddr := ""
	if len(c.Server.Listeners) > 0 && c.Server.Listeners[0].Address != "" {
		listenAddr = c.Server.Listeners[0].Address
	}
	if listenAddr == "" {
		listenAddr = "0.0.0.0:9042"
	}
	nativePort := c.SystemTables.NativePort
	if nativePort == 0 {
		if _, port, err := net.SplitHostPort(listenAddr); err == nil {
			if parsed, err := strconv.Atoi(port); err == nil && parsed > 0 {
				nativePort = parsed
			}
		}
	}
	if nativePort == 0 {
		nativePort = 9042
	}
	storagePort := c.SystemTables.StoragePort
	if storagePort == 0 {
		storagePort = 7000
	}

	rows := []struct {
		Name  string
		Value string
	}{
		// Server settings
		{"listen_address", listenAddr},
		{"max_connections", strconv.Itoa(c.Server.MaxConnections)},
		{"rpc_address", listenAddr},
		{"debug_address", c.Server.Admin.Address},
		{"server_drain_readiness_delay", c.Server.Drain.ReadinessDelay.String()},
		{"server_drain_wait_timeout", c.Server.Drain.WaitTimeout.String()},

		// MySQL backend settings
		{"mysql_host", c.MySQL.Host},
		{"mysql_max_open_conns", strconv.Itoa(c.MySQL.MaxOpenConns)},
		{"mysql_max_idle_conns", strconv.Itoa(c.MySQL.MaxIdleConns)},
		{"mysql_connect_timeout", c.MySQL.ConnectTimeout.String()},
		{"mysql_read_timeout", c.MySQL.ReadTimeout.String()},
		{"mysql_write_timeout", c.MySQL.WriteTimeout.String()},
		{"mysql_conn_max_lifetime", c.MySQL.ConnMaxLifetime.String()},
		{"mysql_conn_max_idle_time", c.MySQL.ConnMaxIdleTime.String()},
		{"mysql_max_index_bytes", strconv.Itoa(c.MySQL.MaxIndexBytes)},
		{"mysql_min_index_prefix_chars", strconv.Itoa(c.MySQL.MinIndexPrefixChars)},
		{"mysql_min_index_prefix_bytes", strconv.Itoa(c.MySQL.MinIndexPrefixBytes)},
		{"mysql_enable_clustered_index", boolToString(c.MySQL.EnableClusteredIndex)},
		{"mysql_prepared_enabled", boolToString(c.MySQL.Prepared.Enabled)},
		{"mysql_prepared_max_statements", strconv.Itoa(c.MySQL.Prepared.MaxStatements)},
		{"mysql_prepared_idle_ttl", c.MySQL.Prepared.IdleTTL.String()},
		{"mysql_prepared_cleanup_interval", c.MySQL.Prepared.CleanupInterval.String()},
		{"mysql_prepared_prewarm_on_prepare", boolToString(c.MySQL.Prepared.PrewarmOnPrepare)},
		{"executor_unlogged_batch_parallel", boolToString(c.Executor.UnloggedBatchParallel)},
		{"executor_unlogged_merge_before_parallel", boolToString(c.Executor.UnloggedMergeBeforeParallelEnabled())},
		{"executor_unlogged_batch_parallelism", strconv.Itoa(c.Executor.UnloggedBatchParallelism)},
		{"executor_unlogged_batch_max_mysql_connections", strconv.Itoa(c.Executor.UnloggedBatchMaxMySQLConnections)},
		{"executor_batch_insert_merge_max_rows", strconv.Itoa(c.Executor.BatchInsertMergeMaxRows)},
		{"executor_batch_insert_merge_max_bytes", strconv.Itoa(c.Executor.BatchInsertMergeMaxBytes)},
		{"executor_batch_group_by_tidb_hash_partition", boolToString(c.Executor.BatchGroupByTiDBHashPartition)},
		{"executor_protocol_mode", c.Executor.ProtocolModeNormalized()},
		{"executor_insert_duplicate_increment_columns", strings.Join(c.Executor.InsertDuplicateIncrementColumns, ",")},

		// Logging settings
		{"log_level", c.Logging.Level},
		{"log_output", c.Logging.Output},
		{"log_file_path", c.Logging.FilePath},
		{"log_development", boolToString(c.Logging.Development)},
		{"log_redact_queries", boolToString(c.Logging.RedactQueriesEnabled())},

		// Cluster settings (mock values for Cassandra compatibility)
		{"cluster_name", c.SystemTables.ClusterName},
		{"system_tables_host_id", c.SystemTables.HostID},
		{"storage_port", strconv.Itoa(storagePort)},
		{"ssl_storage_port", "7001"},
		{"start_native_transport", "true"},
		{"native_transport_port", strconv.Itoa(nativePort)},
		{"hinted_handoff_enabled", "true"},
		{"auto_snapshot", "true"},
		{"incremental_backups", "false"},
		{"enable_transient_replication", "false"},
	}

	return rows
}

// boolToString converts a boolean to a string representation
func boolToString(b bool) string {
	if b {
		return "true"
	}
	return "false"
}

func boolPtr(v bool) *bool {
	return &v
}

// UnloggedMergeBeforeParallelEnabled returns true when unlogged merge-before-parallel is enabled or unset.
func (c ExecutorConfig) UnloggedMergeBeforeParallelEnabled() bool {
	if c.UnloggedMergeBeforeParallel == nil {
		return true
	}
	return *c.UnloggedMergeBeforeParallel
}

// ProtocolModeNormalized returns the effective executor protocol mode.
func (c ExecutorConfig) ProtocolModeNormalized() string {
	return normalizeExecutorProtocolMode(c.ProtocolMode)
}

// ExtendedProtocolModeEnabled returns true when non-compatible executor extensions are enabled.
func (c ExecutorConfig) ExtendedProtocolModeEnabled() bool {
	return c.ProtocolModeNormalized() == ExecutorProtocolModeExtendedSQL
}

// RedactQueriesEnabled returns true when query redaction is enabled or unset.
func (c LoggingConfig) RedactQueriesEnabled() bool {
	if c.RedactQueries == nil {
		return true
	}
	return *c.RedactQueries
}

// EnabledValue returns true when schema sync is enabled or unset.
func (c SchemaSyncConfig) EnabledValue() bool {
	if c.Enabled == nil {
		return true
	}
	return *c.Enabled
}

func normalizeExecutorProtocolMode(mode string) string {
	switch strings.ToLower(strings.TrimSpace(mode)) {
	case "", ExecutorProtocolModeStrictCassandra:
		return ExecutorProtocolModeStrictCassandra
	case ExecutorProtocolModeExtendedSQL, ExecutorProtocolModeExtendedTiDB:
		return ExecutorProtocolModeExtendedSQL
	default:
		return ExecutorProtocolModeStrictCassandra
	}
}

func normalizeExecutorQualifiedColumns(values []string) []string {
	if len(values) == 0 {
		return nil
	}
	out := make([]string, 0, len(values))
	seen := make(map[string]struct{}, len(values))
	for _, value := range values {
		trimmed := strings.TrimSpace(value)
		if trimmed == "" {
			continue
		}
		lower := strings.ToLower(trimmed)
		if _, ok := seen[lower]; ok {
			continue
		}
		seen[lower] = struct{}{}
		out = append(out, lower)
	}
	if len(out) == 0 {
		return nil
	}
	return out
}