package keyspace

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"net/url"
	"testing"

	"github.com/stretchr/testify/require"
	serverlessv1alpha1 "github.com/tidbcloud/aws-shared-provider/apis/serverless/v1alpha1"
	"github.com/tidbcloud/aws-shared-provider/pkg/scheme"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/pd"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func TestMaxConcurrentReconciles(t *testing.T) {
	tests := []struct {
		name  string
		value string
		want  int
	}{
		{
			name: "uses default",
			want: MaxConcurrentReconciles,
		},
		{
			name:  "uses configured value",
			value: "42",
			want:  42,
		},
		{
			name:  "ignores invalid value",
			value: "invalid",
			want:  MaxConcurrentReconciles,
		},
		{
			name:  "ignores non-positive value",
			value: "0",
			want:  MaxConcurrentReconciles,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Setenv("MAX_CONCURRENT_KEYSPACES", tt.value)
			r := &Reconciler{}
			if got := r.maxConcurrentReconciles(); got != tt.want {
				t.Fatalf("maxConcurrentReconciles() = %d, want %d", got, tt.want)
			}
		})
	}
}

func TestReconcileDeletingKeyspaceRemovesPD(t *testing.T) {
	const keyspaceName = "keyspace-token"

	tests := []struct {
		name       string
		bindCRType serverlessv1alpha1.BindType
	}{
		{
			name:       "branch keyspace",
			bindCRType: serverlessv1alpha1.TypeBranch,
		},
		{
			name:       "cluster keyspace",
			bindCRType: serverlessv1alpha1.TypeCluster,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			requests := make([]string, 0, 3)
			httpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				requests = append(requests, r.Method+" "+r.URL.Path)
				if r.Method == http.MethodGet {
					require.NoError(t, json.NewEncoder(w).Encode(&pd.KeyspaceMeta{
						Name:  keyspaceName,
						State: keyspaceEnabled,
					}))
					return
				}
				w.WriteHeader(http.StatusOK)
			}))
			defer httpServer.Close()

			serverURL, err := url.Parse(httpServer.URL)
			require.NoError(t, err)

			deletionTimestamp := metav1.Now()
			keyspace := &serverlessv1alpha1.Keyspace{
				ObjectMeta: metav1.ObjectMeta{
					Name:              "kt-test",
					Namespace:         "tidb-admin",
					Finalizers:        []string{keyspacePoolFinalizer},
					DeletionTimestamp: &deletionTimestamp,
				},
				Spec: serverlessv1alpha1.KeyspaceSpec{
					KeyspaceName: keyspaceName,
				},
				Status: serverlessv1alpha1.KeyspaceStatus{
					BindCluster: "bound-cluster",
					BindCRType:  tt.bindCRType,
				},
			}
			kubeClient := fake.NewClientBuilder().
				WithScheme(scheme.Scheme).
				WithRuntimeObjects(keyspace).
				Build()
			reconciler := &Reconciler{
				Client:   kubeClient,
				PdClient: pd.NewClient(serverURL.Host, nil),
				Recorder: record.NewFakeRecorder(10),
			}

			_, err = reconciler.Reconcile(context.Background(), ctrl.Request{
				NamespacedName: client.ObjectKeyFromObject(keyspace),
			})
			require.NoError(t, err)
			require.Len(t, requests, 3)

			got := &serverlessv1alpha1.Keyspace{}
			require.True(t, apierrors.IsNotFound(kubeClient.Get(
				context.Background(), client.ObjectKeyFromObject(keyspace), got,
			)))
		})
	}
}