package cluster

import (
	"context"
	"fmt"
	"time"

	"github.com/Masterminds/semver"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/core-portal/internal/central/config"
	"github.com/tidbcloud/core-portal/internal/central/db"
	dbutil "github.com/tidbcloud/core-portal/internal/central/db/dbutil"
	"github.com/tidbcloud/core-portal/internal/central/infra-api"
	"github.com/tidbcloud/core-portal/internal/central/metrics"
	"github.com/tidbcloud/core-portal/internal/central/model/cluster/clusterproxy/tidbcloud"
	modelutil "github.com/tidbcloud/core-portal/internal/central/model/util"
	entity "github.com/tidbcloud/core-svc/model"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	"github.com/tidbcloud/pkgv2/tracing"
	"go.opentelemetry.io/otel/attribute"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

var (
	topSQLVersionConstraint, _   = semver.NewConstraint(">= v5.2.2")
	auditLogVersionConstraint, _ = semver.NewConstraint(">= v5.4.0")
	enableTopSQLStatements       = []string{
		"set GLOBAL tidb_enable_top_sql=1;",
		// TODO(Huaxi): remove the below two statements after TiKV fix the bug.
		"set config tikv `resource-metering.enabled`=false;",
		"set config tikv `resource-metering.enabled`=true;",
	}
)

func (m Model) Run(ctx context.Context) error {
	var clusters []entity.Cluster
	logger := m.log.With(log.FieldsFromContext(ctx)...).Sugar()

	err := db.Transaction(func(tx *gorm.DB) error {
		var err error
		clusters, err = dbutil.FindClusters(tx, dbutil.ClusterFilter{
			Statuss: []entity.ClusterStatus{
				entity.ClusterStatusLaunching,
			},
		})
		return err
	})
	if err != nil {
		return err
	}

	startTime := time.Now()
	logger.Infof("[global]sync cluster with infra, start. cnt=%d", len(clusters))
	var errs error
	for _, cluster := range clusters {
		c, err := GetBgContext(m, cluster.TenantID)
		if err != nil {
			logger.Errorf("failed to get bg context, clusterID=%d, err=%s", cluster.ID, err)
			errs = multierr.Append(errs, errors.Annotate(err, "get bg context"))
			continue
		}

		oldStatus := cluster.Status
		err = c.Sync(ctx, &cluster)
		if err != nil {
			errs = multierr.Append(errs, errors.Annotatef(err, "clusterID: %d", cluster.ID.Uint64()))
		}

		if cluster.Status != oldStatus {
			logger.Info("cluster status change",
				zap.Uint64("id", cluster.ID.Uint64()),
				zap.String("name", cluster.Name),
				zap.String("old", string(oldStatus)),
				zap.String("new", string(cluster.Status)),
			)
		}
	}

	logger.Infof("[global]sync cluster with infra, done. cost=%s,errs=%s", time.Since(startTime), errs)
	return errs
}

// Sync do all the work need according the current status in metadb.
// the flowing entity of cluster should be preload
// cluster.Provider
// cluster.Components
// cluster.Connections
//
// status may change:
// model.ClusterStatusInitializing: model.ClusterStatusPreparing
// model.ClusterStatusPreparing: model.ClusterStatusLaunching
// model.ClusterStatusLaunching: model.ClusterStatusNormal
// model.ClusterStatusScaling: model.ClusterStatusNormal
// model.ClusterStatusDeleted: model.ClusterStatusDeletecomplete
// also sync need to sync the read only status of the cluster
func (c BgContext) Sync(ctx context.Context, cluster *entity.Cluster) (err error) {
	ctx, span := Tracer.Start(ctx, "BgContext.Sync")
	span.SetAttributes(attribute.String("cluster_id", fmt.Sprintf("%v", cluster.ID)), attribute.String("cluster_name", cluster.Name))
	defer func() { tracing.EndSpan(span, err) }()
	logger := c.Log().With(zap.Uint64("id", cluster.ID.Uint64()), zap.String("name", cluster.Name)).With(log.FieldsFromContext(ctx)...)
	logger.Info("sync cluster")

	var cr *cloudv1.Cluster
	if cluster.Status == entity.ClusterStatusLaunching ||
		cluster.Status == entity.ClusterStatusPreparing {
		cr, err = c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
		if err != nil {
			return errors.AddStack(err)
		}
	}

	switch cluster.Status {
	case entity.ClusterStatusLaunching:
		return c.handleLaunching(ctx, cluster, cr)
	default:
		logger.Sugar().Errorf("invalid status, status=%s", cluster.Status)
		return nil
	}
}

func (c BgContext) changeClusterPassword(ctx context.Context, cluster entity.Cluster, adminPwd encrypt.Secret, newPwd string) error {
	cr, err := c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		return err
	}

	proxy, err := tidbcloud.BuildClusterProxy(cluster.ID.Uint64(), cr.Status, config.DefaultClusterProxyTimeout)
	if err != nil {
		return err
	}

	err = proxy.ChangePassword(ctx, adminPwd, newPwd)
	if err != nil {
		return err
	}

	return nil
}

// initCluster need to be idempotent.
func (c BgContext) initCluster(ctx context.Context, cluster *entity.Cluster, cr *cloudv1.Cluster) (err error) {
	ctx, span := Tracer.Start(ctx, "BgContext.initCluster")
	defer func() { tracing.EndSpan(span, err) }()
	logger := c.Log().With(log.FieldsFromContext(ctx)...)

	// cluster.EncryptedRootPassword is empty, indicating that the initCluster method has been successfully executed before
	if cluster.EncryptedRootPassword == "" {
		return nil
	}

	proxy, err := tidbcloud.BuildClusterProxy(cluster.ID.Uint64(), cr.Status, config.DefaultClusterProxyTimeout)
	if err != nil {
		return err
	}
	span.AddEvent("build cluster proxy")

	adminPwd := encrypt.Secret("")
	adminPwd, err = modelutil.DecryptCloudAdminPassword(c.encryptor, *cluster, cr.Status)
	if err != nil {
		return errors.AddStack(err)
	}

	if adminPwd != "" {
		err = proxy.CreateCloudAdmin(ctx, adminPwd)
		if err != nil {
			return errors.Annotate(err, "create cloud_admin user")
		}
	}

	rootPassword, err := encrypt.DecryptBase64(c.encryptor, cluster.EncryptedRootPassword)
	if err != nil {
		return errors.Annotate(err, "decrypt encrypted root password")
	}
	err = c.changeClusterPassword(ctx, *cluster, adminPwd, string(rootPassword))
	if err != nil {
		return errors.Annotate(err, "changeClusterPassword")
	}
	logger.Info("root password is set", zap.Uint64("clusterID", cluster.ID.Uint64()))
	span.AddEvent("decrypt password")

	logger.Debug("EnableTopSQL",
		zap.Bool("switch", config.Global().InfraConfig.EnableTopSQL),
		zap.String("TiDB version", cr.Status.DB.Version),
	)

	if config.Global().InfraConfig.EnableTopSQL && topSQLSupported(cr.Status.DB.Version) {
		if err := enableTopSQL(ctx, proxy, adminPwd); err != nil {
			logger.Warn("failed to enable TopSQL", zap.Uint64("clusterID", cluster.ID.Uint64()))
		} else {
			logger.Info("TopSQL is enabled", zap.Uint64("clusterID", cluster.ID.Uint64()))
		}
	}

	cluster.EncryptedRootPassword = ""
	err = db.Transaction(func(tx *gorm.DB) error {
		return cluster.Update(tx, entity.ClusterDBSchema.EncryptedRootPassword)
	})
	return errors.AddStack(err)
}

func (c BgContext) handleLaunching(ctx context.Context, cluster *entity.Cluster, cr *cloudv1.Cluster) (err error) {
	ctx, span := Tracer.Start(ctx, "BgContext.handleLaunching")
	defer func() { tracing.EndSpan(span, err) }()
	span.SetAttributes(attribute.String("cluster_id", fmt.Sprintf("%v", cluster.ID)), attribute.String("cluster_name", cluster.Name))
	logger := c.Log().With(zap.Uint64("id", cluster.ID.Uint64()), zap.String("name", cluster.Name)).With(log.FieldsFromContext(ctx)...)

	// just wait cluster to be created.
	if cr.Status.Phase != cloudv1.ClusterPhaseRunning {
		logger.Info("cluster at infra-api still not become running phase")
		return nil
	}

	// only check public endpoint if cluster is non-xaccount-deploy
	if cluster.CrossAccountDeploymentID == 0 && cr.Status.SqlEndpoints.PublicEndpoint == nil {
		// we enable it in spec, so should not be nil
		err := errors.Errorf("can not get public endpoint from infra api")
		return err
	}

	if cr.Status.Proxy == nil || cr.Status.Proxy.Endpoint == "" {
		logger.Info("proxy endpoint not ready")
		return nil
	}

	var isAuditSupported bool
	// Dev Tier doesn't support audit log
	if cluster.Free {
		isAuditSupported = false
	} else {
		if cr.Status.DB != nil && cr.Status.DB.Version != "" {
			isAuditSupported, err = CheckIfAuditLogSupported(ctx, cr)
			if err != nil {
				return errors.Annotate(err, "check if audit log supported")
			}
			logger.Info("check if audit log supported", zap.Bool("versionIsSupport", isAuditSupported))
		} else {
			logger.Info("cluster has no db version")
			return nil
		}
	}

	if isAuditSupported {
		audit := cloudv1.Audit{
			Enable:    true,
			BucketURL: "",
			Region:    "",
			RoleArn:   "",
		}
		cr.Spec.Cluster.TiDB.Audit = &audit
	}

	err = c.initCluster(ctx, cluster, cr)
	if err != nil {
		return errors.AddStack(err)
	}

	infraapi.AddBillingLabelWhenClusterNormal(cr)
	err = errors.AddStack(c.api.UpdateCluster(ctx, cr))
	if err != nil {
		return err
	}

	useSem, err := modelutil.UseSemForCluster(cr.Status)
	if err != nil {
		logger.Error("failed to check if sem is used", zap.Error(err))
		return err
	}
	if useSem && isAuditSupported {
		proxy, err := tidbcloud.BuildClusterProxy(cluster.ID.Uint64(), cr.Status, config.DefaultClusterProxyTimeout)
		if err != nil {
			return err
		}

		adminPwd, err := modelutil.DecryptCloudAdminPassword(c.encryptor, *cluster, cr.Status)
		if err != nil {
			return errors.AddStack(err)
		}

		if err = proxy.ExecuteSQL(ctx, adminPwd, config.DisableAuditPluginClause); err != nil {
			return errors.Annotate(err, "failed to disable audit plugin")
		}
	}

	status := entity.ClusterStatusNormal
	err = errors.AddStack(db.Transaction(func(tx *gorm.DB) error { // check whether need to restore the cluster
		restore, errGet := dbutil.GetRestoreByClusterID(tx, cluster.TenantID, cluster.ID)
		if errGet != nil && errors.Cause(errGet) != gorm.ErrRecordNotFound {
			return errGet
		} else if errGet == nil && restore.Status != entity.RestoreStatus_SUCCESS {
			status = entity.ClusterStatusRecovering // restore background worker will update the cluster to NORMAL after restored
		}
		return nil
	}))
	if err != nil {
		return err
	}

	err = transStatus(ctx, cluster, status)
	if err != nil {
		return err
	}

	if cluster.Free && status == entity.ClusterStatusNormal {
		err = errors.Annotate(db.Transaction(func(tx *gorm.DB) error {
			return dbutil.DeleteOldDevTierClustersBackups(ctx, tx, c.tenantAndUser.TenantID, cluster.ID)
		}), "delete old cluster backups")
		if err != nil {
			return err
		}
	}

	return nil
}

// enableTopSQL executes some SQL statements to enable the feature when creating a new TiDB cluster.
func enableTopSQL(ctx context.Context, proxy clusterproxy.ClusterProxy, adminPwd encrypt.Secret) error {
	for _, s := range enableTopSQLStatements {
		if err := proxy.ExecuteSQL(ctx, adminPwd, s); err != nil {
			return fmt.Errorf("failed to execute the `%s` to enable TopSQL: %s", s, err)
		}
	}

	return nil
}

// Service like delete cluster will change the cluster status as deleted directly.
// so we only update the status if the status is not changed.
func transStatus(ctx context.Context, cluster *entity.Cluster, status entity.ClusterStatus) (err error) {
	_, span := Tracer.Start(ctx, "transStatus")
	defer func() { tracing.EndSpan(span, err) }()
	oldStatus := cluster.Status
	cluster.Status = status
	span.SetAttributes(attribute.String("cluster_id", fmt.Sprintf("%v", cluster.ID)),
		attribute.String("cluster_old_status", string(oldStatus)),
		attribute.String("cluster_status", string(status)))
	err = db.Transaction(func(tx *gorm.DB) error {
		if status == entity.ClusterStatusNormal {
			now := time.Now()
			metrics.CreateClusterDuration.WithLabelValues(metrics.GetProviderType(cluster.Free)).Observe(time.Since(cluster.CreatedAt).Seconds())
			if errUpdate := dbutil.ClusterUpdateCreationFinishedTime(tx, cluster.ID, &now); errUpdate != nil {
				return errUpdate
			}
		}
		return entity.NewClusterQuerySet(tx).IDEq(cluster.ID).StatusEq(oldStatus).GetUpdater().SetStatus(status).Update()
	})
	if err != nil {
		return errors.AddStack(err)
	}

	return nil
}

func setNetworkThresholdAnnotations(cr *cloudv1.Cluster, ingressThrottled, egressThrottled bool) {
	if cr.Annotations == nil {
		cr.Annotations = map[string]string{}
	}

	if ingressThrottled {
		cr.Annotations[ingressTrafficAnnKey] = config.ThrottledDevTierBandwidth
	} else {
		delete(cr.Annotations, ingressTrafficAnnKey)
	}
	if egressThrottled {
		cr.Annotations[egressTrafficAnnKey] = config.ThrottledDevTierBandwidth
	} else {
		delete(cr.Annotations, egressTrafficAnnKey)
	}
}

// topSQLSupported checks if the given version meets the constraints of TopSQL.
func topSQLSupported(version string) bool {
	v, err := semver.NewVersion(version)
	if err != nil {
		return false
	}

	return topSQLVersionConstraint.Check(v)
}

func CheckIfAuditLogSupported(ctx context.Context, cr *cloudv1.Cluster) (bool, error) {
	if cr.Status.DB == nil || cr.Status.DB.Version == "" {
		return false, errors.Errorf("cluster %s has no db version", cr.Name)
	}

	version, err := semver.NewVersion(cr.Status.DB.Version)
	if err != nil {
		return false, err
	}

	return auditLogVersionConstraint.Check(version), nil
}

func SyncAuditLogStatusToInfra(ctx context.Context, api *infraapi.API, projectID, clusterID uint64, bucketURl, region, roleARN string) error {
	cr, err := api.GetCluster(ctx, projectID, clusterID)
	if err != nil {
		return err
	}

	audit := cloudv1.Audit{
		Enable:    true,
		BucketURL: bucketURl,
		Region:    region,
		RoleArn:   roleARN,
	}
	cr.Spec.Cluster.TiDB.Audit = &audit

	return api.UpdateCluster(ctx, cr)
}

func UpdateAuditLogStatus(ctx context.Context, api *infraapi.API, encryptor encrypt.Encryptor, clusterID, projectID entity.ID, isAuditEnabled bool) error {
	return db.Transaction(func(tx *gorm.DB) error {
		cluster, err := dbutil.GetCluster(tx, clusterID)
		if err != nil {
			return err
		}

		redactLogStatus := config.AuditRedactLogStatusON // ON by default
		if isAuditEnabled {
			redactLogStatus = config.AuditRedactLogStatusOFF
		}

		var clauses []string
		if isAuditEnabled {
			clauses = append(clauses, config.EnableAuditPluginClause)
		} else {
			clauses = append(clauses, config.DisableAuditPluginClause)
		}

		clauses = append(clauses, fmt.Sprintf(config.SetAuditLogRedactStatusClause, redactLogStatus), config.SetAuditLogMaxSize, config.FlushAuditLogAccessClause)

		return modelutil.ExecuteMultiSqlCommon(ctx, &cluster, clauses, api, encryptor, tidbcloud.BuildClusterProxy, config.AuditLogClusterproxyTimeout)
	})
}