//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"testing"
	"time"
)

func TestSystemTablesAndDescribe(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()
	})

	table := "sys_meta"
	if err := session.Query(createTableCQL(keyspace, table)).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}
	index := "sys_meta_idx_idx"
	if err := session.Query(createIndexCQL(index, keyspace, table, "idx")).Exec(); err != nil {
		t.Fatalf("create index failed: %v", err)
	}

	waitFor(t, "system.local row", 5*time.Second, func() (bool, error) {
		rows, err := scanRows(session, "SELECT key FROM system.local")
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["key"]) == "local" {
				return true, nil
			}
		}
		return false, nil
	})

	waitFor(t, "system.peers empty", 5*time.Second, func() (bool, error) {
		rows, err := scanRows(session, "SELECT peer FROM system.peers")
		if err != nil {
			return false, err
		}
		return len(rows) == 0, nil
	})

	waitFor(t, "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) == 0 {
			return false, nil
		}
		peerRows, err := scanRows(session, "SELECT schema_version FROM system.peers")
		if err != nil {
			return false, err
		}
		local := asString(localRows[0]["schema_version"])
		if len(peerRows) == 0 {
			return local != "", nil
		}
		peer := asString(peerRows[0]["schema_version"])
		return local != "" && local == peer, nil
	})

	waitFor(t, "system_schema.keyspaces entry", 5*time.Second, func() (bool, error) {
		rows, err := scanRows(session, "SELECT keyspace_name FROM system_schema.keyspaces")
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["keyspace_name"]) == keyspace {
				return true, nil
			}
		}
		return false, nil
	})

	waitFor(t, "system_schema.tables entry", 5*time.Second, func() (bool, error) {
		rows, err := scanRows(session, "SELECT keyspace_name, table_name FROM system_schema.tables")
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["keyspace_name"]) == keyspace && asString(row["table_name"]) == table {
				return true, nil
			}
		}
		return false, nil
	})

	waitFor(t, "system_schema.columns entry", 5*time.Second, func() (bool, error) {
		rows, err := scanRows(session, "SELECT keyspace_name, table_name, column_name FROM system_schema.columns")
		if err != nil {
			return false, err
		}
		foundID := false
		foundName := false
		for _, row := range rows {
			if asString(row["keyspace_name"]) != keyspace || asString(row["table_name"]) != table {
				continue
			}
			switch asString(row["column_name"]) {
			case "id":
				foundID = true
			case "name":
				foundName = true
			}
		}
		return foundID && foundName, nil
	})

	waitFor(t, "system_schema.indexes entry", 5*time.Second, func() (bool, error) {
		rows, err := scanRows(session, "SELECT keyspace_name, table_name, index_name FROM system_schema.indexes")
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["keyspace_name"]) == keyspace &&
				asString(row["table_name"]) == table &&
				asString(row["index_name"]) == index {
				return true, nil
			}
		}
		return false, nil
	})

	waitFor(t, "DESCRIBE KEYSPACE output", 5*time.Second, func() (bool, error) {
		rows, err := scanRows(session, fmt.Sprintf("DESCRIBE KEYSPACE %s", keyspace))
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["keyspace_name"]) == keyspace && asString(row["create_statement"]) != "" {
				return true, nil
			}
		}
		return false, nil
	})

	waitFor(t, "DESCRIBE TABLE output", 5*time.Second, func() (bool, error) {
		rows, err := scanRows(session, fmt.Sprintf("DESCRIBE TABLE %s.%s", keyspace, table))
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["keyspace_name"]) == keyspace &&
				asString(row["name"]) == table &&
				asString(row["create_statement"]) != "" {
				return true, nil
			}
		}
		return false, nil
	})
}

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

	rows, err := scanRows(session, "SELECT name, value FROM system_views.settings")
	if err != nil {
		t.Fatalf("system_views.settings query failed: %v", err)
	}
	if len(rows) != 0 {
		t.Fatalf("expected system_views.settings to be empty")
	}
}

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

	rows, err := scanRows(session, "DESCRIBE KEYSPACES")
	if err != nil {
		t.Fatalf("describe keyspaces failed: %v", err)
	}
	if len(rows) == 0 {
		t.Fatalf("expected rows from DESCRIBE KEYSPACES")
	}
	for _, row := range rows {
		if asString(row["keyspace_name"]) == "system" {
			return
		}
	}
	t.Fatalf("expected system keyspace in DESCRIBE KEYSPACES output")
}

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

	rows, err := scanRows(session, "DESCRIBE KEYSPACE mysql")
	if err != nil {
		t.Fatalf("describe keyspace failed: %v", err)
	}
	for _, row := range rows {
		if asString(row["keyspace_name"]) == "mysql" {
			t.Fatalf("expected internal keyspace mysql to be filtered from describe output")
		}
	}
}

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

	rows, err := scanRows(session, "SELECT cluster_name, data_center, schema_version FROM system.local WHERE key = 'local'")
	if err != nil {
		t.Fatalf("system.local query failed: %v", err)
	}
	if len(rows) == 0 {
		t.Fatalf("expected system.local rows")
	}
	row := rows[0]
	if asString(row["cluster_name"]) != "Cassandra" {
		t.Fatalf("unexpected cluster_name: %v", row["cluster_name"])
	}
	if asString(row["data_center"]) != "datacenter1" {
		t.Fatalf("unexpected data_center: %v", row["data_center"])
	}
	if asString(row["schema_version"]) == "" {
		t.Fatalf("expected non-empty schema_version")
	}
}