package ack

import (
	"context"
	"testing"

	ossv1alpha1 "github.com/crossplane-contrib/provider-upjet-alibabacloud/apis/oss/v1alpha1"
	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"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/repository"
	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 := &ossv1alpha1.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 := &ossv1alpha1.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.Bucket)
	require.Equal(t, expectedName, *physicalBucket.Spec.ForProvider.Tags["Name"])
	require.Equal(t, "provider-config", physicalBucket.Spec.ProviderConfigReference.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 OSS 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 := &ossv1alpha1.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 := &ossv1alpha1.Bucket{}
	require.NoError(t, localClient.Get(ctx, types.NamespacedName{Name: "selected-after-conflict"}, physicalBucket))
	require.Equal(t, "selected-after-conflict", *physicalBucket.Spec.ForProvider.Bucket)
	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, v1alpha1.AddToScheme(scheme))
	require.NoError(t, ossv1alpha1.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:      "ap-southeast-1",
			Tags: map[string]string{
				"env": "test",
			},
			AccountRef: &v1alpha1.AccountReference{
				ProviderConfigName: "provider-config",
			},
		},
	}

	baseObjects := []client.Object{
		&v1alpha1.DataPlane{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: "dataplane-metadata",
				Name:      "dp-1",
			},
			Spec: v1alpha1.DataPlaneSpec{
				AccountRef: v1alpha1.AccountReference{
					ProviderConfigName: "provider-config",
					AccountID:          "123456789012",
				},
			},
			Status: v1alpha1.DataPlaneStatus{
				AppBox: &v1alpha1.Reference{Name: "appbox-1"},
			},
		},
		&v1alpha1.APPBox{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: "dataplane-metadata",
				Name:      "appbox-1",
			},
			Status: v1alpha1.APPBoxStatus{
				IAM: v1alpha1.IAMStatus{
					NodeRoleName: "node-role",
				},
			},
		},
	}
	allObjects := append(baseObjects, objects...)
	localClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(allObjects...).Build()
	taskContext := &buckettask.Context{
		Local:      localClient,
		DataPlaneR: repository.NewDataPlaneReader(localClient),
		Config: config.RootConfig{
			Namespace:          "dataplane-metadata",
			ResourceNamePrefix: "tidbcloud",
		},
		Bucket: logicalBucket,
	}

	return NewBucket(taskContext), localClient, logicalBucket
}

func newPhysicalBucket(name string, bucketLabels map[string]string) *ossv1alpha1.Bucket {
	return &ossv1alpha1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: bucketLabels,
		},
		Spec: ossv1alpha1.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",
	}
}