package cql

import (
	"net"
	"testing"

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

func TestSchemaChangeEventFrame(t *testing.T) {
	change := &types.SchemaChange{
		ChangeType: "CREATED",
		Target:     "TABLE",
		Keyspace:   "ks",
		Name:       "tbl",
	}

	frame, err := buildSchemaChangeEventFrame(ProtoVersion4, change)
	if err != nil {
		t.Fatalf("buildSchemaChangeEventFrame failed: %v", err)
	}
	if frame.Opcode != OpcodeEvent {
		t.Fatalf("opcode = %d, want %d", frame.Opcode, OpcodeEvent)
	}
	if frame.Stream != -1 {
		t.Fatalf("stream = %d, want -1", frame.Stream)
	}
	if frame.Version != ProtoVersion4Response {
		t.Fatalf("version = 0x%02x, want 0x%02x", frame.Version, ProtoVersion4Response)
	}

	pos := 0
	eventType, n, err := readString(frame.Body[pos:])
	if err != nil {
		t.Fatalf("read event type failed: %v", err)
	}
	pos += n
	if eventType != schemaChangeEventType {
		t.Fatalf("event type = %q, want %q", eventType, schemaChangeEventType)
	}
	changeType, n, err := readString(frame.Body[pos:])
	if err != nil {
		t.Fatalf("read change type failed: %v", err)
	}
	pos += n
	if changeType != change.ChangeType {
		t.Fatalf("change type = %q, want %q", changeType, change.ChangeType)
	}
	target, n, err := readString(frame.Body[pos:])
	if err != nil {
		t.Fatalf("read target failed: %v", err)
	}
	pos += n
	if target != change.Target {
		t.Fatalf("target = %q, want %q", target, change.Target)
	}
	keyspace, n, err := readString(frame.Body[pos:])
	if err != nil {
		t.Fatalf("read keyspace failed: %v", err)
	}
	pos += n
	if keyspace != change.Keyspace {
		t.Fatalf("keyspace = %q, want %q", keyspace, change.Keyspace)
	}
	name, _, err := readString(frame.Body[pos:])
	if err != nil {
		t.Fatalf("read name failed: %v", err)
	}
	if name != change.Name {
		t.Fatalf("name = %q, want %q", name, change.Name)
	}
}

func TestBroadcastSchemaChange(t *testing.T) {
	server := NewServer("127.0.0.1:0", nil, zap.NewNop(), nil)
	serverConn, clientConn := net.Pipe()
	defer serverConn.Close()
	defer clientConn.Close()

	conn := NewConn(serverConn, server)
	conn.AddEventSubscriptions([]string{"schema_change"})

	otherConn := NewConn(clientConn, server)
	server.mu.Lock()
	server.connections["a"] = conn
	server.connections["b"] = otherConn
	server.mu.Unlock()

	change := &types.SchemaChange{
		ChangeType: "UPDATED",
		Target:     "KEYSPACE",
		Keyspace:   "ks",
	}
	server.BroadcastSchemaChange(change)

	select {
	case frame := <-conn.eventCh:
		if frame == nil {
			t.Fatalf("received nil frame")
		}
		if frame.Opcode != OpcodeEvent {
			t.Fatalf("opcode = %d, want %d", frame.Opcode, OpcodeEvent)
		}
		if frame.Version != ProtoVersion4Response {
			t.Fatalf("version = 0x%02x, want 0x%02x", frame.Version, ProtoVersion4Response)
		}
	default:
		t.Fatalf("expected event for subscribed connection")
	}

	select {
	case <-otherConn.eventCh:
		t.Fatalf("unexpected event for unsubscribed connection")
	default:
	}
}

func TestBroadcastSchemaChangeForTenant(t *testing.T) {
	server := NewServer("127.0.0.1:0", nil, zap.NewNop(), nil)
	serverConn, clientConn := net.Pipe()
	defer serverConn.Close()
	defer clientConn.Close()

	conn := NewConn(serverConn, server)
	conn.SetTenant("tenant-a", nil)
	conn.AddEventSubscriptions([]string{"schema_change"})

	otherConn := NewConn(clientConn, server)
	otherConn.SetTenant("tenant-b", nil)
	otherConn.AddEventSubscriptions([]string{"schema_change"})

	server.mu.Lock()
	server.connections["a"] = conn
	server.connections["b"] = otherConn
	server.mu.Unlock()

	change := &types.SchemaChange{
		ChangeType: "UPDATED",
		Target:     "KEYSPACE",
		Keyspace:   "ks",
	}
	server.BroadcastSchemaChangeForTenant(change, "tenant-a")

	select {
	case frame := <-conn.eventCh:
		if frame == nil {
			t.Fatalf("received nil frame")
		}
		if frame.Opcode != OpcodeEvent {
			t.Fatalf("opcode = %d, want %d", frame.Opcode, OpcodeEvent)
		}
	default:
		t.Fatalf("expected event for tenant-a connection")
	}

	select {
	case <-otherConn.eventCh:
		t.Fatalf("unexpected event for tenant-b connection")
	default:
	}
}