//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"strings"
	"testing"
	"time"

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

type systemTableName struct {
	keyspace string
	table    string
}

type columnExpectation struct {
	kind    string
	order   string
	cqlType string
}

func TestSystemTableVirtualizationMissingBuiltins(t *testing.T) {
	session := newSession(t, sessionOptions{disableHostLookup: true})

	missing := []systemTableName{
		{keyspace: "system", table: "IndexInfo"},
		{keyspace: "system", table: "batches"},
		{keyspace: "system", table: "built_views"},
		{keyspace: "system", table: "consensus_migration_state"},
		{keyspace: "system", table: "local_metadata_log"},
		{keyspace: "system", table: "metadata_snapshots"},
		{keyspace: "system", table: "paxos"},
		{keyspace: "system", table: "paxos_repair_history"},
		{keyspace: "system", table: "repairs"},
		{keyspace: "system", table: "table_estimates"},
		{keyspace: "system", table: "top_partitions"},
		{keyspace: "system", table: "transferred_ranges"},
		{keyspace: "system", table: "transferred_ranges_v2"},
		{keyspace: "system", table: "view_builds_in_progress"},
		{keyspace: "system_accord", table: "commands_for_key"},
		{keyspace: "system_accord", table: "journal"},
		{keyspace: "system_cluster_metadata", table: "distributed_metadata_log"},
		{keyspace: "system_schema", table: "aggregates"},
		{keyspace: "system_schema", table: "column_masks"},
		{keyspace: "system_schema", table: "functions"},
		{keyspace: "system_schema", table: "triggers"},
		{keyspace: "system_schema", table: "types"},
		{keyspace: "system_traces", table: "events"},
		{keyspace: "system_traces", table: "sessions"},
	}

	for _, table := range missing {
		name := fmt.Sprintf("%s.%s", table.keyspace, table.table)
		t.Run(name, func(t *testing.T) {
			query := fmt.Sprintf(
				"SELECT * FROM %s.%s LIMIT 1",
				cqlIdent(table.keyspace),
				cqlIdent(table.table),
			)
			if err := session.Query(query).Exec(); err != nil {
				t.Fatalf("query failed for %s: %v", name, err)
			}
		})
	}
}

func TestSystemTableVirtualizationDeterministicEmptyTables(t *testing.T) {
	session := newSession(t, sessionOptions{disableHostLookup: true})

	emptyTables := []systemTableName{
		{keyspace: "system", table: "IndexInfo"},
		{keyspace: "system", table: "batches"},
		{keyspace: "system", table: "paxos"},
		{keyspace: "system", table: "table_estimates"},
		{keyspace: "system", table: "transferred_ranges"},
		{keyspace: "system", table: "transferred_ranges_v2"},
		{keyspace: "system", table: "view_builds_in_progress"},
		{keyspace: "system_accord", table: "journal"},
		{keyspace: "system_cluster_metadata", table: "distributed_metadata_log"},
		{keyspace: "system_schema", table: "functions"},
		{keyspace: "system_schema", table: "aggregates"},
		{keyspace: "system_schema", table: "column_masks"},
		{keyspace: "system_traces", table: "events"},
		{keyspace: "system_traces", table: "sessions"},
	}

	for _, table := range emptyTables {
		name := fmt.Sprintf("%s.%s", table.keyspace, table.table)
		t.Run(name, func(t *testing.T) {
			query := fmt.Sprintf(
				"SELECT * FROM %s.%s LIMIT 1",
				cqlIdent(table.keyspace),
				cqlIdent(table.table),
			)
			rows, err := scanRows(session, query)
			if err != nil {
				t.Fatalf("query failed for %s: %v", name, err)
			}
			if len(rows) != 0 {
				t.Fatalf("expected empty rows for %s, got %d", name, len(rows))
			}
		})
	}
}

func TestSystemTableBlockedKeyspaces(t *testing.T) {
	session := newSession(t, sessionOptions{disableHostLookup: true})

	blocked := []systemTableName{
		{keyspace: "system_distributed", table: "repair_history"},
	}

	for _, table := range blocked {
		name := fmt.Sprintf("%s.%s", table.keyspace, table.table)
		t.Run(name, func(t *testing.T) {
			query := fmt.Sprintf(
				"SELECT * FROM %s.%s LIMIT 1",
				cqlIdent(table.keyspace),
				cqlIdent(table.table),
			)
			if err := session.Query(query).Exec(); err == nil {
				t.Fatalf("expected blocked access for %s", name)
			}
		})
	}
}

func TestSystemSchemaMetadataFromCQLMeta(t *testing.T) {
	session := newSession(t, sessionOptions{disableHostLookup: true})

	keyspace := uniqueKeyspace()
	if err := session.Query(createKeyspaceCQL(keyspace)).Exec(); err != nil {
		t.Fatalf("create keyspace failed: %v", err)
	}
	t.Cleanup(func() {
		_ = session.Query(dropKeyspaceCQL(keyspace)).Exec()
	})

	typeName := "addr"
	createType := fmt.Sprintf("CREATE TYPE %s.%s (street text, zip int)", keyspace, typeName)
	if err := session.Query(createType).Exec(); err != nil {
		t.Fatalf("create type failed: %v", err)
	}

	tableName := "meta_tbl"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (id int, bucket int, ts timestamp, name text, idx int, addr frozen<%s>, PRIMARY KEY ((id, bucket), ts)) WITH CLUSTERING ORDER BY (ts DESC)",
		keyspace, tableName, typeName,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	indexName := "meta_tbl_idx_idx"
	createIndex := fmt.Sprintf("CREATE INDEX %s ON %s.%s (idx)", indexName, keyspace, tableName)
	if err := session.Query(createIndex).Exec(); err != nil {
		t.Fatalf("create index failed: %v", err)
	}

	triggerName := "tr_meta"
	withFeatureFlags(t, func(features *config.FeatureConfig) {
		features.MetadataOnlyTriggers = true
	})
	createTrigger := fmt.Sprintf("CREATE TRIGGER %s ON %s.%s USING 'com.example.Trigger'", triggerName, keyspace, tableName)
	if err := session.Query(createTrigger).Exec(); err != nil {
		t.Fatalf("create trigger failed: %v", err)
	}

	waitFor(t, "system_schema.tables metadata", 5*time.Second, func() (bool, error) {
		query := fmt.Sprintf("SELECT table_name FROM system_schema.tables WHERE keyspace_name = '%s'", keyspace)
		rows, err := scanRows(session, query)
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["table_name"]) == tableName {
				return true, nil
			}
		}
		return false, nil
	})

	expectedType := fmt.Sprintf("frozen<%s.%s>", keyspace, typeName)
	expectedColumns := map[string]columnExpectation{
		"id":     {kind: "partition_key", order: "NONE", cqlType: "int"},
		"bucket": {kind: "partition_key", order: "NONE", cqlType: "int"},
		"ts":     {kind: "clustering", order: "DESC", cqlType: "timestamp"},
		"name":   {kind: "regular", order: "NONE", cqlType: "text"},
		"idx":    {kind: "regular", order: "NONE", cqlType: "int"},
		"addr":   {kind: "regular", order: "NONE", cqlType: expectedType},
	}

	waitFor(t, "system_schema.columns metadata", 5*time.Second, func() (bool, error) {
		query := fmt.Sprintf(
			"SELECT column_name, kind, clustering_order, type FROM system_schema.columns WHERE keyspace_name = '%s' AND table_name = '%s'",
			keyspace, tableName,
		)
		rows, err := scanRows(session, query)
		if err != nil {
			return false, err
		}
		return matchColumnMeta(rows, expectedColumns), nil
	})

	waitFor(t, "system_schema.indexes metadata", 5*time.Second, func() (bool, error) {
		query := fmt.Sprintf(
			"SELECT index_name FROM system_schema.indexes WHERE keyspace_name = '%s' AND table_name = '%s'",
			keyspace, tableName,
		)
		rows, err := scanRows(session, query)
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["index_name"]) == indexName {
				return true, nil
			}
		}
		return false, nil
	})

	waitFor(t, "system_schema.types metadata", 5*time.Second, func() (bool, error) {
		query := fmt.Sprintf(
			"SELECT type_name, field_names, field_types FROM system_schema.types WHERE keyspace_name = '%s'",
			keyspace,
		)
		rows, err := scanRows(session, query)
		if err != nil {
			return false, err
		}
		return matchTypeRow(rows, typeName, []string{"street", "zip"}, []string{"text", "int"}), nil
	})

	waitFor(t, "system_schema.triggers metadata", 5*time.Second, func() (bool, error) {
		query := fmt.Sprintf(
			"SELECT trigger_name, table_name FROM system_schema.triggers WHERE keyspace_name = '%s'",
			keyspace,
		)
		rows, err := scanRows(session, query)
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["trigger_name"]) == triggerName && asString(row["table_name"]) == tableName {
				return true, nil
			}
		}
		return false, nil
	})

	emptyQueries := []string{
		fmt.Sprintf("SELECT * FROM system_schema.functions WHERE keyspace_name = '%s'", keyspace),
		fmt.Sprintf("SELECT * FROM system_schema.aggregates WHERE keyspace_name = '%s'", keyspace),
		fmt.Sprintf("SELECT * FROM system_schema.column_masks WHERE keyspace_name = '%s'", keyspace),
	}
	for _, query := range emptyQueries {
		rows, err := scanRows(session, query)
		if err != nil {
			t.Fatalf("query failed for %s: %v", query, err)
		}
		if len(rows) != 0 {
			t.Fatalf("expected empty rows for %s, got %d", query, len(rows))
		}
	}
}

func TestSystemTableDeterministicMinimalRows(t *testing.T) {
	session := newSession(t, sessionOptions{disableHostLookup: true})

	waitFor(t, "system.local deterministic row", 5*time.Second, func() (bool, error) {
		rows, err := scanRows(session, "SELECT key, cluster_name, schema_version FROM system.local")
		if err != nil {
			return false, err
		}
		if len(rows) != 1 {
			return false, nil
		}
		if asString(rows[0]["key"]) != "local" {
			return false, nil
		}
		if asString(rows[0]["cluster_name"]) != "Cassandra" {
			return false, nil
		}
		return asString(rows[0]["schema_version"]) != "", nil
	})

	waitFor(t, "system.peers schema_version agreement", 5*time.Second, func() (bool, error) {
		localRows, err := scanRows(session, "SELECT schema_version FROM system.local WHERE key = 'local'")
		if err != nil {
			return false, err
		}
		if len(localRows) != 1 {
			return false, nil
		}
		localVersion := asString(localRows[0]["schema_version"])
		if localVersion == "" {
			return false, nil
		}
		peerRows, err := scanRows(session, "SELECT schema_version FROM system.peers")
		if err != nil {
			return false, err
		}
		if len(peerRows) == 0 {
			return true, nil
		}
		for _, row := range peerRows {
			if asString(row["schema_version"]) != localVersion {
				return false, nil
			}
		}
		return true, nil
	})
}

func cqlIdent(name string) string {
	for i := 0; i < len(name); i++ {
		ch := name[i]
		if (ch < 'a' || ch > 'z') && (ch < '0' || ch > '9') && ch != '_' {
			return `"` + strings.ReplaceAll(name, `"`, `""`) + `"`
		}
	}
	return name
}

func matchColumnMeta(rows []map[string]interface{}, expected map[string]columnExpectation) bool {
	seen := make(map[string]columnExpectation, len(rows))
	for _, row := range rows {
		name := rowString(row["column_name"])
		if name == "" {
			continue
		}
		seen[name] = columnExpectation{
			kind:    strings.ToLower(rowString(row["kind"])),
			order:   strings.ToUpper(rowString(row["clustering_order"])),
			cqlType: strings.ToLower(rowString(row["type"])),
		}
	}
	for name, want := range expected {
		got, ok := seen[name]
		if !ok {
			return false
		}
		if want.kind != "" && got.kind != strings.ToLower(want.kind) {
			return false
		}
		if want.order != "" && got.order != strings.ToUpper(want.order) {
			return false
		}
		if want.cqlType != "" && got.cqlType != strings.ToLower(want.cqlType) {
			return false
		}
	}
	return true
}

func rowString(value interface{}) string {
	if value == nil {
		return ""
	}
	return asString(value)
}