package cluster

import (
	"context"
	"fmt"
	"strings"

	"github.com/aws/aws-sdk-go-v2/aws"
	kmsapi "github.com/aws/aws-sdk-go-v2/service/kms"
	kmstypes "github.com/aws/aws-sdk-go-v2/service/kms/types"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	awskmsv1alpha1 "github.com/crossplane/provider-aws/apis/kms/v1alpha1"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"go.opentelemetry.io/otel/trace"
	"golang.org/x/mod/semver"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/controllers/common"
	reconcileerr "github.com/tidbcloud/infra-api/controllers/common/error"
	"github.com/tidbcloud/infra-api/controllers/mixeddeploypd"
	"github.com/tidbcloud/infra-api/controllers/utils"
	"github.com/tidbcloud/infra-api/pkg/audit"
	"github.com/tidbcloud/infra-api/pkg/consts"
	"github.com/tidbcloud/infra-api/pkg/k8s"
	tidbclusterutils "github.com/tidbcloud/infra-api/pkg/pingcap/util/tidbcluster"
	awsapi "github.com/tidbcloud/infra-api/pkg/providers/aws"
	tracerutil "github.com/tidbcloud/infra-api/pkg/tracer"
)

const (
	defaultKMSRegionForGCP = "us-west-2"

	kmsPermissionGrantedAnno = "key.cloud.pingcap.com/granted"
	kmsRotationDisabledAnno  = "key.cloud.pingcap.com/rotation-disabled"

	dashboardSessionSecretName   = "db-dashboard-session-secret" // #nosec G101
	tidbBackupServiceAccountName = "tidb-backup-manager"

	gcpKMSSecretName string = "gcp-kms-credentials" // #nosec G101
)

var (
	waitNodeReady = fmt.Errorf("wait all node ready")
)

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

	// both AWS and GCP clusters use AWS KMS key
	region := r.Network.Spec.Region
	if r.ProviderType == consts.ProviderTypeGCP {
		region = defaultKMSRegionForGCP
	}
	k := &awskmsv1alpha1.Key{
		ObjectMeta: metav1.ObjectMeta{
			Name:   r.Target.ClusterScopeName(),
			Labels: r.SubResourceLabels(),
		},
		Spec: awskmsv1alpha1.KeySpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: r.providerGardenerSpec.AWS.CrossplaneProvider,
				},
			},
			ForProvider: awskmsv1alpha1.KeyParameters{
				Region: region,
			},
		},
	}
	if _, err := controllerutil.CreateOrUpdate(spanCtx, r.Client, k, func() error {
		k.Labels = r.SubResourceLabels()
		return nil
	}); err != nil {
		return errors.Annotate(err, "ensure kms key")
	}
	r.KmsRegion = &region
	return nil
}

func (r *reconcileContext) setKMSFromKeyResource(ctx context.Context) error {
	k := &awskmsv1alpha1.Key{}
	keyName := types.NamespacedName{Name: r.Target.ClusterScopeName()}
	err := r.Client.Get(ctx, keyName, k)
	if err != nil {
		return err
	}

	r.KmsRegion = &k.Spec.ForProvider.Region
	r.KmsKeyId = k.Status.AtProvider.KeyID
	return nil
}

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

	var err error

	// get key
	k := &awskmsv1alpha1.Key{}
	keyName := types.NamespacedName{Name: r.Target.ClusterScopeName()}
	if err = r.Client.Get(spanCtx, keyName, k); err != nil {
		return k8s.RequeueIf(err, errors.IsNotFound, defaultWaitInterval)
	}
	if k.Status.AtProvider.KeyID == nil {
		span.AddEvent("kms key not ready, requeuing")
		r.Log.Info("kms key not ready, requeuing")
		return &reconcileerr.RequeueAfterError{Caused: nil, RequeueAfter: defaultWaitInterval}
	}

	// disable key rotation
	span.AddEvent("Disable key rotation")
	if err = common.AtLeastOnce(spanCtx, r.Client, r.Target, kmsRotationDisabledAnno, func() error {
		_, err = r.AWS.KMS().DisableKeyRotation(spanCtx, &kmsapi.DisableKeyRotationInput{
			KeyId: k.Status.AtProvider.KeyID,
		}, awsapi.WithKMSRegion(k.Spec.ForProvider.Region))
		return err
	}); err != nil {
		return errors.Annotate(err, "disable key rotation")
	}

	// grant key permission to TiKV
	switch r.ProviderType {
	case consts.ProviderTypeAWS:
		// grant permission to node role for AWS shoot
		span.AddEvent("Grant permission to node role for AWS shoot")
		if err = common.AtLeastOnce(spanCtx, r.Client, r.Target, kmsPermissionGrantedAnno, func() error {
			_, err = r.AWS.KMS().CreateGrant(spanCtx, &kmsapi.CreateGrantInput{
				KeyId:            k.Status.AtProvider.KeyID,
				GranteePrincipal: aws.String(r.networkGardenStatus.NodeRole),
				Operations: []kmstypes.GrantOperation{
					kmstypes.GrantOperationEncrypt,
					kmstypes.GrantOperationDecrypt,
					kmstypes.GrantOperationGenerateDataKey,
					kmstypes.GrantOperationGenerateDataKeyWithoutPlaintext,
				},
			}, awsapi.WithKMSRegion(k.Spec.ForProvider.Region))
			return err
		}); err != nil {
			return errors.Annotate(err, "grant key permission to AWS shoot node role")
		}
	case consts.ProviderTypeGCP:
		// grant permission to node role for GCP shoot
		span.AddEvent("Grant permission to node role for GCP shoot")
		if r.networkGardenStatus.GCP.TiKVAWSUserArn == nil || r.networkGardenStatus.GCP.KmsSecretName == nil {
			return errors.New("waiting the AWS user arn of GCP TiKV")
		}
		if err = common.AtLeastOnce(spanCtx, r.Client, r.Target, kmsPermissionGrantedAnno, func() error {
			if legacyUser, ok := r.Network.Annotations[cloudv1.LegacyTiKVUsername]; ok {
				// non-project before infra-api will have legacy tikv user that we should maintain permissions
				_, err = r.AWS.KMS().CreateGrant(spanCtx, &kmsapi.CreateGrantInput{
					KeyId:            k.Status.AtProvider.KeyID,
					GranteePrincipal: aws.String(legacyUser),
					Operations: []kmstypes.GrantOperation{
						kmstypes.GrantOperationEncrypt,
						kmstypes.GrantOperationDecrypt,
						kmstypes.GrantOperationGenerateDataKey,
						kmstypes.GrantOperationGenerateDataKeyWithoutPlaintext,
					},
				}, awsapi.WithKMSRegion(k.Spec.ForProvider.Region))
				if err != nil {
					return err
				}
			}
			_, err = r.AWS.KMS().CreateGrant(spanCtx, &kmsapi.CreateGrantInput{
				KeyId:            k.Status.AtProvider.KeyID,
				GranteePrincipal: r.networkGardenStatus.GCP.TiKVAWSUserArn,
				Operations: []kmstypes.GrantOperation{
					kmstypes.GrantOperationEncrypt,
					kmstypes.GrantOperationDecrypt,
					kmstypes.GrantOperationGenerateDataKey,
					kmstypes.GrantOperationGenerateDataKeyWithoutPlaintext,
				},
			}, awsapi.WithKMSRegion(k.Spec.ForProvider.Region))
			if err != nil {
				return err
			}
			return r.ensureKmsAccessSecretForGCPTiKV()
		}); err != nil {
			return errors.Annotate(err, "grant key permission to GCP kms user")
		}
	default:
		return errors.Errorf("unknown provider type %s", r.ProviderType)
	}

	r.KmsKeyId = k.Status.AtProvider.KeyID
	return nil
}

func (r *reconcileContext) ensureKmsAccessSecretForGCPTiKV() error {
	secret := &corev1.Secret{}
	namespacedName := types.NamespacedName{
		Namespace: r.networkGardenStatus.AdminNamespace,
		Name:      *r.networkGardenStatus.GCP.KmsSecretName,
	}
	err := r.shootClient.Get(r.ctx, namespacedName, secret)
	if err != nil {
		return errors.Annotate(err, "get kms access secret")
	}
	secretCopy := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      gcpKMSSecretName,
			Namespace: r.Target.PrefixedId(),
		},
		Data: secret.Data,
		Type: corev1.SecretTypeOpaque,
	}
	_, err = controllerutil.CreateOrUpdate(r.ctx, r.shootClient, secretCopy, func() error {
		secretCopy.Data = secret.Data
		return nil
	})
	if err != nil {
		return errors.Annotate(err, "create KMS access key secret in tidb cluster namespace")
	}
	return nil
}

func (r *reconcileContext) ensureClusterNamespace() error {
	ns := &corev1.Namespace{
		ObjectMeta: metav1.ObjectMeta{
			Name: r.Target.PrefixedId(),
		},
	}
	if _, err := k8s.CreateIgnoreExists(r.ctx, r.shootClient, ns); err != nil {
		return errors.Annotate(err, "ensure TiDB namespace")
	}
	return nil
}

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

	if r.KmsKeyId == nil || r.KmsRegion == nil {
		return errors.New("waiting kms key ready")
	}

	span.AddEvent("assembleTidbClusterValues")
	tc, err := r.AssembleTidbClusterValues(r.PDMixedDeployConfig)
	if err != nil {
		return errors.Annotate(err, "assemble TiDB cluster")
	}

	_, err = audit.Reconcile(context.TODO(), r.Network, r.Target, r.Client, tc, r.shootClient, true /*advanceSTS*/)
	if err != nil {
		return err
	}

	span.AddEvent("ensureDashboardSessionSecret")
	if err := r.ensureDashboardSessionSecret(tc.Namespace); err != nil {
		return errors.Annotate(err, "ensure dashboard session secret")
	}

	span.AddEvent("createTLSCertBetweenComponents")
	if err := r.createTLSCertBetweenComponents(tc.Namespace, tc.Name); err != nil {
		return errors.Annotate(err, "ensure tls between components")
	}

	span.AddEvent("createTLSCertForMySQLClient")
	if err := r.createTLSCertForMySQLClient(tc.Namespace, tc.Name, r.publicEndpointAddress()); err != nil {
		return errors.Annotate(err, "ensure tls of tidb-server")
	}

	if r.ProviderType == consts.ProviderTypeGCP {
		span.AddEvent("copySecretForBR")
		if err := r.copySecretForBR(tc.Namespace); err != nil {
			return errors.Annotate(err, "copy GCP Application Default Credentials secret for backup and restore")
		}
	}

	tenantID, projectID, _ := GetIDs(&r.ClusterContext)
	if cloudv1.FeatureEnabled(r.FeatureGates, cloudv1.TiDBExposeOOMPathForNewCluster, tenantID, projectID) {
		r.MountOOMPathToHost(tc)
	}
	if cloudv1.FeatureEnabled(r.FeatureGates, cloudv1.TiDBExposeReplayerPathForNewCluster, tenantID, projectID) {
		r.MountPlanReplayerPathToHost(tc)
	}

	if _, err := k8s.CreateIgnoreExists(spanCtx, r.shootClient, tc); err != nil {
		return errors.Annotate(err, "ensure tidbcluster resource")
	}

	span.AddEvent("refreshTidbClusterStatus")
	if err := r.refreshTidbClusterStatus(); err != nil {
		return err
	}
	return nil
}

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

	enableSEM := semSupported(r.tc.Spec.Version)
	enableAnalyze := analyzeSupported(r.tc)

	tenantID := utils.ParsingLegacyId(r.Target, consts.TenantTagKey)
	if tenantID == nil {
		dummyId := uint64(0)
		tenantID = &dummyId
	}
	projectID := utils.ParsingLegacyId(r.Target, consts.ProjectTagKey)
	if projectID == nil {
		dummyId := uint64(0)
		projectID = &dummyId
	}
	r.Log.Info(fmt.Sprintf("tenantID is %d, projectId is %d for this cluster", *tenantID, *projectID))
	experimentalEnabling := experimentalSupported(r.ExpFeatureGates, r.tc.Spec.Version, *tenantID, *projectID)
	initializer := NewDefaultTidbInitializer(r.tc.Namespace, enableSEM, enableAnalyze, experimentalEnabling)

	return r.createSubresourceOfTidbCluster(initializer)
}

func experimentalSupported(expFeatures []cloudv1.ExpFeatureGate, curVersion string, tenantID, projectID uint64) string {
	supported := make([]string, 0, len(expFeatures))
	for _, expFeature := range expFeatures {
		if semver.Compare(curVersion, expFeature.MinSupportVersion) >= 0 &&
			cloudv1.IsFeatureEnabled(expFeature.FeatureGate, tenantID, projectID) {
			commands := strings.Split(expFeature.Command, ";")
			for _, cmd := range commands {
				trimed := strings.TrimSpace(cmd)
				if len(trimed) > 0 {
					supported = append(supported, trimed)
				}
			}
		}
	}
	return strings.Join(supported, ";\n") + ";\n"
}

func (r *reconcileContext) ensureDashboardSessionSecret(ns string) error {
	dashboardEncryptionKey := cloudv1.RandomString(32)

	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      dashboardSessionSecretName,
			Namespace: ns,
			Labels:    r.SubResourceLabels(),
		},
		Data: map[string][]byte{
			"encryption_key": []byte(dashboardEncryptionKey),
		},
	}
	_, err := k8s.CreateIgnoreExists(r.ctx, r.shootClient, secret)
	return err
}

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

	if r.tc == nil {
		return errors.New("ensure admin: wait for tidb cluster to be ensured")
	}

	// skip to ensure extra tidb related resources when SEM is supported
	if semSupported(r.tc.Spec.Version) {
		initName := getInitializerName(r.tc.Name)
		_, err := r.getInitializer(types.NamespacedName{Namespace: r.tc.Namespace, Name: initName})
		return err
	}

	secret := newExtraSecret(r.tc)
	if err := r.createSubresourceOfTidbCluster(secret); err != nil {
		return err
	}
	cm := newExtraConfigMap(r.tc)
	if err := r.createSubresourceOfTidbCluster(cm); err != nil {
		return err
	}
	dep := newExtraAccessDeployment(r.tc, r.Target.PrefixedId())
	if err := r.createSubresourceOfTidbCluster(dep); err != nil {
		return err
	}
	svc := newExtraAccessSvc(r.tc)
	if err := r.createSubresourceOfTidbCluster(svc); err != nil {
		return err
	}

	return nil
}

func (r *reconcileContext) ensureReady(_ context.Context) (errResult error) {
	if err := r.refreshTidbClusterStatus(); err != nil {
		return err
	}
	tcReady := tidbclusterutils.GetTidbClusterReadyCondition(r.tc)
	if tcReady == nil {
		if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "waiting tidbcluster being reconciled"); err != nil {
			return err
		}
		return errors.New("tc is not yet get reconciled")
	}
	if tcReady.Status != corev1.ConditionTrue {
		if err := r.updateProgress(cloudv1.ClusterPhaseCreating, fmt.Sprintf("waiting tidbcluster, current phase: %s", tcReady.Message)); err != nil {
			return err
		}
		return &reconcileerr.RequeueAfterError{
			RequeueAfter: defaultWaitInterval,
			Caused:       errors.Errorf("TiDB cluster not ready yet"),
		}
	}
	return nil
}

// When creating a new cluster, security enhanced mode is disabled in the config.
// After the initializer succeeded, the sem will be enabled during reconcile
func (r *reconcileContext) ensureSecurityEnhancedMode(_ context.Context) error {
	if !semSupported(r.tc.Spec.Version) {
		return nil
	}
	// Wait until the initializer is completed
	succeeded, init, err := r.initSucceeded(r.tc)
	// r.Log.V(4).Info(fmt.Sprintf("Set phase: SEM Supported %v %+v %+v", succeeded, err, init))
	if err != nil {
		return errors.AddStack(err)
	}
	if !succeeded {
		if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "waiting tidbcluster enabling SEM"); err != nil {
			return err
		}
		if init != nil && init.Status.Phase == v1alpha1.InitializePhaseFailed {
			return errors.Errorf("Initializer failed %+v", init.Status)
		} else {
			err := errors.Errorf("Wait for initializer to complete %+v", init.Status)
			return &reconcileerr.RequeueAfterError{
				RequeueAfter: defaultWaitInterval,
				Caused:       err,
			}
		}
	}

	if r.tc.Spec.TiDB.Config == nil {
		r.tc.Spec.TiDB.Config = v1alpha1.NewTiDBConfig()
	}
	r.tc.Spec.TiDB.Config.Set("security.enable-sem", true)
	if err := r.shootClient.Update(r.ctx, r.tc); err != nil {
		return errors.Annotate(err, "enable-sem=true")
	}

	return nil
}

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

	return r.ensureStatus(func() error {
		r.Target.Status.Phase = cloudv1.ClusterPhaseRunning
		r.Target.Status.Reason = "cluster is running"
		switch r.ProviderType {
		case consts.ProviderTypeGCP:
			if err := cloudv1.MutateGcpClusterStatus(r.Target, func(status *cloudv1.GcpClusterStatus) error {
				if status.Kubernetes == nil {
					status.Kubernetes = &cloudv1.KubernetesStatus{}
				}
				status.Kubernetes.Namespace = r.tc.Namespace
				return nil
			}); err != nil {
				return err
			}
		case consts.ProviderTypeAWS:
			if err := cloudv1.MutateAwsClusterStatus(r.Target, func(status *cloudv1.AwsClusterStatus) error {
				if status.Kubernetes == nil {
					status.Kubernetes = &cloudv1.KubernetesStatus{}
				}
				status.Kubernetes.Namespace = r.tc.Namespace
				return nil
			}); err != nil {
				return err
			}
		default:
			return fmt.Errorf("unsupported provider type %s", r.ProviderType)
		}
		return nil
	})
}

// observeNodeProgress observes the cluster node provision provision progress and collect
// the possible reason if the nodes are not fully provisioned
func (r *reconcileContext) observeNodeProgressAndReason() error {
	if err := r.observeNodeProgress(); err != nil {
		if err != waitNodeReady {
			// unexpected error
			return err
		}
		// node not ready, try check the reason
		err := r.checkNodeNotProvisionedReason()
		if err != nil {
			return err
		}
	}
	return nil
}

// observeNodeProgress observes the cluster node provision provision progress
func (r *reconcileContext) observeNodeProgress() error {
	gardenerStatus, err := gardenerStatusFromCluster(r.Target)
	if err != nil {
		return err
	}
	if gardenerStatus.NodeProvisioned != nil && *gardenerStatus.NodeProvisioned {
		// node has been setup, skip
		return nil
	}
	countReady := func(nodes []corev1.Node) int {
		count := 0
		for _, node := range nodes {
			for _, item := range node.Status.Conditions {
				if item.Type == corev1.NodeReady && item.Status == corev1.ConditionTrue {
					count++
				}
			}
		}
		return count
	}
	c := r.Target.Spec.Cluster
	componentExpectedCount := func(comp cloudv1.ClusterComponent) (int, error) {
		expected := comp.GetReplicas()
		if comp.GetComponentType() != cloudv1.ComponentPD {
			return expected, nil
		}
		zeroPDNodes, _, err := mixeddeploypd.ShouldUseZeroPDNodes(
			expected,
			c.TiKV.GetReplicas(),
			c.TiDB.GetReplicas(),
			c.TiFlash.GetReplicas(),
			"node progress",
			r.tc,
			r.Provider,
			r.GetComponentNodeLabels,
			r.shootClient,
			r.Target,
			r.PDMixedDeployConfig)
		if err != nil {
			return expected, err
		}
		if zeroPDNodes {
			expected = 0
		}
		return expected, nil
	}
	componentNodeProvisioned := func(comp cloudv1.ClusterComponent) error {
		nodes, err := k8s.SelectNodes(r.ctx, r.shootClient, r.GetComponentNodeLabels(comp.GetComponentType()))
		if err != nil {
			return err
		}
		readyCount := countReady(nodes)
		expectedCount, err := componentExpectedCount(comp)
		if err != nil {
			return err
		}
		if readyCount < expectedCount {
			return waitNodeReady
		}
		return nil
	}
	for _, comp := range []cloudv1.ClusterComponent{&c.PD, &c.TiDB, &c.TiKV, &c.TiFlash} {
		if err := componentNodeProvisioned(comp); err != nil {
			return err
		}
	}
	return r.ensureStatus(func() error {
		gardenerStatus.NodeProvisioned = pointer.Bool(true)
		return setGardenerStatusToCluster(r.Target, *gardenerStatus)
	})
}

func (r *reconcileContext) checkNodeNotProvisionedReason() error {
	gardenerStatus, err := gardenerStatusFromCluster(r.Target)
	if err != nil {
		return err
	}
	// check if there is any shoot error blocking node provision
	shoot, err := r.Gardener.GetShoot(r.ctx, r.networkGardenStatus.ShootName)
	if err != nil {
		return errors.AddStack(err)
	}
	if err := r.ensureStatus(func() error {
		var errs []cloudv1.ShootLastError
		for _, lastErr := range shoot.Status.LastErrors {
			var codes []string
			for _, code := range lastErr.Codes {
				codes = append(codes, string(code))
			}
			errs = append(errs, cloudv1.ShootLastError{
				Description:    lastErr.Description,
				TaskID:         lastErr.TaskID,
				Codes:          codes,
				LastUpdateTime: lastErr.LastUpdateTime,
			})
		}
		gardenerStatus.ShootLastErrors = errs
		return setGardenerStatusToCluster(r.Target, *gardenerStatus)
	}); err != nil {
		return errors.AddStack(err)
	}
	return nil
}

func (r *reconcileContext) createSubresourceOfTidbCluster(obj client.Object) error {
	if err := controllerutil.SetOwnerReference(r.tc, obj, r.Scheme); err != nil {
		return err
	}
	_, err := k8s.CreateIgnoreExists(r.ctx, r.shootClient, obj)
	return err
}