package eks

import (
	"context"
	"testing"

	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	buckettask "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/bucket/task"
	s3v1beta2 "github.com/upbound/provider-aws/apis/s3/v1beta2"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func TestBucket_CreateIsIdempotentWithDeterministicName(t *testing.T) {
	ctx := context.Background()
	task, localClient, logicalBucket := newTestBucketTask(t)

	expectedName, err := task.GetSubResourceName()
	require.NoError(t, err)

	result := task.Create(ctx)
	require.NoError(t, result.RawError())
	require.True(t, result.IsDoneState())
	require.Equal(t, expectedName, logicalBucket.Status.Bucket.Name)

	result = task.Create(ctx)
	require.NoError(t, result.RawError())
	require.True(t, result.IsDoneState())
	require.Equal(t, expectedName, logicalBucket.Status.Bucket.Name)

	physicalBuckets := &s3v1beta2.BucketList{}
	require.NoError(t, localClient.List(ctx, physicalBuckets, &client.ListOptions{
		LabelSelector: labels.SelectorFromSet(map[string]string{
			consts.DataplaneLabelKey:       logicalBucket.Spec.DataplaneId,
			"tags.tidbcloud.com/bucket-id": logicalBucket.Spec.Id,
		}),
	}))
	require.Len(t, physicalBuckets.Items, 1)
	require.Equal(t, expectedName, physicalBuckets.Items[0].Name)

	physicalBucket := &s3v1beta2.Bucket{}
	require.NoError(t, localClient.Get(ctx, types.NamespacedName{Name: expectedName}, physicalBucket))
	require.Equal(t, logicalBucket.Spec.Region, *physicalBucket.Spec.ForProvider.Region)
	require.Equal(t, expectedName, *physicalBucket.Spec.ForProvider.Tags["Name"])
}

func TestBucket_CreateFailsWhenMultipleBucketsMatchLogicalBucket(t *testing.T) {
	ctx := context.Background()
	task, _, logicalBucket := newTestBucketTask(t,
		newPhysicalBucket("bucket-a", logicalBucketLabels()),
		newPhysicalBucket("bucket-b", logicalBucketLabels()),
	)

	result := task.Create(ctx)

	require.Error(t, result.RawError())
	require.Contains(t, result.RawError().Error(), "found multiple Buckets")
	require.Contains(t, result.RawError().Error(), logicalBucket.Spec.DataplaneId)
	require.Contains(t, result.RawError().Error(), logicalBucket.Spec.Id)
	require.Nil(t, logicalBucket.Status.Bucket)
}

func TestBucket_CreateUsesDeterministicAlternateNameAfterConflict(t *testing.T) {
	ctx := context.Background()
	conflictingBucket := newPhysicalBucket("conflicting-bucket", logicalBucketLabels())
	conflictingBucket.Status.Conditions = []xpv1.Condition{
		{
			Type:    xpv1.TypeSynced,
			Status:  corev1.ConditionFalse,
			Message: "BucketAlreadyExists",
		},
	}
	task, localClient, logicalBucket := newTestBucketTask(t, conflictingBucket)
	defaultName, err := task.GetSubResourceName()
	require.NoError(t, err)
	expectedName, err := task.GetSubResourceNameAfterConflict(conflictingBucket.Name)
	require.NoError(t, err)
	require.NotEqual(t, defaultName, expectedName)

	result := task.Create(ctx)
	require.NoError(t, result.RawError())
	require.True(t, result.IsDoneState())
	require.Equal(t, expectedName, logicalBucket.Status.Bucket.Name)

	physicalBuckets := &s3v1beta2.BucketList{}
	require.NoError(t, localClient.List(ctx, physicalBuckets, &client.ListOptions{
		LabelSelector: labels.SelectorFromSet(logicalBucketLabels()),
	}))
	require.Len(t, physicalBuckets.Items, 1)
	require.Equal(t, expectedName, physicalBuckets.Items[0].Name)
}

func TestBucket_CreateReusesStatusBucketNameWhenListIsEmpty(t *testing.T) {
	ctx := context.Background()
	task, localClient, logicalBucket := newTestBucketTask(t)
	logicalBucket.Status.Bucket = &v1alpha1.Reference{Name: "selected-after-conflict"}

	result := task.Create(ctx)
	require.NoError(t, result.RawError())
	require.True(t, result.IsDoneState())
	require.Equal(t, "selected-after-conflict", logicalBucket.Status.Bucket.Name)

	physicalBucket := &s3v1beta2.Bucket{}
	require.NoError(t, localClient.Get(ctx, types.NamespacedName{Name: "selected-after-conflict"}, physicalBucket))
	require.Equal(t, "selected-after-conflict", *physicalBucket.Spec.ForProvider.Tags["Name"])
}

func newTestBucketTask(t *testing.T, objects ...client.Object) (*Bucket, client.Client, *v1alpha1.Bucket) {
	t.Helper()

	scheme := runtime.NewScheme()
	require.NoError(t, s3v1beta2.AddToScheme(scheme))

	logicalBucket := &v1alpha1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name:   "dp-1-bucket-1",
			Labels: logicalBucketLabels(),
		},
		Spec: v1alpha1.BucketSpec{
			DataplaneId: "dp-1",
			Id:          "bucket-1",
			Region:      "us-west-2",
			Tags: map[string]string{
				"env": "test",
			},
			AccountRef: &v1alpha1.AccountReference{
				ProviderConfigName: "provider-config",
			},
		},
	}

	localClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build()
	taskContext := &buckettask.Context{
		Local: localClient,
		Config: config.RootConfig{
			ResourceNamePrefix: "tidbcloud",
		},
		Bucket: logicalBucket,
	}

	return NewBucket(taskContext), localClient, logicalBucket
}

func newPhysicalBucket(name string, bucketLabels map[string]string) *s3v1beta2.Bucket {
	return &s3v1beta2.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: bucketLabels,
		},
		Spec: s3v1beta2.BucketSpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: "provider-config"},
			},
		},
	}
}

func logicalBucketLabels() map[string]string {
	return map[string]string{
		consts.DataplaneLabelKey:       "dp-1",
		"tags.tidbcloud.com/bucket-id": "bucket-1",
	}
}