package cluster

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/Masterminds/semver"
	"github.com/pingcap/dbaas/central/config"
	"github.com/pingcap/dbaas/central/db"
	"github.com/pingcap/dbaas/central/db/dbutil"
	"github.com/pingcap/dbaas/central/db/entity"
	infraapi "github.com/pingcap/dbaas/central/infra-api"
	"github.com/pingcap/dbaas/central/metrics"
	"github.com/pingcap/dbaas/central/model/cluster/clusterproxy/tidbcloud"
	"github.com/pingcap/dbaas/central/model/util"
	modelutil "github.com/pingcap/dbaas/central/model/util"
	"github.com/pingcap/errors"
	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"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/utils/pointer"
)

// StorageSize
const (
	MB = 1024 * 1024
	GB = 1024 * 1024 * 1024
)

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;",
	}
)

var (
	ErrClusterFinalBackupRunning = errors.New("final backup of cluster is running")
)

func (c BgContext) Run(ctx context.Context) error {
	var clusters []entity.Cluster
	logger := c.Log().With(log.FieldsFromContext(ctx)...)

	err := db.Transaction(func(tx *gorm.DB) error {
		var err error
		clusters, err = dbutil.FindClusters(tx, dbutil.ClusterFilter{
			TenantID: &c.tenantAndUser.TenantID,
			Statuss: []entity.ClusterStatus{
				entity.ClusterStatusInitializing,
				entity.ClusterStatusPreparing,
				entity.ClusterStatusLaunching,
				entity.ClusterStatusNormal,
				entity.ClusterStatusScaling,
				entity.ClusterStatusDeleted,
			},
		})
		return err
	})
	if err != nil {
		return err
	}

	var errs error
	for _, cluster := range clusters {
		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)),
			)
		}
	}

	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:
// entity.ClusterStatusInitializing: entity.ClusterStatusPreparing
// entity.ClusterStatusPreparing: entity.ClusterStatusLaunching
// entity.ClusterStatusLaunching: entity.ClusterStatusNormal
// entity.ClusterStatusScaling: entity.ClusterStatusNormal
// entity.ClusterStatusDeleted: entity.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.ClusterStatusInitializing:
		// already migrated into core
		return nil
	case entity.ClusterStatusPreparing:
		// already migrated into core
		return nil
	case entity.ClusterStatusLaunching:
		return c.handleLaunching(ctx, cluster, cr)
	case entity.ClusterStatusNormal:
		// already migrated into core
		return nil
	case entity.ClusterStatusScaling:
		// already migrated into core
		return nil
	case entity.ClusterStatusDeleted:
		return c.handleDeleted(ctx, cluster)
	case entity.ClusterStatusDeletecomplete:
		return nil
	default:
		return nil
	}
}

func (c BgContext) ensureTiCDCDeleted(ctx context.Context, cluster *entity.Cluster) error {
	var (
		err error
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		return entity.NewCDCTaskQuerySet(tx).ClusterIDEq(cluster.ID).StatusNe(entity.CDCStatusDeleted).GetUpdater().SetStatus(entity.CDCStatusDeleting).Update()
	})
	if err != nil {
		return err
	}

	_, err = c.api.GetTicdc(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil && !apierrors.IsNotFound(err) {
		return err
	} else if err != nil && apierrors.IsNotFound(err) {
		return nil
	}

	return errors.Errorf("ticdc of cluster %d is not deleted", cluster.ID.Uint64())
}

func (c BgContext) createFinalBackup(ctx context.Context, cluster entity.Cluster) error {
	var backupSchedule entity.BackupSchedule
	// Query cluster final backup created by tenant system user
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewBackupScheduleQuerySet(tx).ClusterIDEq(cluster.ID).NameEq(config.FinalBackupName).UserIDEq(c.tenantAndUser.ID).One(&backupSchedule)
	})
	if err != nil {
		if nil != errors.Find(err, func(e error) bool { return e == gorm.ErrRecordNotFound }) {
			err = disableClusterEndpoint(context.TODO(), c.api, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
			if err != nil {
				return errors.Annotatef(err, "failed to disable connection of cluster %d", cluster.ID)
			}

			err = db.Transaction(func(tx *gorm.DB) error {
				var err error
				now := time.Now()
				newBackupSchedule := &entity.BackupSchedule{
					Name:           config.FinalBackupName,
					Type:           entity.BackupType_Manual,
					ClusterID:      cluster.ID,
					UserID:         c.tenantAndUser.ID,
					Method:         entity.BackupMethod_Full,
					NextBackupTime: &now,
				}
				err = newBackupSchedule.Create(tx)
				if err != nil {
					return errors.Annotate(err, "faild to create backup schedule")
				}
				backupSchedule.ID = newBackupSchedule.ID
				err = dbutil.CreateBackup(ctx, tx, *newBackupSchedule, cluster)
				if err != nil {
					return errors.Annotate(err, "faild to create backup")
				}
				return dbutil.AddAccessLogNow(tx, c.tenantAndUser.User.TenantID, c.tenantAndUser.Email, fmt.Sprintf("create backup schedule for cluster: %d, name: %s, schedule: %s", cluster.ID, newBackupSchedule.Name, newBackupSchedule.Schedule))
			})
			if err != nil {
				c.Log().Error("failed to create final backup for cluster", zap.Uint64("clusterID", cluster.ID.Uint64()), zap.Error(err))
				return errors.Annotatef(err, "failed to create final backup for cluster %d", cluster.ID)
			}
			c.Log().Info("create final backup for cluster", zap.Uint64("clusterID", cluster.ID.Uint64()))
		} else {
			return errors.Annotatef(err, "failed to query backup schedule of cluster %d", cluster.ID)
		}
	}

	var backup entity.Backup
	err = db.Transaction(func(tx *gorm.DB) error {
		return entity.NewBackupQuerySet(tx).BackupScheduleIDEq(backupSchedule.ID).One(&backup)
	})
	if err != nil {
		return errors.Annotatef(err, "failed to get final backup by backup_schedule_id %d", backupSchedule.ID)
	}
	switch backup.Status {
	case entity.BackupStatus_SUCCESS:
		return nil
	case entity.BackupStatus_FAILED:
		c.Log().Error("failed to do final backup for cluster", zap.String("clusterName", cluster.Name), zap.Uint64("clusterID", uint64(cluster.ID)))
		return errors.Annotatef(err, "failed to do final backup for cluster %d", cluster.ID)
	default:
		return ErrClusterFinalBackupRunning
	}
}

// 1. if cluster.FinalBackupBeforeDelete create final backup and make sure it success
// 2. ensure ticdc resource is deleted in infra-api
// 3. delete cluster resource in infra-api
// 4. trans status to entity.ClusterStatusDeletecomplete(it's the backup model's resposibility to trans to entity.ClusterStatusCleared
func (c BgContext) handleDeleted(ctx context.Context, cluster *entity.Cluster) (err error) {
	ctx, span := Tracer.Start(ctx, "BgContext.handleDeleted")
	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(log.FieldsFromContext(ctx)...)
	// If the final backup of the poc cluster fails, the cluster will not be deleted.
	if cluster.FinalBackupBeforeDelete {
		err := c.createFinalBackup(ctx, *cluster)
		if err != nil {
			if err == ErrClusterFinalBackupRunning {
				// just let it re-check next time
				return nil
			}
			return errors.AddStack(err)
		}
	}

	logger.Info("tearing down cluster", zap.Uint64("clusterID", cluster.ID.Uint64()))

	if err := c.ensureTiCDCDeleted(ctx, cluster); err != nil {
		return err
	}

	err = c.api.DeleteClusterIfExist(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		return err
	}

	return transStatus(ctx, cluster, entity.ClusterStatusDeletecomplete)
}

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: "s3://",
			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
}

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

	updated := false

	if cr.Spec.SqlEndpoints.Public.IsEnabled() {
		cr.Spec.SqlEndpoints.Public.Enabled = pointer.BoolPtr(false)
		updated = true
	}

	if cr.Spec.SqlEndpoints.Private != nil && cr.Spec.SqlEndpoints.Private.IsEnabled() {
		cr.Spec.SqlEndpoints.Private.Enabled = pointer.BoolPtr(false)
		updated = true
	}

	if !updated {
		return nil
	}

	return api.UpdateCluster(ctx, cr)
}

// 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)
	}

	// TODO(chenfei): remove this check after we support audit log in eks-provider
	if strings.Contains(cr.Spec.NetworkName, "eks") {
		return false, nil
	}

	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 util.ExecuteMultiSqlCommon(ctx, &cluster, clauses, api, encryptor, tidbcloud.BuildClusterProxy, config.AuditLogClusterproxyTimeout)
	})
}