//go:build integration
// +build integration

package integration_test

import (
	"encoding/binary"
	"fmt"
	"strings"
	"testing"

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

type rowsResult struct {
	flags          int32
	pagingState    []byte
	globalKeyspace string
	globalTable    string
	rows           [][][]byte
}

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

	insert := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)
	if err := session.Query(insert, 1, "alpha", 1).Exec(); err != nil {
		t.Fatalf("insert failed: %v", err)
	}

	client := newRawProtocolConn(t, cql.ProtoVersion4, rawProtocolOptions{})
	query := fmt.Sprintf("SELECT name FROM %s.%s WHERE id = ?", keyspace, table)
	frame := cql.BuildQueryFrame(query, cql.ConsistencyOne, cql.QueryFlagValues, []cql.QueryValue{
		{Value: encodeInt32(1)},
	}, 0, nil, 0, 0, "", nil)
	frame.Stream = 1
	client.writeFrame(frame)
	resp := client.readFrame()
	if resp.Opcode == cql.OpcodeError {
		code, msg := parseError(t, resp)
		t.Fatalf("query failed: code=%d msg=%q", code, msg)
	}

	result := parseRowsResult(t, resp)
	if len(result.rows) != 1 || len(result.rows[0]) != 1 {
		t.Fatalf("expected 1 row/1 column, got %d rows", len(result.rows))
	}
	name := string(result.rows[0][0])
	if name != "alpha" {
		t.Fatalf("unexpected name: %q", name)
	}
}

func TestProtocolQueryValuesNamed(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 := "named_values"
	create := fmt.Sprintf("CREATE TABLE %s.%s (id int, idx int, name text, PRIMARY KEY (id, idx))", keyspace, table)
	if err := session.Query(create).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}
	insert := fmt.Sprintf("INSERT INTO %s.%s (id, idx, name) VALUES (?, ?, ?)", keyspace, table)
	if err := session.Query(insert, 1, 2, "beta").Exec(); err != nil {
		t.Fatalf("insert failed: %v", err)
	}

	client := newRawProtocolConn(t, cql.ProtoVersion4, rawProtocolOptions{})
	query := fmt.Sprintf("SELECT name FROM %s.%s WHERE id = :id AND idx = :idx", keyspace, table)
	flags := uint32(cql.QueryFlagValues | cql.QueryFlagWithNameValues)
	values := []cql.QueryValue{
		{Name: "idx", Value: encodeInt32(2)},
		{Name: "id", Value: encodeInt32(1)},
	}
	frame := cql.BuildQueryFrame(query, cql.ConsistencyOne, flags, values, 0, nil, 0, 0, "", nil)
	frame.Stream = 2
	client.writeFrame(frame)
	resp := client.readFrame()
	if resp.Opcode == cql.OpcodeError {
		code, msg := parseError(t, resp)
		t.Fatalf("query failed: code=%d msg=%q", code, msg)
	}

	result := parseRowsResult(t, resp)
	if len(result.rows) != 1 || len(result.rows[0]) != 1 {
		t.Fatalf("expected 1 row/1 column, got %d rows", len(result.rows))
	}
	name := string(result.rows[0][0])
	if name != "beta" {
		t.Fatalf("unexpected name: %q", name)
	}
}

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

	client := newRawProtocolConn(t, cql.ProtoVersion4, rawProtocolOptions{})
	prepareStmt := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (:id, :name, :idx)", keyspace, table)
	prepare := buildPrepareFrame(client.version, prepareStmt, "")
	prepare.Stream = 1
	client.writeFrame(prepare)
	prepared := client.readFrame()
	stmtID := parsePreparedID(t, prepared)

	flags := uint32(cql.QueryFlagValues | cql.QueryFlagWithNameValues)
	values := []cql.QueryValue{
		{Name: "name", Value: encodeText("gamma")},
		{Name: "idx", Value: encodeInt32(3)},
		{Name: "id", Value: encodeInt32(3)},
	}
	execute := cql.BuildExecuteFrame(stmtID, nil, cql.ConsistencyOne, flags, values, 0, nil, 0, 0, "", nil)
	execute.Stream = 2
	client.writeFrame(execute)
	resp := client.readFrame()
	if parseResultKind(t, resp) != cql.ResultKindVoid {
		t.Fatalf("expected VOID result")
	}

	var name string
	query := fmt.Sprintf("SELECT name FROM %s.%s WHERE id = ?", keyspace, table)
	if err := session.Query(query, 3).Scan(&name); err != nil {
		t.Fatalf("select failed: %v", err)
	}
	if name != "gamma" {
		t.Fatalf("unexpected name: %q", name)
	}
}

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

	frame := cql.BuildQueryFrame("SELECT key FROM system.local", cql.ConsistencyOne, 0, nil, 0, nil, 0, 0, "", nil)
	frame.Stream = 1
	frame.CustomPayload = cql.CustomPayload{
		"trace-id": []byte("payload"),
	}
	client.writeFrame(frame)
	resp := client.readFrame()
	if resp.Opcode != cql.OpcodeResult {
		t.Fatalf("expected RESULT, got opcode %d", resp.Opcode)
	}
	got := resp.CustomPayload["trace-id"]
	if string(got) != "payload" {
		t.Fatalf("unexpected custom payload: %q", string(got))
	}
}

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

	client := newRawProtocolConn(t, cql.ProtoVersion4, rawProtocolOptions{})
	register := cql.BuildRegisterFrame([]string{cql.EventSchemaChange})
	register.Stream = 1
	client.writeFrame(register)
	resp := client.readFrame()
	if resp.Opcode != cql.OpcodeReady {
		t.Fatalf("expected READY, got opcode %d", resp.Opcode)
	}

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

	event := client.readFrame()
	if event.Opcode != cql.OpcodeEvent {
		t.Fatalf("expected EVENT, got opcode %d", event.Opcode)
	}
	change := parseSchemaChangeEvent(t, event)
	if change.changeType != "CREATED" || change.target != "KEYSPACE" || change.keyspace != keyspace {
		t.Fatalf("unexpected schema change event: %#v", change)
	}
}

func TestProtocolPagingStateRoundTrip(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 := "paging_state"
	create := fmt.Sprintf("CREATE TABLE %s.%s (id int, idx int, name text, PRIMARY KEY (id, idx))", keyspace, table)
	if err := session.Query(create).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	insert := fmt.Sprintf("INSERT INTO %s.%s (id, idx, name) VALUES (?, ?, ?)", keyspace, table)
	for i := 1; i <= 3; i++ {
		if err := session.Query(insert, 1, i, fmt.Sprintf("n%d", i)).Exec(); err != nil {
			t.Fatalf("insert failed: %v", err)
		}
	}

	client := newRawProtocolConn(t, cql.ProtoVersion4, rawProtocolOptions{})
	query := fmt.Sprintf("SELECT idx FROM %s.%s WHERE id = ?", keyspace, table)
	values := []cql.QueryValue{{Value: encodeInt32(1)}}
	flags := uint32(cql.QueryFlagValues | cql.QueryFlagPageSize)
	frame := cql.BuildQueryFrame(query, cql.ConsistencyOne, flags, values, 2, nil, 0, 0, "", nil)
	frame.Stream = 3
	client.writeFrame(frame)
	resp := client.readFrame()
	if resp.Opcode == cql.OpcodeError {
		code, msg := parseError(t, resp)
		t.Fatalf("page query failed: code=%d msg=%q", code, msg)
	}

	firstPage := parseRowsResult(t, resp)
	if firstPage.flags&cql.ResultMetadataFlagHasMorePages == 0 {
		t.Fatalf("expected paging flag on first page")
	}
	if len(firstPage.pagingState) == 0 {
		t.Fatalf("expected paging state on first page")
	}
	if len(firstPage.rows) != 2 {
		t.Fatalf("expected 2 rows on first page, got %d", len(firstPage.rows))
	}
	if decodeInt32(firstPage.rows[0][0]) != 1 || decodeInt32(firstPage.rows[1][0]) != 2 {
		t.Fatalf("unexpected first page rows: %v", firstPage.rows)
	}

	flags = cql.QueryFlagValues | cql.QueryFlagPageSize | cql.QueryFlagWithPagingState
	frame = cql.BuildQueryFrame(query, cql.ConsistencyOne, flags, values, 2, firstPage.pagingState, 0, 0, "", nil)
	frame.Stream = 4
	client.writeFrame(frame)
	resp = client.readFrame()
	if resp.Opcode == cql.OpcodeError {
		code, msg := parseError(t, resp)
		t.Fatalf("page query failed: code=%d msg=%q", code, msg)
	}
	secondPage := parseRowsResult(t, resp)
	if secondPage.flags&cql.ResultMetadataFlagHasMorePages != 0 {
		t.Fatalf("expected no paging flag on second page")
	}
	if len(secondPage.rows) != 1 {
		t.Fatalf("expected 1 row on second page, got %d", len(secondPage.rows))
	}
	if decodeInt32(secondPage.rows[0][0]) != 3 {
		t.Fatalf("unexpected second page row: %v", secondPage.rows)
	}
}

func TestProtocolGlobalTableSpecFlag(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 := "global_spec"
	if err := session.Query(createTableCQL(keyspace, table)).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}
	insert := fmt.Sprintf("INSERT INTO %s.%s (id, name, idx) VALUES (?, ?, ?)", keyspace, table)
	if err := session.Query(insert, 1, "delta", 1).Exec(); err != nil {
		t.Fatalf("insert failed: %v", err)
	}

	client := newRawProtocolConn(t, cql.ProtoVersion4, rawProtocolOptions{})
	query := fmt.Sprintf("SELECT id, name FROM %s.%s WHERE id = ?", keyspace, table)
	frame := cql.BuildQueryFrame(query, cql.ConsistencyOne, cql.QueryFlagValues, []cql.QueryValue{
		{Value: encodeInt32(1)},
	}, 0, nil, 0, 0, "", nil)
	frame.Stream = 5
	client.writeFrame(frame)
	resp := client.readFrame()
	if resp.Opcode == cql.OpcodeError {
		code, msg := parseError(t, resp)
		t.Fatalf("query failed: code=%d msg=%q", code, msg)
	}

	result := parseRowsResult(t, resp)
	if result.flags&cql.ResultMetadataFlagGlobalTableSpec == 0 {
		t.Fatalf("expected GLOBAL_TABLES_SPEC flag")
	}
	if result.globalKeyspace != keyspace || result.globalTable != table {
		t.Fatalf("unexpected global spec: %s.%s", result.globalKeyspace, result.globalTable)
	}
}

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

	client := newRawProtocolConn(t, cql.ProtoVersion5, rawProtocolOptions{})
	prepare := buildPrepareFrame(client.version, fmt.Sprintf("SELECT name FROM %s.%s WHERE id = ?", keyspace, table), "")
	prepare.Stream = 1
	client.writeFrame(prepare)
	resp := client.readFrame()
	if resp.Opcode == cql.OpcodeError {
		code, msg := parseError(t, resp)
		t.Fatalf("prepare failed: code=%d msg=%q", code, msg)
	}

	parser := cql.NewFrameBodyParser(resp.Body)
	kind, err := parser.ReadInt()
	if err != nil {
		t.Fatalf("read result kind failed: %v", err)
	}
	if kind != int32(cql.ResultKindPrepared) {
		t.Fatalf("expected PREPARED result, got %d", kind)
	}
	if _, err := parser.ReadShortBytes(); err != nil {
		t.Fatalf("read prepared id failed: %v", err)
	}
	metaID, err := parser.ReadShortBytes()
	if err != nil {
		t.Fatalf("read metadata id failed: %v", err)
	}
	if len(metaID) == 0 {
		t.Fatalf("expected non-empty metadata id")
	}
}

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

	client := newRawProtocolConn(t, cql.ProtoVersion4, rawProtocolOptions{})
	prepare := buildPrepareFrame(client.version, fmt.Sprintf("SELECT * FROM %s.%s WHERE name = ?", keyspace, table), "")
	prepare.Stream = 1
	client.writeFrame(prepare)
	resp := client.readFrame()
	if resp.Opcode != cql.OpcodeError {
		t.Fatalf("expected ERROR frame, got %d", resp.Opcode)
	}

	code, msg := parseError(t, resp)
	if code != cql.ErrorCodeInvalid {
		t.Fatalf("expected invalid code, got %d with message %q", code, msg)
	}
	if !strings.Contains(msg, "ALLOW FILTERING") {
		t.Fatalf("expected ALLOW FILTERING hint, got %q", msg)
	}
	if strings.Contains(strings.ToUpper(msg), "SQL") {
		t.Fatalf("expected sanitized prepare error without SQL details, got %q", msg)
	}
}

func parseRowsResult(t *testing.T, frame *cql.Frame) rowsResult {
	t.Helper()
	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.ResultKindRows) {
		t.Fatalf("expected ROWS result, got %d", kind)
	}
	flags, err := parser.ReadInt()
	if err != nil {
		t.Fatalf("read metadata flags failed: %v", err)
	}
	colCount, err := parser.ReadInt()
	if err != nil {
		t.Fatalf("read column count failed: %v", err)
	}

	result := rowsResult{flags: flags}
	if flags&cql.ResultMetadataFlagHasMorePages != 0 {
		state, err := parser.ReadBytes()
		if err != nil {
			t.Fatalf("read paging state failed: %v", err)
		}
		result.pagingState = state
	}
	global := flags&cql.ResultMetadataFlagGlobalTableSpec != 0
	if global {
		keyspace, err := parser.ReadString()
		if err != nil {
			t.Fatalf("read global keyspace failed: %v", err)
		}
		table, err := parser.ReadString()
		if err != nil {
			t.Fatalf("read global table failed: %v", err)
		}
		result.globalKeyspace = keyspace
		result.globalTable = table
	}
	if flags&cql.ResultMetadataFlagNoMetaData == 0 {
		for i := 0; i < int(colCount); i++ {
			if !global {
				if _, err := parser.ReadString(); err != nil {
					t.Fatalf("read column keyspace failed: %v", err)
				}
				if _, err := parser.ReadString(); err != nil {
					t.Fatalf("read column table failed: %v", err)
				}
			}
			if _, err := parser.ReadString(); err != nil {
				t.Fatalf("read column name failed: %v", err)
			}
			if err := skipTypeInfo(parser); err != nil {
				t.Fatalf("read column type failed: %v", err)
			}
		}
	}
	rowCount, err := parser.ReadInt()
	if err != nil {
		t.Fatalf("read row count failed: %v", err)
	}
	rows := make([][][]byte, 0, int(rowCount))
	for i := 0; i < int(rowCount); i++ {
		row := make([][]byte, int(colCount))
		for j := 0; j < int(colCount); j++ {
			val, err := parser.ReadBytes()
			if err != nil {
				t.Fatalf("read row value failed: %v", err)
			}
			row[j] = val
		}
		rows = append(rows, row)
	}
	result.rows = rows
	return result
}

func decodeInt32(raw []byte) int32 {
	if len(raw) < 4 {
		return 0
	}
	return int32(binary.BigEndian.Uint32(raw))
}

func parseSchemaChangeEvent(t *testing.T, frame *cql.Frame) schemaChangeInfo {
	t.Helper()
	parser := cql.NewFrameBodyParser(frame.Body)
	eventType, err := parser.ReadString()
	if err != nil {
		t.Fatalf("read event type failed: %v", err)
	}
	if eventType != cql.EventSchemaChange {
		t.Fatalf("unexpected event type: %q", eventType)
	}
	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,
	}
}