//go:build integration
// +build integration

package integration_test

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

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

func TestTypeDDLSystemSchema(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 := "address"
	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)
	}

	waitFor(t, "system_schema.types create", 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
	})

	alterAdd := fmt.Sprintf("ALTER TYPE %s.%s ADD city text", keyspace, typeName)
	if err := session.Query(alterAdd).Exec(); err != nil {
		t.Fatalf("alter type add failed: %v", err)
	}
	waitFor(t, "system_schema.types add field", 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", "city"}, []string{"text", "int", "text"}), nil
	})

	alterRename := fmt.Sprintf("ALTER TYPE %s.%s RENAME street TO street_name", keyspace, typeName)
	if err := session.Query(alterRename).Exec(); err != nil {
		t.Fatalf("alter type rename failed: %v", err)
	}
	waitFor(t, "system_schema.types rename field", 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_name", "zip", "city"}, []string{"text", "int", "text"}), nil
	})

	table := "udt_users"
	createTable := fmt.Sprintf("CREATE TABLE %s.%s (id int PRIMARY KEY, info frozen<%s>)", keyspace, table, typeName)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	dropType := fmt.Sprintf("DROP TYPE %s.%s", keyspace, typeName)
	err := session.Query(dropType).Exec()
	requireErrorCode(t, err, cql.ErrorCodeInvalid)

	if err := session.Query(fmt.Sprintf("DROP TABLE %s.%s", keyspace, table)).Exec(); err != nil {
		t.Fatalf("drop table failed: %v", err)
	}
	if err := session.Query(dropType).Exec(); err != nil {
		t.Fatalf("drop type failed: %v", err)
	}

	waitFor(t, "system_schema.types drop", 5*time.Second, func() (bool, error) {
		query := fmt.Sprintf("SELECT type_name FROM system_schema.types WHERE keyspace_name = '%s'", keyspace)
		rows, err := scanRows(session, query)
		if err != nil {
			return false, err
		}
		return !typeRowExists(rows, typeName), nil
	})
}

func TestViewDDLSystemSchema(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 := "view_base"
	if err := session.Query(createTableCQL(keyspace, table)).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	view := "mv_basic"
	createView := fmt.Sprintf(
		"CREATE MATERIALIZED VIEW %s.%s AS SELECT id, name FROM %s.%s WHERE id IS NOT NULL PRIMARY KEY (id)",
		keyspace, view, keyspace, table,
	)
	if err := session.Query(createView).Exec(); err != nil {
		t.Fatalf("create view failed: %v", err)
	}

	waitFor(t, "system_schema.views create", 5*time.Second, func() (bool, error) {
		query := fmt.Sprintf("SELECT view_name, base_table_name, where_clause FROM system_schema.views WHERE keyspace_name = '%s'", keyspace)
		rows, err := scanRows(session, query)
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["view_name"]) != view {
				continue
			}
			if asString(row["base_table_name"]) != table {
				return false, nil
			}
			whereClause := strings.ToLower(asString(row["where_clause"]))
			return strings.Contains(whereClause, "id is not null"), nil
		}
		return false, nil
	})

	if err := session.Query(fmt.Sprintf("DROP MATERIALIZED VIEW %s.%s", keyspace, view)).Exec(); err != nil {
		t.Fatalf("drop view failed: %v", err)
	}
	waitFor(t, "system_schema.views drop", 5*time.Second, func() (bool, error) {
		query := fmt.Sprintf("SELECT view_name FROM system_schema.views WHERE keyspace_name = '%s'", keyspace)
		rows, err := scanRows(session, query)
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["view_name"]) == view {
				return false, nil
			}
		}
		return true, nil
	})
}

func TestViewMetadataOnlyWarnings(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 := "view_meta_only"
	if err := session.Query(createTableCQL(keyspace, table)).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	withFeatureFlags(t, func(features *config.FeatureConfig) {
		features.MetadataOnlyViews = true
	})

	view := "mv_meta_only"
	createView := fmt.Sprintf(
		"CREATE MATERIALIZED VIEW %s.%s AS SELECT id FROM %s.%s WHERE now() > 0 PRIMARY KEY (id)",
		keyspace, view, keyspace, table,
	)

	client := newRawProtocolConn(t, cql.ProtoVersion4, rawProtocolOptions{})
	resp := executeRawQuery(t, client, createView, 1)
	if resp.Opcode == cql.OpcodeError {
		code, msg := parseError(t, resp)
		t.Fatalf("create view failed: code=%d msg=%q", code, msg)
	}
	expected := "view definition stored in metadata only; MySQL view not created"
	if !containsWarning(resp.Warnings, expected) {
		t.Fatalf("expected warning %q, got %v", expected, resp.Warnings)
	}

	waitFor(t, "system_schema.views metadata-only create", 5*time.Second, func() (bool, error) {
		query := fmt.Sprintf("SELECT view_name, base_table_name, where_clause FROM system_schema.views WHERE keyspace_name = '%s'", keyspace)
		rows, err := scanRows(session, query)
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["view_name"]) != view {
				continue
			}
			if asString(row["base_table_name"]) != table {
				return false, nil
			}
			whereClause := strings.ToLower(asString(row["where_clause"]))
			return strings.Contains(whereClause, "now() > 0"), nil
		}
		return false, nil
	})

	if err := session.Query(fmt.Sprintf("DROP MATERIALIZED VIEW %s.%s", keyspace, view)).Exec(); err != nil {
		t.Fatalf("drop view failed: %v", err)
	}
	waitFor(t, "system_schema.views metadata-only drop", 5*time.Second, func() (bool, error) {
		query := fmt.Sprintf("SELECT view_name FROM system_schema.views WHERE keyspace_name = '%s'", keyspace)
		rows, err := scanRows(session, query)
		if err != nil {
			return false, err
		}
		for _, row := range rows {
			if asString(row["view_name"]) == view {
				return false, nil
			}
		}
		return true, nil
	})
}

func TestTriggerDDLSystemSchema(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 := "trigger_table"
	if err := session.Query(createTableCQL(keyspace, table)).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	trigger := "tr_meta"
	createTrigger := fmt.Sprintf("CREATE TRIGGER %s ON %s.%s USING 'com.example.Trigger'", trigger, keyspace, table)
	err := session.Query(createTrigger).Exec()
	requireErrorCode(t, err, cql.ErrorCodeInvalid)

	withFeatureFlags(t, func(features *config.FeatureConfig) {
		features.MetadataOnlyTriggers = true
	})

	client := newRawProtocolConn(t, cql.ProtoVersion4, rawProtocolOptions{})
	resp := executeRawQuery(t, client, createTrigger, 1)
	if resp.Opcode == cql.OpcodeError {
		code, msg := parseError(t, resp)
		t.Fatalf("create trigger failed: code=%d msg=%q", code, msg)
	}
	expected := "trigger stored in metadata only; triggers are not executed"
	if !containsWarning(resp.Warnings, expected) {
		t.Fatalf("expected warning %q, got %v", expected, resp.Warnings)
	}

	waitFor(t, "system_schema.triggers create", 5*time.Second, func() (bool, error) {
		query := fmt.Sprintf("SELECT trigger_name, table_name, options 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"]) != trigger {
				continue
			}
			if asString(row["table_name"]) != table {
				return false, nil
			}
			return optionClass(row["options"]) == "com.example.Trigger", nil
		}
		return false, nil
	})

	dropTrigger := fmt.Sprintf("DROP TRIGGER %s ON %s.%s", trigger, keyspace, table)
	if err := session.Query(dropTrigger).Exec(); err != nil {
		t.Fatalf("drop trigger failed: %v", err)
	}
	waitFor(t, "system_schema.triggers drop", 5*time.Second, func() (bool, error) {
		query := fmt.Sprintf("SELECT trigger_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"]) == trigger {
				return false, nil
			}
		}
		return true, nil
	})
}

func withFeatureFlags(t *testing.T, update func(*config.FeatureConfig)) {
	t.Helper()
	current := config.GetConfig()
	if current == nil {
		current = config.DefaultConfig()
	}
	clone := *current
	update(&clone.Features)
	config.SetConfig(&clone)
	t.Cleanup(func() {
		config.SetConfig(current)
	})
}

func matchTypeRow(rows []map[string]interface{}, typeName string, fieldNames, fieldTypes []string) bool {
	for _, row := range rows {
		if asString(row["type_name"]) != typeName {
			continue
		}
		names := stringList(row["field_names"])
		types := stringList(row["field_types"])
		return stringSliceEqual(names, fieldNames) && stringSliceEqual(types, fieldTypes)
	}
	return false
}

func typeRowExists(rows []map[string]interface{}, typeName string) bool {
	for _, row := range rows {
		if asString(row["type_name"]) == typeName {
			return true
		}
	}
	return false
}

func stringList(value interface{}) []string {
	switch v := value.(type) {
	case []string:
		return v
	case []interface{}:
		out := make([]string, 0, len(v))
		for _, item := range v {
			out = append(out, asString(item))
		}
		return out
	default:
		return nil
	}
}

func stringSliceEqual(a, b []string) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

func containsWarning(warnings []string, target string) bool {
	for _, warning := range warnings {
		if warning == target {
			return true
		}
	}
	return false
}

func optionClass(value interface{}) string {
	switch v := value.(type) {
	case map[string]string:
		return v["class"]
	case map[string]interface{}:
		if raw, ok := v["class"]; ok {
			return asString(raw)
		}
	}
	return ""
}