package cluster

import (
	"context"
	"fmt"

	"github.com/pingcap/errors"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/multierr"
	corev1 "k8s.io/api/core/v1"
	rbacv1beta1 "k8s.io/api/rbac/v1beta1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	reconcilerr "github.com/tidbcloud/infra-api/controllers/common/error"
	"github.com/tidbcloud/infra-api/pkg/consts"
	"github.com/tidbcloud/infra-api/pkg/k8s"
	tracerutil "github.com/tidbcloud/infra-api/pkg/tracer"
)

const (
	tidbServiceTargetPort = 4000

	legacyPrivateRecordSuffix = "-inpublic"
)

var (
	waitPublicService  = fmt.Errorf("wait public tidb service")
	waitPrivateService = fmt.Errorf("wait private tidb service")
)

// reconcileTCSuspendActions reconciles the suspend
// actions for an existing TidbCluster resource.
// Please note that we can suspend the cluster
// at any phase from the API view.
func (r *reconcileContext) reconcileTCSuspendActions(ctx context.Context) (errResult error) {
	_, span := tracer.Start(ctx, "reconcileTCSuspendActions")
	defer func() { tracerutil.EndSpan(span, errResult) }()
	// Update suspendActions
	// No need to check the cluster status since
	// we should be able to suspend at any phase
	if _, err := r.mutateTidbCluster(r.overlaySuspendActions); err != nil {
		return &reconcilerr.RequeueAfterError{RequeueAfter: defaultWaitInterval, Caused: err}
	}

	return nil
}

// reconcileTidbCluster reconciles an existing TidbCluster resource
// ensureTidbCluster() and reconcileTidbCluster() are separated for backward compatibility since we
// will only mutate replicas, suspendActions, and configs after the cluster is created. We will finally merge these
// two functions with future improvements.
func (r *reconcileContext) reconcileTidbCluster(ctx context.Context) (errResult error) {
	_, span := tracer.Start(ctx, "reconcileTidbCluster")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	// since we're not the only owner of tidbcluster resource (gitops may also manage tc), do an observation
	// to get the latest real state
	if err := r.observeTidbClusterStatus(); err != nil {
		return err
	}
	// for reliability, some operations are only taken when the cluster is running normally
	whenRunning := whenPhaseIn([]cloudv1.ClusterPhase{cloudv1.ClusterPhaseRunning})

	ignoreMutatedBool := func(_ bool, err error) error {
		return err
	}
	errs := multierr.Combine(
		ignoreMutatedBool(r.mutateTidbCluster(r.overlayTags)),
		ignoreMutatedBool(r.mutateTidbCluster(r.overlayAnnotations)),
		ignoreMutatedBool(r.mutateTidbCluster(r.overlayBillingLabels)),
	)
	if errs != nil {
		return errs
	}

	// For r.overlayTiFlash() when computing config of tiflash.
	if err := r.setKMSFromKeyResource(ctx); err != nil {
		return err
	}
	if err := r.ensureClusterStorageClass(ctx); err != nil {
		return err
	}
	// reconcile tiflash
	if mutated, err := r.mutateTidbCluster(r.OverlayTiFlash, whenRunning); err != nil {
		return err
	} else if mutated {
		if err := r.updateProgress(cloudv1.ClusterPhaseScaling, "tiflash is toggled"); err != nil {
			return err
		}
	}

	// createTLSCertBetweenComponents create cert only for components enabled.
	// so we need to recheck it.
	err := r.createTLSCertBetweenComponents(r.tc.Namespace, r.tc.Name)
	if err != nil {
		return err
	}

	// if deployment are changed (mixed to normal), new PD nodes will be created, wait for them before mutate cluster
	if err := r.observeNodeProgressAndReason(); err != nil {
		return &reconcilerr.RequeueAfterError{RequeueAfter: defaultWaitInterval, Caused: err}
	}

	// scale tidbcluster, with respect of cross zone balancing
	if mutated, err := r.mutateTidbCluster(r.scaleCluster, whenRunning); err != nil {
		return &reconcilerr.RequeueAfterError{RequeueAfter: defaultWaitInterval, Caused: err}
	} else if mutated {
		if err := r.updateProgress(cloudv1.ClusterPhaseScaling, "cluster scaling requested"); err != nil {
			return err
		}
	}

	// scale storage
	if _, err := r.mutateTidbCluster(r.scaleStorage, whenRunning); err != nil {
		return &reconcilerr.RequeueAfterError{RequeueAfter: defaultWaitInterval, Caused: err}
	}

	// Update suspendActions
	// No need to check the cluster status since
	// we should be able to suspend at any phase
	if _, err := r.mutateTidbCluster(r.overlaySuspendActions); err != nil {
		return &reconcilerr.RequeueAfterError{RequeueAfter: defaultWaitInterval, Caused: err}
	}

	// upgrading
	// no-op since the current upgrading reconciliation is handled by config-service
	// TODO: migrate upgrading logics from config-service
	return nil
}

func (r *reconcileContext) reconcilePublicEndpoint(ctx context.Context) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "reconcilePublicEndpoint")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	if err := r.refreshTidbClusterStatus(); err != nil {
		return err
	}
	// update tidb endpoint spec in tc to match the cluster spec
	if _, err := r.mutateTidbCluster(r.overlayTiDBEndpoint); err != nil {
		return errors.AddStack(err)
	}

	needPublicEndpoint := r.Target.Spec.SqlEndpoints.Public.IsEnabled()
	publicEndpointReady := r.Target.Status.SqlEndpoints.PublicEndpoint != nil

	if needPublicEndpoint && (!publicEndpointReady) {
		// 1. wait service loadbalancer reconciled
		span.AddEvent("Wait service loadbalancer reconciled")
		observed := &corev1.Service{}
		svcKey := types.NamespacedName{Namespace: r.tc.Namespace, Name: fmt.Sprintf("%s-tidb", r.tc.Name)}
		if err := r.shootClient.Get(spanCtx, svcKey, observed); err != nil {
			return &reconcilerr.RequeueAfterError{RequeueAfter: defaultWaitInterval, Caused: waitPublicService}

		}
		if len(observed.Status.LoadBalancer.Ingress) == 0 {
			return &reconcilerr.RequeueAfterError{RequeueAfter: defaultWaitInterval, Caused: waitPublicService}
		}
		ingress := observed.Status.LoadBalancer.Ingress[0]

		// 2. ensure heath check point
		span.AddEvent("Ensure heath check point")
		if r.providerGardenerSpec.GetEnableGracefulShutdownWebhook() {
			if _, exist := observed.Annotations[consts.AnnHealthCheckPort]; !exist {
				_, err := controllerutil.CreateOrUpdate(spanCtx, r.shootClient, observed, func() error {
					return setHealthCheckPortAnnotation(observed)
				})
				if err != nil {
					return errors.AddStack(err)
				}
			}
		}

		// 3. add an dns alias to the loadbalancer
		span.AddEvent("Add an dns alias to the loadbalancer")

		dnsName := r.publicEndpointAddress()

		switch r.ProviderType {
		case consts.ProviderTypeAWS:
			dns := DnsAlias{
				Name:         r.publicDnsRecordName(),
				HostedZoneId: r.publicHostedZoneID,
				Alias:        dnsName,
				Address:      ingress.Hostname,
			}

			if isUseLegacyDNS(r.Target) {
				dns.HostedZoneId = r.providerGardenerSpec.AWS.PublicHostedZoneId
			}

			if err := r.ensureAwsDnsRecord(dns); err != nil {
				return err
			}
		case consts.ProviderTypeGCP:
			if isUseLegacyDNS(r.Target) {
				hostedZoneId := r.providerGardenerSpec.GCP.PublicHostedZoneId
				if err := r.ensureLegacyGcpDnsRecord(DnsAlias{
					Name:         r.publicDnsRecordName(),
					HostedZoneId: hostedZoneId,
					Alias:        dnsName,
					Address:      ingress.IP,
				}); err != nil {
					return err
				}
			} else {
				if err := r.ensureGcpDnsRecord(DnsAlias{
					Name:         r.publicDnsRecordName(),
					HostedZoneId: r.publicHostedZoneID,
					Alias:        dnsName,
					Address:      ingress.IP,
				}); err != nil {
					return err
				}
			}

		default:
			return fmt.Errorf("unsupported provider %s", r.ProviderType)
		}

		// 4. checkpoint the alias dns name to cluster status
		span.AddEvent("Checkpoint the alias dns name to cluster status")
		if err := r.ensureStatus(func() error {
			r.Target.Status.SqlEndpoints.PublicEndpoint = &cloudv1.SqlEndpointStatus{
				Host: dnsName,
				Port: int(r.Target.GetPublicSQLEndpointPort()),
			}
			return nil
		}); err != nil {
			return err
		}
	}
	if (!needPublicEndpoint) && publicEndpointReady {
		// FIXME(aylei): if the public endpoint creation process is not succeed, the semi-state would not be cleaned
		// 1. delete public endpoints
		span.AddEvent("Delete public endpoints")
		if err := r.deletePublicEndpoints(); err != nil {
			return errors.AddStack(err)
		}
		if err := r.ensureStatus(func() error {
			r.Target.Status.SqlEndpoints.PublicEndpoint = nil
			return nil
		}); err != nil {
			return err
		}
	}
	return nil
}

func (r *reconcileContext) reconcilePrivateEndpoint(ctx context.Context) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "reconcilePrivateEndpoint")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	if err := r.refreshTidbClusterStatus(); err != nil {
		return err
	}
	needPrivateEndpoint := r.Target.Spec.SqlEndpoints.Private != nil && r.Target.Spec.SqlEndpoints.Private.IsEnabled()
	privateEndpointReady := r.Target.Status.SqlEndpoints.PrivateEndpoint != nil
	// try enable private service is not yet ready
	if needPrivateEndpoint && (!privateEndpointReady) {
		// 1. ensure service

		span.AddEvent("Ensure private service")
		svc, err := r.newPrivateService(r.tc, r.Provider)
		if err != nil {
			return err
		}
		if err := r.createSubresourceOfTidbCluster(svc); err != nil {
			return err
		}
		observed := &corev1.Service{}
		if err := r.shootClient.Get(r.ctx, types.NamespacedName{Namespace: svc.Namespace, Name: svc.Name}, observed); err != nil {
			return err
		}

		// 2. wait service loadbalancer reconciled
		span.AddEvent("Wait service loadbalancer reconciled")
		if len(observed.Status.LoadBalancer.Ingress) == 0 {
			return &reconcilerr.RequeueAfterError{RequeueAfter: defaultWaitInterval, Caused: waitPrivateService}
		}
		ingress := observed.Status.LoadBalancer.Ingress[0]

		// 3. ensure heath check point
		span.AddEvent("Ensure heath check point")
		if r.providerGardenerSpec.GetEnableGracefulShutdownWebhook() {
			if _, exist := observed.Annotations[consts.AnnHealthCheckPort]; !exist {
				_, err := controllerutil.CreateOrUpdate(spanCtx, r.shootClient, observed, func() error {
					return setHealthCheckPortAnnotation(observed)
				})
				if err != nil {
					return errors.AddStack(err)
				}
			}
		}

		// 4. add an dns alias record to the loadbalancer to avoid clients access the real address directly (which is
		// out of our control)
		dnsName := getPrivateEndpointAddress(r.publicEndpointAddress())
		switch r.ProviderType {
		case consts.ProviderTypeAWS:
			// no need to create dns record in aws private hostedzone
			dns := DnsAlias{
				Name:         r.privateDnsRecordName() + legacyPrivateRecordSuffix,
				HostedZoneId: r.publicHostedZoneID,
				Alias:        dnsName,
				Address:      ingress.Hostname,
			}
			if isUseLegacyDNS(r.Target) {
				dns.HostedZoneId = r.providerGardenerSpec.AWS.PublicHostedZoneId
			}

			if err = r.ensureAwsDnsRecord(dns); err != nil {
				return errors.Annotate(err, "create private AWS dns record in public hosted zone")
			}

		case consts.ProviderTypeGCP:
			if isUseLegacyDNS(r.Target) {
				if r.networkGardenStatus.GCP.HostedZoneID == nil {
					return errors.New("wait GCP shoot private hosted zone ready")
				}
				hostedZoneId := *r.networkGardenStatus.GCP.HostedZoneID
				span.AddEvent("ensureLegacyGcpDnsRecord")
				if err := r.ensureLegacyGcpDnsRecord(DnsAlias{
					Name:         r.privateDnsRecordName(),
					HostedZoneId: hostedZoneId,
					Alias:        dnsName,
					Address:      ingress.IP,
				}); err != nil {
					return errors.Annotate(err, "create private GCP dns record in private hosted zone")
				}
				// FIXME: this is a legacy bug that we should keep in infra-api https://jira.tidbcloud.com/browse/DM-2516
				if err := r.ensureLegacyGcpDnsRecord(DnsAlias{
					Name:         r.privateDnsRecordName() + legacyPrivateRecordSuffix,
					HostedZoneId: r.providerGardenerSpec.GCP.PublicHostedZoneId,
					Alias:        dnsName,
					Address:      ingress.IP,
				}); err != nil {
					return errors.Annotate(err, "create private GCP dns record in public hosted zone")
				}
			} else {
				if err := r.ensureGcpDnsRecord(DnsAlias{
					Name:         r.privateDnsRecordName() + legacyPrivateRecordSuffix,
					HostedZoneId: r.publicHostedZoneID,
					Alias:        dnsName,
					Address:      ingress.IP,
				}); err != nil {
					return errors.Annotate(err, "create private GCP dns record in public hosted zone")
				}
			}

		default:
			return fmt.Errorf("unsupported provider %s", r.ProviderType)
		}

		// 4. checkpoint the alias dns name to cluster status
		span.AddEvent("Checkpoint the alias dns name to cluster status")
		if err := r.ensureStatus(func() error {
			r.Target.Status.SqlEndpoints.PrivateEndpoint = &cloudv1.SqlEndpointStatus{
				Host: dnsName,
				Port: int(r.Target.GetPrivateSQLEndpointPort()),
			}
			return nil
		}); err != nil {
			return err
		}
	}
	if !needPrivateEndpoint && (privateEndpointReady) {
		span.AddEvent("Delete private endpoint")
		if err := r.deletePrivateEndpoints(); err != nil {
			return errors.AddStack(err)
		}
		if err := r.ensureStatus(func() error {
			r.Target.Status.SqlEndpoints.PrivateEndpoint = nil
			return nil
		}); err != nil {
			return err
		}
	}
	return nil
}

func (r *reconcileContext) reconcileBackupRBAC(ctx context.Context) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "reconcileBackupRBAC", trace.WithSpanKind(trace.SpanKindClient))
	defer func() { tracerutil.EndSpan(span, errResult) }()

	namespace := r.Target.PrefixedId()
	sa := &corev1.ServiceAccount{
		ObjectMeta: metav1.ObjectMeta{
			Name:      tidbBackupServiceAccountName,
			Namespace: namespace,
		},
	}
	if _, err := k8s.CreateIgnoreExists(spanCtx, r.shootClient, sa); err != nil {
		return err
	}

	role := &rbacv1beta1.Role{
		ObjectMeta: metav1.ObjectMeta{
			Name:      tidbBackupServiceAccountName,
			Labels:    r.SubResourceLabels(),
			Namespace: namespace,
		},
	}
	if _, err := controllerutil.CreateOrUpdate(spanCtx, r.shootClient, role, func() error {
		role.Rules = []rbacv1beta1.PolicyRule{
			{
				APIGroups: []string{""},
				Resources: []string{"events"},
				Verbs:     []string{"*"},
			},
			{
				APIGroups: []string{"pingcap.com"},
				Resources: []string{"backups", "restores"},
				Verbs:     []string{"get", "watch", "list", "update"},
			},
		}
		return nil
	}); err != nil {
		return err
	}

	roleBinding := &rbacv1beta1.RoleBinding{
		ObjectMeta: metav1.ObjectMeta{
			Name:      tidbBackupServiceAccountName,
			Labels:    r.SubResourceLabels(),
			Namespace: namespace,
		},
	}
	if _, err := controllerutil.CreateOrUpdate(spanCtx, r.shootClient, roleBinding, func() error {
		roleBinding.Subjects = []rbacv1beta1.Subject{
			{
				Kind: rbacv1beta1.ServiceAccountKind,
				Name: tidbBackupServiceAccountName,
			},
		}
		roleBinding.RoleRef = rbacv1beta1.RoleRef{
			APIGroup: "rbac.authorization.k8s.io",
			Kind:     "Role",
			Name:     tidbBackupServiceAccountName,
		}
		return nil
	}); err != nil {
		return err
	}
	return nil
}