package privatelinkservice

import (
	"context"
	"encoding/json"
	"fmt"
	"testing"
	"time"

	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	route53v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/route53/v1alpha1"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"

	regionalapiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
)

// newPurgeTestScheme returns a runtime scheme with every type the purge path touches.
func newPurgeTestScheme(t *testing.T) *runtime.Scheme {
	t.Helper()
	s := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(s))
	require.NoError(t, ec2v1alpha1.SchemeBuilder.AddToScheme(s))
	require.NoError(t, route53v1alpha1.SchemeBuilder.AddToScheme(s))
	require.NoError(t, v1alpha1.AddToScheme(s))
	return s
}

// pausedPLS returns a paused AWS PrivateLinkService fixture with the given crossplane
// name annotations set. Caller may mutate before handing off to the fake client.
func pausedPLS(clusterID, plsID, namespace, vpcesvcName, rrsetName string) *v1alpha1.PrivateLinkService {
	pls := &v1alpha1.PrivateLinkService{
		ObjectMeta: metav1.ObjectMeta{
			Name:       plsID,
			Namespace:  namespace,
			Finalizers: []string{v1alpha1.Finalizer},
			Labels: map[string]string{
				v1alpha1.ClusterIDKey: clusterID,
			},
			Annotations: map[string]string{
				v1alpha1.AnnotationKeyReconciliationPaused: "true",
			},
		},
		Spec: v1alpha1.PrivateLinkServiceSpec{
			Provider: consts.ProviderTypeEKS,
			Region:   "us-west-2",
		},
	}
	if vpcesvcName != "" {
		pls.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameVPCEService] = vpcesvcName
	}
	if rrsetName != "" {
		pls.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameResourceRecordSet] = rrsetName
	}
	return pls
}

// fakeRW is a minimal PrivateLinkServiceReaderWriter that delegates to a real
// controller-runtime fake client so the purge flow can exercise the PLS end-to-end.
type fakeRW struct {
	cli client.Client
}

func (f *fakeRW) Get(ctx context.Context, key client.ObjectKey, pls *v1alpha1.PrivateLinkService) error {
	return f.cli.Get(ctx, key, pls)
}
func (f *fakeRW) List(ctx context.Context, list *v1alpha1.PrivateLinkServiceList, opts ...client.ListOption) error {
	return f.cli.List(ctx, list, opts...)
}
func (f *fakeRW) Create(ctx context.Context, pls *v1alpha1.PrivateLinkService) error {
	return f.cli.Create(ctx, pls)
}
func (f *fakeRW) Update(ctx context.Context, pls *v1alpha1.PrivateLinkService) error {
	return f.cli.Update(ctx, pls)
}
func (f *fakeRW) Delete(ctx context.Context, pls *v1alpha1.PrivateLinkService) error {
	return f.cli.Delete(ctx, pls)
}
func (f *fakeRW) UpdateStatus(ctx context.Context, pls *v1alpha1.PrivateLinkService) error {
	return f.cli.Status().Update(ctx, pls)
}

func newPurgeService(t *testing.T, cli client.Client, provider string) *PrivateLinkServiceService {
	t.Helper()
	return NewPrivateLinkServiceService(cli, &fakeRW{cli: cli}, nil, nil, config.RootConfig{
		ControllerConfig: config.ControllerConfig{
			Provider: provider,
			RegionConfig: config.RegionConfig{
				Name: "us-west-2",
			},
		},
	})
}

func TestPurgePrivateLinkMetadata_UnsupportedProvider(t *testing.T) {
	cli := fake.NewClientBuilder().WithScheme(newPurgeTestScheme(t)).Build()
	svc := newPurgeService(t, cli, consts.ProviderTypeGKE)

	err := svc.PurgePrivateLinkMetadata(context.Background(), &regionalapiv1.PurgePrivateLinkMetadataRequest{
		ClusterId:            "c1",
		PrivateLinkServiceId: "p1",
	})
	assert.ErrorContains(t, err, "only supported for AWS")
}

// PLS not found -> purge is a no-op success.
func TestPurgePrivateLinkMetadata_NotFoundIsSuccess(t *testing.T) {
	cli := fake.NewClientBuilder().WithScheme(newPurgeTestScheme(t)).Build()
	svc := newPurgeService(t, cli, consts.ProviderTypeEKS)

	err := svc.PurgePrivateLinkMetadata(context.Background(), &regionalapiv1.PurgePrivateLinkMetadataRequest{
		ClusterId:            "c1",
		PrivateLinkServiceId: "p1",
	})
	require.NoError(t, err)
}

// cluster id mismatch is treated as not-found (security) -> no-op success.
func TestPurgePrivateLinkMetadata_ClusterIDMismatchIsSuccess(t *testing.T) {
	namer := utils.NewNamerWithClusterID("c1", "us-west-2", "")
	pls := pausedPLS("other-cluster", "p1", namer.GetNamespace(), "ves-1", "rrs-1")
	cli := fake.NewClientBuilder().WithScheme(newPurgeTestScheme(t)).WithObjects(pls).Build()
	svc := newPurgeService(t, cli, consts.ProviderTypeEKS)

	err := svc.PurgePrivateLinkMetadata(context.Background(), &regionalapiv1.PurgePrivateLinkMetadataRequest{
		ClusterId:            "c1",
		PrivateLinkServiceId: "p1",
	})
	require.NoError(t, err)

	// Nothing mutated.
	got := &v1alpha1.PrivateLinkService{}
	require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(pls), got))
	assert.Contains(t, got.Finalizers, v1alpha1.Finalizer)
}

// PLS already has DeletionTimestamp -> purge is a no-op success.
func TestPurgePrivateLinkMetadata_AlreadyDeletingIsSuccess(t *testing.T) {
	namer := utils.NewNamerWithClusterID("c1", "us-west-2", "")
	pls := pausedPLS("c1", "p1", namer.GetNamespace(), "ves-1", "rrs-1")
	// A non-nil DeletionTimestamp can only be set through a real Delete call + a
	// finalizer. Bootstrap the PLS with a finalizer, Delete it, then run purge.
	ves := &ec2v1alpha1.VPCEndpointServiceConfiguration{ObjectMeta: metav1.ObjectMeta{Name: "ves-1"}}
	rrs := &route53v1alpha1.ResourceRecordSet{ObjectMeta: metav1.ObjectMeta{Name: "rrs-1"}}
	cli := fake.NewClientBuilder().WithScheme(newPurgeTestScheme(t)).
		WithObjects(pls, ves, rrs).Build()
	require.NoError(t, cli.Delete(context.Background(), pls.DeepCopy()))

	// Verify setup: PLS has DeletionTimestamp but is still present (finalizer blocking).
	got := &v1alpha1.PrivateLinkService{}
	require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(pls), got))
	require.False(t, got.DeletionTimestamp.IsZero())

	svc := newPurgeService(t, cli, consts.ProviderTypeEKS)
	err := svc.PurgePrivateLinkMetadata(context.Background(), &regionalapiv1.PurgePrivateLinkMetadataRequest{
		ClusterId:            "c1",
		PrivateLinkServiceId: "p1",
	})
	require.NoError(t, err)

	// Crossplane CRs must still be intact; finalizer on PLS unchanged.
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{Name: "ves-1"}, &ec2v1alpha1.VPCEndpointServiceConfiguration{}))
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{Name: "rrs-1"}, &route53v1alpha1.ResourceRecordSet{}))
	require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(pls), got))
	assert.Contains(t, got.Finalizers, v1alpha1.Finalizer)

	// Snapshot must NOT have been written.
	cm := &corev1.ConfigMap{}
	err = cli.Get(context.Background(),
		client.ObjectKey{Namespace: pls.Namespace, Name: purgeSnapshotCMName("p1")}, cm)
	assert.True(t, apierrors.IsNotFound(err))
}

func TestPurgePrivateLinkMetadata_PreconditionFailed_NotPaused(t *testing.T) {
	namer := utils.NewNamerWithClusterID("c1", "us-west-2", "")
	pls := pausedPLS("c1", "p1", namer.GetNamespace(), "ves-1", "rrs-1")
	delete(pls.Annotations, v1alpha1.AnnotationKeyReconciliationPaused)
	cli := fake.NewClientBuilder().WithScheme(newPurgeTestScheme(t)).WithObjects(pls).Build()
	svc := newPurgeService(t, cli, consts.ProviderTypeEKS)

	err := svc.PurgePrivateLinkMetadata(context.Background(), &regionalapiv1.PurgePrivateLinkMetadataRequest{
		ClusterId:            "c1",
		PrivateLinkServiceId: "p1",
	})
	require.Error(t, err)
	assert.Equal(t, bizerr.PreconditionFailed, err.(bizerr.Error).Code())

	// Nothing mutated.
	got := &v1alpha1.PrivateLinkService{}
	require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(pls), got))
	assert.Contains(t, got.Finalizers, v1alpha1.Finalizer)

	// No snapshot written.
	cm := &corev1.ConfigMap{}
	err = cli.Get(context.Background(),
		client.ObjectKey{Namespace: pls.Namespace, Name: purgeSnapshotCMName("p1")}, cm)
	assert.True(t, apierrors.IsNotFound(err))
}

// Happy path: snapshot is written, crossplane CRs and PLS are deleted via finalizer
// clearing, cloud-facing finalizers never fired.
func TestPurgePrivateLinkMetadata_HappyPath(t *testing.T) {
	namer := utils.NewNamerWithClusterID("c1", "us-west-2", "")
	pls := pausedPLS("c1", "p1", namer.GetNamespace(), "ves-1", "rrs-1")
	// Give the crossplane CRs finalizers to simulate crossplane managing them.
	ves := &ec2v1alpha1.VPCEndpointServiceConfiguration{
		ObjectMeta: metav1.ObjectMeta{
			Name:       "ves-1",
			Finalizers: []string{"finalizer.managedresource.crossplane.io"},
		},
	}
	rrs := &route53v1alpha1.ResourceRecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:       "rrs-1",
			Finalizers: []string{"finalizer.managedresource.crossplane.io"},
		},
	}
	cli := fake.NewClientBuilder().WithScheme(newPurgeTestScheme(t)).
		WithObjects(pls, ves, rrs).Build()
	svc := newPurgeService(t, cli, consts.ProviderTypeEKS)

	err := svc.PurgePrivateLinkMetadata(context.Background(), &regionalapiv1.PurgePrivateLinkMetadataRequest{
		ClusterId:            "c1",
		PrivateLinkServiceId: "p1",
	})
	require.NoError(t, err)

	// PLS gone.
	assert.True(t, apierrors.IsNotFound(cli.Get(context.Background(),
		client.ObjectKeyFromObject(pls), &v1alpha1.PrivateLinkService{})))
	// Crossplane CRs gone.
	assert.True(t, apierrors.IsNotFound(cli.Get(context.Background(),
		client.ObjectKey{Name: "ves-1"}, &ec2v1alpha1.VPCEndpointServiceConfiguration{})))
	assert.True(t, apierrors.IsNotFound(cli.Get(context.Background(),
		client.ObjectKey{Name: "rrs-1"}, &route53v1alpha1.ResourceRecordSet{})))

	// Snapshot CM written.
	cm := &corev1.ConfigMap{}
	require.NoError(t, cli.Get(context.Background(),
		client.ObjectKey{Namespace: pls.Namespace, Name: purgeSnapshotCMName("p1")}, cm))
	assert.Equal(t, purgeSnapshotLabelOwnedByVal, cm.Labels[purgeSnapshotLabelOwnedBy])
	assert.Equal(t, "c1", cm.Labels[purgeSnapshotLabelClusterID])
	assert.Equal(t, "p1", cm.Labels[purgeSnapshotLabelPLSID])
	assert.NotEmpty(t, cm.Annotations[purgeSnapshotAnnoPurgedAt])

	var snap purgeSnapshot
	require.NoError(t, json.Unmarshal([]byte(cm.Data[purgeSnapshotDataKey]), &snap))
	assert.Equal(t, purgeSnapshotSchemaVersion, snap.SchemaVersion)
	assert.Equal(t, consts.ProviderTypeEKS, snap.Provider)
	assert.Equal(t, "c1", snap.ClusterID)
	assert.Equal(t, "p1", snap.PrivateLinkID)
	require.NotNil(t, snap.PrivateLinkService)
	require.Len(t, snap.CrossplaneRefs, 2)
	assert.Equal(t, "VPCEndpointServiceConfiguration", snap.CrossplaneRefs[0].Kind)
	assert.True(t, snap.CrossplaneRefs[0].Existed)
	assert.NotNil(t, snap.CrossplaneRefs[0].ManifestBeforePurge)
	assert.Equal(t, "ResourceRecordSet", snap.CrossplaneRefs[1].Kind)
	assert.True(t, snap.CrossplaneRefs[1].Existed)
}

// Snapshot already exists with a different clusterId -> ResourceConflict, no mutation.
func TestPurgePrivateLinkMetadata_SnapshotConflict(t *testing.T) {
	namer := utils.NewNamerWithClusterID("c1", "us-west-2", "")
	pls := pausedPLS("c1", "p1", namer.GetNamespace(), "ves-1", "rrs-1")
	ves := &ec2v1alpha1.VPCEndpointServiceConfiguration{ObjectMeta: metav1.ObjectMeta{Name: "ves-1"}}
	rrs := &route53v1alpha1.ResourceRecordSet{ObjectMeta: metav1.ObjectMeta{Name: "rrs-1"}}

	stale := purgeSnapshot{
		SchemaVersion: purgeSnapshotSchemaVersion,
		Provider:      consts.ProviderTypeEKS,
		ClusterID:     "other-cluster",
		PrivateLinkID: "p1",
		PurgedAt:      "2020-01-01T00:00:00Z",
	}
	raw, err := json.MarshalIndent(&stale, "", "  ")
	require.NoError(t, err)
	staleCM := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      purgeSnapshotCMName("p1"),
			Namespace: namer.GetNamespace(),
		},
		Data: map[string]string{purgeSnapshotDataKey: string(raw)},
	}

	cli := fake.NewClientBuilder().WithScheme(newPurgeTestScheme(t)).
		WithObjects(pls, ves, rrs, staleCM).Build()
	svc := newPurgeService(t, cli, consts.ProviderTypeEKS)

	err = svc.PurgePrivateLinkMetadata(context.Background(), &regionalapiv1.PurgePrivateLinkMetadataRequest{
		ClusterId:            "c1",
		PrivateLinkServiceId: "p1",
	})
	require.Error(t, err)
	assert.Equal(t, bizerr.ResourceConflict, err.(bizerr.Error).Code())

	gotPLS := &v1alpha1.PrivateLinkService{}
	require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(pls), gotPLS))
	assert.Contains(t, gotPLS.Finalizers, v1alpha1.Finalizer)
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{Name: "ves-1"}, &ec2v1alpha1.VPCEndpointServiceConfiguration{}))
	require.NoError(t, cli.Get(context.Background(), client.ObjectKey{Name: "rrs-1"}, &route53v1alpha1.ResourceRecordSet{}))
}

// Legacy PLS without child-name annotations should still resolve both downstream
// resources using the same fallback logic as the controller tasks.
func TestPurgePrivateLinkMetadata_UsesControllerNamingFallbacks(t *testing.T) {
	namer := utils.NewNamerWithClusterID("c1", "us-west-2", "")
	pls := pausedPLS("c1", "p1", namer.GetNamespace(), "", "")
	pls.Status.EndpointService.PrivateDNSName = "rrs-fallback"

	ves := &ec2v1alpha1.VPCEndpointServiceConfiguration{
		ObjectMeta: metav1.ObjectMeta{Name: utils.GetEndpointServiceName(pls)},
	}
	rrs := &route53v1alpha1.ResourceRecordSet{
		ObjectMeta: metav1.ObjectMeta{Name: pls.Status.EndpointService.PrivateDNSName},
	}
	cli := fake.NewClientBuilder().WithScheme(newPurgeTestScheme(t)).
		WithObjects(pls, ves, rrs).Build()
	svc := newPurgeService(t, cli, consts.ProviderTypeEKS)

	err := svc.PurgePrivateLinkMetadata(context.Background(), &regionalapiv1.PurgePrivateLinkMetadataRequest{
		ClusterId:            "c1",
		PrivateLinkServiceId: "p1",
	})
	require.NoError(t, err)

	cm := &corev1.ConfigMap{}
	require.NoError(t, cli.Get(context.Background(),
		client.ObjectKey{Namespace: pls.Namespace, Name: purgeSnapshotCMName("p1")}, cm))
	var snap purgeSnapshot
	require.NoError(t, json.Unmarshal([]byte(cm.Data[purgeSnapshotDataKey]), &snap))
	require.Len(t, snap.CrossplaneRefs, 2)
	assert.Equal(t, "VPCEndpointServiceConfiguration", snap.CrossplaneRefs[0].Kind)
	assert.Equal(t, utils.GetEndpointServiceName(pls), snap.CrossplaneRefs[0].Name)
	assert.Equal(t, "ResourceRecordSet", snap.CrossplaneRefs[1].Kind)
	assert.Equal(t, pls.Status.EndpointService.PrivateDNSName, snap.CrossplaneRefs[1].Name)
}

// Critical retry contract: after a first call fails mid-way, a second call MUST reuse
// the existing snapshot and converge, NOT fail with ResourceConflict.
func TestPurgePrivateLinkMetadata_ResumeAfterPartialFailure(t *testing.T) {
	namer := utils.NewNamerWithClusterID("c1", "us-west-2", "")
	pls := pausedPLS("c1", "p1", namer.GetNamespace(), "ves-1", "rrs-1")
	ves := &ec2v1alpha1.VPCEndpointServiceConfiguration{
		ObjectMeta: metav1.ObjectMeta{
			Name:       "ves-1",
			Finalizers: []string{"finalizer.managedresource.crossplane.io"},
		},
	}
	rrs := &route53v1alpha1.ResourceRecordSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:       "rrs-1",
			Finalizers: []string{"finalizer.managedresource.crossplane.io"},
		},
	}

	// Interceptor: fail Delete on ResourceRecordSet ONLY on the first call.
	failOnce := true
	cli := fake.NewClientBuilder().
		WithScheme(newPurgeTestScheme(t)).
		WithObjects(pls, ves, rrs).
		WithInterceptorFuncs(interceptor.Funcs{
			Delete: func(ctx context.Context, c client.WithWatch, obj client.Object, opts ...client.DeleteOption) error {
				if _, ok := obj.(*route53v1alpha1.ResourceRecordSet); ok && failOnce {
					failOnce = false
					return fmt.Errorf("injected transient error")
				}
				return c.Delete(ctx, obj, opts...)
			},
		}).Build()
	svc := newPurgeService(t, cli, consts.ProviderTypeEKS)

	err := svc.PurgePrivateLinkMetadata(context.Background(), &regionalapiv1.PurgePrivateLinkMetadataRequest{
		ClusterId:            "c1",
		PrivateLinkServiceId: "p1",
	})
	require.Error(t, err)
	assert.Equal(t, bizerr.K8SClientError, err.(bizerr.Error).Code())

	// Snapshot was written before the failure; ves-1 was finalizer-stripped + deleted;
	// rrs-1 had its finalizer stripped but Delete failed -> rrs-1 still present.
	cm := &corev1.ConfigMap{}
	require.NoError(t, cli.Get(context.Background(),
		client.ObjectKey{Namespace: pls.Namespace, Name: purgeSnapshotCMName("p1")}, cm))
	assert.True(t, apierrors.IsNotFound(cli.Get(context.Background(),
		client.ObjectKey{Name: "ves-1"}, &ec2v1alpha1.VPCEndpointServiceConfiguration{})))

	// Second call MUST succeed by resuming from the existing snapshot.
	err = svc.PurgePrivateLinkMetadata(context.Background(), &regionalapiv1.PurgePrivateLinkMetadataRequest{
		ClusterId:            "c1",
		PrivateLinkServiceId: "p1",
	})
	require.NoError(t, err, "retry after partial failure must converge, not return ResourceConflict")

	assert.True(t, apierrors.IsNotFound(cli.Get(context.Background(),
		client.ObjectKeyFromObject(pls), &v1alpha1.PrivateLinkService{})))
	assert.True(t, apierrors.IsNotFound(cli.Get(context.Background(),
		client.ObjectKey{Name: "rrs-1"}, &route53v1alpha1.ResourceRecordSet{})))
	// Snapshot still present.
	require.NoError(t, cli.Get(context.Background(),
		client.ObjectKey{Namespace: pls.Namespace, Name: purgeSnapshotCMName("p1")}, cm))
}

// Second call after a fully successful purge: PLS is gone; purge returns OK (no-op).
func TestPurgePrivateLinkMetadata_Repeat_AfterFullSuccess(t *testing.T) {
	namer := utils.NewNamerWithClusterID("c1", "us-west-2", "")
	existingSnap := purgeSnapshot{
		SchemaVersion: purgeSnapshotSchemaVersion,
		Provider:      consts.ProviderTypeEKS,
		ClusterID:     "c1",
		PrivateLinkID: "p1",
		PurgedAt:      time.Now().UTC().Format(time.RFC3339),
	}
	raw, err := json.MarshalIndent(&existingSnap, "", "  ")
	require.NoError(t, err)
	cm := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      purgeSnapshotCMName("p1"),
			Namespace: namer.GetNamespace(),
		},
		Data: map[string]string{purgeSnapshotDataKey: string(raw)},
	}
	cli := fake.NewClientBuilder().WithScheme(newPurgeTestScheme(t)).WithObjects(cm).Build()
	svc := newPurgeService(t, cli, consts.ProviderTypeEKS)

	err = svc.PurgePrivateLinkMetadata(context.Background(), &regionalapiv1.PurgePrivateLinkMetadataRequest{
		ClusterId:            "c1",
		PrivateLinkServiceId: "p1",
	})
	assert.NoError(t, err)
}