package service

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	opv2 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/cluster-agent/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func TestParseStorageUsage(t *testing.T) {
	tests := []struct {
		name          string
		respBody      string
		expectedCap   uint64
		expectedAvail uint64
		expectedError string
	}{
		{
			name: "valid input with both metrics",
			respBody: `tikv_store_size_bytes{instance="tikv-1",type="capacity"} 1000000000
tikv_store_size_bytes{instance="tikv-1",type="available"} 500000000`,
			expectedCap:   1000000000,
			expectedAvail: 500000000,
			expectedError: "",
		},
		{
			name: "valid input with different order",
			respBody: `tikv_store_size_bytes{instance="tikv-1",type="available"} 750000000
tikv_store_size_bytes{instance="tikv-1",type="capacity"} 2000000000`,
			expectedCap:   2000000000,
			expectedAvail: 750000000,
			expectedError: "",
		},
		{
			name: "valid input with additional labels",
			respBody: `tikv_store_size_bytes{cluster="test",instance="tikv-1",job="tikv",type="capacity"} 3000000000
tikv_store_size_bytes{cluster="test",instance="tikv-1",job="tikv",type="available"} 1500000000`,
			expectedCap:   3000000000,
			expectedAvail: 1500000000,
			expectedError: "",
		},
		{
			name: "valid input with large values",
			respBody: `tikv_store_size_bytes{instance="tikv-1",type="capacity"} 18446744073709551615
tikv_store_size_bytes{instance="tikv-1",type="available"} 9223372036854775807`,
			expectedCap:   18446744073709551615,
			expectedAvail: 9223372036854775807,
			expectedError: "",
		},
		{
			name:          "missing capacity metric",
			respBody:      `tikv_store_size_bytes{instance="tikv-1",type="available"} 500000000`,
			expectedCap:   0,
			expectedAvail: 0,
			expectedError: "should have 2 matches, got 1",
		},
		{
			name:          "missing available metric",
			respBody:      `tikv_store_size_bytes{instance="tikv-1",type="capacity"} 1000000000`,
			expectedCap:   0,
			expectedAvail: 0,
			expectedError: "should have 2 matches, got 1",
		},
		{
			name:          "no metrics found",
			respBody:      `some_other_metric{instance="tikv-1"} 1000000000`,
			expectedCap:   0,
			expectedAvail: 0,
			expectedError: "should have 2 matches, got 0",
		},
		{
			name:          "empty response body",
			respBody:      ``,
			expectedCap:   0,
			expectedAvail: 0,
			expectedError: "should have 2 matches, got 0",
		},
		{
			name: "invalid metric value",
			respBody: `tikv_store_size_bytes{instance="tikv-1",type="capacity"} not-a-number
tikv_store_size_bytes{instance="tikv-1",type="available"} 500000000`,
			expectedCap:   0,
			expectedAvail: 0,
			expectedError: "failed to parse capacity",
		},
		{
			name: "duplicate capacity metric",
			respBody: `tikv_store_size_bytes{instance="tikv-1",type="capacity"} 1000000000
tikv_store_size_bytes{instance="tikv-1",type="capacity"} 2000000000
tikv_store_size_bytes{instance="tikv-1",type="available"} 500000000`,
			expectedCap:   0,
			expectedAvail: 0,
			expectedError: "duplicate capacity metric",
		},
		{
			name: "duplicate available metric",
			respBody: `tikv_store_size_bytes{instance="tikv-1",type="capacity"} 1000000000
tikv_store_size_bytes{instance="tikv-1",type="available"} 500000000
tikv_store_size_bytes{instance="tikv-1",type="available"} 750000000`,
			expectedCap:   0,
			expectedAvail: 0,
			expectedError: "duplicate available metric",
		},
		{
			name: "wrong metric name",
			respBody: `tikv_store_size_wrong{instance="tikv-1",type="capacity"} 1000000000
tikv_store_size_wrong{instance="tikv-1",type="available"} 500000000`,
			expectedCap:   0,
			expectedAvail: 0,
			expectedError: "should have 2 matches, got 0",
		},
		{
			name: "wrong type label value",
			respBody: `tikv_store_size_bytes{instance="tikv-1",type="wrong"} 1000000000
tikv_store_size_bytes{instance="tikv-1",type="also_wrong"} 500000000`,
			expectedCap:   0,
			expectedAvail: 0,
			expectedError: "should have 2 matches, got 0",
		},
		{
			name: "mixed valid and invalid metrics",
			respBody: `tikv_store_size_bytes{instance="tikv-1",type="capacity"} 1000000000
tikv_store_size_bytes{instance="tikv-1",type="available"} 500000000
some_other_metric{instance="tikv-1"} 2000000000`,
			expectedCap:   1000000000,
			expectedAvail: 500000000,
			expectedError: "",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			capacity, available, err := parseStorageUsage(strings.NewReader(tt.respBody))

			if tt.expectedError != "" {
				if err == nil {
					t.Errorf("expected error containing '%s', got nil", tt.expectedError)
					return
				}
				if !strings.Contains(err.Error(), tt.expectedError) {
					t.Errorf("expected error containing '%s', got '%s'", tt.expectedError, err.Error())
				}
				return
			}

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

			if capacity != tt.expectedCap {
				t.Errorf("expected capacity %d, got %d", tt.expectedCap, capacity)
			}

			if available != tt.expectedAvail {
				t.Errorf("expected available %d, got %d", tt.expectedAvail, available)
			}
		})
	}
}

func TestParseStorageUsage_ScanError(t *testing.T) {
	capacity, available, err := parseStorageUsage(errReadCloser{err: errors.New("scan failed")})
	require.Error(t, err)
	assert.Zero(t, capacity)
	assert.Zero(t, available)
	assert.Contains(t, err.Error(), "failed to scan response body")
}

func TestBuildTiKVMetricsURL(t *testing.T) {
	tests := []struct {
		name    string
		host    string
		wantURL string
		wantErr bool
	}{
		{
			name:    "valid host",
			host:    "db-0.db-cluster.tidb123.svc",
			wantURL: "https://db-0.db-cluster.tidb123.svc:20180/metrics",
			wantErr: false,
		},
		{
			name:    "invalid host with scheme",
			host:    "https://example.com",
			wantURL: "",
			wantErr: true,
		},
		{
			name:    "invalid host with slash",
			host:    "example.com/path",
			wantURL: "",
			wantErr: true,
		},
		{
			name:    "invalid host with port",
			host:    "example.com:20180",
			wantURL: "",
			wantErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			gotURL, err := buildTiKVMetricsURL(tt.host)
			if tt.wantErr {
				if err == nil {
					t.Fatalf("expected error, got nil")
				}
				return
			}
			if err != nil {
				t.Fatalf("expected no error, got %v", err)
			}
			if gotURL != tt.wantURL {
				t.Fatalf("unexpected URL, got %q, want %q", gotURL, tt.wantURL)
			}
		})
	}
}

func TestGetClusterTopology_WithTiKVUsage(t *testing.T) {
	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))

	pdPod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "db-pd-0",
			Namespace: "tidb123",
			Labels: map[string]string{
				opv2.LabelKeyCluster:   clusterName,
				opv2.LabelKeyComponent: PDComponentType,
			},
		},
	}
	tikvPod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "db-tikv-0",
			Namespace: "tidb123",
			Labels: map[string]string{
				opv2.LabelKeyCluster:   clusterName,
				opv2.LabelKeyComponent: TiKVComponentType,
			},
		},
		Spec: corev1.PodSpec{NodeName: "node-a"},
	}

	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(pdPod, tikvPod).Build()

	var requestedURL string
	svc := NewAgentService(cli, nil,
		WithTiKVHTTPClient(&http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
			requestedURL = req.URL.String()
			return &http.Response{
				StatusCode: http.StatusOK,
				Body: io.NopCloser(strings.NewReader(`tikv_store_size_bytes{instance="tikv-1",type="capacity"} 1000000000
tikv_store_size_bytes{instance="tikv-1",type="available"} 500000000`)),
			}, nil
		})}),
	)

	resp, err := svc.GetClusterTopology(context.Background(), &apiv1.GetClusterTopologyReq{
		ClusterId:        "123",
		IncludeTikvUsage: true,
	})
	require.NoError(t, err)
	require.Len(t, resp.PdNodes, 1)
	require.Len(t, resp.TikvNodes, 1)
	assert.Equal(t, "https://db-tikv-0.db-cluster.tidb123.svc:20180/metrics", requestedURL)
	assert.EqualValues(t, 1000000000, resp.TikvNodes[0].StorageCapacityBytes)
	assert.EqualValues(t, 500000000, resp.TikvNodes[0].StorageAvailableBytes)
}

func TestGetClusterTopology_WithTiKVUsageNilSharedClient(t *testing.T) {
	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))

	tikvPod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "db-tikv-0",
			Namespace: "tidb123",
			Labels: map[string]string{
				opv2.LabelKeyCluster:   clusterName,
				opv2.LabelKeyComponent: TiKVComponentType,
			},
		},
	}

	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tikvPod).Build()
	svc := NewAgentService(cli, nil)

	resp, err := svc.GetClusterTopology(context.Background(), &apiv1.GetClusterTopologyReq{
		ClusterId:        "123",
		IncludeTikvUsage: true,
	})
	require.Error(t, err)
	assert.Nil(t, resp)
	assert.Contains(t, err.Error(), "tikv HTTP client unavailable")
}

func TestFetchTiKVUsage_BoundedConcurrency(t *testing.T) {
	t.Parallel()

	var inFlight int32
	var maxInFlight int32

	svc := NewAgentService(nil, nil,
		WithTiKVHTTPClient(&http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
			current := atomic.AddInt32(&inFlight, 1)
			for {
				max := atomic.LoadInt32(&maxInFlight)
				if current <= max {
					break
				}
				if atomic.CompareAndSwapInt32(&maxInFlight, max, current) {
					break
				}
			}

			time.Sleep(20 * time.Millisecond)
			atomic.AddInt32(&inFlight, -1)

			return &http.Response{
				StatusCode: http.StatusOK,
				Body: io.NopCloser(strings.NewReader(`tikv_store_size_bytes{instance="tikv-1",type="capacity"} 1000000000
tikv_store_size_bytes{instance="tikv-1",type="available"} 500000000`)),
			}, nil
		})}),
	)

	resp := &apiv1.GetClusterTopologyResp{}
	for i := 0; i < 9; i++ {
		resp.TikvNodes = append(resp.TikvNodes, &apiv1.GetClusterTopologyResp_TiKVNodeInfo{
			Name: fmt.Sprintf("tikv-%d", i),
			Host: fmt.Sprintf("tikv-%d.db-cluster.tidb123.svc", i),
		})
	}

	err := svc.fetchTiKVUsage(context.Background(), resp)
	require.NoError(t, err)
	assert.Greater(t, atomic.LoadInt32(&maxInFlight), int32(1))
	assert.LessOrEqual(t, atomic.LoadInt32(&maxInFlight), int32(tikvUsageFetchMaxConcurrency))

	for _, node := range resp.TikvNodes {
		assert.EqualValues(t, 1000000000, node.StorageCapacityBytes)
		assert.EqualValues(t, 500000000, node.StorageAvailableBytes)
	}
}