//go:build integration
// +build integration

package integration_test

import (
	"fmt"
	"strings"
	"testing"

	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"go.uber.org/zap/zaptest/observer"
)

func TestPreparedSelectMetadataBindsWhereAndLimitMarkers(t *testing.T) {
	requireMySQLConfig(t)

	core, observed := observer.New(zapcore.DebugLevel)
	log := zap.New(core)
	host, port, stop, _, err := startProxyWithConfigAndPoolLogger(mysqlCfg, nil, log)
	if err != nil {
		t.Fatalf("proxy start failed: %v", err)
	}
	defer stop()

	session := newSessionForHostPort(t, host, port)
	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 := "prepared_meta_markers"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (user_id int, session_id int, session_type int, name text, PRIMARY KEY (user_id, session_id, session_type))",
		keyspace,
		table,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}
	insert := fmt.Sprintf("INSERT INTO %s.%s (user_id, session_id, session_type, name) VALUES (?, ?, ?, ?)", keyspace, table)
	if err := session.Query(insert, 7, 100, 2, "alpha").Exec(); err != nil {
		t.Fatalf("insert row failed: %v", err)
	}

	client := newRawProtocolConnForAddr(t, host, port, cql.ProtoVersion4, rawProtocolOptions{})
	stmt := fmt.Sprintf(
		"SELECT name FROM %s.%s WHERE user_id = ? AND session_id = ? AND session_type = ? LIMIT ?",
		keyspace,
		table,
	)
	prepare := buildPrepareFrame(client.version, stmt, "")
	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)
	}

	result := parsePreparedResult(t, resp)
	if len(result.columns) != 1 {
		t.Fatalf("expected prepared result metadata for one column, got %+v", result.columns)
	}
	if result.columns[0].name != "name" {
		t.Fatalf("expected result column name, got %+v", result.columns[0])
	}

	execute := buildExecuteFrame(
		client.version,
		result.id,
		nil,
		cql.ConsistencyOne,
		cql.QueryFlagValues,
		[][]byte{encodeInt32(7), encodeInt32(100), encodeInt32(2), encodeInt32(1)},
		0,
		"",
	)
	execute.Stream = 2
	client.writeFrame(execute)
	execResp := client.readFrame()
	if execResp.Opcode == cql.OpcodeError {
		code, msg := parseError(t, execResp)
		t.Fatalf("execute failed: code=%d msg=%q", code, msg)
	}

	for _, entry := range observed.All() {
		if entry.Message == "failed to get column metadata" {
			t.Fatalf("metadata query failed during PREPARE: %v", entry.Context)
		}
		for _, field := range entry.Context {
			value := field.String
			if field.Interface != nil {
				value = fmt.Sprint(field.Interface)
			}
			if strings.Contains(value, "unexpected '?'") {
				t.Fatalf("unexpected TiDB parser error in logs: msg=%q field=%s value=%q", entry.Message, field.Key, value)
			}
		}
	}
}

func TestPreparedSelectMetadataBindsTextAndBlobMarkers(t *testing.T) {
	requireMySQLConfig(t)

	host, port, stop, err := startProxyWithConfig(mysqlCfg, nil)
	if err != nil {
		t.Fatalf("proxy start failed: %v", err)
	}
	defer stop()

	session := newSessionForHostPort(t, host, port)
	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 := "prepared_meta_text_blob"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (id int PRIMARY KEY, body text, payload blob)",
		keyspace,
		table,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	client := newRawProtocolConnForAddr(t, host, port, cql.ProtoVersion4, rawProtocolOptions{})
	stmt := fmt.Sprintf(
		"SELECT id FROM %s.%s WHERE body = ? AND payload = ? ALLOW FILTERING",
		keyspace,
		table,
	)
	prepare := buildPrepareFrame(client.version, stmt, "")
	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)
	}

	result := parsePreparedResult(t, resp)
	if len(result.columns) != 1 || result.columns[0].name != "id" {
		t.Fatalf("expected prepared result metadata for id column, got %+v", result.columns)
	}
}

func TestPreparedParameterMetadataUsesCQLColumnTypes(t *testing.T) {
	requireMySQLConfig(t)

	host, port, stop, err := startProxyWithConfig(mysqlCfg, nil)
	if err != nil {
		t.Fatalf("proxy start failed: %v", err)
	}
	defer stop()

	session := newSessionForHostPort(t, host, port)
	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 := "prepared_param_types"
	createTable := fmt.Sprintf(
		"CREATE TABLE %s.%s (user_id bigint, session_id bigint, body text, PRIMARY KEY (user_id, session_id))",
		keyspace,
		table,
	)
	if err := session.Query(createTable).Exec(); err != nil {
		t.Fatalf("create table failed: %v", err)
	}

	client := newRawProtocolConnForAddr(t, host, port, cql.ProtoVersion4, rawProtocolOptions{})
	stmt := fmt.Sprintf(
		"SELECT body FROM %s.%s WHERE user_id = ? AND session_id = ? LIMIT ?",
		keyspace,
		table,
	)
	prepare := buildPrepareFrame(client.version, stmt, "")
	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)
	}

	result := parsePreparedResult(t, resp)
	if len(result.params) != 3 {
		t.Fatalf("expected 3 prepared params, got %+v", result.params)
	}
	expected := []struct {
		typ uint16
	}{
		{typ: cql.TypeBigint},
		{typ: cql.TypeBigint},
		{typ: cql.TypeInt},
	}
	for i, want := range expected {
		got := result.params[i]
		if got.typ != want.typ {
			t.Fatalf("param %d = {name:%q type:0x%04x}, want type:0x%04x; all params=%+v", i, got.name, got.typ, want.typ, result.params)
		}
	}
}

type preparedResultColumn struct {
	keyspace string
	table    string
	name     string
	typ      uint16
}

type preparedResultInfo struct {
	id      []byte
	params  []preparedResultColumn
	columns []preparedResultColumn
}

func parsePreparedResult(t *testing.T, frame *cql.Frame) preparedResultInfo {
	t.Helper()
	parser := cql.NewFrameBodyParser(frame.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)
	}
	id, err := parser.ReadShortBytes()
	if err != nil {
		t.Fatalf("read prepared id failed: %v", err)
	}

	// Prepared metadata.
	if _, err := parser.ReadInt(); err != nil {
		t.Fatalf("read prepared metadata flags failed: %v", err)
	}
	preparedCount, err := parser.ReadInt()
	if err != nil {
		t.Fatalf("read prepared metadata count failed: %v", err)
	}
	pkCount, err := parser.ReadInt()
	if err != nil {
		t.Fatalf("read prepared pk count failed: %v", err)
	}
	for i := int32(0); i < pkCount; i++ {
		if _, err := parser.ReadShort(); err != nil {
			t.Fatalf("read prepared pk index failed: %v", err)
		}
	}
	params := make([]preparedResultColumn, 0, preparedCount)
	for i := int32(0); i < preparedCount; i++ {
		params = append(params, readColumnSpec(t, parser, false, "", ""))
	}

	// Result metadata.
	flags, err := parser.ReadInt()
	if err != nil {
		t.Fatalf("read result metadata flags failed: %v", err)
	}
	columnCount, err := parser.ReadInt()
	if err != nil {
		t.Fatalf("read result metadata count failed: %v", err)
	}
	global := flags&0x0001 != 0
	globalKeyspace := ""
	globalTable := ""
	if global {
		globalKeyspace, err = parser.ReadString()
		if err != nil {
			t.Fatalf("read global keyspace failed: %v", err)
		}
		globalTable, err = parser.ReadString()
		if err != nil {
			t.Fatalf("read global table failed: %v", err)
		}
	}
	columns := make([]preparedResultColumn, 0, columnCount)
	for i := int32(0); i < columnCount; i++ {
		columns = append(columns, readColumnSpec(t, parser, global, globalKeyspace, globalTable))
	}
	return preparedResultInfo{id: id, params: params, columns: columns}
}

func skipColumnSpec(t *testing.T, parser *cql.FrameBodyParser, global bool, globalKeyspace, globalTable string) {
	t.Helper()
	_ = readColumnSpec(t, parser, global, globalKeyspace, globalTable)
}

func readColumnSpec(t *testing.T, parser *cql.FrameBodyParser, global bool, globalKeyspace, globalTable string) preparedResultColumn {
	t.Helper()
	keyspace := globalKeyspace
	table := globalTable
	var err error
	if !global {
		keyspace, err = parser.ReadString()
		if err != nil {
			t.Fatalf("read column keyspace failed: %v", err)
		}
		table, err = parser.ReadString()
		if err != nil {
			t.Fatalf("read column table failed: %v", err)
		}
	}
	name, err := parser.ReadString()
	if err != nil {
		t.Fatalf("read column name failed: %v", err)
	}
	typ, err := readTypeOption(parser)
	if err != nil {
		t.Fatalf("skip column type failed: %v", err)
	}
	return preparedResultColumn{keyspace: keyspace, table: table, name: name, typ: typ}
}

func skipTypeOption(parser *cql.FrameBodyParser) error {
	_, err := readTypeOption(parser)
	return err
}

func readTypeOption(parser *cql.FrameBodyParser) (uint16, error) {
	typ, err := parser.ReadShort()
	if err != nil {
		return 0, err
	}
	switch typ {
	case 0x0020, 0x0021:
		_, err := readTypeOption(parser)
		return typ, err
	case 0x0022:
		if _, err := readTypeOption(parser); err != nil {
			return typ, err
		}
		_, err := readTypeOption(parser)
		return typ, err
	case 0x0030:
		_, err = parser.ReadString()
		return typ, err
	case 0x0031:
		_, err = parser.ReadShort()
		return typ, err
	default:
		return typ, nil
	}
}