package mysqlbackend

import (
	"context"
	"fmt"
	"sort"
	"sync"
	"time"
)

// TableInfo describes a MySQL table.
type TableInfo struct {
	Schema  string
	Name    string
	Comment string
}

// ColumnInfo describes a MySQL column.
type ColumnInfo struct {
	Schema           string
	Table            string
	Name             string
	DataType         string
	NumericPrecision int
	NumericScale     int
	Ordinal          int
}

// IndexInfo describes a MySQL index column.
type IndexInfo struct {
	Schema    string
	Table     string
	Name      string
	Column    string
	Seq       int
	NonUnique bool
	Collation string
	IsPrimary bool
}

// UserInfo describes a TiDB user account visible in mysql.user.
type UserInfo struct {
	Name string
}

// SchemaManager provides cached access to information_schema metadata.
type SchemaManager struct {
	poolManager *PoolManager
	cache       *SchemaCache
	metaCache   *CQLMetaCache
	metaMu      sync.Mutex
	metaEnsured map[userKey]struct{}
}

// NewSchemaManager creates a SchemaManager with configurable TTL caches.
func NewSchemaManager(poolManager *PoolManager, schemaTTL, metaTTL time.Duration) *SchemaManager {
	return &SchemaManager{
		poolManager: poolManager,
		cache:       NewSchemaCache(schemaTTL),
		metaCache:   NewCQLMetaCache(metaTTL),
		metaEnsured: make(map[userKey]struct{}),
	}
}

// Invalidate clears cached schema metadata for the user.
func (m *SchemaManager) Invalidate(user, password string) {
	if m == nil || m.cache == nil {
		return
	}
	m.cache.InvalidateUser(user, password)
	if m.metaCache != nil {
		m.metaCache.InvalidateUser(user, password)
	}
}

// InvalidateAll clears cached schema metadata for all users.
func (m *SchemaManager) InvalidateAll() {
	if m == nil || m.cache == nil {
		return
	}
	m.cache.InvalidateAll()
	if m.metaCache != nil {
		m.metaCache.InvalidateAll()
	}
}

// ListSchemas returns all schemas visible to the user.
func (m *SchemaManager) ListSchemas(ctx context.Context, user, password string) ([]string, error) {
	if m == nil || m.cache == nil {
		return nil, fmt.Errorf("schema manager not initialized")
	}
	entry, err := m.cache.Schemas(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, err
	}
	return entry.list, nil
}

// SchemaExists reports whether a schema exists.
func (m *SchemaManager) SchemaExists(ctx context.Context, user, password, schema string) (bool, error) {
	entry, err := m.cache.Schemas(ctx, m.poolManager, user, password)
	if err != nil {
		return false, err
	}
	if _, ok := entry.set[schema]; ok {
		return true, nil
	}
	_, ok := entry.lowerSet[normalizeLookup(schema)]
	return ok, nil
}

// ListTables returns all base tables visible to the user.
func (m *SchemaManager) ListTables(ctx context.Context, user, password string) ([]TableInfo, error) {
	entry, err := m.cache.Tables(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, err
	}
	return entry.list, nil
}

// TableExists reports whether a table exists.
func (m *SchemaManager) TableExists(ctx context.Context, user, password, schema, table string) (bool, error) {
	entry, err := m.cache.Tables(ctx, m.poolManager, user, password)
	if err != nil {
		return false, err
	}
	if _, ok := entry.tableSet[tableKey{schema: schema, table: table}]; ok {
		return true, nil
	}
	_, ok := entry.tableSetLower[normalizeTableKey(schema, table)]
	return ok, nil
}

// ResolveTable returns the table info for a schema/table, matching case-insensitively when needed.
func (m *SchemaManager) ResolveTable(ctx context.Context, user, password, schema, table string) (TableInfo, bool, error) {
	entry, err := m.cache.Tables(ctx, m.poolManager, user, password)
	if err != nil {
		return TableInfo{}, false, err
	}
	if tables, ok := entry.bySchema[schema]; ok {
		for _, tbl := range tables {
			if tbl.Name == table {
				return tbl, true, nil
			}
		}
	}
	keyLower := normalizeTableKey(schema, table)
	if _, ok := entry.tableLowerAmbiguous[keyLower]; ok {
		return TableInfo{}, false, nil
	}
	if info, ok := entry.tableByLower[keyLower]; ok {
		return info, true, nil
	}
	normSchema := normalizeLookup(schema)
	normTable := normalizeLookup(table)
	var match TableInfo
	matched := false
	for _, tbl := range entry.list {
		if normalizeLookup(tbl.Schema) != normSchema {
			continue
		}
		if normalizeLookup(tbl.Name) != normTable {
			continue
		}
		if matched {
			return TableInfo{}, false, nil
		}
		match = tbl
		matched = true
	}
	if matched {
		return match, true, nil
	}
	return TableInfo{}, false, nil
}

// ListColumns returns all columns visible to the user.
func (m *SchemaManager) ListColumns(ctx context.Context, user, password string) ([]ColumnInfo, error) {
	entry, err := m.cache.Columns(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, err
	}
	return entry.list, nil
}

// ColumnsForTable returns columns for a specific table ordered by ordinal position.
func (m *SchemaManager) ColumnsForTable(ctx context.Context, user, password, schema, table string) ([]ColumnInfo, error) {
	entry, err := m.cache.Columns(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, err
	}
	key := tableKey{schema: schema, table: table}
	if cols, ok := entry.byTable[key]; ok {
		return cols, nil
	}
	return entry.byTableLower[normalizeTableKey(schema, table)], nil
}

// ColumnExists reports whether a column exists in a table.
func (m *SchemaManager) ColumnExists(ctx context.Context, user, password, schema, table, column string) (bool, error) {
	entry, err := m.cache.Columns(ctx, m.poolManager, user, password)
	if err != nil {
		return false, err
	}
	key := tableKey{schema: schema, table: table}
	cols, ok := entry.byTable[key]
	if !ok {
		cols = entry.byTableLower[normalizeTableKey(schema, table)]
	}
	for i := range cols {
		if cols[i].Name == column {
			return true, nil
		}
	}
	return false, nil
}

// ListIndexes returns all indexes visible to the user.
func (m *SchemaManager) ListIndexes(ctx context.Context, user, password string) ([]IndexInfo, error) {
	entry, err := m.cache.Indexes(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, err
	}
	return entry.list, nil
}

// ListUsers returns distinct TiDB user names visible to the user.
func (m *SchemaManager) ListUsers(ctx context.Context, user, password string) ([]UserInfo, error) {
	if m == nil || m.poolManager == nil {
		return nil, fmt.Errorf("schema manager not initialized")
	}
	conn, release, err := getConn(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, err
	}
	if release != nil {
		defer release()
	}
	rows, err := conn.QueryContext(ctx, "SELECT User FROM mysql.user WHERE User <> '' GROUP BY User ORDER BY User")
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	users := make([]UserInfo, 0)
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return nil, err
		}
		users = append(users, UserInfo{Name: name})
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return users, nil
}

// IndexesForTable returns all index columns for a table.
func (m *SchemaManager) IndexesForTable(ctx context.Context, user, password, schema, table string) ([]IndexInfo, error) {
	entry, err := m.cache.Indexes(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, err
	}
	key := tableKey{schema: schema, table: table}
	if indexes, ok := entry.byTable[key]; ok {
		return indexes, nil
	}
	return entry.byTableLower[normalizeTableKey(schema, table)], nil
}

// PrimaryKeyColumns returns primary key columns ordered by sequence.
func (m *SchemaManager) PrimaryKeyColumns(ctx context.Context, user, password, schema, table string) ([]IndexInfo, error) {
	indexes, err := m.IndexesForTable(ctx, user, password, schema, table)
	if err != nil {
		return nil, err
	}
	pk := make([]IndexInfo, 0)
	for _, idx := range indexes {
		if idx.IsPrimary {
			pk = append(pk, idx)
		}
	}
	sort.Slice(pk, func(i, j int) bool {
		return pk[i].Seq < pk[j].Seq
	})
	return pk, nil
}

// IndexExists reports whether an index exists on a table.
func (m *SchemaManager) IndexExists(ctx context.Context, user, password, schema, table, index string) (bool, error) {
	indexes, err := m.IndexesForTable(ctx, user, password, schema, table)
	if err != nil {
		return false, err
	}
	lookup := normalizeLookup(index)
	for _, idx := range indexes {
		if idx.Name == index || normalizeLookup(idx.Name) == lookup {
			return true, nil
		}
	}
	return false, nil
}

// FindIndexesByName returns indexes matching schema and index name.
func (m *SchemaManager) FindIndexesByName(ctx context.Context, user, password, schema, index string) ([]IndexInfo, error) {
	entry, err := m.cache.Indexes(ctx, m.poolManager, user, password)
	if err != nil {
		return nil, err
	}
	if indexes, ok := entry.byName[indexKey{schema: schema, name: index}]; ok {
		return indexes, nil
	}
	return entry.byNameLower[normalizeIndexKey(schema, index)], nil
}