package systemtables

import (
	"context"
	"strings"
	"testing"

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

// TestSystemTableIntegration tests the complete system table query flow
func TestSystemTableIntegration(t *testing.T) {
	registry := NewRegistry()
	executor := NewSystemTableExecutor(registry)
	ctx := context.Background()

	columnsFor := func(name string) int {
		table, ok := registry.Lookup(name)
		if !ok {
			return 0
		}
		return len(table.Columns)
	}

	keyspaceCount := func() int {
		seen := make(map[string]struct{})
		for _, ks := range keyspace.BuiltInKeyspaces() {
			if keyspace.IsBlocked(ks) && ks != "system_auth" {
				continue
			}
			seen[ks] = struct{}{}
		}
		for _, table := range registry.ListAll() {
			if keyspace.IsBlocked(table.Keyspace) && table.Keyspace != "system_auth" {
				continue
			}
			seen[table.Keyspace] = struct{}{}
		}
		return len(seen)
	}

	tableCount := func() int {
		count := 0
		for _, table := range registry.ListAll() {
			if keyspace.IsBlocked(table.Keyspace) && table.Keyspace != "system_auth" {
				continue
			}
			count++
		}
		return count
	}

	columnCount := func() int {
		count := 0
		for _, table := range registry.ListAll() {
			if keyspace.IsBlocked(table.Keyspace) && table.Keyspace != "system_auth" {
				continue
			}
			count += len(table.Columns)
		}
		return count
	}

	tests := []struct {
		name          string
		query         string
		expectRows    int
		expectColumns int
		expectError   bool
	}{
		{
			name:          "system.local",
			query:         "SELECT * FROM system.local",
			expectRows:    1,
			expectColumns: columnsFor("system.local"),
			expectError:   false,
		},
		{
			name:          "system.peers",
			query:         "SELECT * FROM system.peers",
			expectRows:    0,
			expectColumns: columnsFor("system.peers"),
			expectError:   false,
		},
		{
			name:          "system_schema.keyspaces",
			query:         "SELECT * FROM system_schema.keyspaces",
			expectRows:    keyspaceCount(),
			expectColumns: columnsFor("system_schema.keyspaces"),
			expectError:   false,
		},
		{
			name:          "system_schema.tables",
			query:         "SELECT * FROM system_schema.tables",
			expectRows:    tableCount(),
			expectColumns: columnsFor("system_schema.tables"),
			expectError:   false,
		},
		{
			name:          "system_schema.columns",
			query:         "SELECT * FROM system_schema.columns",
			expectRows:    columnCount(),
			expectColumns: columnsFor("system_schema.columns"),
			expectError:   false,
		},
		{
			name:          "system.compaction_history",
			query:         "SELECT * FROM system.compaction_history",
			expectRows:    0,
			expectColumns: columnsFor("system.compaction_history"),
			expectError:   false,
		},
		{
			name:          "system.size_estimates",
			query:         "SELECT * FROM system.size_estimates",
			expectRows:    0,
			expectColumns: columnsFor("system.size_estimates"),
			expectError:   false,
		},
		{
			name:          "system_distributed.repair_history",
			query:         "SELECT * FROM system_distributed.repair_history",
			expectRows:    0,
			expectColumns: columnsFor("system_distributed.repair_history"),
			expectError:   false,
		},
		{
			name:          "system_auth.roles",
			query:         "SELECT * FROM system_auth.roles",
			expectRows:    0,
			expectColumns: columnsFor("system_auth.roles"),
			expectError:   false,
		},
		{
			name:          "dse_security.role_options",
			query:         "SELECT * FROM dse_security.role_options",
			expectRows:    0,
			expectColumns: 2,
			expectError:   false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			// Parse the query
			stmt, err := parser.NewParser(tt.query).Parse()
			if err != nil {
				t.Fatalf("Failed to parse query: %v", err)
			}

			selectStmt, ok := stmt.(*parser.SelectStatement)
			if !ok {
				t.Fatalf("Expected SelectStatement, got %T", stmt)
			}

			// Check if it's detected as a system table query
			if !IsSystemTableQuery(selectStmt) {
				t.Errorf("Query not detected as system table query: %s", tt.query)
			}

			// Execute the query
			result, err := executor.Execute(ctx, selectStmt, "test_user", "test_password")
			if tt.expectError {
				if err == nil {
					t.Errorf("Expected error but got none")
				}
				return
			}
			if err != nil {
				t.Fatalf("Execute failed: %v", err)
			}

			// Verify results
			if tt.expectRows >= 0 && len(result.Rows) != tt.expectRows {
				t.Errorf("Expected %d rows, got %d", tt.expectRows, len(result.Rows))
			}
			if tt.expectColumns >= 0 && len(result.Columns) != tt.expectColumns {
				t.Errorf("Expected %d columns, got %d", tt.expectColumns, len(result.Columns))
			}
		})
	}
}

func TestSystemTablePeersConfigured(t *testing.T) {
	cfg := config.DefaultConfig()
	cfg.SystemTables.AdvertisedAddress = "127.0.0.1"
	cfg.SystemTables.Peers = []config.SystemTablePeerCfg{
		{
			AdvertisedAddress: "127.0.0.2",
			NativePort:        9042,
			StoragePort:       7000,
		},
	}

	current := config.GetConfig()
	config.SetConfig(cfg)
	t.Cleanup(func() {
		config.SetConfig(current)
	})

	registry := NewRegistry()
	executor := NewSystemTableExecutor(registry)
	ctx := context.Background()

	stmt, err := parser.NewParser("SELECT * FROM system.peers").Parse()
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	selectStmt := stmt.(*parser.SelectStatement)
	result, err := executor.Execute(ctx, selectStmt, "test_user", "test_password")
	if err != nil {
		t.Fatalf("execute failed: %v", err)
	}
	if len(result.Rows) != 1 {
		t.Fatalf("expected 1 peer row, got %d", len(result.Rows))
	}
}

// TestSystemTableFiltering tests WHERE clause filtering
func TestSystemTableFiltering(t *testing.T) {
	registry := NewRegistry()
	executor := NewSystemTableExecutor(registry)
	ctx := context.Background()

	// Test filtering on system.local
	query := "SELECT * FROM system.local WHERE key = 'local'"
	stmt, err := parser.NewParser(query).Parse()
	if err != nil {
		t.Fatalf("Failed to parse query: %v", err)
	}

	selectStmt := stmt.(*parser.SelectStatement)
	result, err := executor.Execute(ctx, selectStmt, "test_user", "test_password")
	if err != nil {
		t.Fatalf("Execute failed: %v", err)
	}

	if len(result.Rows) != 1 {
		t.Errorf("Expected 1 row after filtering, got %d", len(result.Rows))
	}

	// Test filtering on system_schema.keyspaces
	query = "SELECT * FROM system_schema.keyspaces WHERE keyspace_name = 'system_schema'"
	stmt, err = parser.NewParser(query).Parse()
	if err != nil {
		t.Fatalf("Failed to parse query: %v", err)
	}

	selectStmt = stmt.(*parser.SelectStatement)
	result, err = executor.Execute(ctx, selectStmt, "test_user", "test_password")
	if err != nil {
		t.Fatalf("Execute failed: %v", err)
	}

	if len(result.Rows) != 1 {
		t.Errorf("Expected 1 row after filtering, got %d", len(result.Rows))
	}
}

// TestSystemTableLimit tests LIMIT clause
func TestSystemTableLimit(t *testing.T) {
	registry := NewRegistry()
	executor := NewSystemTableExecutor(registry)
	ctx := context.Background()

	query := "SELECT * FROM system_schema.tables LIMIT 2"
	stmt, err := parser.NewParser(query).Parse()
	if err != nil {
		t.Fatalf("Failed to parse query: %v", err)
	}

	selectStmt := stmt.(*parser.SelectStatement)
	result, err := executor.Execute(ctx, selectStmt, "test_user", "test_password")
	if err != nil {
		t.Fatalf("Execute failed: %v", err)
	}

	if len(result.Rows) != 2 {
		t.Errorf("Expected 2 rows with LIMIT 2, got %d", len(result.Rows))
	}
}

// TestSystemTableColumnProjection tests column projection
func TestSystemTableColumnProjection(t *testing.T) {
	registry := NewRegistry()
	executor := NewSystemTableExecutor(registry)
	ctx := context.Background()

	query := "SELECT keyspace_name, table_name FROM system_schema.tables"
	stmt, err := parser.NewParser(query).Parse()
	if err != nil {
		t.Fatalf("Failed to parse query: %v", err)
	}

	selectStmt := stmt.(*parser.SelectStatement)
	result, err := executor.Execute(ctx, selectStmt, "test_user", "test_password")
	if err != nil {
		t.Fatalf("Execute failed: %v", err)
	}

	if len(result.Columns) != 2 {
		t.Errorf("Expected 2 columns, got %d", len(result.Columns))
	}
	if result.Columns[0].Name != "keyspace_name" || result.Columns[1].Name != "table_name" {
		t.Errorf("Column names incorrect: %v", result.Columns)
	}
}

// TestIsSystemTableQuery tests the detection function
func TestIsSystemTableQuery(t *testing.T) {
	tests := []struct {
		query    string
		expected bool
	}{
		{"SELECT * FROM system.local", true},
		{"SELECT * FROM system.peers", true},
		{"SELECT * FROM system_schema.keyspaces", true},
		{"SELECT * FROM system_schema.tables", true},
		{"SELECT * FROM system_distributed.repair_history", true},
		{"SELECT * FROM system_auth.roles", true},
		{"SELECT * FROM system_traces.sessions", true},
		{"SELECT * FROM system_cluster_metadata.distributed_metadata_log", true},
		{"SELECT * FROM system_accord.journal", true},
		{"SELECT * FROM dse_security.role_options", true},
		{"SELECT * FROM system_custom.local", false},
		{"SELECT * FROM users", false},
		{"SELECT * FROM my_keyspace.my_table", false},
		{"SELECT * FROM virtual_table", false},
	}

	for _, tt := range tests {
		t.Run(tt.query, func(t *testing.T) {
			stmt, err := parser.NewParser(tt.query).Parse()
			if err != nil {
				t.Fatalf("Failed to parse: %v", err)
			}
			selectStmt, ok := stmt.(*parser.SelectStatement)
			if !ok {
				t.Fatalf("Not a select statement")
			}

			result := IsSystemTableQuery(selectStmt)
			if result != tt.expected {
				t.Errorf("IsSystemTableQuery(%s) = %v, want %v", tt.query, result, tt.expected)
			}
		})
	}
}

// TestRegistryLookup tests the registry
func TestRegistryLookup(t *testing.T) {
	registry := NewRegistry()

	tests := []struct {
		name       string
		tableName  string
		shouldFind bool
	}{
		{"system.local", "system.local", true},
		{"system.peers", "system.peers", true},
		{"system_schema.keyspaces", "system_schema.keyspaces", true},
		{"unknown table", "system.unknown", false},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			table, found := registry.Lookup(tt.tableName)
			if tt.shouldFind && !found {
				t.Errorf("Expected to find table %s, but didn't", tt.tableName)
			}
			if !tt.shouldFind && found {
				t.Errorf("Expected not to find table %s, but did", tt.tableName)
			}
			if found && table == nil {
				t.Errorf("Found table but it's nil")
			}
		})
	}
}

// TestAllTablesRegistered verifies all system tables are registered
func TestAllTablesRegistered(t *testing.T) {
	registry := NewRegistry()
	tables := registry.ListAll()

	expected := len(builtinSystemTableDefs) + 3
	if len(tables) != expected {
		t.Errorf("Expected %d tables, got %d", expected, len(tables))
	}

	// Verify all tables are read-only
	for _, table := range tables {
		if !table.IsReadOnly {
			t.Errorf("Table %s should be read-only", table.Name)
		}
		if table.Provider == nil {
			t.Errorf("Table %s has no provider", table.Name)
		}
		if len(table.Columns) == 0 {
			t.Errorf("Table %s has no columns", table.Name)
		}
	}
}

// TestProviderDataTypes verifies data types in providers
func TestProviderDataTypes(t *testing.T) {
	registry := NewRegistry()
	ctx := context.Background()

	// Test system.local - should have inet, uuid types
	table, _ := registry.Lookup("system.local")
	result, err := table.Provider(ctx, QueryOptions{})
	if err != nil {
		t.Fatalf("Provider failed: %v", err)
	}

	if len(result) == 0 {
		t.Fatal("Expected at least one row from system.local")
	}

	// Verify the first row has correct number of values
	if len(result[0].Values) != len(table.Columns) {
		t.Errorf("Row value count mismatch: expected %d, got %d", len(table.Columns), len(result[0].Values))
	}

	// Test that inet values are strings (127.0.0.1)
	colIndex := make(map[string]int)
	for i, col := range table.Columns {
		colIndex[col.Name] = i
	}
	broadcastIdx, ok := colIndex["broadcast_address"]
	if !ok {
		t.Fatalf("broadcast_address column not found")
	}
	broadcastAddr := result[0].Values[broadcastIdx]
	if _, ok := broadcastAddr.(string); !ok {
		t.Errorf("Expected broadcast_address to be string, got %T", broadcastAddr)
	}
	if !strings.Contains(broadcastAddr.(string), ".") {
		t.Errorf("Expected IP address format, got %v", broadcastAddr)
	}

	// Test that uuid values are strings
	hostIdx, ok := colIndex["host_id"]
	if !ok {
		t.Fatalf("host_id column not found")
	}
	hostID := result[0].Values[hostIdx]
	if _, ok := hostID.(string); !ok {
		t.Errorf("Expected host_id to be string, got %T", hostID)
	}
}

func TestProviderFactoriesCoverBuiltins(t *testing.T) {
	registry := NewRegistry()
	providers := registry.providerFactories()

	var missing []string
	for _, def := range builtinSystemTableDefs {
		fullName := def.Keyspace + "." + def.Table
		if _, ok := providers[fullName]; !ok {
			missing = append(missing, fullName)
		}
	}
	if len(missing) > 0 {
		t.Fatalf("missing providers for: %s", strings.Join(missing, ", "))
	}
}

func TestBuiltinSystemTableQueries(t *testing.T) {
	registry := NewRegistry()
	executor := NewSystemTableExecutor(registry)
	ctx := context.Background()

	quoteIfNeeded := func(name string) string {
		for i := 0; i < len(name); i++ {
			ch := name[i]
			if ch >= 'A' && ch <= 'Z' {
				return `"` + strings.ReplaceAll(name, `"`, `""`) + `"`
			}
		}
		return name
	}

	for _, def := range builtinSystemTableDefs {
		keyspaceName := quoteIfNeeded(def.Keyspace)
		tableName := quoteIfNeeded(def.Table)
		query := "SELECT * FROM " + keyspaceName + "." + tableName
		stmt, err := parser.NewParser(query).Parse()
		if err != nil {
			t.Fatalf("parse failed for %s.%s: %v", def.Keyspace, def.Table, err)
		}
		selectStmt, ok := stmt.(*parser.SelectStatement)
		if !ok {
			t.Fatalf("expected select for %s.%s", def.Keyspace, def.Table)
		}
		result, err := executor.Execute(ctx, selectStmt, "test_user", "test_password")
		if err != nil {
			t.Fatalf("execute failed for %s.%s: %v", def.Keyspace, def.Table, err)
		}
		if len(result.Columns) != len(def.Columns) {
			t.Errorf("column mismatch for %s.%s: expected %d, got %d", def.Keyspace, def.Table, len(def.Columns), len(result.Columns))
		}
	}
}