package systemtables

import (
	"fmt"

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

// SystemTable represents a system table definition
//
// PartitionKeys and ClusteringKeys are used for system_schema metadata.
type SystemTable struct {
	Name            string             // Full name (e.g., "system.local")
	Keyspace        string             // Keyspace name (e.g., "system")
	Table           string             // Table name (e.g., "local")
	Columns         []types.ColumnInfo // Column schema
	PartitionKeys   []string           // Primary key partition columns
	ClusteringKeys  []string           // Primary key clustering columns
	ClusteringOrder map[string]string  // Column -> ASC/DESC
	Provider        DataProvider       // Data provider function
	Comment         string             // Table description
	IsReadOnly      bool               // Always true for system tables
}

// Registry manages all system table definitions
type Registry struct {
	tables map[string]*SystemTable
	state  *SystemTableState
}

// NewRegistry creates and initializes a system table registry
func NewRegistry() *Registry {
	state, err := NewSystemTableState(config.GetConfig(), nil)
	if err != nil {
		panic(fmt.Sprintf("system table state init failed: %v", err))
	}
	return NewRegistryWithState(state)
}

// NewRegistryWithState creates a registry with an explicit state instance.
func NewRegistryWithState(state *SystemTableState) *Registry {
	r := &Registry{
		tables: make(map[string]*SystemTable),
		state:  state,
	}
	r.registerBuiltins()
	r.registerDSESecurityTables()
	return r
}

// Register adds a system table to the registry
func (r *Registry) Register(st *SystemTable) {
	r.tables[st.Name] = st
	r.tables[fmt.Sprintf("%s.%s", st.Keyspace, st.Table)] = st
}

// Lookup retrieves a system table by name
func (r *Registry) Lookup(name string) (*SystemTable, bool) {
	st, ok := r.tables[name]
	return st, ok
}

// ListAll returns all registered system tables
func (r *Registry) ListAll() []*SystemTable {
	tables := make([]*SystemTable, 0, len(r.tables)/2)
	seen := make(map[string]bool)
	for _, st := range r.tables {
		if !seen[st.Name] {
			tables = append(tables, st)
			seen[st.Name] = true
		}
	}
	return tables
}

func (r *Registry) registerBuiltins() {
	providers := r.providerFactories()
	for _, def := range builtinSystemTableDefs {
		columns := make([]types.ColumnInfo, len(def.Columns))
		for i, col := range def.Columns {
			columns[i] = types.ColumnInfo{
				Keyspace: def.Keyspace,
				Table:    def.Table,
				Name:     col.Name,
				Type:     col.Type,
			}
		}

		fullName := fmt.Sprintf("%s.%s", def.Keyspace, def.Table)
		provider := provideEmpty
		if factory, ok := providers[fullName]; ok {
			provider = factory(columns)
		}

		r.Register(&SystemTable{
			Name:            fullName,
			Keyspace:        def.Keyspace,
			Table:           def.Table,
			Columns:         columns,
			PartitionKeys:   def.PartitionKeys,
			ClusteringKeys:  def.ClusteringKeys,
			ClusteringOrder: def.ClusteringOrder,
			Provider:        provider,
			Comment:         "",
			IsReadOnly:      true,
		})
	}
}

// registerDSESecurityTables registers dse_security keyspace tables
func (r *Registry) registerDSESecurityTables() {
	// 1. role_options - Role options
	r.Register(&SystemTable{
		Name:     "dse_security.role_options",
		Keyspace: "dse_security",
		Table:    "role_options",
		Columns: []types.ColumnInfo{
			{Keyspace: "dse_security", Table: "role_options", Name: "role", Type: "text"},
			{Keyspace: "dse_security", Table: "role_options", Name: "options", Type: "frozen<map<text, text>>"},
		},
		Provider:   provideDSERoleOptions,
		Comment:    "Role options",
		IsReadOnly: true,
	})

	// 2. digest_tokens - Kerberos digest tokens
	r.Register(&SystemTable{
		Name:     "dse_security.digest_tokens",
		Keyspace: "dse_security",
		Table:    "digest_tokens",
		Columns: []types.ColumnInfo{
			{Keyspace: "dse_security", Table: "digest_tokens", Name: "id", Type: "uuid"},
			{Keyspace: "dse_security", Table: "digest_tokens", Name: "password", Type: "text"},
		},
		Provider:   provideDSEDigestTokens,
		Comment:    "Kerberos digest tokens",
		IsReadOnly: true,
	})

	// 3. spark_security - Spark shared secret
	r.Register(&SystemTable{
		Name:     "dse_security.spark_security",
		Keyspace: "dse_security",
		Table:    "spark_security",
		Columns: []types.ColumnInfo{
			{Keyspace: "dse_security", Table: "spark_security", Name: "dc", Type: "text"},
			{Keyspace: "dse_security", Table: "spark_security", Name: "shared_secret", Type: "text"},
		},
		Provider:   provideDSPSparkSecurity,
		Comment:    "Spark shared secret",
		IsReadOnly: true,
	})
}