package branch

import (
	"net/http"
	"net/http/httptest"
	"sync/atomic"
	"testing"

	"github.com/stretchr/testify/require"
	serverlessv1alpha1 "github.com/tidbcloud/aws-shared-provider/apis/serverless/v1alpha1"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/clientfake"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

func TestDeleteBranchSkipClusterDeleteEtcdAndPd(t *testing.T) {
	const (
		keyspaceName      = "keyspace-token"
		keyspaceNamespace = "tidb-admin"
	)

	tests := []struct {
		name                string
		skip                bool
		wantPDDeleteCalls   int32
		wantEtcdDeleted     bool
		wantKeyspaceDeleted bool
	}{
		{
			name:                "delete etcd keyspace and pd resource group",
			wantPDDeleteCalls:   1,
			wantEtcdDeleted:     true,
			wantKeyspaceDeleted: true,
		},
		{
			name:                "skip etcd keyspace and pd resource group",
			skip:                true,
			wantPDDeleteCalls:   0,
			wantEtcdDeleted:     false,
			wantKeyspaceDeleted: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			var pdDeleteCalls int32
			httpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				if r.Method == http.MethodDelete {
					atomic.AddInt32(&pdDeleteCalls, 1)
				}
				w.WriteHeader(http.StatusOK)
			}))
			defer httpServer.Close()

			branch := &cloudv1.Branch{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "branch",
					Namespace: defaultNamespace,
				},
			}
			keyspace := &serverlessv1alpha1.Keyspace{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "kt-test",
					Namespace: keyspaceNamespace,
				},
				Spec: serverlessv1alpha1.KeyspaceSpec{
					KeyspaceName: keyspaceName,
				},
			}

			reconciler := newFakeReconciler(
				[]runtime.Object{keyspace},
				[]runtime.Object{branch},
				httpServer,
			)
			r := newFakeReconcileContext(branch, reconciler, httpServer)
			r.keyspaceNs = keyspaceNamespace
			r.keyspaceToken = keyspaceName
			r.keyspaceID = 42
			r.SkipClusterDeleteEtcdAndPd = tt.skip

			etcdKeys := []string{
				r.EtcdTokenPrefixMap[defaultTidbPool] + keyspaceName + "/" + utils.KeyReplicas,
				r.EtcdTokenPrefixMap["other-pool"] + keyspaceName + "/" + utils.KeyEndpoints,
			}
			r.EtcdKV = clientfake.NewEtcdKV(clientfake.BuildArg{
				DB: map[string]string{
					etcdKeys[0]: "1",
					etcdKeys[1]: "endpoint",
				},
			})

			require.NoError(t, r.deleteBranch())
			require.Equal(t, tt.wantPDDeleteCalls, atomic.LoadInt32(&pdDeleteCalls))

			for _, etcdKey := range etcdKeys {
				resp, err := r.EtcdKV.Get(r.ctx, etcdKey)
				require.NoError(t, err)
				if tt.wantEtcdDeleted {
					require.Empty(t, resp.Kvs)
				} else {
					require.Len(t, resp.Kvs, 1)
				}
			}

			gotKeyspace := &serverlessv1alpha1.Keyspace{}
			err := r.Client.Get(r.ctx, client.ObjectKeyFromObject(keyspace), gotKeyspace)
			if tt.wantKeyspaceDeleted {
				require.True(t, apierrors.IsNotFound(err), "expected keyspace to be deleted, got %v", err)
			} else {
				require.NoError(t, err)
			}
		})
	}
}