package streaming

import (
	"context"
	"errors"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/segmentio/kafka-go"

	"github.com/tidbcloud/service-proxy/pkg/checker/core"
)

func TestTestSingleBrokerReachability(t *testing.T) {
	lis, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("listen error: %v", err)
	}
	defer lis.Close()

	done := make(chan struct{})
	go func() {
		defer close(done)
		conn, err := lis.Accept()
		if err != nil {
			return
		}
		defer conn.Close()

		// Ensure the client closes promptly.
		_ = conn.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
		buf := make([]byte, 1)
		_, err = conn.Read(buf)
		if err != nil {
			if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) || errors.Is(err, context.DeadlineExceeded) {
				return
			}
			if ne, ok := err.(net.Error); ok && ne.Timeout() {
				return
			}
			t.Errorf("server side read error: %v", err)
		}
	}()

	host, portStr, _ := net.SplitHostPort(lis.Addr().String())
	port, _ := strconv.Atoi(portStr)

	if err := testSingleBrokerReachability(context.Background(), host, port); err != nil {
		t.Fatalf("expected reachable broker, got error: %v", err)
	}

	<-done

	// After closing listener, dialing should fail quickly.
	_ = lis.Close()
	if err := testSingleBrokerReachability(context.Background(), host, port); err == nil {
		t.Fatal("expected error after listener closed, got nil")
	}
}

func TestTestSingleBrokerReachabilityCancelledContext(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	err := testSingleBrokerReachability(ctx, "127.0.0.1", 1)
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("expected context canceled, got %v", err)
	}
}

func TestCreateKafkaConnectionCancelledContext(t *testing.T) {
	checker := NewKafkaChecker(&core.KafkaParams{
		Brokers: "127.0.0.1:1,127.0.0.1:2",
	})
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	start := time.Now()
	_, _, err := checker.createKafkaConnection(ctx)

	if !errors.Is(err, context.Canceled) {
		t.Fatalf("expected context canceled, got %v", err)
	}
	if elapsed := time.Since(start); elapsed > 500*time.Millisecond {
		t.Fatalf("expected canceled context to return quickly, took %s", elapsed)
	}
}

func TestCreateKafkaConnectionAggregatesEndpointErrorsInOrder(t *testing.T) {
	first := closedLocalAddr(t)
	second := closedLocalAddr(t)
	checker := NewKafkaChecker(&core.KafkaParams{
		Brokers: strings.Join([]string{first, second}, ","),
	})

	_, _, err := checker.createKafkaConnection(context.Background())
	if err == nil {
		t.Fatal("expected error, got nil")
	}

	errText := err.Error()
	firstText := "connect to '" + first + "' failed"
	secondText := "connect to '" + second + "' failed"
	firstIdx := strings.Index(errText, firstText)
	secondIdx := strings.Index(errText, secondText)
	if firstIdx < 0 {
		t.Fatalf("error %q does not contain first endpoint %q", errText, firstText)
	}
	if secondIdx < 0 {
		t.Fatalf("error %q does not contain second endpoint %q", errText, secondText)
	}
	if firstIdx > secondIdx {
		t.Fatalf("expected endpoint errors in input order, got %q", errText)
	}
}

func TestCheckBrokersReachabilityCancelledContext(t *testing.T) {
	checker := NewKafkaChecker(&core.KafkaParams{
		Brokers: "127.0.0.1:1",
	})
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	err := checker.checkBrokersReachability(ctx, []kafka.Broker{
		{Host: "127.0.0.1", Port: 1},
		{Host: "127.0.0.1", Port: 2},
	})
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("expected context canceled, got %v", err)
	}
}

func TestValidateSchemaRegistryEndpointIfNeeded(t *testing.T) {
	makeServer := func(body string) *httptest.Server {
		handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			w.WriteHeader(http.StatusOK)
			_, _ = w.Write([]byte(body))
		})
		return httptest.NewServer(handler)
	}

	tests := []struct {
		name        string
		endpoint    string
		wantErr     bool
		errContains string
		setup       func() *httptest.Server
	}{
		{
			name:     "empty endpoint",
			endpoint: "",
			wantErr:  false,
			setup:    func() *httptest.Server { return nil },
		},
		{
			name:    "valid response (empty json object)",
			wantErr: false,
			setup: func() *httptest.Server {
				return makeServer("{}")
			},
		},
		{
			name:        "invalid response (not empty json object)",
			wantErr:     true,
			errContains: "unexpected schema registry response",
			setup: func() *httptest.Server {
				return makeServer(`{"some":"data"}`)
			},
		},
		{
			name:        "unreachable endpoint",
			endpoint:    "http://127.0.0.1:1", // unlikely to be listening
			wantErr:     true,
			errContains: "connect schema registry endpoint failed",
			setup:       func() *httptest.Server { return nil },
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			server := tt.setup()
			if server != nil {
				t.Cleanup(server.Close)
			}

			endpoint := tt.endpoint
			if endpoint == "" && server != nil {
				endpoint = server.URL
			}

			err := validateSchemaRegistryEndpointIfNeeded(context.Background(), endpoint)

			if tt.wantErr {
				if err == nil {
					t.Fatal("expected error, got nil")
				}
				if tt.errContains != "" && !strings.Contains(err.Error(), tt.errContains) {
					t.Fatalf("error %q does not contain %q", err.Error(), tt.errContains)
				}
				return
			}

			if err != nil {
				t.Fatalf("unexpected error: %v", err)
			}
		})
	}
}

func TestValidateSchemaRegistryEndpointIfNeededRespectsContext(t *testing.T) {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		<-r.Context().Done()
	}))
	t.Cleanup(server.Close)

	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
	defer cancel()

	start := time.Now()
	err := validateSchemaRegistryEndpointIfNeeded(ctx, server.URL)

	if !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("expected context deadline exceeded, got %v", err)
	}
	if elapsed := time.Since(start); elapsed > time.Second {
		t.Fatalf("expected schema registry request to return quickly after context deadline, took %s", elapsed)
	}
}

func closedLocalAddr(t *testing.T) string {
	t.Helper()

	lis, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("listen error: %v", err)
	}
	addr := lis.Addr().String()
	if err := lis.Close(); err != nil {
		t.Fatalf("close listener: %v", err)
	}
	return addr
}