//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"testing"

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

type schemaChangeInfo struct {
	changeType string
	target     string
	keyspace   string
	name       string
}

func TestSchemaChangeResultFrames(t *testing.T) {
	client := newRawProtocolConn(t, cql.ProtoVersion4, rawProtocolOptions{})

	keyspace := uniqueKeyspace()
	resp := executeRawQuery(t, client, createKeyspaceCQL(keyspace), 1)
	info := parseSchemaChange(t, resp)
	if info.changeType != "CREATED" || info.target != "KEYSPACE" || info.keyspace != keyspace {
		t.Fatalf("unexpected keyspace schema change: %#v", info)
	}

	table := "schema_changes"
	createTable := fmt.Sprintf("CREATE TABLE %s.%s (id int PRIMARY KEY, name text, idx int)", keyspace, table)
	resp = executeRawQuery(t, client, createTable, 2)
	info = parseSchemaChange(t, resp)
	if info.changeType != "CREATED" || info.target != "TABLE" || info.keyspace != keyspace || info.name != table {
		t.Fatalf("unexpected table schema change: %#v", info)
	}

	index := "schema_changes_idx"
	createIndex := createIndexCQL(index, keyspace, table, "idx")
	resp = executeRawQuery(t, client, createIndex, 3)
	info = parseSchemaChange(t, resp)
	if info.changeType != "UPDATED" || info.target != "TABLE" || info.keyspace != keyspace || info.name != table {
		t.Fatalf("unexpected index create schema change: %#v", info)
	}

	dropIndex := fmt.Sprintf("DROP INDEX %s.%s", keyspace, index)
	resp = executeRawQuery(t, client, dropIndex, 4)
	info = parseSchemaChange(t, resp)
	if info.changeType != "UPDATED" || info.target != "TABLE" || info.keyspace != keyspace || info.name != table {
		t.Fatalf("unexpected index drop schema change: %#v", info)
	}

	dropTable := fmt.Sprintf("DROP TABLE %s.%s", keyspace, table)
	resp = executeRawQuery(t, client, dropTable, 5)
	info = parseSchemaChange(t, resp)
	if info.changeType != "DROPPED" || info.target != "TABLE" || info.keyspace != keyspace || info.name != table {
		t.Fatalf("unexpected table drop schema change: %#v", info)
	}

	dropKeyspace := fmt.Sprintf("DROP KEYSPACE %s", keyspace)
	resp = executeRawQuery(t, client, dropKeyspace, 6)
	info = parseSchemaChange(t, resp)
	if info.changeType != "DROPPED" || info.target != "KEYSPACE" || info.keyspace != keyspace {
		t.Fatalf("unexpected keyspace drop schema change: %#v", info)
	}
}

func parseSchemaChange(t *testing.T, frame *cql.Frame) schemaChangeInfo {
	t.Helper()
	if frame.Opcode == cql.OpcodeError {
		code, msg := parseError(t, frame)
		t.Fatalf("schema change failed: code=%d msg=%q", code, msg)
	}
	if frame.Opcode != cql.OpcodeResult {
		t.Fatalf("expected RESULT frame, got %d", frame.Opcode)
	}
	parser := cql.NewFrameBodyParser(frame.Body)
	kind, err := parser.ReadInt()
	if err != nil {
		t.Fatalf("read result kind failed: %v", err)
	}
	if kind != int32(cql.ResultKindSchemaChange) {
		t.Fatalf("expected schema change result, got %d", kind)
	}
	changeType, err := parser.ReadString()
	if err != nil {
		t.Fatalf("read change type failed: %v", err)
	}
	target, err := parser.ReadString()
	if err != nil {
		t.Fatalf("read target failed: %v", err)
	}
	keyspace, err := parser.ReadString()
	if err != nil {
		t.Fatalf("read keyspace failed: %v", err)
	}
	name := ""
	if target == "TABLE" || target == "TYPE" {
		name, err = parser.ReadString()
		if err != nil {
			t.Fatalf("read name failed: %v", err)
		}
	}
	return schemaChangeInfo{
		changeType: changeType,
		target:     target,
		keyspace:   keyspace,
		name:       name,
	}
}