package tioperator

import (
	"context"
	"strconv"
	"testing"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func TestCreatePlacementPolicyCreatesCR(t *testing.T) {
	ctx := context.Background()
	sharedPoolID := "123"
	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).Build()

	got, err := CreatePlacementPolicy(ctx, cli, sharedPoolID, &entity.PlacementPolicy{
		SharedPoolId:      sharedPoolID,
		PlacementPolicyId: "1001",
		ClusterId:         "1001",
		TikvGroupId:       "blue",
	}, 42)
	require.NoError(t, err)
	require.NotNil(t, got)
	assert.Equal(t, "1001", got.GetPlacementPolicyId())
	assert.Equal(t, "1001", got.GetClusterId())
	assert.Equal(t, "blue", got.GetTikvGroupId())
	assert.Equal(t, apiv1.PlacementPolicy_CREATING, got.GetPhase())

	cr := &v1alpha1.PlacementPolicy{}
	err = cli.Get(ctx, types.NamespacedName{
		Namespace: dataplaneNamespace(sharedPoolID),
		Name:      "1001",
	}, cr)
	require.NoError(t, err)
	assert.Equal(t, clusterCRName, cr.Spec.Cluster.Name)
	assert.Equal(t, sharedPoolID, cr.Labels[schemav1.LabelSharedPoolKey])
	assert.Equal(t, "1001", cr.Labels[schemav1.LabelUserClusterKey])
	assert.Equal(t, placementPolicyTiKVGroupName("blue"), cr.Labels[placementPolicyLabelTiKVGroup])
	assert.NotContains(t, cr.Labels, "tags.tidbcloud.com/keyspace")
	require.Len(t, cr.Spec.GroupRefs, 1)
	assert.Equal(t, "db-blue", cr.Spec.GroupRefs[0].Name)
	require.Len(t, cr.Spec.Rules, 1)
	assert.Equal(t, v1alpha1.PlacementPolicyRoleVoter, cr.Spec.Rules[0].Role)
	assert.Equal(t, placementPolicyVoterCount, cr.Spec.Rules[0].Count)
	require.NotNil(t, cr.Spec.Rules[0].Selector.Keyspace)
	assert.Equal(t, []string{"42"}, cr.Spec.Rules[0].Selector.Keyspace.IDs)
}

func TestCreatePlacementPolicyRejectsExistingCR(t *testing.T) {
	ctx := context.Background()
	sharedPoolID := "123"
	namespace := dataplaneNamespace(sharedPoolID)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
		newPlacementPolicyCR(namespace, "1001", "blue", 42),
	).Build()

	_, err := CreatePlacementPolicy(ctx, cli, sharedPoolID, &entity.PlacementPolicy{
		SharedPoolId:      sharedPoolID,
		PlacementPolicyId: "1001",
		ClusterId:         "1001",
		TikvGroupId:       "green",
	}, 43)
	require.Error(t, err)
	assert.Equal(t, codes.AlreadyExists, status.Code(err))
}

func TestUpdatePlacementPolicyReturnsNotFoundForMissingCR(t *testing.T) {
	ctx := context.Background()
	sharedPoolID := "123"
	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).Build()

	_, err := UpdatePlacementPolicy(ctx, cli, sharedPoolID, &entity.PlacementPolicy{
		SharedPoolId:      sharedPoolID,
		PlacementPolicyId: "1001",
		ClusterId:         "1001",
		TikvGroupId:       "blue",
	}, 42)
	require.Error(t, err)
	assert.Equal(t, codes.NotFound, status.Code(err))

	cr := &v1alpha1.PlacementPolicy{}
	err = cli.Get(ctx, types.NamespacedName{
		Namespace: dataplaneNamespace(sharedPoolID),
		Name:      "1001",
	}, cr)
	require.True(t, apierrors.IsNotFound(err))
}

func TestUpdatePlacementPolicyUpdatesExistingCR(t *testing.T) {
	ctx := context.Background()
	sharedPoolID := "123"
	namespace := dataplaneNamespace(sharedPoolID)
	existing := newPlacementPolicyCR(namespace, "1001", "blue", 42)
	existing.Labels = map[string]string{
		"keep":                         "true",
		schemav1.LabelSharedPoolKey:    "old-shared-pool",
		schemav1.LabelUserClusterKey:   "old-logical-cluster",
		placementPolicyLabelTiKVGroup:  placementPolicyTiKVGroupName("blue"),
		"tags.tidbcloud.com/untouched": "yes",
	}

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(existing).Build()

	_, err := UpdatePlacementPolicy(ctx, cli, sharedPoolID, &entity.PlacementPolicy{
		PlacementPolicyId: "1001",
		ClusterId:         "1001",
		TikvGroupId:       "green",
	}, 43)
	require.NoError(t, err)

	cr := &v1alpha1.PlacementPolicy{}
	err = cli.Get(ctx, types.NamespacedName{Namespace: namespace, Name: "1001"}, cr)
	require.NoError(t, err)
	assert.Equal(t, "true", cr.Labels["keep"])
	assert.Equal(t, "yes", cr.Labels["tags.tidbcloud.com/untouched"])
	assert.Equal(t, sharedPoolID, cr.Labels[schemav1.LabelSharedPoolKey])
	assert.Equal(t, "1001", cr.Labels[schemav1.LabelUserClusterKey])
	assert.Equal(t, placementPolicyTiKVGroupName("green"), cr.Labels[placementPolicyLabelTiKVGroup])
	assert.NotContains(t, cr.Labels, "tags.tidbcloud.com/keyspace")
	assert.Equal(t, "db-green", cr.Spec.GroupRefs[0].Name)
	assert.Equal(t, []string{"43"}, cr.Spec.Rules[0].Selector.Keyspace.IDs)
	assert.Equal(t, placementPolicyVoterCount, cr.Spec.Rules[0].Count)
}

func TestListPlacementPoliciesUsesTiKVGroupLabelSelector(t *testing.T) {
	ctx := context.Background()
	sharedPoolID := "123"
	namespace := dataplaneNamespace(sharedPoolID)
	specBlueLabelGreen := newPlacementPolicyCR(namespace, "1003", "blue", 44)
	specBlueLabelGreen.Labels[placementPolicyLabelTiKVGroup] = placementPolicyTiKVGroupName("green")

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
		newPlacementPolicyCR(namespace, "1001", "blue", 42),
		newPlacementPolicyCR(namespace, "1002", "green", 43),
		specBlueLabelGreen,
		newPlacementPolicyCR("tidb999", "1003", "blue", 44),
	).Build()

	all, err := ListPlacementPolicies(ctx, cli, sharedPoolID, "")
	require.NoError(t, err)
	assert.Len(t, all, 3)

	byGroup, err := ListPlacementPolicies(ctx, cli, sharedPoolID, "blue")
	require.NoError(t, err)
	require.Len(t, byGroup, 1)
	assert.Equal(t, "1001", byGroup[0].GetClusterId())
	assert.Equal(t, apiv1.PlacementPolicy_AVAILABLE, byGroup[0].GetPhase())
}

func TestGetAndDeletePlacementPolicy(t *testing.T) {
	ctx := context.Background()
	sharedPoolID := "123"
	namespace := dataplaneNamespace(sharedPoolID)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
		newPlacementPolicyCR(namespace, "1001", "blue", 42),
	).Build()

	got, err := GetPlacementPolicy(ctx, cli, sharedPoolID, "1001")
	require.NoError(t, err)
	assert.Equal(t, "1001", got.GetPlacementPolicyId())
	assert.Equal(t, "1001", got.GetClusterId())
	assert.Equal(t, "blue", got.GetTikvGroupId())
	assert.Equal(t, apiv1.PlacementPolicy_AVAILABLE, got.GetPhase())

	err = DeletePlacementPolicy(ctx, cli, sharedPoolID, "1001")
	require.NoError(t, err)

	_, err = GetPlacementPolicy(ctx, cli, sharedPoolID, "1001")
	require.Error(t, err)
	assert.Contains(t, err.Error(), "not found")

	err = DeletePlacementPolicy(ctx, cli, sharedPoolID, "1001")
	require.Error(t, err)
	assert.Equal(t, codes.NotFound, status.Code(err))
	assert.Contains(t, err.Error(), "placement policy 1001 not found")
	assert.NotContains(t, err.Error(), "failed to delete")
}

func TestGetPlacementPolicyRejectsGroupRefWithoutTiKVGroupPrefix(t *testing.T) {
	ctx := context.Background()
	sharedPoolID := "123"
	namespace := dataplaneNamespace(sharedPoolID)
	policy := newPlacementPolicyCR(namespace, "1001", "blue", 42)
	policy.Spec.GroupRefs[0].Name = "blue"

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(policy).Build()

	_, err := GetPlacementPolicy(ctx, cli, sharedPoolID, "1001")
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, err.Error(), "invalid group ref name")
	assert.Contains(t, err.Error(), "db- prefix")
}

func TestGetPlacementPolicyRejectsMismatchedPlacementPolicyAndClusterIDs(t *testing.T) {
	ctx := context.Background()
	sharedPoolID := "123"
	namespace := dataplaneNamespace(sharedPoolID)
	policy := newPlacementPolicyCR(namespace, "1001", "blue", 42)
	policy.Labels[schemav1.LabelUserClusterKey] = "1002"

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(policy).Build()

	_, err := GetPlacementPolicy(ctx, cli, sharedPoolID, "1001")
	require.Error(t, err)
	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
	assert.Contains(t, err.Error(), "cluster id 1002 must match placement policy id")
}

func newPlacementPolicyCR(namespace, clusterID, tikvGroupID string, keyspaceID uint32) *v1alpha1.PlacementPolicy {
	return &v1alpha1.PlacementPolicy{
		TypeMeta: metav1.TypeMeta{
			APIVersion: placementPolicyAPIVersion,
			Kind:       placementPolicyKind,
		},
		ObjectMeta: metav1.ObjectMeta{
			Namespace: namespace,
			Name:      clusterID,
			Labels: map[string]string{
				schemav1.LabelUserClusterKey:  clusterID,
				placementPolicyLabelTiKVGroup: placementPolicyTiKVGroupName(tikvGroupID),
			},
		},
		Spec: v1alpha1.PlacementPolicySpec{
			Cluster: v1alpha1.ClusterReference{Name: clusterCRName},
			GroupRefs: []v1alpha1.PlacementPolicyGroupRef{
				{
					Group: placementPolicyGroup,
					Kind:  placementPolicyTiKVGroupKind,
					Name:  placementPolicyTiKVGroupName(tikvGroupID),
				},
			},
			Rules: []v1alpha1.PlacementPolicyRule{
				{
					Name:  placementPolicyRuleName,
					Role:  v1alpha1.PlacementPolicyRoleVoter,
					Count: placementPolicyVoterCount,
					Selector: v1alpha1.PlacementPolicySelector{
						Keyspace: &v1alpha1.PlacementPolicyKeyspaceSelector{
							IDs: []string{strconv.FormatUint(uint64(keyspaceID), 10)},
						},
					},
				},
			},
		},
		Status: v1alpha1.PlacementPolicyStatus{
			Conditions: []metav1.Condition{
				{
					Type:               v1alpha1.CondSynced,
					Status:             metav1.ConditionTrue,
					Reason:             "Synced",
					ObservedGeneration: 1,
				},
			},
		},
	}
}