//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"testing"
	"time"

	"github.com/apache/cassandra-gocql-driver/v2"
)

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

	baseline, err := readSchemaVersion(session)
	if err != nil {
		t.Fatalf("read schema_version failed: %v", err)
	}
	if baseline == "" {
		t.Fatalf("expected non-empty schema_version")
	}

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

	waitFor(t, "schema_version bump", 5*time.Second, func() (bool, error) {
		current, err := readSchemaVersion(session)
		if err != nil {
			return false, err
		}
		return current != "" && current != baseline, nil
	})
}

func readSchemaVersion(session *gocql.Session) (string, error) {
	rows, err := scanRows(session, "SELECT schema_version FROM system.local WHERE key = 'local'")
	if err != nil {
		return "", err
	}
	if len(rows) == 0 {
		return "", fmt.Errorf("no system.local rows")
	}
	return asString(rows[0]["schema_version"]), nil
}