package privatelinkservice

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

	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	route53v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/route53/v1alpha1"
	"github.com/crossplane/crossplane-runtime/pkg/meta"
	"github.com/pingcap/log"
	"go.uber.org/zap"
	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/types"
	"sigs.k8s.io/controller-runtime/pkg/client"

	regionalapiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	inframetrics "github.com/tidbcloud/infra-provider/pkg/metrics"
	"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"
)

const (
	// purgeSnapshotCMNamePrefix is the deterministic prefix for the snapshot ConfigMap
	// name: <prefix><private_link_service_id>.
	purgeSnapshotCMNamePrefix = "pls-purge-snapshot-"
	// purgeSnapshotDataKey is the key under which the serialised snapshot JSON is stored.
	purgeSnapshotDataKey = "snapshot.json"
	// purgeSnapshotSchemaVersion bumps whenever the snapshot shape changes in an
	// incompatible way.
	purgeSnapshotSchemaVersion = 1

	// Snapshot CM labels / annotations (indexing + audit).
	purgeSnapshotLabelOwnedBy    = "cloud.pingcap.com/owned-by"
	purgeSnapshotLabelOwnedByVal = "private-link-purge"
	purgeSnapshotLabelClusterID  = "cloud.pingcap.com/cluster-id"
	purgeSnapshotLabelPLSID      = "cloud.pingcap.com/pls-id"
	purgeSnapshotAnnoPurgedAt    = "cloud.pingcap.com/purged-at"

	// Metric result labels.
	purgeResultSuccess            = "success"
	purgeResultPreconditionFailed = "precondition_failed"
	purgeResultConflict           = "conflict"
	purgeResultError              = "error"
)

// purgeCrossplaneRef describes one downstream crossplane CR the PLS owns,
// captured before deletion so the snapshot can fully reconstruct it later.
type purgeCrossplaneRef struct {
	Kind                string                 `json:"kind"`
	APIVersion          string                 `json:"apiVersion"`
	Name                string                 `json:"name"`
	AnnotationKey       string                 `json:"annotationKey"`
	CloudID             string                 `json:"cloudID,omitempty"`
	Existed             bool                   `json:"existed"`
	ManifestBeforePurge map[string]interface{} `json:"manifestBeforePurge,omitempty"`
}

// purgeSnapshot is the payload stored under data["snapshot.json"] of the snapshot ConfigMap.
// Any field addition/removal MUST bump SchemaVersion.
type purgeSnapshot struct {
	SchemaVersion      int                          `json:"schemaVersion"`
	Provider           string                       `json:"provider"`
	ClusterID          string                       `json:"clusterId"`
	PrivateLinkID      string                       `json:"privateLinkServiceId"`
	PurgedAt           string                       `json:"purgedAt"`
	PrivateLinkService *v1alpha1.PrivateLinkService `json:"privateLinkService"`
	CrossplaneRefs     []purgeCrossplaneRef         `json:"crossplaneRefs"`
}

// PurgePrivateLinkMetadata removes the PrivateLinkService CR and its AWS crossplane CRs
// (VPCEndpointServiceConfiguration, Route53 ResourceRecordSet) from the cluster WITHOUT
// touching the underlying AWS cloud resources.
//
// Deletion mechanism: we clear finalizers on each CR and then issue a client-go Delete.
// Because the crossplane controller never runs its delete handler, no AWS API call is
// made and the cloud resource survives.
//
// Flow:
//  1. Provider must be AWS, else ParameterInvalid.
//  2. PLS NotFound               -> return success (nothing to purge).
//  3. PLS.DeletionTimestamp set  -> return success (already being purged).
//  4. PLS not paused             -> return PreconditionFailed.
//  5. Snapshot CM missing        -> collect refs from live state and create snapshot.
//     Snapshot CM present        -> reuse it as the authoritative plan (idempotent retry).
//  6. For each ref in the snapshot: clear finalizers + Delete (skip if not found).
//  7. Clear finalizers + Delete the PLS.
//
// Each destructive step is individually idempotent, so a retry after any failure converges.
func (s *PrivateLinkServiceService) PurgePrivateLinkMetadata(ctx context.Context, req *regionalapiv1.PurgePrivateLinkMetadataRequest) error {
	logger := log.With(
		zap.String("cluster_id", req.ClusterId),
		zap.String("pls_id", req.PrivateLinkServiceId),
	)
	logger.Warn("PurgePrivateLinkMetadata invoked")

	if s.config.Provider != consts.ProviderTypeEKS {
		s.recordPurgeResult(purgeResultError)
		return bizerr.New(bizerr.ParameterInvalid, "purge private link metadata is only supported for AWS")
	}

	namer := utils.NewNamerWithClusterID(req.ClusterId, s.config.RegionConfig.Name, s.config.Env)
	plsKey := client.ObjectKey{Name: req.PrivateLinkServiceId, Namespace: namer.GetNamespace()}

	// Step 2-4: load PLS, early-return on NotFound / already-deleting, enforce Paused.
	pls := &v1alpha1.PrivateLinkService{}
	if err := s.privatelinkservicerw.Get(ctx, plsKey, pls); err != nil {
		if apierrors.IsNotFound(err) {
			logger.Info("privatelinkservice not found, nothing to purge")
			s.recordPurgeResult(purgeResultSuccess)
			return nil
		}
		logger.Error("failed to get privatelinkservice", zap.Error(err))
		s.recordPurgeResult(purgeResultError)
		return bizerr.New(bizerr.K8SClientError, "failed to get privatelinkservice")
	}
	if pls.GetLabels()[v1alpha1.ClusterIDKey] != req.ClusterId {
		// Treat a cluster-id mismatch as not-found for security: do not leak whether
		// a PLS with this id exists in a different cluster.
		logger.Info("privatelinkservice cluster id mismatch, treating as not found",
			zap.String("actual_cluster_id", pls.GetLabels()[v1alpha1.ClusterIDKey]),
			zap.String("expected_cluster_id", req.ClusterId))
		s.recordPurgeResult(purgeResultSuccess)
		return nil
	}
	if !pls.DeletionTimestamp.IsZero() {
		logger.Info("privatelinkservice is already being deleted, returning success")
		s.recordPurgeResult(purgeResultSuccess)
		return nil
	}
	if !utils.IsPaused(pls) {
		logger.Warn("privatelinkservice is not paused, purge refused")
		s.recordPurgeResult(purgeResultPreconditionFailed)
		return bizerr.New(bizerr.PreconditionFailed,
			"privatelinkservice is not paused; call PausePrivateLinkService first")
	}

	// Step 5: ensure we have an authoritative snapshot. If one already exists for this
	// PLS, reuse it (write-once audit record); otherwise build & create one.
	snapshot, found, err := s.loadPurgeSnapshot(ctx, plsKey.Namespace, req.PrivateLinkServiceId)
	if err != nil {
		logger.Error("failed to load existing snapshot", zap.Error(err))
		s.recordPurgeResult(purgeResultError)
		return bizerr.New(bizerr.K8SClientError, "failed to load existing snapshot")
	}
	if found {
		if err := validateExistingSnapshot(snapshot, req); err != nil {
			logger.Error("existing snapshot is incompatible", zap.Error(err))
			s.recordPurgeResult(purgeResultConflict)
			return err
		}
		logger.Info("reusing existing snapshot", zap.String("purged_at", snapshot.PurgedAt),
			zap.Int("refs_count", len(snapshot.CrossplaneRefs)))
	} else {
		refs, err := s.collectAWSCrossplaneRefs(ctx, pls)
		if err != nil {
			logger.Error("failed to collect crossplane refs", zap.Error(err))
			s.recordPurgeResult(purgeResultError)
			return bizerr.New(bizerr.K8SClientError, "failed to collect crossplane refs")
		}
		snapshot = &purgeSnapshot{
			SchemaVersion:      purgeSnapshotSchemaVersion,
			Provider:           s.config.Provider,
			ClusterID:          req.ClusterId,
			PrivateLinkID:      req.PrivateLinkServiceId,
			PurgedAt:           time.Now().UTC().Format(time.RFC3339),
			PrivateLinkService: pls.DeepCopy(),
			CrossplaneRefs:     refs,
		}
		if err := s.writePurgeSnapshot(ctx, plsKey.Namespace, snapshot); err != nil {
			if bizerr.Is(err, bizerr.ResourceConflict) {
				s.recordPurgeResult(purgeResultConflict)
			} else {
				s.recordPurgeResult(purgeResultError)
			}
			return err
		}
	}

	refSummaries := make([]string, 0, len(snapshot.CrossplaneRefs))
	for _, r := range snapshot.CrossplaneRefs {
		refSummaries = append(refSummaries, fmt.Sprintf("%s/%s(existed=%t)", r.Kind, r.Name, r.Existed))
	}
	logger.Warn("purging PrivateLinkService metadata via finalizer removal",
		zap.Strings("crossplane_refs", refSummaries))

	// Step 6: purge each crossplane CR. Finalizer-removal + Delete bypasses the
	// crossplane controller entirely so the cloud resource is never touched.
	for _, ref := range snapshot.CrossplaneRefs {
		if !ref.Existed {
			continue
		}
		if err := s.purgeCrossplaneRef(ctx, ref); err != nil {
			logger.Error("failed to purge crossplane ref",
				zap.String("kind", ref.Kind), zap.String("name", ref.Name), zap.Error(err))
			s.recordPurgeResult(purgeResultError)
			return bizerr.New(bizerr.K8SClientError,
				fmt.Sprintf("failed to purge %s/%s: %v", ref.Kind, ref.Name, err))
		}
	}

	// Step 7: purge the PLS CR itself.
	if err := s.purgePLS(ctx, pls); err != nil {
		logger.Error("failed to purge PrivateLinkService", zap.Error(err))
		s.recordPurgeResult(purgeResultError)
		return bizerr.New(bizerr.K8SClientError, "failed to purge privatelinkservice")
	}

	logger.Warn("PurgePrivateLinkMetadata completed")
	s.recordPurgeResult(purgeResultSuccess)
	return nil
}

// validateExistingSnapshot returns a ResourceConflict error if the snapshot found on
// cluster is not compatible with the current request (wrong identity / schema).
func validateExistingSnapshot(snap *purgeSnapshot, req *regionalapiv1.PurgePrivateLinkMetadataRequest) error {
	if snap.SchemaVersion != purgeSnapshotSchemaVersion {
		return bizerr.New(bizerr.ResourceConflict,
			fmt.Sprintf("snapshot schemaVersion %d is not supported (expected %d)",
				snap.SchemaVersion, purgeSnapshotSchemaVersion))
	}
	if snap.ClusterID != req.ClusterId || snap.PrivateLinkID != req.PrivateLinkServiceId {
		return bizerr.New(bizerr.ResourceConflict,
			"snapshot configmap already exists for a different cluster/privatelinkservice")
	}
	return nil
}

// collectAWSCrossplaneRefs resolves downstream AWS crossplane CR names using the same
// logic as the controller tasks, then fetches each CR so its full manifest can be
// embedded in the snapshot.
func (s *PrivateLinkServiceService) collectAWSCrossplaneRefs(ctx context.Context, pls *v1alpha1.PrivateLinkService) ([]purgeCrossplaneRef, error) {
	var refs []purgeCrossplaneRef

	vpceServiceName := utils.GetEndpointServiceName(pls)
	ref := purgeCrossplaneRef{
		Kind:          "VPCEndpointServiceConfiguration",
		APIVersion:    ec2v1alpha1.GroupVersion.String(),
		Name:          vpceServiceName,
		AnnotationKey: v1alpha1.AnnotationKeyOwnCrossplaneNameVPCEService,
	}
	ves := &ec2v1alpha1.VPCEndpointServiceConfiguration{}
	if err := s.cli.Get(ctx, types.NamespacedName{Name: vpceServiceName}, ves); err != nil {
		if !apierrors.IsNotFound(err) {
			return nil, fmt.Errorf("get VPCEndpointServiceConfiguration %s: %w", vpceServiceName, err)
		}
	} else {
		ref.Existed = true
		if ves.Status.AtProvider.ServiceConfiguration != nil && ves.Status.AtProvider.ServiceConfiguration.ServiceID != nil {
			ref.CloudID = *ves.Status.AtProvider.ServiceConfiguration.ServiceID
		}
		manifest, err := toGenericMap(ves)
		if err != nil {
			return nil, fmt.Errorf("marshal VPCEndpointServiceConfiguration %s: %w", vpceServiceName, err)
		}
		ref.ManifestBeforePurge = manifest
	}
	refs = append(refs, ref)

	recordSetName := pls.Status.EndpointService.PrivateDNSName
	if name := pls.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameResourceRecordSet]; name != "" {
		recordSetName = name
	}
	if recordSetName == "" {
		log.Info("ResourceRecordSet name cannot be resolved from annotation or status, skipping",
			zap.String("pls", pls.Name),
			zap.String("annotation", v1alpha1.AnnotationKeyOwnCrossplaneNameResourceRecordSet))
		return refs, nil
	}

	ref = purgeCrossplaneRef{
		Kind:          "ResourceRecordSet",
		APIVersion:    route53v1alpha1.SchemeGroupVersion.String(),
		Name:          recordSetName,
		AnnotationKey: v1alpha1.AnnotationKeyOwnCrossplaneNameResourceRecordSet,
	}
	rrs := &route53v1alpha1.ResourceRecordSet{}
	if err := s.cli.Get(ctx, types.NamespacedName{Name: recordSetName}, rrs); err != nil {
		if !apierrors.IsNotFound(err) {
			return nil, fmt.Errorf("get ResourceRecordSet %s: %w", recordSetName, err)
		}
	} else {
		ref.Existed = true
		manifest, err := toGenericMap(rrs)
		if err != nil {
			return nil, fmt.Errorf("marshal ResourceRecordSet %s: %w", recordSetName, err)
		}
		ref.ManifestBeforePurge = manifest
	}
	refs = append(refs, ref)

	return refs, nil
}

// purgeCrossplaneRef removes a single crossplane CR via finalizer clearing. Before
// clearing finalizers it also sets the `crossplane.io/paused` annotation, as a
// belt-and-suspenders against a racing crossplane reconcile.
//
// Finalizer-clearing is the non-invasive path: once the finalizer list is empty, the
// Delete operation succeeds purely at the k8s level without any provider code running,
// so the underlying AWS resource (VPC endpoint service, Route53 record) is preserved.
func (s *PrivateLinkServiceService) purgeCrossplaneRef(ctx context.Context, ref purgeCrossplaneRef) error {
	switch ref.Kind {
	case "VPCEndpointServiceConfiguration":
		obj := &ec2v1alpha1.VPCEndpointServiceConfiguration{}
		return s.finalizerClearAndDelete(ctx, obj, types.NamespacedName{Name: ref.Name})
	case "ResourceRecordSet":
		obj := &route53v1alpha1.ResourceRecordSet{}
		return s.finalizerClearAndDelete(ctx, obj, types.NamespacedName{Name: ref.Name})
	default:
		return fmt.Errorf("unknown crossplane kind %q", ref.Kind)
	}
}

// finalizerClearAndDelete is the core primitive: pause (best-effort), strip
// all finalizers, then Delete. Each step is idempotent and NotFound-tolerant.
func (s *PrivateLinkServiceService) finalizerClearAndDelete(ctx context.Context, obj client.Object, key types.NamespacedName) error {
	if err := s.cli.Get(ctx, key, obj); err != nil {
		if apierrors.IsNotFound(err) {
			return nil
		}
		return err
	}
	if err := s.pauseAndStripFinalizers(ctx, obj); err != nil {
		return err
	}
	if err := s.cli.Delete(ctx, obj); err != nil && !apierrors.IsNotFound(err) {
		return err
	}
	return nil
}

// pauseAndStripFinalizers patches obj to (a) set crossplane.io/paused=true and (b)
// drop all finalizers. Both happen in a single patch to minimise races with the
// crossplane controller. It's a no-op if both are already in the target state.
func (s *PrivateLinkServiceService) pauseAndStripFinalizers(ctx context.Context, obj client.Object) error {
	base := obj.DeepCopyObject().(client.Object)

	changed := false
	if len(obj.GetFinalizers()) != 0 {
		obj.SetFinalizers(nil)
		changed = true
	}
	anno := obj.GetAnnotations()
	if anno == nil {
		anno = map[string]string{}
	}
	if anno[meta.AnnotationKeyReconciliationPaused] != "true" {
		anno[meta.AnnotationKeyReconciliationPaused] = "true"
		obj.SetAnnotations(anno)
		changed = true
	}
	if !changed {
		return nil
	}
	if err := s.cli.Patch(ctx, obj, client.MergeFrom(base)); err != nil && !apierrors.IsNotFound(err) {
		return err
	}
	return nil
}

// purgePLS removes the PrivateLinkService CR via finalizer clearing + Delete.
func (s *PrivateLinkServiceService) purgePLS(ctx context.Context, pls *v1alpha1.PrivateLinkService) error {
	if len(pls.GetFinalizers()) != 0 {
		base := pls.DeepCopy()
		pls.SetFinalizers(nil)
		if err := s.cli.Patch(ctx, pls, client.MergeFrom(base)); err != nil && !apierrors.IsNotFound(err) {
			return err
		}
	}
	if err := s.privatelinkservicerw.Delete(ctx, pls); err != nil && !apierrors.IsNotFound(err) {
		return err
	}
	return nil
}

// writePurgeSnapshot creates the snapshot ConfigMap. The snapshot is the write-once
// authoritative record of the purge plan; the caller MUST have verified that no snapshot
// already exists (via loadPurgeSnapshot). We still tolerate a racy AlreadyExists from a
// concurrent Create by re-loading and validating identity.
func (s *PrivateLinkServiceService) writePurgeSnapshot(ctx context.Context, namespace string, snapshot *purgeSnapshot) error {
	payload, err := json.MarshalIndent(snapshot, "", "  ")
	if err != nil {
		return bizerr.New(bizerr.InternalServerError, fmt.Sprintf("marshal snapshot: %v", err))
	}
	cm := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      purgeSnapshotCMName(snapshot.PrivateLinkID),
			Namespace: namespace,
			Labels: map[string]string{
				purgeSnapshotLabelOwnedBy:   purgeSnapshotLabelOwnedByVal,
				purgeSnapshotLabelClusterID: snapshot.ClusterID,
				purgeSnapshotLabelPLSID:     snapshot.PrivateLinkID,
			},
			Annotations: map[string]string{
				purgeSnapshotAnnoPurgedAt: snapshot.PurgedAt,
			},
		},
		Data: map[string]string{
			purgeSnapshotDataKey: string(payload),
		},
	}
	if err := s.cli.Create(ctx, cm); err != nil {
		if !apierrors.IsAlreadyExists(err) {
			return bizerr.New(bizerr.K8SClientError, fmt.Sprintf("create snapshot configmap: %v", err))
		}
		// Race: another concurrent purge created the snapshot between our load and
		// write. Verify the existing snapshot targets the same PLS and return OK.
		existing, found, loadErr := s.loadPurgeSnapshot(ctx, namespace, snapshot.PrivateLinkID)
		if loadErr != nil {
			return bizerr.New(bizerr.K8SClientError, fmt.Sprintf("load snapshot after race: %v", loadErr))
		}
		if !found {
			return bizerr.New(bizerr.K8SClientError, "snapshot race: create reported AlreadyExists but subsequent Get returned NotFound")
		}
		if existing.ClusterID != snapshot.ClusterID || existing.PrivateLinkID != snapshot.PrivateLinkID {
			return bizerr.New(bizerr.ResourceConflict,
				"snapshot configmap already exists for a different cluster/privatelinkservice")
		}
	}
	return nil
}

// loadPurgeSnapshot fetches a snapshot ConfigMap for the given PLS. It returns
// (snapshot, found, error); found=false with nil error means the CM does not exist yet.
func (s *PrivateLinkServiceService) loadPurgeSnapshot(ctx context.Context, namespace, plsID string) (*purgeSnapshot, bool, error) {
	cm := &corev1.ConfigMap{}
	if err := s.cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: purgeSnapshotCMName(plsID)}, cm); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, false, nil
		}
		return nil, false, err
	}
	raw, ok := cm.Data[purgeSnapshotDataKey]
	if !ok {
		return nil, false, fmt.Errorf("snapshot configmap %q is missing data key %q", cm.Name, purgeSnapshotDataKey)
	}
	var snap purgeSnapshot
	if err := json.Unmarshal([]byte(raw), &snap); err != nil {
		return nil, false, fmt.Errorf("unmarshal snapshot %q: %w", cm.Name, err)
	}
	return &snap, true, nil
}

func purgeSnapshotCMName(plsID string) string {
	return purgeSnapshotCMNamePrefix + plsID
}

// toGenericMap serialises an object via JSON into a map[string]interface{}, so that the
// snapshot is self-describing and doesn't hard-depend on the Go types at restore time.
func toGenericMap(obj client.Object) (map[string]interface{}, error) {
	raw, err := json.Marshal(obj)
	if err != nil {
		return nil, err
	}
	var m map[string]interface{}
	if err := json.Unmarshal(raw, &m); err != nil {
		return nil, err
	}
	return m, nil
}

func (s *PrivateLinkServiceService) recordPurgeResult(result string) {
	inframetrics.PrivateLinkPurgeMetadataTotal.WithLabelValues(s.config.Provider, result).Inc()
}