package cql

import (
	"net"
	"testing"
	"time"

	"go.uber.org/zap"
)

func TestConnServeV5ClassicFramesRemainClassicAfterStartup(t *testing.T) {
	handler := NewMessageHandler(zap.NewNop(), &AllowAllAuthenticator{})
	server := NewServer("127.0.0.1:0", handler, zap.NewNop(), &ServerOptions{
		Authenticator: &AllowAllAuthenticator{},
	})

	clientConn, serverConn := net.Pipe()
	defer clientConn.Close()
	defer serverConn.Close()

	conn := NewConn(serverConn, server)
	errCh := make(chan error, 1)
	go func() {
		errCh <- conn.Serve()
	}()

	readResp := func() *Frame {
		t.Helper()
		_ = clientConn.SetReadDeadline(time.Now().Add(2 * time.Second))
		frame, err := ReadFrame(clientConn)
		if err != nil {
			t.Fatalf("read response failed: %v", err)
		}
		return frame
	}

	startup := BuildStartupFrame("3.4.0", "", "", "")
	startup.Version = ProtoVersion5
	startup.Stream = 1
	if err := startup.WriteFrame(clientConn); err != nil {
		t.Fatalf("write startup failed: %v", err)
	}
	startupResp := readResp()
	if startupResp.Opcode != OpcodeReady {
		t.Fatalf("expected READY for startup, got %s", MessageType(startupResp.Opcode).String())
	}

	options := BuildOptionsFrame()
	options.Version = ProtoVersion5
	options.Stream = 2
	if err := options.WriteFrame(clientConn); err != nil {
		t.Fatalf("write options failed: %v", err)
	}
	optionsResp := readResp()
	if optionsResp.Opcode != OpcodeSupported {
		t.Fatalf("expected SUPPORTED for options, got %s", MessageType(optionsResp.Opcode).String())
	}

	_ = clientConn.Close()
	select {
	case err := <-errCh:
		if err != nil {
			t.Fatalf("serve returned error: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for serve loop to exit")
	}
}

func TestConnServeReturnsNilWhenClientClosesBeforeFirstFrame(t *testing.T) {
	handler := NewMessageHandler(zap.NewNop(), &AllowAllAuthenticator{})
	server := NewServer("127.0.0.1:0", handler, zap.NewNop(), &ServerOptions{
		Authenticator: &AllowAllAuthenticator{},
	})

	clientConn, serverConn := net.Pipe()
	defer serverConn.Close()

	conn := NewConn(serverConn, server)
	errCh := make(chan error, 1)
	go func() {
		errCh <- conn.Serve()
	}()

	_ = clientConn.Close()
	select {
	case err := <-errCh:
		if err != nil {
			t.Fatalf("serve returned error: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for serve loop to exit")
	}
}

func TestConnServeV5ClassicAuthResponseAllowedBeforeCredentialsSet(t *testing.T) {
	auth := NewPasswordAuthenticator()
	handler := NewMessageHandler(zap.NewNop(), auth)
	server := NewServer("127.0.0.1:0", handler, zap.NewNop(), &ServerOptions{
		Authenticator: auth,
	})

	clientConn, serverConn := net.Pipe()
	defer clientConn.Close()
	defer serverConn.Close()

	conn := NewConn(serverConn, server)
	errCh := make(chan error, 1)
	go func() {
		errCh <- conn.Serve()
	}()

	readResp := func() *Frame {
		t.Helper()
		_ = clientConn.SetReadDeadline(time.Now().Add(2 * time.Second))
		frame, err := ReadFrame(clientConn)
		if err != nil {
			t.Fatalf("read response failed: %v", err)
		}
		return frame
	}

	startup := BuildStartupFrame("3.4.0", "", "", "")
	startup.Version = ProtoVersion5
	startup.Stream = 1
	if err := startup.WriteFrame(clientConn); err != nil {
		t.Fatalf("write startup failed: %v", err)
	}
	startupResp := readResp()
	if startupResp.Opcode != OpcodeAuthenticate {
		t.Fatalf("expected AUTHENTICATE for startup, got %s", MessageType(startupResp.Opcode).String())
	}

	authResp := BuildAuthResponseFrame("abc.root", "xxxx")
	authResp.Version = ProtoVersion5
	authResp.Stream = 2
	if err := authResp.WriteFrame(clientConn); err != nil {
		t.Fatalf("write auth response failed: %v", err)
	}
	loginResp := readResp()
	if loginResp.Opcode != OpcodeAuthSuccess {
		t.Fatalf("expected AUTH_SUCCESS, got %s", MessageType(loginResp.Opcode).String())
	}
	deadline := time.Now().Add(200 * time.Millisecond)
	for !conn.IsStartupComplete() && time.Now().Before(deadline) {
		time.Sleep(5 * time.Millisecond)
	}
	if !conn.IsStartupComplete() {
		t.Fatal("expected startup complete after AUTH_SUCCESS")
	}

	_ = clientConn.Close()
	select {
	case err := <-errCh:
		if err != nil {
			t.Fatalf("serve returned error: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for serve loop to exit")
	}
}

func TestConnServeStartupCompleteRequiresReadyOrAuthSuccess(t *testing.T) {
	auth := NewPasswordAuthenticator()
	handler := NewMessageHandler(zap.NewNop(), auth)
	server := NewServer("127.0.0.1:0", handler, zap.NewNop(), &ServerOptions{
		Authenticator: auth,
	})

	clientConn, serverConn := net.Pipe()
	defer clientConn.Close()
	defer serverConn.Close()

	conn := NewConn(serverConn, server)
	errCh := make(chan error, 1)
	go func() {
		errCh <- conn.Serve()
	}()

	readResp := func() *Frame {
		t.Helper()
		_ = clientConn.SetReadDeadline(time.Now().Add(2 * time.Second))
		frame, err := ReadFrame(clientConn)
		if err != nil {
			t.Fatalf("read response failed: %v", err)
		}
		return frame
	}

	startup := BuildStartupFrame("3.4.0", "", "", "")
	startup.Version = ProtoVersion5
	startup.Stream = 1
	if err := startup.WriteFrame(clientConn); err != nil {
		t.Fatalf("write startup failed: %v", err)
	}
	startupResp := readResp()
	if startupResp.Opcode != OpcodeAuthenticate {
		t.Fatalf("expected AUTHENTICATE for startup, got %s", MessageType(startupResp.Opcode).String())
	}
	if conn.IsStartupComplete() {
		t.Fatal("startup should not be complete before AUTH_SUCCESS")
	}

	authResp := BuildAuthResponseFrame("abc.root", "xxxx")
	authResp.Version = ProtoVersion5
	authResp.Stream = 2
	if err := authResp.WriteFrame(clientConn); err != nil {
		t.Fatalf("write auth response failed: %v", err)
	}
	loginResp := readResp()
	if loginResp.Opcode != OpcodeAuthSuccess {
		t.Fatalf("expected AUTH_SUCCESS, got %s", MessageType(loginResp.Opcode).String())
	}
	deadline := time.Now().Add(200 * time.Millisecond)
	for !conn.IsStartupComplete() && time.Now().Before(deadline) {
		time.Sleep(5 * time.Millisecond)
	}
	if !conn.IsStartupComplete() {
		t.Fatal("expected startup complete after AUTH_SUCCESS")
	}

	_ = clientConn.Close()
	select {
	case err := <-errCh:
		if err != nil {
			t.Fatalf("serve returned error: %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for serve loop to exit")
	}
}