package cluster

import (
	"context"
	"fmt"
	"net/http"
	"net/http/httptest"
	"sync/atomic"
	"testing"
	"time"

	. "github.com/onsi/gomega"
	"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"
	infra "github.com/tidbcloud/tidb-management-service/api/serverless_infra"
	"github.com/tidbcloud/tidb-management-service/api/serverless_infra/meta"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

func TestPendingDelete(t *testing.T) {
	g := NewGomegaWithT(t)
	scheme := runtime.NewScheme()
	g.Expect(cloudv1.AddToScheme(scheme)).To(Succeed())

	// deleted cluster
	cluster := &infra.Cluster{
		ObjectMeta: meta.ObjectMeta{
			Name:      "cluster",
			Namespace: "default",
		},
		Status: infra.ClusterStatus{
			Phase: infra.ClusterPhasePendingDelete,
		},
	}

	// all import will deleted
	ipt := &cloudv1.Import{
		ObjectMeta: v1.ObjectMeta{
			Name:      "restore",
			Namespace: cluster.Namespace,
		},
		Status: cloudv1.ImportStatus{
			Phase: cloudv1.ImportPrepare,
		},
	}

	httpServer := newFakeHTTP()
	defer httpServer.Close()

	r := newFakeReconcileContext(cluster, nil, httpServer)
	g.Expect(r.InfraClient.Create(context.TODO(), ipt)).To(Succeed())

	g.Eventually(func() error {
		if err := r.pendingDeleteCluster(); err != nil {
			return err
		}

		if err := r.InfraClusterCli.Get(r.ctx, cluster); err != nil {
			return err
		}
		if cluster.Status.Phase != infra.ClusterPhaseDeleting {
			return fmt.Errorf("cluster phase is not deleting")
		}
		return nil
	}, 10*time.Second, 1*time.Second).Should(Succeed(), "import will clean up")
}

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

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

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

			cluster := &infra.Cluster{
				ObjectMeta: meta.ObjectMeta{
					Name:      "cluster",
					Namespace: defaultNamespace,
				},
			}
			keyspace := &serverlessv1alpha1.Keyspace{
				ObjectMeta: v1.ObjectMeta{
					Name:      "kt-test",
					Namespace: keyspaceNamespace,
				},
				Spec: serverlessv1alpha1.KeyspaceSpec{
					KeyspaceName: keyspaceName,
				},
			}
			reconciler := newFakeReconciler([]runtime.Object{keyspace}, []infra.Object{cluster}, httpServer)
			r := newFakeReconcileContext(cluster, 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.deleteCluster())
			require.Equal(t, tt.wantPDRequests, atomic.LoadInt32(&pdRequests))

			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)
			}
		})
	}
}