package systemtables

import (
	"context"
	"fmt"
	"sort"

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

// QueryOptions contains options for system table queries
type QueryOptions struct {
	WhereClause interface{} // Parsed WHERE clause (from parser.Expression)
	Limit       *int64
	Offset      *int64
}

// DataProvider is a function that generates rows for a system table
type DataProvider func(ctx context.Context, opts QueryOptions) ([]types.Row, error)

type providerFactory func(columns []types.ColumnInfo) DataProvider

func buildRow(columns []types.ColumnInfo, values map[string]interface{}) types.Row {
	row := types.Row{Values: make([]interface{}, len(columns))}
	for i, col := range columns {
		if val, ok := values[col.Name]; ok {
			row.Values[i] = val
		}
	}
	return row
}

func buildRows(columns []types.ColumnInfo, values []map[string]interface{}) []types.Row {
	rows := make([]types.Row, 0, len(values))
	for _, row := range values {
		rows = append(rows, buildRow(columns, row))
	}
	return rows
}

func columnIndexMap(columns []types.ColumnInfo) map[string]int {
	index := make(map[string]int, len(columns))
	for i, col := range columns {
		index[col.Name] = i
	}
	return index
}

func provideEmpty(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

func emptyProviderFactory(_ []types.ColumnInfo) DataProvider {
	return provideEmpty
}

func staticProviderFactory(provider DataProvider) providerFactory {
	return func(_ []types.ColumnInfo) DataProvider {
		return provider
	}
}

func (r *Registry) providerFactories() map[string]providerFactory {
	state := r.state
	return map[string]providerFactory{
		"system.local": func(columns []types.ColumnInfo) DataProvider {
			return provideLocal(state, columns)
		},
		"system.peers": func(columns []types.ColumnInfo) DataProvider {
			return providePeers(state, columns)
		},
		"system.peers_v2": func(columns []types.ColumnInfo) DataProvider {
			return providePeersV2(state, columns)
		},
		"system.peer_events":                               providePeerEvents,
		"system.peer_events_v2":                            providePeerEventsV2,
		"system.compaction_history":                        provideCompactionHistory,
		"system.prepared_statements":                       providePreparedStatements,
		"system.size_estimates":                            provideSizeEstimates,
		"system.sstable_activity":                          provideSSTableActivity,
		"system.sstable_activity_v2":                       provideSSTableActivityV2,
		"system.available_ranges":                          provideAvailableRanges,
		"system.available_ranges_v2":                       provideAvailableRangesV2,
		"system.batches":                                   staticProviderFactory(provideBatches),
		"system.built_views":                               staticProviderFactory(provideBuiltViews),
		"system.consensus_migration_state":                 emptyProviderFactory,
		"system.local_metadata_log":                        emptyProviderFactory,
		"system.metadata_snapshots":                        emptyProviderFactory,
		"system.paxos":                                     staticProviderFactory(providePaxos),
		"system.paxos_repair_history":                      emptyProviderFactory,
		"system.repairs":                                   emptyProviderFactory,
		"system.table_estimates":                           emptyProviderFactory,
		"system.top_partitions":                            emptyProviderFactory,
		"system.transferred_ranges":                        emptyProviderFactory,
		"system.transferred_ranges_v2":                     emptyProviderFactory,
		"system.view_builds_in_progress":                   staticProviderFactory(provideViewsBuildsInProgress),
		"system.IndexInfo":                                 staticProviderFactory(provideIndexInfo),
		"system_schema.keyspaces":                          r.provideSchemaKeyspaces,
		"system_schema.tables":                             r.provideSchemaTables,
		"system_schema.columns":                            r.provideSchemaColumns,
		"system_schema.dropped_columns":                    provideSchemaDroppedColumns,
		"system_schema.indexes":                            provideSchemaIndexes,
		"system_schema.views":                              provideSchemaViews,
		"system_schema.aggregates":                         staticProviderFactory(provideSchemaAggregates),
		"system_schema.column_masks":                       emptyProviderFactory,
		"system_schema.functions":                          staticProviderFactory(provideSchemaFunctions),
		"system_schema.triggers":                           emptyProviderFactory,
		"system_schema.types":                              staticProviderFactory(provideSchemaTypes),
		"system_auth.role_permissions":                     provideAuthRolePermissions,
		"system_auth.role_members":                         provideAuthRoleMembers,
		"system_auth.roles":                                provideAuthRoles,
		"system_auth.cidr_groups":                          emptyProviderFactory,
		"system_auth.cidr_permissions":                     emptyProviderFactory,
		"system_auth.identity_to_role":                     emptyProviderFactory,
		"system_auth.network_permissions":                  emptyProviderFactory,
		"system_auth.resource_role_permissons_index":       emptyProviderFactory,
		"system_distributed.repair_history":                provideDistributedRepairHistory,
		"system_distributed.auto_repair_history":           emptyProviderFactory,
		"system_distributed.auto_repair_priority":          emptyProviderFactory,
		"system_distributed.compression_dictionaries":      emptyProviderFactory,
		"system_distributed.parent_repair_history":         staticProviderFactory(provideDistributedParentRepairHistory),
		"system_distributed.partition_denylist":            emptyProviderFactory,
		"system_distributed.view_build_status":             staticProviderFactory(provideDistributedViewBuildStatus),
		"system_cluster_metadata.distributed_metadata_log": emptyProviderFactory,
		"system_accord.commands_for_key":                   emptyProviderFactory,
		"system_accord.journal":                            emptyProviderFactory,
		"system_traces.events":                             emptyProviderFactory,
		"system_traces.sessions":                           emptyProviderFactory,
	}
}

// ============================================
// System Keyspace Providers (selected tables)
// ============================================

// provideLocal - system.local
func provideLocal(state *SystemTableState, columns []types.ColumnInfo) DataProvider {
	colIndex := columnIndexMap(columns)
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		if state == nil {
			return nil, fmt.Errorf("system table state not initialized")
		}
		local, err := state.localNode(ctx)
		if err != nil {
			return nil, err
		}
		row := types.Row{Values: make([]interface{}, len(columns))}
		set := func(name string, value interface{}) {
			if idx, ok := colIndex[name]; ok {
				row.Values[idx] = value
			}
		}
		set("key", "local")
		set("bootstrapped", "COMPLETED")
		set("broadcast_address", local.ip)
		set("broadcast_port", local.storagePort)
		set("cluster_name", state.clusterName)
		set("cql_version", state.cqlVersion)
		set("data_center", state.datacenter)
		set("gossip_generation", 1)
		set("host_id", local.hostID)
		set("listen_address", local.ip)
		set("listen_port", local.storagePort)
		set("native_protocol_version", state.nativeProtocolVersion)
		set("partitioner", state.partitioner)
		set("rack", state.rack)
		set("release_version", state.releaseVersion)
		set("rpc_address", local.ip)
		set("rpc_port", local.nativePort)
		set("schema_version", state.SchemaVersion())
		set("tokens", local.tokens)
		return filterAndLimit([]types.Row{row}, opts)
	}
}

// providePeers - system.peers
func providePeers(state *SystemTableState, columns []types.ColumnInfo) DataProvider {
	colIndex := columnIndexMap(columns)
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		if state == nil {
			return nil, fmt.Errorf("system table state not initialized")
		}
		local, err := state.localNode(ctx)
		if err != nil {
			return nil, err
		}
		peers, err := state.peerNodes(ctx, local)
		if err != nil {
			return nil, err
		}
		if len(peers) == 0 {
			return filterAndLimit(nil, opts)
		}
		rows := make([]types.Row, 0, len(peers))
		for _, peer := range peers {
			row := types.Row{Values: make([]interface{}, len(columns))}
			set := func(name string, value interface{}) {
				if idx, ok := colIndex[name]; ok {
					row.Values[idx] = value
				}
			}
			set("peer", peer.ip)
			set("data_center", state.datacenter)
			set("host_id", peer.hostID)
			set("preferred_ip", nil)
			set("rack", state.rack)
			set("release_version", state.releaseVersion)
			set("rpc_address", peer.ip)
			set("schema_version", state.SchemaVersion())
			set("tokens", peer.tokens)
			rows = append(rows, row)
		}
		return filterAndLimit(rows, opts)
	}
}

// providePeersV2 - system.peers_v2
func providePeersV2(state *SystemTableState, columns []types.ColumnInfo) DataProvider {
	colIndex := columnIndexMap(columns)
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		if state == nil {
			return nil, fmt.Errorf("system table state not initialized")
		}
		local, err := state.localNode(ctx)
		if err != nil {
			return nil, err
		}
		peers, err := state.peerNodes(ctx, local)
		if err != nil {
			return nil, err
		}
		if len(peers) == 0 {
			return filterAndLimit(nil, opts)
		}
		rows := make([]types.Row, 0, len(peers))
		for _, peer := range peers {
			row := types.Row{Values: make([]interface{}, len(columns))}
			set := func(name string, value interface{}) {
				if idx, ok := colIndex[name]; ok {
					row.Values[idx] = value
				}
			}
			set("peer", peer.ip)
			set("peer_port", peer.storagePort)
			set("data_center", state.datacenter)
			set("host_id", peer.hostID)
			set("preferred_ip", nil)
			set("preferred_port", nil)
			set("rack", state.rack)
			set("release_version", state.releaseVersion)
			set("native_address", peer.ip)
			set("native_port", peer.nativePort)
			set("schema_version", state.SchemaVersion())
			set("tokens", peer.tokens)
			rows = append(rows, row)
		}
		return filterAndLimit(rows, opts)
	}
}

// providePeerEvents - system.peer_events
func providePeerEvents(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// providePeerEventsV2 - system.peer_events_v2
func providePeerEventsV2(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// provideBatches - system.batches
func provideBatches(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideBatchlog - system.batchlog
func provideBatchlog(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideCompactionHistory - system.compaction_history
func provideCompactionHistory(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// provideIndexInfo - system.IndexInfo
func provideIndexInfo(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// providePaxos - system.paxos
func providePaxos(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// providePreparedStatements - system.prepared_statements
func providePreparedStatements(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// provideRangeXfers - system.range_xfers
func provideRangeXfers(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideSizeEstimates - system.size_estimates
func provideSizeEstimates(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// provideSSTableActivity - system.sstable_activity
func provideSSTableActivity(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// provideSSTableActivityV2 - system.sstable_activity_v2
func provideSSTableActivityV2(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// provideViewsBuildsInProgress - system.views_builds_in_progress
func provideViewsBuildsInProgress(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideBuiltViews - system.built_views
func provideBuiltViews(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideAvailableRanges - system.available_ranges
func provideAvailableRanges(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// provideAvailableRangesV2 - system.available_ranges_v2
func provideAvailableRangesV2(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// ============================================
// System Schema Providers (selected tables)
// ============================================

// provideSchemaKeyspaces - system_schema.keyspaces
func (r *Registry) provideSchemaKeyspaces(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		keyspaceSet := make(map[string]struct{})
		for _, ks := range keyspace.BuiltInKeyspaces() {
			if keyspace.IsBlocked(ks) && ks != "system_auth" {
				continue
			}
			keyspaceSet[ks] = struct{}{}
		}
		for _, table := range r.ListAll() {
			if keyspace.IsBlocked(table.Keyspace) && table.Keyspace != "system_auth" {
				continue
			}
			keyspaceSet[table.Keyspace] = struct{}{}
		}
		keyspaces := make([]string, 0, len(keyspaceSet))
		for ks := range keyspaceSet {
			keyspaces = append(keyspaces, ks)
		}
		sort.Strings(keyspaces)

		rows := make([]types.Row, 0, len(keyspaces))
		for _, ks := range keyspaces {
			rows = append(rows, buildRow(columns, map[string]interface{}{
				"keyspace_name":  ks,
				"durable_writes": true,
				"replication": map[string]string{
					"class":              "SimpleStrategy",
					"replication_factor": "1",
				},
				"fast_path": false,
			}))
		}
		return filterAndLimit(rows, opts)
	}
}

// provideSchemaTables - system_schema.tables
func (r *Registry) provideSchemaTables(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		tables := r.sortedTables()
		rows := make([]types.Row, 0, len(tables))

		for _, table := range tables {
			id, err := tableIDFor(table.Keyspace, table.Table)
			if err != nil {
				return nil, err
			}
			values := map[string]interface{}{
				"keyspace_name":               table.Keyspace,
				"table_name":                  table.Table,
				"allow_auto_snapshot":         true,
				"bloom_filter_fp_chance":      0.01,
				"caching":                     map[string]string{"keys": "ALL"},
				"comment":                     table.Comment,
				"compaction":                  map[string]string{"class": "LeveledCompactionStrategy"},
				"compression":                 map[string]string{"class": "LZ4Compressor"},
				"memtable":                    map[string]string{},
				"crc_check_chance":            1.0,
				"dclocal_read_repair_chance":  0.0,
				"default_time_to_live":        0,
				"extensions":                  map[string]string{},
				"flags":                       []string{"compound"},
				"gc_grace_seconds":            0,
				"incremental_backups":         false,
				"id":                          id,
				"max_index_interval":          2048,
				"memtable_flush_period_in_ms": 0,
				"min_index_interval":          128,
				"read_repair_chance":          0.0,
				"speculative_retry":           "99PERCENTILE",
				"additional_write_policy":     "99PERCENTILE",
				"cdc":                         false,
				"read_repair":                 "NONE",
				"fast_path":                   false,
				"auto_repair":                 false,
			}
			rows = append(rows, buildRow(columns, values))
		}

		return filterAndLimit(rows, opts)
	}
}

// provideSchemaColumns - system_schema.columns
func (r *Registry) provideSchemaColumns(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		tables := r.sortedTables()
		rows := make([]types.Row, 0)

		for _, table := range tables {
			for _, col := range table.Columns {
				kind, position, order := columnKind(table, col.Name)
				rows = append(rows, buildRow(columns, map[string]interface{}{
					"keyspace_name":     table.Keyspace,
					"table_name":        table.Table,
					"column_name":       col.Name,
					"clustering_order":  order,
					"column_name_bytes": []byte(col.Name),
					"kind":              kind,
					"position":          position,
					"type":              col.Type,
				}))
			}
		}

		return filterAndLimit(rows, opts)
	}
}

// provideSchemaAggregates - system_schema.aggregates
func provideSchemaAggregates(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideSchemaDroppedColumns - system_schema.dropped_columns
func provideSchemaDroppedColumns(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// provideSchemaFunctions - system_schema.functions
func provideSchemaFunctions(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideSchemaIndexes - system_schema.indexes
func provideSchemaIndexes(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// provideSchemaTypes - system_schema.types
func provideSchemaTypes(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideSchemaViews - system_schema.views
func provideSchemaViews(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// ============================================
// System Distributed Providers (5 tables)
// ============================================

// provideDistributedRepairHistory - system_distributed.repair_history
func provideDistributedRepairHistory(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// provideDistributedNodesyncStatus - system_distributed.nodesync_status
func provideDistributedNodesyncStatus(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideDistributedNodesyncUserValidations - system_distributed.nodesync_user_validations
func provideDistributedNodesyncUserValidations(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideDistributedParentRepairHistory - system_distributed.parent_repair_history
func provideDistributedParentRepairHistory(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideDistributedViewBuildStatus - system_distributed.view_build_status
func provideDistributedViewBuildStatus(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// ============================================
// System Auth Providers (3 tables)
// ============================================

// provideAuthRolePermissions - system_auth.role_permissions
func provideAuthRolePermissions(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// provideAuthRoleMembers - system_auth.role_members
func provideAuthRoleMembers(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// provideAuthRoles - system_auth.roles
func provideAuthRoles(columns []types.ColumnInfo) DataProvider {
	return func(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
		_ = ctx
		return filterAndLimit(nil, opts)
	}
}

// ============================================
// DSE Security Providers (3 tables)
// ============================================

// provideDSERoleOptions - dse_security.role_options
func provideDSERoleOptions(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideDSEDigestTokens - dse_security.digest_tokens
func provideDSEDigestTokens(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// provideDSPSparkSecurity - dse_security.spark_security
func provideDSPSparkSecurity(ctx context.Context, opts QueryOptions) ([]types.Row, error) {
	return filterAndLimit(nil, opts)
}

// ============================================
// Helper Functions
// ============================================

// filterAndLimit applies WHERE filtering and LIMIT to rows
func filterAndLimit(rows []types.Row, opts QueryOptions) ([]types.Row, error) {
	var result []types.Row

	// Apply WHERE filtering if present
	if opts.WhereClause != nil {
		// Note: The actual filtering is done in executor.go using FilterEngine
		// This function is called by providers, but the executor handles filtering
		// For now, we'll just return all rows and let the executor filter
		result = rows
	} else {
		result = rows
	}

	// Apply LIMIT if present
	if opts.Limit != nil && int64(len(result)) > *opts.Limit {
		result = result[:*opts.Limit]
	}

	return result, nil
}

func (r *Registry) sortedTables() []*SystemTable {
	tables := r.ListAll()
	filtered := make([]*SystemTable, 0, len(tables))
	for _, table := range tables {
		if keyspace.IsBlocked(table.Keyspace) && table.Keyspace != "system_auth" {
			continue
		}
		filtered = append(filtered, table)
	}
	sort.Slice(filtered, func(i, j int) bool {
		if filtered[i].Keyspace == filtered[j].Keyspace {
			return filtered[i].Table < filtered[j].Table
		}
		return filtered[i].Keyspace < filtered[j].Keyspace
	})
	return filtered
}

func columnKind(table *SystemTable, name string) (string, int, string) {
	for i, pk := range table.PartitionKeys {
		if pk == name {
			return "partition_key", i, "NONE"
		}
	}
	for i, ck := range table.ClusteringKeys {
		if ck == name {
			order := "ASC"
			if table.ClusteringOrder != nil {
				if v, ok := table.ClusteringOrder[ck]; ok {
					order = v
				}
			}
			return "clustering", i, order
		}
	}
	return "regular", -1, "NONE"
}