package mysqlbackend

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"sync"
	"time"

	mysql "github.com/go-sql-driver/mysql"

	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/metrics"
)

// CQLKeyspaceInfo describes a keyspace stored in cql_meta.
type CQLKeyspaceInfo struct {
	Name            string
	DurableWrites   bool
	ReplicationJSON []byte
}

// CQLTableInfo describes a table stored in cql_meta.
type CQLTableInfo struct {
	Keyspace    string
	Table       string
	OptionsJSON []byte
	Comment     string
}

// CQLColumnInfo describes a column stored in cql_meta.
type CQLColumnInfo struct {
	Keyspace        string
	Table           string
	Name            string
	CQLType         string
	DefaultExpr     string
	OnUpdateExpr    string
	Kind            string
	Position        int
	ClusteringOrder string
}

// CQLTypeInfo describes a user-defined type stored in cql_meta.
type CQLTypeInfo struct {
	Keyspace   string
	Name       string
	FieldNames []string
	FieldTypes []string
}

// CQLIndexInfo describes an index stored in cql_meta.
type CQLIndexInfo struct {
	Keyspace    string
	Table       string
	Name        string
	Kind        string
	OptionsJSON []byte
}

// CQLViewInfo describes a view stored in cql_meta.
type CQLViewInfo struct {
	Keyspace    string
	Name        string
	BaseTable   string
	WhereClause string
	OptionsJSON []byte
}

// CQLTriggerInfo describes a trigger stored in cql_meta.
type CQLTriggerInfo struct {
	Keyspace    string
	Table       string
	Name        string
	OptionsJSON []byte
}

type cqlKeyspaceCacheEntry struct {
	expires   time.Time
	list      []CQLKeyspaceInfo
	available bool
}

type cqlTableCacheEntry struct {
	expires       time.Time
	list          []CQLTableInfo
	bySchema      map[string][]CQLTableInfo
	bySchemaLower map[string][]CQLTableInfo
	available     bool
}

type cqlColumnCacheEntry struct {
	expires      time.Time
	list         []CQLColumnInfo
	byTable      map[tableKey][]CQLColumnInfo
	byTableLower map[tableKey][]CQLColumnInfo
	available    bool
}

type cqlTypeCacheEntry struct {
	expires     time.Time
	list        []CQLTypeInfo
	byName      map[typeKey]CQLTypeInfo
	byNameLower map[typeKey]CQLTypeInfo
	available   bool
}

type cqlIndexCacheEntry struct {
	expires      time.Time
	list         []CQLIndexInfo
	byTable      map[tableKey][]CQLIndexInfo
	byTableLower map[tableKey][]CQLIndexInfo
	available    bool
}

type cqlViewCacheEntry struct {
	expires     time.Time
	list        []CQLViewInfo
	byName      map[viewKey]CQLViewInfo
	byNameLower map[viewKey]CQLViewInfo
	available   bool
}

type cqlTriggerCacheEntry struct {
	expires     time.Time
	list        []CQLTriggerInfo
	byName      map[triggerKey]CQLTriggerInfo
	byNameLower map[triggerKey]CQLTriggerInfo
	available   bool
}

// CQLMetaCache caches cql_meta metadata with a short TTL.
type CQLMetaCache struct {
	ttl           time.Duration
	keyspaces     map[userKey]cqlKeyspaceCacheEntry
	tables        map[userKey]cqlTableCacheEntry
	columns       map[userKey]cqlColumnCacheEntry
	types         map[userKey]cqlTypeCacheEntry
	indexes       map[userKey]cqlIndexCacheEntry
	views         map[userKey]cqlViewCacheEntry
	triggers      map[userKey]cqlTriggerCacheEntry
	mu            sync.RWMutex
	refreshMu     sync.Mutex
	keyspaceLocks map[userKey]*sync.Mutex
	tableLocks    map[userKey]*sync.Mutex
	columnLocks   map[userKey]*sync.Mutex
	typeLocks     map[userKey]*sync.Mutex
	indexLocks    map[userKey]*sync.Mutex
	viewLocks     map[userKey]*sync.Mutex
	triggerLocks  map[userKey]*sync.Mutex
}

// NewCQLMetaCache creates a CQLMetaCache with the provided TTL.
func NewCQLMetaCache(ttl time.Duration) *CQLMetaCache {
	if ttl <= 0 {
		ttl = defaultSchemaCacheTTL
	}
	return &CQLMetaCache{
		ttl:       ttl,
		keyspaces: make(map[userKey]cqlKeyspaceCacheEntry),
		tables:    make(map[userKey]cqlTableCacheEntry),
		columns:   make(map[userKey]cqlColumnCacheEntry),
		types:     make(map[userKey]cqlTypeCacheEntry),
		indexes:   make(map[userKey]cqlIndexCacheEntry),
		views:     make(map[userKey]cqlViewCacheEntry),
		triggers:  make(map[userKey]cqlTriggerCacheEntry),
	}
}

func (c *CQLMetaCache) lockFor(m *map[userKey]*sync.Mutex, key userKey) *sync.Mutex {
	if c == nil {
		return nil
	}
	c.refreshMu.Lock()
	defer c.refreshMu.Unlock()
	if *m == nil {
		*m = make(map[userKey]*sync.Mutex)
	}
	lock := (*m)[key]
	if lock == nil {
		lock = &sync.Mutex{}
		(*m)[key] = lock
	}
	return lock
}

func (c *CQLMetaCache) keyspaceLock(key userKey) *sync.Mutex {
	return c.lockFor(&c.keyspaceLocks, key)
}

func (c *CQLMetaCache) tableLock(key userKey) *sync.Mutex {
	return c.lockFor(&c.tableLocks, key)
}

func (c *CQLMetaCache) columnLock(key userKey) *sync.Mutex {
	return c.lockFor(&c.columnLocks, key)
}

func (c *CQLMetaCache) typeLock(key userKey) *sync.Mutex {
	return c.lockFor(&c.typeLocks, key)
}

func (c *CQLMetaCache) indexLock(key userKey) *sync.Mutex {
	return c.lockFor(&c.indexLocks, key)
}

func (c *CQLMetaCache) viewLock(key userKey) *sync.Mutex {
	return c.lockFor(&c.viewLocks, key)
}

func (c *CQLMetaCache) triggerLock(key userKey) *sync.Mutex {
	return c.lockFor(&c.triggerLocks, key)
}

// InvalidateUser clears cached metadata for a specific user.
func (c *CQLMetaCache) InvalidateUser(user, password string) {
	if c == nil {
		return
	}
	key := userKey{user: user, password: password}
	c.mu.Lock()
	delete(c.keyspaces, key)
	delete(c.tables, key)
	delete(c.columns, key)
	delete(c.types, key)
	delete(c.indexes, key)
	delete(c.views, key)
	delete(c.triggers, key)
	c.mu.Unlock()

	c.refreshMu.Lock()
	delete(c.keyspaceLocks, key)
	delete(c.tableLocks, key)
	delete(c.columnLocks, key)
	delete(c.typeLocks, key)
	delete(c.indexLocks, key)
	delete(c.viewLocks, key)
	delete(c.triggerLocks, key)
	c.refreshMu.Unlock()
}

// InvalidateAll clears cached metadata for all users.
func (c *CQLMetaCache) InvalidateAll() {
	if c == nil {
		return
	}
	c.mu.Lock()
	c.keyspaces = make(map[userKey]cqlKeyspaceCacheEntry)
	c.tables = make(map[userKey]cqlTableCacheEntry)
	c.columns = make(map[userKey]cqlColumnCacheEntry)
	c.types = make(map[userKey]cqlTypeCacheEntry)
	c.indexes = make(map[userKey]cqlIndexCacheEntry)
	c.views = make(map[userKey]cqlViewCacheEntry)
	c.triggers = make(map[userKey]cqlTriggerCacheEntry)
	c.mu.Unlock()

	c.refreshMu.Lock()
	c.keyspaceLocks = nil
	c.tableLocks = nil
	c.columnLocks = nil
	c.typeLocks = nil
	c.indexLocks = nil
	c.viewLocks = nil
	c.triggerLocks = nil
	c.refreshMu.Unlock()
}

// Keyspaces returns cached keyspace metadata.
func (c *CQLMetaCache) Keyspaces(ctx context.Context, pm *PoolManager, user, password string) (cqlKeyspaceCacheEntry, error) {
	if c == nil {
		return cqlKeyspaceCacheEntry{}, fmt.Errorf("cql meta cache not initialized")
	}
	key := userKey{user: user, password: password}
	now := time.Now()

	c.mu.RLock()
	entry, ok := c.keyspaces[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	lock := c.keyspaceLock(key)
	if lock != nil {
		lock.Lock()
		defer lock.Unlock()
	}

	now = time.Now()
	c.mu.RLock()
	entry, ok = c.keyspaces[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	start := time.Now()
	list, available, err := fetchCQLKeyspaces(ctx, pm, user, password)
	if err != nil {
		return cqlKeyspaceCacheEntry{}, err
	}

	entry = cqlKeyspaceCacheEntry{
		expires:   now.Add(c.ttl),
		list:      list,
		available: available,
	}

	metrics.ObserveStageFromContext(ctx, metrics.StageCQLMetaCache, time.Since(start))

	c.mu.Lock()
	c.keyspaces[key] = entry
	c.mu.Unlock()

	return entry, nil
}

// Tables returns cached table metadata.
func (c *CQLMetaCache) Tables(ctx context.Context, pm *PoolManager, user, password string) (cqlTableCacheEntry, error) {
	if c == nil {
		return cqlTableCacheEntry{}, fmt.Errorf("cql meta cache not initialized")
	}
	key := userKey{user: user, password: password}
	now := time.Now()

	c.mu.RLock()
	entry, ok := c.tables[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	lock := c.tableLock(key)
	if lock != nil {
		lock.Lock()
		defer lock.Unlock()
	}

	now = time.Now()
	c.mu.RLock()
	entry, ok = c.tables[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	start := time.Now()
	list, available, err := fetchCQLTables(ctx, pm, user, password)
	if err != nil {
		return cqlTableCacheEntry{}, err
	}
	bySchema := make(map[string][]CQLTableInfo)
	bySchemaLower := make(map[string][]CQLTableInfo)
	for _, tbl := range list {
		bySchema[tbl.Keyspace] = append(bySchema[tbl.Keyspace], tbl)
		bySchemaLower[normalizeLookup(tbl.Keyspace)] = append(bySchemaLower[normalizeLookup(tbl.Keyspace)], tbl)
	}

	entry = cqlTableCacheEntry{
		expires:       now.Add(c.ttl),
		list:          list,
		bySchema:      bySchema,
		bySchemaLower: bySchemaLower,
		available:     available,
	}

	metrics.ObserveStageFromContext(ctx, metrics.StageCQLMetaCache, time.Since(start))

	c.mu.Lock()
	c.tables[key] = entry
	c.mu.Unlock()

	return entry, nil
}

// Columns returns cached column metadata.
func (c *CQLMetaCache) Columns(ctx context.Context, pm *PoolManager, user, password string) (cqlColumnCacheEntry, error) {
	if c == nil {
		return cqlColumnCacheEntry{}, fmt.Errorf("cql meta cache not initialized")
	}
	key := userKey{user: user, password: password}
	now := time.Now()

	c.mu.RLock()
	entry, ok := c.columns[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	lock := c.columnLock(key)
	if lock != nil {
		lock.Lock()
		defer lock.Unlock()
	}

	now = time.Now()
	c.mu.RLock()
	entry, ok = c.columns[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	start := time.Now()
	list, available, err := fetchCQLColumns(ctx, pm, user, password)
	if err != nil {
		return cqlColumnCacheEntry{}, err
	}
	byTable := make(map[tableKey][]CQLColumnInfo)
	byTableLower := make(map[tableKey][]CQLColumnInfo)
	for _, col := range list {
		k := tableKey{schema: col.Keyspace, table: col.Table}
		byTable[k] = append(byTable[k], col)
		kLower := normalizeTableKey(col.Keyspace, col.Table)
		byTableLower[kLower] = append(byTableLower[kLower], col)
	}

	entry = cqlColumnCacheEntry{
		expires:      now.Add(c.ttl),
		list:         list,
		byTable:      byTable,
		byTableLower: byTableLower,
		available:    available,
	}

	metrics.ObserveStageFromContext(ctx, metrics.StageCQLMetaCache, time.Since(start))

	c.mu.Lock()
	c.columns[key] = entry
	c.mu.Unlock()

	return entry, nil
}

// Types returns cached UDT metadata.
func (c *CQLMetaCache) Types(ctx context.Context, pm *PoolManager, user, password string) (cqlTypeCacheEntry, error) {
	if c == nil {
		return cqlTypeCacheEntry{}, fmt.Errorf("cql meta cache not initialized")
	}
	key := userKey{user: user, password: password}
	now := time.Now()

	c.mu.RLock()
	entry, ok := c.types[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	lock := c.typeLock(key)
	if lock != nil {
		lock.Lock()
		defer lock.Unlock()
	}

	now = time.Now()
	c.mu.RLock()
	entry, ok = c.types[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	start := time.Now()
	list, available, err := fetchCQLTypes(ctx, pm, user, password)
	if err != nil {
		return cqlTypeCacheEntry{}, err
	}
	byName := make(map[typeKey]CQLTypeInfo)
	byNameLower := make(map[typeKey]CQLTypeInfo)
	for _, typ := range list {
		byName[typeKey{schema: typ.Keyspace, name: typ.Name}] = typ
		keyLower := normalizeTypeKey(typ.Keyspace, typ.Name)
		if _, ok := byNameLower[keyLower]; !ok {
			byNameLower[keyLower] = typ
		}
	}

	entry = cqlTypeCacheEntry{
		expires:     now.Add(c.ttl),
		list:        list,
		byName:      byName,
		byNameLower: byNameLower,
		available:   available,
	}

	metrics.ObserveStageFromContext(ctx, metrics.StageCQLMetaCache, time.Since(start))

	c.mu.Lock()
	c.types[key] = entry
	c.mu.Unlock()

	return entry, nil
}

// Indexes returns cached index metadata.
func (c *CQLMetaCache) Indexes(ctx context.Context, pm *PoolManager, user, password string) (cqlIndexCacheEntry, error) {
	if c == nil {
		return cqlIndexCacheEntry{}, fmt.Errorf("cql meta cache not initialized")
	}
	key := userKey{user: user, password: password}
	now := time.Now()

	c.mu.RLock()
	entry, ok := c.indexes[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	lock := c.indexLock(key)
	if lock != nil {
		lock.Lock()
		defer lock.Unlock()
	}

	now = time.Now()
	c.mu.RLock()
	entry, ok = c.indexes[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	start := time.Now()
	list, available, err := fetchCQLIndexes(ctx, pm, user, password)
	if err != nil {
		return cqlIndexCacheEntry{}, err
	}
	byTable := make(map[tableKey][]CQLIndexInfo)
	byTableLower := make(map[tableKey][]CQLIndexInfo)
	for _, idx := range list {
		key := tableKey{schema: idx.Keyspace, table: idx.Table}
		byTable[key] = append(byTable[key], idx)
		keyLower := normalizeTableKey(idx.Keyspace, idx.Table)
		byTableLower[keyLower] = append(byTableLower[keyLower], idx)
	}

	entry = cqlIndexCacheEntry{
		expires:      now.Add(c.ttl),
		list:         list,
		byTable:      byTable,
		byTableLower: byTableLower,
		available:    available,
	}

	metrics.ObserveStageFromContext(ctx, metrics.StageCQLMetaCache, time.Since(start))

	c.mu.Lock()
	c.indexes[key] = entry
	c.mu.Unlock()

	return entry, nil
}

// Views returns cached view metadata.
func (c *CQLMetaCache) Views(ctx context.Context, pm *PoolManager, user, password string) (cqlViewCacheEntry, error) {
	if c == nil {
		return cqlViewCacheEntry{}, fmt.Errorf("cql meta cache not initialized")
	}
	key := userKey{user: user, password: password}
	now := time.Now()

	c.mu.RLock()
	entry, ok := c.views[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	lock := c.viewLock(key)
	if lock != nil {
		lock.Lock()
		defer lock.Unlock()
	}

	now = time.Now()
	c.mu.RLock()
	entry, ok = c.views[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	start := time.Now()
	list, available, err := fetchCQLViews(ctx, pm, user, password)
	if err != nil {
		return cqlViewCacheEntry{}, err
	}
	byName := make(map[viewKey]CQLViewInfo)
	byNameLower := make(map[viewKey]CQLViewInfo)
	for _, view := range list {
		byName[viewKey{schema: view.Keyspace, name: view.Name}] = view
		keyLower := normalizeViewKey(view.Keyspace, view.Name)
		if _, ok := byNameLower[keyLower]; !ok {
			byNameLower[keyLower] = view
		}
	}

	entry = cqlViewCacheEntry{
		expires:     now.Add(c.ttl),
		list:        list,
		byName:      byName,
		byNameLower: byNameLower,
		available:   available,
	}

	metrics.ObserveStageFromContext(ctx, metrics.StageCQLMetaCache, time.Since(start))

	c.mu.Lock()
	c.views[key] = entry
	c.mu.Unlock()

	return entry, nil
}

// Triggers returns cached trigger metadata.
func (c *CQLMetaCache) Triggers(ctx context.Context, pm *PoolManager, user, password string) (cqlTriggerCacheEntry, error) {
	if c == nil {
		return cqlTriggerCacheEntry{}, fmt.Errorf("cql meta cache not initialized")
	}
	key := userKey{user: user, password: password}
	now := time.Now()

	c.mu.RLock()
	entry, ok := c.triggers[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	lock := c.triggerLock(key)
	if lock != nil {
		lock.Lock()
		defer lock.Unlock()
	}

	now = time.Now()
	c.mu.RLock()
	entry, ok = c.triggers[key]
	c.mu.RUnlock()
	if ok && now.Before(entry.expires) {
		return entry, nil
	}

	start := time.Now()
	list, available, err := fetchCQLTriggers(ctx, pm, user, password)
	if err != nil {
		return cqlTriggerCacheEntry{}, err
	}
	byName := make(map[triggerKey]CQLTriggerInfo)
	byNameLower := make(map[triggerKey]CQLTriggerInfo)
	for _, trigger := range list {
		byName[triggerKey{schema: trigger.Keyspace, table: trigger.Table, name: trigger.Name}] = trigger
		keyLower := normalizeTriggerKey(trigger.Keyspace, trigger.Table, trigger.Name)
		if _, ok := byNameLower[keyLower]; !ok {
			byNameLower[keyLower] = trigger
		}
	}

	entry = cqlTriggerCacheEntry{
		expires:     now.Add(c.ttl),
		list:        list,
		byName:      byName,
		byNameLower: byNameLower,
		available:   available,
	}

	metrics.ObserveStageFromContext(ctx, metrics.StageCQLMetaCache, time.Since(start))

	c.mu.Lock()
	c.triggers[key] = entry
	c.mu.Unlock()

	return entry, nil
}

type typeKey struct {
	schema string
	name   string
}

type viewKey struct {
	schema string
	name   string
}

type triggerKey struct {
	schema string
	table  string
	name   string
}

func normalizeTypeKey(schema, name string) typeKey {
	return typeKey{
		schema: normalizeLookup(schema),
		name:   normalizeLookup(name),
	}
}

func normalizeViewKey(schema, name string) viewKey {
	return viewKey{
		schema: normalizeLookup(schema),
		name:   normalizeLookup(name),
	}
}

func normalizeTriggerKey(schema, table, name string) triggerKey {
	return triggerKey{
		schema: normalizeLookup(schema),
		table:  normalizeLookup(table),
		name:   normalizeLookup(name),
	}
}

func fetchCQLKeyspaces(ctx context.Context, pm *PoolManager, user, password string) ([]CQLKeyspaceInfo, bool, error) {
	conn, release, err := getConn(ctx, pm, user, password)
	if err != nil {
		return nil, false, err
	}
	if release != nil {
		defer release()
	}

	rows, err := queryContextWithMetrics(ctx, conn, metrics.MySQLOpQueryRead, `
		SELECT keyspace_name, durable_writes, replication_json
		FROM cql_meta.keyspaces
		ORDER BY keyspace_name
	`)
	if err != nil {
		if isCQLMetaUnavailable(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("failed to query cql_meta.keyspaces: %w", err)
	}
	defer rows.Close()

	list := make([]CQLKeyspaceInfo, 0)
	for rows.Next() {
		var info CQLKeyspaceInfo
		var replication []byte
		if err := rows.Scan(&info.Name, &info.DurableWrites, &replication); err != nil {
			return nil, false, fmt.Errorf("failed to scan cql_meta.keyspaces: %w", err)
		}
		info.ReplicationJSON = cloneJSONBytes(replication)
		list = append(list, info)
	}
	if err := rows.Err(); err != nil {
		return nil, false, fmt.Errorf("failed to read cql_meta.keyspaces: %w", err)
	}

	return list, true, nil
}

func fetchCQLTables(ctx context.Context, pm *PoolManager, user, password string) ([]CQLTableInfo, bool, error) {
	conn, release, err := getConn(ctx, pm, user, password)
	if err != nil {
		return nil, false, err
	}
	if release != nil {
		defer release()
	}

	rows, err := queryContextWithMetrics(ctx, conn, metrics.MySQLOpQueryRead, `
		SELECT keyspace_name, table_name, options_json, comment
		FROM cql_meta.tables
		ORDER BY keyspace_name, table_name
	`)
	if err != nil {
		if isCQLMetaUnavailable(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("failed to query cql_meta.tables: %w", err)
	}
	defer rows.Close()

	list := make([]CQLTableInfo, 0)
	for rows.Next() {
		var info CQLTableInfo
		var options []byte
		if err := rows.Scan(&info.Keyspace, &info.Table, &options, &info.Comment); err != nil {
			return nil, false, fmt.Errorf("failed to scan cql_meta.tables: %w", err)
		}
		info.OptionsJSON = cloneJSONBytes(options)
		list = append(list, info)
	}
	if err := rows.Err(); err != nil {
		return nil, false, fmt.Errorf("failed to read cql_meta.tables: %w", err)
	}

	return list, true, nil
}

func fetchCQLColumns(ctx context.Context, pm *PoolManager, user, password string) ([]CQLColumnInfo, bool, error) {
	conn, rows, release, err := queryMetadataContextWithRetry(ctx, pm, user, password, `
		SELECT keyspace_name, table_name, column_name, cql_type, default_expr, on_update_expr, kind, position, clustering_order
		FROM cql_meta.columns
		ORDER BY keyspace_name, table_name, position
	`)
	defer releaseRows(rows, release)
	if err != nil {
		if isCQLMetaUnavailable(err) {
			return nil, false, nil
		}
		if isCQLMetaUnknownColumn(err) {
			return fetchCQLColumnsLegacy(ctx, conn)
		}
		return nil, false, fmt.Errorf("failed to query cql_meta.columns: %w", err)
	}

	list := make([]CQLColumnInfo, 0)
	for rows.Next() {
		var info CQLColumnInfo
		var defaultExpr sql.NullString
		var onUpdateExpr sql.NullString
		if err := rows.Scan(&info.Keyspace, &info.Table, &info.Name, &info.CQLType, &defaultExpr, &onUpdateExpr, &info.Kind, &info.Position, &info.ClusteringOrder); err != nil {
			return nil, false, fmt.Errorf("failed to scan cql_meta.columns: %w", err)
		}
		info.DefaultExpr = nullableStringValue(defaultExpr)
		info.OnUpdateExpr = nullableStringValue(onUpdateExpr)
		list = append(list, info)
	}
	if err := rows.Err(); err != nil {
		return nil, false, fmt.Errorf("failed to read cql_meta.columns: %w", err)
	}

	return list, true, nil
}

func fetchCQLColumnsLegacy(ctx context.Context, conn *sql.Conn) ([]CQLColumnInfo, bool, error) {
	rows, err := queryContextWithMetrics(ctx, conn, metrics.MySQLOpQueryRead, `
		SELECT keyspace_name, table_name, column_name, cql_type, kind, position, clustering_order
		FROM cql_meta.columns
		ORDER BY keyspace_name, table_name, position
	`)
	if err != nil {
		if isCQLMetaUnavailable(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("failed to query legacy cql_meta.columns: %w", err)
	}
	defer rows.Close()

	list := make([]CQLColumnInfo, 0)
	for rows.Next() {
		var info CQLColumnInfo
		if err := rows.Scan(&info.Keyspace, &info.Table, &info.Name, &info.CQLType, &info.Kind, &info.Position, &info.ClusteringOrder); err != nil {
			return nil, false, fmt.Errorf("failed to scan legacy cql_meta.columns: %w", err)
		}
		list = append(list, info)
	}
	if err := rows.Err(); err != nil {
		return nil, false, fmt.Errorf("failed to read legacy cql_meta.columns: %w", err)
	}
	return list, true, nil
}

func nullableStringValue(value sql.NullString) string {
	if !value.Valid {
		return ""
	}
	return value.String
}

func fetchCQLTypes(ctx context.Context, pm *PoolManager, user, password string) ([]CQLTypeInfo, bool, error) {
	conn, release, err := getConn(ctx, pm, user, password)
	if err != nil {
		return nil, false, err
	}
	if release != nil {
		defer release()
	}

	rows, err := queryContextWithMetrics(ctx, conn, metrics.MySQLOpQueryRead, `
		SELECT keyspace_name, type_name, field_names_json, field_types_json
		FROM cql_meta.types
		ORDER BY keyspace_name, type_name
	`)
	if err != nil {
		if isCQLMetaUnavailable(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("failed to query cql_meta.types: %w", err)
	}
	defer rows.Close()

	list := make([]CQLTypeInfo, 0)
	for rows.Next() {
		var info CQLTypeInfo
		var rawNames []byte
		var rawTypes []byte
		if err := rows.Scan(&info.Keyspace, &info.Name, &rawNames, &rawTypes); err != nil {
			return nil, false, fmt.Errorf("failed to scan cql_meta.types: %w", err)
		}
		names, err := parseJSONStringArray(rawNames)
		if err != nil {
			return nil, false, fmt.Errorf("failed to decode cql_meta.types field names: %w", err)
		}
		types, err := parseJSONStringArray(rawTypes)
		if err != nil {
			return nil, false, fmt.Errorf("failed to decode cql_meta.types field types: %w", err)
		}
		info.FieldNames = names
		info.FieldTypes = types
		list = append(list, info)
	}
	if err := rows.Err(); err != nil {
		return nil, false, fmt.Errorf("failed to read cql_meta.types: %w", err)
	}

	return list, true, nil
}

func fetchCQLIndexes(ctx context.Context, pm *PoolManager, user, password string) ([]CQLIndexInfo, bool, error) {
	conn, release, err := getConn(ctx, pm, user, password)
	if err != nil {
		return nil, false, err
	}
	if release != nil {
		defer release()
	}

	rows, err := queryContextWithMetrics(ctx, conn, metrics.MySQLOpQueryRead, `
		SELECT keyspace_name, table_name, index_name, kind, options_json
		FROM cql_meta.indexes
		ORDER BY keyspace_name, table_name, index_name
	`)
	if err != nil {
		if isCQLMetaUnavailable(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("failed to query cql_meta.indexes: %w", err)
	}
	defer rows.Close()

	list := make([]CQLIndexInfo, 0)
	for rows.Next() {
		var info CQLIndexInfo
		var options []byte
		if err := rows.Scan(&info.Keyspace, &info.Table, &info.Name, &info.Kind, &options); err != nil {
			return nil, false, fmt.Errorf("failed to scan cql_meta.indexes: %w", err)
		}
		info.OptionsJSON = cloneJSONBytes(options)
		list = append(list, info)
	}
	if err := rows.Err(); err != nil {
		return nil, false, fmt.Errorf("failed to read cql_meta.indexes: %w", err)
	}

	return list, true, nil
}

func fetchCQLViews(ctx context.Context, pm *PoolManager, user, password string) ([]CQLViewInfo, bool, error) {
	conn, release, err := getConn(ctx, pm, user, password)
	if err != nil {
		return nil, false, err
	}
	if release != nil {
		defer release()
	}

	rows, err := queryContextWithMetrics(ctx, conn, metrics.MySQLOpQueryRead, `
		SELECT keyspace_name, view_name, base_table, where_clause, options_json
		FROM cql_meta.views
		ORDER BY keyspace_name, view_name
	`)
	if err != nil {
		if isCQLMetaUnavailable(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("failed to query cql_meta.views: %w", err)
	}
	defer rows.Close()

	list := make([]CQLViewInfo, 0)
	for rows.Next() {
		var info CQLViewInfo
		var options []byte
		if err := rows.Scan(&info.Keyspace, &info.Name, &info.BaseTable, &info.WhereClause, &options); err != nil {
			return nil, false, fmt.Errorf("failed to scan cql_meta.views: %w", err)
		}
		info.OptionsJSON = cloneJSONBytes(options)
		list = append(list, info)
	}
	if err := rows.Err(); err != nil {
		return nil, false, fmt.Errorf("failed to read cql_meta.views: %w", err)
	}

	return list, true, nil
}

func fetchCQLTriggers(ctx context.Context, pm *PoolManager, user, password string) ([]CQLTriggerInfo, bool, error) {
	conn, release, err := getConn(ctx, pm, user, password)
	if err != nil {
		return nil, false, err
	}
	if release != nil {
		defer release()
	}

	rows, err := queryContextWithMetrics(ctx, conn, metrics.MySQLOpQueryRead, `
		SELECT keyspace_name, table_name, trigger_name, options_json
		FROM cql_meta.triggers
		ORDER BY keyspace_name, table_name, trigger_name
	`)
	if err != nil {
		if isCQLMetaUnavailable(err) {
			return nil, false, nil
		}
		return nil, false, fmt.Errorf("failed to query cql_meta.triggers: %w", err)
	}
	defer rows.Close()

	list := make([]CQLTriggerInfo, 0)
	for rows.Next() {
		var info CQLTriggerInfo
		var options []byte
		if err := rows.Scan(&info.Keyspace, &info.Table, &info.Name, &options); err != nil {
			return nil, false, fmt.Errorf("failed to scan cql_meta.triggers: %w", err)
		}
		info.OptionsJSON = cloneJSONBytes(options)
		list = append(list, info)
	}
	if err := rows.Err(); err != nil {
		return nil, false, fmt.Errorf("failed to read cql_meta.triggers: %w", err)
	}

	return list, true, nil
}

func cloneJSONBytes(raw []byte) []byte {
	if len(raw) == 0 {
		return []byte("{}")
	}
	out := make([]byte, len(raw))
	copy(out, raw)
	return out
}

func parseJSONStringArray(raw []byte) ([]string, error) {
	if len(raw) == 0 {
		return []string{}, nil
	}
	var out []string
	if err := json.Unmarshal(raw, &out); err != nil {
		return nil, err
	}
	return out, nil
}

func isCQLMetaMissing(err error) bool {
	var myErr *mysql.MySQLError
	if errors.As(err, &myErr) {
		switch myErr.Number {
		case 1049, 1146: // Unknown database / table
			return true
		}
	}
	return false
}

func isCQLMetaUnavailable(err error) bool {
	return isCQLMetaMissing(err) || IsAccessDenied(err)
}

// IsAccessDenied reports whether an error is a MySQL/TiDB privilege denial.
func IsAccessDenied(err error) bool {
	var myErr *mysql.MySQLError
	if errors.As(err, &myErr) {
		switch myErr.Number {
		case 1044, 1142: // Database access denied / command denied
			return true
		}
	}
	return false
}

func isCQLMetaUnknownColumn(err error) bool {
	var myErr *mysql.MySQLError
	if errors.As(err, &myErr) {
		return myErr.Number == 1054
	}
	return false
}