package export

import (
	"context"
	"fmt"
	"testing"

	"github.com/pingcap/kvproto/pkg/keyspacepb"
	"github.com/stretchr/testify/require"
	providerscheme "github.com/tidbcloud/aws-shared-provider/pkg/scheme"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/pd"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/tools/record"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

type fakeExportPdClient struct {
	putErrs    map[string]error
	postErrs   map[string]error
	deleteErrs map[string]error

	puts    []string
	posts   []string
	deletes []string
}

func (f *fakeExportPdClient) GetClusterID(context.Context) (uint64, error) {
	return 0, nil
}

func (f *fakeExportPdClient) GetKeyspaceByID(context.Context, uint32) (*pd.KeyspaceMeta, error) {
	return nil, nil
}

func (f *fakeExportPdClient) GetKeyspaceByName(context.Context, string) (*pd.KeyspaceMeta, error) {
	return nil, nil
}

func (f *fakeExportPdClient) CreateKeyspace(context.Context, *pd.KeyspaceMeta) (*pd.KeyspaceMeta, error) {
	return nil, nil
}

func (f *fakeExportPdClient) UpdateKeyspaceState(context.Context, string, keyspacepb.KeyspaceState) error {
	return nil
}

func (f *fakeExportPdClient) UpdateKeyspaceConfig(context.Context, string, *pd.KeyspaceConfig) error {
	return nil
}

func (f *fakeExportPdClient) SetKeyspaceRuVersion(context.Context, string, *pd.SetKeyspaceRuVersionRequest) error {
	return nil
}

func (f *fakeExportPdClient) PostResourceGroup(_ context.Context, group *pd.ResourceGroup) error {
	f.posts = append(f.posts, group.Name)
	return f.postErrs[group.Name]
}

func (f *fakeExportPdClient) PutResourceGroup(_ context.Context, group *pd.ResourceGroup) error {
	f.puts = append(f.puts, group.Name)
	return f.putErrs[group.Name]
}

func (f *fakeExportPdClient) DeleteResourceGroup(_ context.Context, groupName string) error {
	f.deletes = append(f.deletes, groupName)
	return f.deleteErrs[groupName]
}

func (f *fakeExportPdClient) GetMetaServiceGroupList(context.Context) (*[]pd.MetaServiceGroup, error) {
	return nil, nil
}

func newResourceGroupTestContext(pdClient pd.HttpPdClient, configMapData map[string]string) *reconcileContext {
	objs := []client.Object{}
	if configMapData != nil {
		objs = append(objs, &corev1.ConfigMap{
			ObjectMeta: metav1.ObjectMeta{
				Name:      ExportConfigCM,
				Namespace: "export-job",
			},
			Data: configMapData,
		})
	}
	kubeClient := fake.NewClientBuilder().
		WithScheme(providerscheme.Scheme).
		WithObjects(objs...).
		Build()
	return &reconcileContext{
		Reconciler: &Reconciler{
			Client:        kubeClient,
			PdClient:      pdClient,
			InfraRecorder: record.NewFakeRecorder(10),
			ExportJobNS:   "export-job",
		},
		ctx:        context.Background(),
		export:     &cloudv1.Export{ObjectMeta: metav1.ObjectMeta{Name: "export-name"}},
		keyspaceID: 42,
	}
}

func TestEnsureResourceGroupUsesOldNameByDefault(t *testing.T) {
	fakePd := &fakeExportPdClient{
		putErrs: map[string]error{"export-name": fmt.Errorf("the export-name resource group does not exist")},
	}
	r := newResourceGroupTestContext(fakePd, map[string]string{})

	resourceGroupName, err := r.exportResourceGroupName()
	require.NoError(t, err)
	require.Equal(t, "export-name", resourceGroupName)
	require.NoError(t, r.updateResourceGroup(resourceGroupName))
	require.Equal(t, []string{"export-name"}, fakePd.puts)
	require.Equal(t, []string{"export-name"}, fakePd.posts)
}

func TestEnsureResourceGroupUsesNewNameWhenEnabled(t *testing.T) {
	fakePd := &fakeExportPdClient{
		putErrs: map[string]error{"exp-42": fmt.Errorf("the exp-42 resource group does not exist")},
	}
	r := newResourceGroupTestContext(fakePd, map[string]string{"enable-export-new-resource-group": "true"})

	resourceGroupName, err := r.exportResourceGroupName()
	require.NoError(t, err)
	require.Equal(t, "exp-42", resourceGroupName)
	require.NoError(t, r.updateResourceGroup(resourceGroupName))
	require.Equal(t, []string{"exp-42"}, fakePd.puts)
	require.Equal(t, []string{"exp-42"}, fakePd.posts)
}

func TestEnsureResourceGroupLimitReturnsPostError(t *testing.T) {
	fakePd := &fakeExportPdClient{
		putErrs:  map[string]error{"export-name": fmt.Errorf("the export-name resource group does not exist")},
		postErrs: map[string]error{"export-name": fmt.Errorf("resource group already exists")},
	}
	r := newResourceGroupTestContext(fakePd, map[string]string{})

	created, err := r.ensureResourceGroupLimit("export-name", exportSpeedLimit)
	require.ErrorContains(t, err, "resource group already exists")
	require.False(t, created)
	require.Equal(t, []string{"export-name"}, fakePd.puts)
	require.Equal(t, []string{"export-name"}, fakePd.posts)
}

func TestDeleteResourceGroupUsesRecordedName(t *testing.T) {
	fakePd := &fakeExportPdClient{}
	r := newResourceGroupTestContext(fakePd, map[string]string{})
	r.export.Annotations = map[string]string{types.AnnoExportResourceGroup: "exp-42"}

	require.NoError(t, r.tryDeleteResourceGroup())
	require.Equal(t, []string{"exp-42"}, fakePd.deletes)
}