// Package cluster encapsulates all cluster functionality
package cluster

import (
	"context"
	"crypto/tls"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/core-portal/internal/central/config"
	"github.com/tidbcloud/core-portal/internal/central/db"
	"github.com/tidbcloud/core-portal/internal/central/db/dbutil"
	clusterDao "github.com/tidbcloud/core-portal/internal/central/db/dbutil/cluster"
	replicationDao "github.com/tidbcloud/core-portal/internal/central/db/dbutil/replication"
	"github.com/tidbcloud/core-portal/internal/central/db/tenant"
	"github.com/tidbcloud/core-portal/internal/central/grpc/grpcli"
	"github.com/tidbcloud/core-portal/internal/central/infra-api"
	"github.com/tidbcloud/core-portal/internal/central/model"
	"github.com/tidbcloud/core-portal/internal/central/model/cluster/clusterproxy/tidbcloud"
	"github.com/tidbcloud/core-portal/internal/central/model/util"
	"github.com/tidbcloud/core-portal/internal/central/pkg/tidb/component"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/base"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/central"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/dataflow"
	entity "github.com/tidbcloud/core-svc/model"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/log"
	mailv2 "github.com/tidbcloud/pkgv2/mail/v2"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"
	"gorm.io/gorm"
	"k8s.io/utils/pointer"
)

// https://cloud.google.com/compute/docs/troubleshooting/troubleshooting-vm-creation#resource_availability
const (
	ErrorInfraGcpZoneResourcePoolExhausted = "ZONE_RESOURCE_POOL_EXHAUSTED_WITH_DETAILS"
	ErrorInfraResourcesDepleted            = "ERR_INFRA_RESOURCES_DEPLETED"

	// get super read_only flag
	GetSuperReadonlyClause = "show global variables where variable_name = 'tidb_super_read_only';"
)

var Tracer = otel.Tracer("central/model/cluster")

// Model is created with NewModel
type Model struct {
	log            *zap.Logger
	encryptor      encrypt.Encryptor
	mailFacade     mailv2.Facade
	AESCrypt       *encrypt.AESCrypt
	ProxyTlsConfig *tls.Config
	api            *infraapi.API
	dao            clusterDao.ClusterDaoFacade
	rpDao          replicationDao.ReplicationDaoFacade
}

// NewModel returns a Model
func NewModel(log *zap.Logger,
	infraAPI *infraapi.API,
	encryptor encrypt.Encryptor,
	mailFacade mailv2.Facade,
	aesCrypt *encrypt.AESCrypt,
	proxyTlsConfig *tls.Config,
	dao clusterDao.ClusterDaoFacade,
	rpDao replicationDao.ReplicationDaoFacade,
) Model {
	return Model{
		log:            log,
		encryptor:      encryptor,
		mailFacade:     mailFacade,
		AESCrypt:       aesCrypt,
		ProxyTlsConfig: proxyTlsConfig,
		api:            infraAPI,
		dao:            dao,
		rpDao:          rpDao,
	}
}

type BaseContext interface {
	model() Model
}

// APIContext attaches a APIKey to a Context.
// This helps audit the APIKey
type APIContext struct {
	Model
	apiKeyEntity *entity.APIKey
	log          *zap.Logger
}

func (c APIContext) Log() *zap.Logger {
	return c.log
}

func (c APIContext) TenantID() uint64 {
	return c.apiKeyEntity.TenantID.Uint64()
}

func (c APIContext) model() Model {
	return c.Model
}

func (m Model) InfraAPI() *infraapi.API {
	return m.api
}

// API returns a APIContext
func (m Model) API(apiKeyEntity *entity.APIKey) APIContext {
	return APIContext{
		m,
		apiKeyEntity,
		m.log.With(
			zap.Uint64("tenantID", apiKeyEntity.TenantID.Uint64()),
			zap.String("apiKeyName", apiKeyEntity.Name),
			zap.String("module", "cluster")),
	}
}

// UserContext attaches a User to a Context.
// This helps audit the User
type UserContext struct {
	Model
	tenantAndUser *tenant.User
	log           *zap.Logger
}

func (c UserContext) TenantID() uint64 {
	return c.tenantAndUser.TenantID.Uint64()
}

func (c UserContext) UserID() uint64 {
	return c.tenantAndUser.User.ID.Uint64()
}

func (m Model) Encryptor() encrypt.Encryptor {
	return m.encryptor
}

// User returns a UserContext
func (m Model) User(ctx context.Context, user *tenant.User) UserContext {
	return UserContext{
		Model:         m,
		tenantAndUser: user,
		log: m.log.With(zap.Uint64("tenantID", user.TenantID.Uint64()),
			zap.Uint64("userID", uint64(user.ID)), zap.String("module", "cluster")).With(log.FieldsFromContext(ctx)...),
	}
}

func (m Model) TenantLog(tenantID entity.ID) *zap.Logger {
	return m.log.With(zap.Uint64("tenantID", uint64(tenantID)), zap.String("module", "cluster"))
}

func (c UserContext) Log() *zap.Logger {
	return c.log
}

func (c UserContext) CheckOrgID(orgID entity.ID) error {
	if c.tenantAndUser.IsImpersonated {
		return nil
	}
	userTenantID := c.tenantAndUser.TenantID
	if orgID != userTenantID {
		return errcode.NewForbiddenErr(errors.New("org_id parameter is inconsistent with the user's org ID"))
	}
	return nil
}

func (c UserContext) CheckOrgIDAndProjectID(orgID, projectID entity.ID) error {
	user := c.tenantAndUser
	if user.IsImpersonated {
		return nil
	}

	if orgID != user.TenantID {
		return errcode.NewForbiddenErr(errors.New("org_id parameter is inconsistent with the user's org ID"))
	}

	return dbutil.CheckProjectID(user.Projects, projectID)
}

func (c UserContext) CheckOrgIDAndProjectIDAndClusterID(ctx context.Context, orgID, projectID, clusterID uint64) error {
	if err := c.CheckOrgIDAndProjectID(entity.ID(orgID), entity.ID(projectID)); err != nil {
		return err
	}
	_, err := grpcli.GetCluster(ctx, orgID, projectID, clusterID)
	return err
}

func (c UserContext) model() Model {
	return c.Model
}

// EndpointStatus indicates if an endpoinDeleteJobt is ready
type EndpointStatus struct {
	Host  string
	IP    string
	Port  int32
	Token string
	Ready bool
}

// Cluster is the model for a TiDB cluster.
type Cluster struct {
	ID              uint64
	Name            string
	Status          central.Cluster_Status
	SecondaryStatus central.Cluster_SecondaryStatus
	Progress        central.Cluster_CreateProgress
	Version         string
	Provider        entity.Cloud
	Region          string
	RegionName      string
	CreatedAt       time.Time
	CreatedBy       *entity.User
	Endpoints       Endpoints
	VPCCidrBlock    string
	Resources       []component.Resources
	Restored        bool
	// user will grant bucket permission to this RoleARN
	// and we will use this role to access user's bucket.
	RoleARN          string
	ImportID         *uint64
	LastErrors       []*central.LastError
	Replications     []*central.Replication
	ClusterType      central.Cluster_ClusterType
	WriteStatus      central.WriteStatus
	IsDevTier        bool
	XAccountDeployId uint64
}

// DeletedCluster is the model for a deleted TiDB cluster.
type DeletedCluster struct {
	ID               uint64
	Name             string
	Provider         string
	Region           string
	RegionName       string
	XAccountDeployId uint64
	DeleteTime       time.Time
	HasBackups       bool
	LatestBackupSize uint64
}

type TenantName int

// Get retrieves the cluster
func (c UserContext) Get(ctx context.Context, orgID, projectID, clusterID entity.ID) (*Cluster, error) {
	rdb := db.NewProvider().ReadDB(ctx)
	cluster, err := grpcli.GetCluster(ctx, orgID.Uint64(), projectID.Uint64(), clusterID.Uint64())
	if err != nil {
		return nil, err
	}

	restored := false
	restores, bizerr := dbutil.RestoresInProgress(rdb, clusterID)
	if bizerr != nil {
		return nil, bizerr
	}
	restored = len(restores) != 0

	cl := c.convertClusterV2(ctx, cluster, restored)
	return &cl, nil
}

// List all deleted clusters
func (c UserContext) ListDeletedClusters(ctx context.Context, provider entity.Cloud, projectID entity.ID) ([]*DeletedCluster, error) {
	var (
		clusters []entity.Cluster
		err      error
	)
	items := []*DeletedCluster{}

	err = db.Transaction(func(tx *gorm.DB) error {
		if string(provider) != "" {
			clusters, err = dbutil.ListDeletedClustersByProjectAndProvider(tx, provider, c.tenantAndUser.TenantID, projectID)
		} else {
			clusters, err = dbutil.ListDeletedClustersByProject(tx, c.tenantAndUser.TenantID, projectID)
		}
		return err
	})

	for _, cluster := range clusters {
		var region *entity.Region
		err = db.Transaction(func(tx *gorm.DB) error {
			region, err = dbutil.GetClusterRegion(ctx, tx, cluster)
			return err
		})
		if err != nil {
			return items, err
		}
		deletedCluster := c.convertDeletedCluster(cluster, region)
		if deletedCluster.HasBackups {
			items = append(items, deletedCluster)
		}
	}
	return items, nil
}

func (c UserContext) convertDeletedCluster(cluster entity.Cluster, region *entity.Region) *DeletedCluster {
	var (
		err     error
		backups []entity.Backup
	)
	deletedCluster := &DeletedCluster{
		ID:               cluster.ID.Uint64(),
		Name:             cluster.Name,
		Provider:         string(cluster.Provider.Name),
		Region:           region.Name,
		RegionName:       region.DisplayName,
		XAccountDeployId: cluster.CrossAccountDeploymentID.Uint64(),
		DeleteTime:       *cluster.DeletedTime,
		HasBackups:       false, // change later
		LatestBackupSize: 0,     // fill later
	}

	err = db.Transaction(func(tx *gorm.DB) error {
		backups, err = dbutil.ClusterBackups(tx, cluster.ID)
		return err
	})
	if err != nil || len(backups) == 0 {
		if err != nil {
			c.Log().Error("failed to list deleted backups", zap.Uint64("cluster id", cluster.ID.Uint64()), zap.Error(err))
		}
		deletedCluster.HasBackups = false
		return deletedCluster
	}
	deletedCluster.HasBackups = true

	dbBackupMap := make(map[string]bool, len(backups))
	for _, item := range backups {
		dbBackupMap[util.BackupCRDName(&item)] = true
	}

	backupCRs, err := c.api.ListBackups(context.TODO(), cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		c.Log().Error("failed to list backups", zap.Uint64("clusterID", cluster.ID.Uint64()), zap.Error(err))
	} else {
		var size uint64
		for _, item := range backupCRs {
			if dbBackupMap[item.Name] && item.Status.BackupSize > 0 {
				size = uint64(item.Status.BackupSize) // FIXME: should compare `BackupTS` or other field to get the real LATEST backup?
			}
		}
		deletedCluster.LatestBackupSize = size
	}

	return deletedCluster
}

func (c UserContext) ListV2(ctx context.Context, projectID entity.ID, provider entity.Cloud) ([]Cluster, int32, error) {
	filter := &core.ClusterFilter{
		TenantIds:  []uint64{c.tenantAndUser.TenantID.Uint64()},
		ProjectIds: []uint64{projectID.Uint64()},
		Deleted:    pointer.BoolPtr(false),
	}
	sorters := []*base.Sorter{
		{OrderBy: "free", Sort: base.Sort_SORT_ASC}, // put dev tier cluster to the tail of list
		{OrderBy: "created_at", Sort: base.Sort_SORT_DESC},
	}
	if provider.String() != "" {
		filter.CloudProviderNames = []string{provider.String()}
	}

	listReq := &core.ListClustersReq{
		Filter: filter,
		// No pager
		Pager:   nil,
		Sorters: sorters,
		BaseReq: nil,
	}
	listResp, err := grpcli.ListClusters(ctx, listReq)
	if err != nil {
		return nil, 0, err
	}
	countResp, err := grpcli.CountClusters(ctx, &core.CountClustersReq{Filter: filter})
	if err != nil {
		return nil, 0, err
	}

	rdb := db.NewProvider().ReadDB(ctx)
	out := make([]Cluster, 0, len(listResp.Clusters))
	for _, clusterId := range listResp.OrderedClusterIds {
		restored := false
		restores, err := dbutil.RestoresInProgress(rdb, entity.ID(clusterId))
		if err != nil {
			return nil, 0, err
		}
		restored = len(restores) != 0
		out = append(out, c.convertClusterV2(ctx, listResp.Clusters[clusterId], restored))
	}
	return out, int32(countResp.Count), nil
}

func assembleReplicationList(ctx context.Context, clusterID entity.ID) ([]*central.Replication, central.Cluster_ClusterType, error) {
	var (
		err            error
		tmpReplication *central.Replication
		rList          []entity.Replication
		retList        []*central.Replication
		clusterType    central.Cluster_ClusterType
	)
	clusterType = central.Cluster_COMMON
	err = db.Transaction(func(tx *gorm.DB) error {
		err = entity.NewReplicationQuerySet(tx).
			PrimaryClusterIDEq(clusterID).
			PreloadPrimaryCluster().
			PreloadSecondaryCluster().
			PreloadDetachUser().
			PreloadSecondaryUser().
			All(&rList)
		if err != nil {
			return err
		}
		for _, r := range rList {
			if r.ReplicationStatus != entity.ReplicationStatusDetached {
				clusterType = central.Cluster_PRIMARY
			}
			tmpReplication, err = assembleReplication(ctx, clusterID, r)
			if err != nil {
				return err
			}
			retList = append(retList, tmpReplication)
		}
		err = entity.NewReplicationQuerySet(tx).
			SecondaryClusterIDEq(clusterID).
			PreloadPrimaryCluster().
			PreloadSecondaryCluster().
			PreloadDetachUser().
			PreloadSecondaryUser().
			All(&rList)
		if err != nil {
			return err
		}
		for _, r := range rList {
			if r.ReplicationStatus != entity.ReplicationStatusDetached {
				clusterType = central.Cluster_SECONDARY
			}
			tmpReplication, err = assembleReplication(ctx, clusterID, r)
			if err != nil {
				return err
			}
			retList = append(retList, tmpReplication)
		}
		return nil
	})
	return retList, clusterType, err
}

func assembleReplication(ctx context.Context, clusterID entity.ID, r entity.Replication) (*central.Replication, error) {
	var (
		err      error
		ret      *central.Replication
		tracking entity.ReplicationTracking
	)
	ret = &central.Replication{
		PrimaryClusterName:   r.PrimaryCluster.Name,
		PrimaryClusterId:     r.PrimaryCluster.ID.Uint64(),
		Status:               string(r.ReplicationStatus),
		ReplicationId:        r.ID.Uint64(),
		ReplicationLag:       "unknow",
		PrimaryWriteStatus:   central.WriteStatus_WRITABLE,
		SecondaryWriteStatus: central.WriteStatus_WRITABLE,
	}
	if r.PrimaryCluster.ReadOnly&entity.ReadOnlyStatus != 0 {
		ret.PrimaryWriteStatus = central.WriteStatus_READONLY
	}
	if r.SecondaryCluster != nil {
		ret.SecondaryClusterName = r.SecondaryCluster.Name
		ret.SecondaryClusterId = r.SecondaryCluster.ID.Uint64()
		if r.SecondaryCluster.ReadOnly&entity.ReadOnlyStatus != 0 {
			ret.SecondaryWriteStatus = central.WriteStatus_READONLY
		}
	}
	if r.SecondaryUser != nil {
		ret.Creator = r.SecondaryUser.Email
	}
	if r.DetachUser != nil {
		ret.DetachUser = r.DetachUser.Email
	}
	if r.LagTimeWhenDetach != nil {
		ret.ReplicationLag = strconv.Itoa(int(*r.LagTimeWhenDetach))
	}
	err = db.Transaction(func(tx *gorm.DB) error {
		return entity.NewReplicationTrackingQuerySet(tx).
			ReplicationIDEq(r.ID).
			StatusEq(entity.ReplicationStatusDetaching).
			One(&tracking)
	})
	if err != nil {
		if err != gorm.ErrRecordNotFound {
			log.Warn(ctx, "failed to get tracking of replication", zap.Uint64("replication id", r.ID.Uint64()))
		}
	} else {
		ret.DetachedTime = timestamppb.New(tracking.CreatedAt)
	}
	ret.CreationTime = timestamppb.New(r.CreatedAt)
	if r.ReplicationStatus == entity.ReplicationStatusCreating {
		ret.CreateProgress = getReplicationProgress(clusterID, &r)
	}

	return ret, nil
}

func (c UserContext) addLastError(cluster *core.Cluster, model *Cluster) error {
	cr, err := c.api.GetCluster(context.TODO(), cluster.ProjectId, cluster.Id)
	if err != nil {
		return errors.Annotate(err, "GetCluster")
	}

	status, err := infraapi.ConvertClusterStatus(cluster.InfraProviderType, cr)
	if err != nil {
		return errors.Annotate(err, "ConvertClusterStatus")
	}

	lastErrors := status.GardenerStatus.ShootLastErrors

	// Currently, we only check ERR_INFRA_RESOURCES_DEPLETED errors
	// https://github.com/gardener/gardener/blob/v1.19.1/pkg/apis/core/v1beta1/types_common.go#L36
	// GCP: check lastError.Description
	for _, lastError := range lastErrors {
		var infraResourcesDepleted bool
		for _, code := range lastError.Codes {
			if code == ErrorInfraResourcesDepleted {
				infraResourcesDepleted = true
				break
			}
		}
		if !infraResourcesDepleted && strings.Contains(lastError.Description, ErrorInfraGcpZoneResourcePoolExhausted) {
			infraResourcesDepleted = true
		}

		if infraResourcesDepleted {
			infraLastError := central.LastError{
				Codes: []central.InfraErrorCode{
					central.InfraErrorCode_INFRA_RESOURCES_DEPLETED,
				},
			}
			model.LastErrors = append(model.LastErrors, &infraLastError)
			break
		}
	}

	return nil
}

func (c UserContext) convertClusterV2(ctx context.Context, cluster *core.Cluster, restored bool) Cluster {
	rdb := db.NewProvider().ReadDB(ctx)
	logger := log.WithContextS(ctx)
	model := Cluster{
		ID:               cluster.Id,
		Name:             cluster.Name,
		Status:           central.Cluster_Status(cluster.UserStatus),
		SecondaryStatus:  central.Cluster_SecondaryStatus(cluster.UserSubStatus),
		Progress:         central.Cluster_COMPLETED,
		Version:          cluster.Version,
		Provider:         entity.Cloud(cluster.CloudProviderName),
		Region:           cluster.RegionName,
		RegionName:       cluster.RegionDisplayName,
		CreatedAt:        time.Unix(cluster.CreatedAt, 0),
		CreatedBy:        nil,
		Endpoints:        Endpoints{},
		VPCCidrBlock:     "",
		Resources:        util.CoreClusterComponents2componentResources(cluster.Components),
		Restored:         restored,
		RoleARN:          "",
		ImportID:         cluster.ImportId,
		LastErrors:       nil,
		Replications:     nil,
		ClusterType:      0,
		WriteStatus:      0,
		IsDevTier:        cluster.IsDevTier,
		XAccountDeployId: cluster.XaccountDeployId,
	}

	if cluster.CreatedBy != 0 {
		creator, err := dbutil.GetUserByID(rdb, entity.ID(cluster.TenantId), entity.ID(cluster.CreatedBy))
		if err != nil {
			logger.Warnf("failed to get creator of cluster %d: %s", cluster.Id, err)
		}
		model.CreatedBy = &creator
	}

	switch cluster.Status {
	case core.ClusterStatus_CLUSTER_STATUS_INITIALIZING:
		model.Progress = central.Cluster_INITIALIZING
	case core.ClusterStatus_CLUSTER_STATUS_PREPARING:
		model.Progress = central.Cluster_PREPARING
	case core.ClusterStatus_CLUSTER_STATUS_LAUNCHING:
		model.Progress = central.Cluster_LAUNCHING
	case core.ClusterStatus_CLUSTER_STATUS_HIBERNATED:
		model.Progress = central.Cluster_HIBERNATED
	}

	var ready bool
	if cluster.UserStatus != core.UserClusterStatus_USER_CLUSTER_STATUS_CREATING {
		ready = true
		model.Endpoints.PrometheusReady = true
		model.Endpoints.PdReady = true
	}
	if cluster.Status == core.ClusterStatus_CLUSTER_STATUS_HIBERNATED {
		model.Endpoints.PrometheusReady = false
		model.Endpoints.PdReady = false
	}

	for _, each := range cluster.Connections {
		if each.Type == core.ClusterConnectionType_CLUSTER_CONNECTION_TYPE_PUBLIC {
			model.Endpoints.MySQL = EndpointStatus{
				Host:  each.Host,
				Port:  each.Port,
				Token: each.Token,
				Ready: ready,
			}
		} else if each.Type == core.ClusterConnectionType_CLUSTER_CONNECTION_TYPE_PRIVATE {
			model.Endpoints.PrivateMySQL = EndpointStatus{
				Host:  each.Host,
				Port:  each.Port,
				Token: each.Token,
				Ready: ready,
			}
		}
	}

	if cluster.UserStatus == core.UserClusterStatus_USER_CLUSTER_STATUS_CREATING {
		if err := c.addLastError(cluster, &model); err != nil {
			c.Log().Error("fail to add last error to TiDB cluster", zap.Error(err))
		}
	}

	// add replications to cluster entity
	var err error
	model.Replications, model.ClusterType, err = assembleReplicationList(ctx, entity.ID(cluster.Id))
	if err != nil {
		c.Log().Error("failed to attach replication to cluster", zap.Uint64("clusterID", cluster.Id))
		return model
	}

	// for the secondary cluster, the status need be modify via the replication status
	// when the replication status is BACKUPING/RESTOREING/REPLICATION_PREPAREING the cluster status is creating
	if model.Progress == central.Cluster_COMPLETED {
		for _, r := range model.Replications {
			if r.Status == string(entity.ReplicationStatusCreating) {
				model.Progress = r.CreateProgress
				if model.ID == r.SecondaryClusterId {
					model.Status = central.Cluster_Status(core.UserClusterStatus_USER_CLUSTER_STATUS_CREATING)
				}
			}
		}
	}

	network, err := c.api.GetNetwork(context.TODO(), cluster.InfraProviderType, cluster.ProjectId, cluster.RegionName)
	if err != nil {
		c.Log().Error("failed to get network", zap.Error(err))
		return model
	}

	networkStatus, err := infraapi.ConvertNetworkStatus(network)
	if err != nil {
		c.Log().Error("failed to convert network status", zap.Error(err))
		return model
	}

	model.VPCCidrBlock = networkStatus.VPCCidrBlock
	switch model.Provider {
	case entity.Aws:
		model.RoleARN = networkStatus.ImportRoleARN
	case entity.Gcp:
		model.RoleARN = networkStatus.ServiceAccountEmail
	}

	model.WriteStatus = central.WriteStatus_WRITABLE
	if cluster.ReadOnly {
		model.WriteStatus = central.WriteStatus_READONLY
	}

	return model
}

// getReplciationProgress return the cluster progress when in create replication progress
// BACKUPING/RESTOREING/REPLICATION_PREPAREING is only show in secondary cluster create progress
func getReplicationProgress(clusterID entity.ID, r *entity.Replication) central.Cluster_CreateProgress {
	if clusterID != r.SecondaryClusterID {
		return central.Cluster_COMPLETED
	}
	if r.BackupID == 0 {
		return central.Cluster_BACKUPING
	} else if r.CDCTaskID == 0 {
		return central.Cluster_RESTORING
	} else {
		return central.Cluster_REPLICATION_PREPARING
	}
}

// BgContext is meant for the system worker/controller
type BgContext UserContext

var GetBgContext = getBgContext

// getBgContext returns a BgContext
func getBgContext(m Model, tenantID entity.ID) (BgContext, error) {
	var (
		systemUser entity.User
		err        error
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		systemUser, err = dbutil.SystemUser(tx, tenantID)
		return err
	})
	if err != nil {
		return BgContext{}, err
	}
	return BgContext{m, &tenant.User{User: systemUser, TenantID: tenantID}, m.log.With(zap.Uint64("tenantID", tenantID.Uint64()), zap.Uint64("userID", uint64(systemUser.ID)), zap.String("module", "cluster"))}, nil
}

func (c BgContext) Log() *zap.Logger {
	return c.log
}

func (c BgContext) TenantID() uint64 {
	return c.tenantAndUser.TenantID.Uint64()
}

func (c BgContext) model() Model {
	return c.Model
}

// Delete soft deletes a cluster in the database. Teardown will remove the running cluster.
func (c UserContext) Delete(ctx context.Context, orgID, projectID, clusterID entity.ID) bizerr.BizError {
	var replications []entity.Replication
	// just check cluster exist
	err := db.Transaction(func(tx *gorm.DB) error {
		model.LockTenant(tx, orgID)

		cluster, err := dbutil.GetClusterByID(tx, orgID, projectID, clusterID)
		if err != nil {
			if errors.IsNotFound(err) {
				return nil
			}
			return bizerr.New(bizerr.ErrCodeDB, "failed to get cluster by id")
		}
		// the primary and secondary clusters can't be deleted before detached
		replications, err = c.Model.rpDao.FindAliveReplicationsByClusterID(cluster.ID.Uint64())
		if err != nil {
			return bizerr.New(bizerr.ErrCodeDB, "failed to find alive replications by cluster id")
		}

		for _, r := range replications {
			// if this cluster is in create replication
			// DBaaS need to detach it first then delete the cluster
			if cluster.ID == r.PrimaryClusterID {
				return bizerr.New(bizerr.ErrCodeParams, "This Cluster is a Primary Cluster of Cross Region Replication. Detach the Secondary Cluster before deleting the Primary Cluster.")
			} else if r.ReplicationStatus == entity.ReplicationStatusReplicating && cluster.ID == r.SecondaryClusterID {
				return bizerr.New(bizerr.ErrCodeParams, "This Cluster is a Secondary Cluster of Cross Region Replication. Detach it before deleting it")
			}

			_, err = c.DetachReplication(orgID, clusterID, r.ID, dataflow.DetachReplicationReq_None)
			if err != nil {
				return bizerr.New(bizerr.ErrCodeDB, "failed to detach replication")
			}
		}

		if err := dbutil.DeleteCluster(ctx, tx, clusterID); err != nil {
			return bizerr.New(bizerr.ErrCodeDB, "failed to delete cluster")
		}
		err = dbutil.AddAccessLogNow(tx, c.tenantAndUser.User.TenantID, c.tenantAndUser.Email, fmt.Sprintf("delete cluster: %s<%d>", cluster.Name, clusterID))
		if err != nil {
			return bizerr.New(bizerr.ErrCodeDB, "failed to add access log")
		}
		return nil
	})
	if err != nil {
		return err.(bizerr.BizError)
	}
	return nil
}

// Endpoints is the collection of endpoints for the Cluster.
type Endpoints struct {
	MySQL           EndpointStatus
	PrometheusReady bool
	PdReady         bool
	PrivateMySQL    EndpointStatus
}

func FindClusterStatus(ctx context.Context, api *infraapi.API, clusterID entity.ID, orgID *entity.ID) (*cloudv1.Cluster, entity.Cluster, error) {
	var cluster entity.Cluster

	err := db.Transaction(func(tx *gorm.DB) error {
		var err error
		cluster, err = dbutil.GetCluster(tx, clusterID)
		if err != nil {
			return err
		}

		if orgID != nil && cluster.TenantID != *orgID {
			return errcode.NewForbiddenErr(fmt.Errorf("cluster %d belongs to tenant %d, does not belong to tenant %d in the request", clusterID, cluster.TenantID, *orgID))
		}

		return nil
	})

	if err != nil {
		return nil, cluster, err
	}

	cr, err := api.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		return nil, cluster, err
	}
	return cr, cluster, nil
}

func FindClusterMetricsBaseURL(ctx context.Context, api *infraapi.API, clusterID entity.ID, orgID *entity.ID) (string, error) {
	cr, _, err := FindClusterStatus(ctx, api, clusterID, orgID)
	if err != nil {
		return "", errors.AddStack(err)
	}

	if cr.Status.Metrics == nil {
		return "", errors.New("no metrics status")
	}

	return cr.Status.Metrics.BaseURL, nil
}

func FindClusterDashboardBaseURL(ctx context.Context, encryptor encrypt.Encryptor, api *infraapi.API, clusterID entity.ID, orgID *entity.ID) (string, error) {
	cr, cluster, err := FindClusterStatus(ctx, api, clusterID, orgID)
	if err != nil {
		return "", errors.Annotatef(err, "did not find cluster status for cluster %d", uint(clusterID))
	}
	if cr.Status.Proxy == nil {
		return "", errors.Errorf("no proxy status for cluster %d", uint(clusterID))
	}

	params := ""
	adminPwd, err := util.DecryptCloudAdminPassword(encryptor, cluster, cr.Status)
	if err != nil {
		return "", errors.Annotate(err, "could not decrypt password")
	}
	if len(adminPwd) > 0 {
		params = "?password=" + base64.StdEncoding.EncodeToString([]byte(string(adminPwd)))
	} else {
		log.Info(ctx, "no cloud admin password for cluster", zap.Uint64("clusterID", uint64(clusterID)))
	}

	return cr.Status.Proxy.Endpoint + params, nil
}

func FindClusterProxyBaseURL(ctx context.Context, api *infraapi.API, clusterID entity.ID, orgID *entity.ID) (string, error) {
	cr, _, err := FindClusterStatus(ctx, api, clusterID, orgID)
	if err != nil {
		return "", errors.AddStack(err)
	}
	if cr.Status.Proxy == nil {
		return "", errors.New("no proxy status")
	}
	return cr.Status.Proxy.Endpoint, nil
}

// getComponentName trims the tc name prefix
// "{tcname}-tidb-0" -> "tidb-0"
func getComponentNodeName(podName string) string {
	ar := strings.Split(podName, "-")
	if len(ar) > 2 {
		ar = ar[len(ar)-2:]
	}
	return strings.Join(ar, "-")
}

func tikvNodeStatus(ctx context.Context, s cloudv1.MemberStatus) central.TiKVNode_Status {
	switch s {
	case cloudv1.MemberStatusNormal:
		return central.TiKVNode_NORMAL
	case cloudv1.MemberStatusCreating:
		return central.TiKVNode_CREATING
	case cloudv1.MemberStatusUnavailable:
		return central.TiKVNode_UNAVAILABLE
	case cloudv1.MemberStatusTerminating:
		return central.TiKVNode_TERMINATING
	case cloudv1.MemberStatusLeaving:
		return central.TiKVNode_LEAVING
	default:
		log.Error(ctx, "unknown member status for tikv", zap.String("status", string(s)))
		return central.TiKVNode_NORMAL
	}
}

func tiflashNodeStatus(ctx context.Context, s cloudv1.MemberStatus) central.TiFlashNode_Status {
	switch s {
	case cloudv1.MemberStatusNormal:
		return central.TiFlashNode_NORMAL
	case cloudv1.MemberStatusCreating:
		return central.TiFlashNode_CREATING
	case cloudv1.MemberStatusUnavailable:
		return central.TiFlashNode_UNAVAILABLE
	case cloudv1.MemberStatusTerminating:
		return central.TiFlashNode_TERMINATING
	case cloudv1.MemberStatusLeaving:
		return central.TiFlashNode_LEAVING
	default:
		log.Error(ctx, "unknown member status for tiflash", zap.String("status", string(s)))
		return central.TiFlashNode_NORMAL
	}
}

func tidbNodeStatus(ctx context.Context, s cloudv1.MemberStatus) central.TiDBNode_Status {
	switch s {
	case cloudv1.MemberStatusNormal:
		return central.TiDBNode_NORMAL
	case cloudv1.MemberStatusCreating:
		return central.TiDBNode_CREATING
	case cloudv1.MemberStatusUnavailable:
		return central.TiDBNode_UNAVAILABLE
	case cloudv1.MemberStatusTerminating:
		return central.TiDBNode_TERMINATING
	default:
		log.Error(ctx, "unknown member status for tidb", zap.String("status", string(s)))
		return central.TiDBNode_NORMAL
	}
}

func (c UserContext) ListTiKVNodesV1(ctx context.Context, orgID, projectID, clusterID entity.ID) (nodes []*central.TiKVNode, err error) {
	var cluster entity.Cluster
	err = db.Transaction(func(tx *gorm.DB) error {
		var err error
		cluster, err = dbutil.GetClusterByID(tx, orgID, projectID, clusterID)
		return err
	})

	if err != nil {
		return nil, err
	}

	cr, err := c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), clusterID.Uint64())
	if err != nil {
		return nil, err
	}

	if cr.Status.DB == nil {
		return nil, nil
	}

	com := cr.Status.DB.TiKV
	if com == nil {
		return nil, nil
	}

	// don't use the storageSize of cr, use component.StorageSizeGi instead.
	// it's not correct for local-storage, due to the some space are reserved by filesystem.
	storageSizeGi := uint32(0)
	for _, each := range cluster.ComponentsV2 {
		if each.ComponentType == entity.TiKV {
			storageSizeGi = each.StorageSizeGi
		}
	}
	mUnit := int64(MemoryUnit)
	for _, m := range com.Members {
		node := &central.TiKVNode{
			Name:      getComponentNodeName(m.Name),
			Status:    tikvNodeStatus(ctx, m.Status),
			Type:      central.TiKVNode_STORAGE,
			StorageGi: storageSizeGi,
			Compute: &central.Compute{
				Cpu:      float32(m.Resources.CPU.Value()),
				MemoryMi: uint32(m.Resources.Memory.Value() / mUnit / mUnit),
			},
			Az: m.Labels["cloud.pingcap.com/zone"],
		}

		nodes = append(nodes, node)
	}

	return nodes, nil
}

func (c UserContext) ListTiFlashNodesV1(ctx context.Context, orgID, projectID, clusterID entity.ID) (nodes []*central.TiFlashNode, err error) {
	var cluster entity.Cluster
	err = db.Transaction(func(tx *gorm.DB) error {
		var err error
		cluster, err = dbutil.GetClusterByID(tx, orgID, projectID, clusterID)
		return err
	})

	if err != nil {
		return nil, err
	}

	cr, err := c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), clusterID.Uint64())
	if err != nil {
		return nil, err
	}

	if cr.Status.DB == nil {
		return nil, nil
	}

	com := cr.Status.DB.TiFlash
	if com == nil {
		return nil, nil
	}

	// don't use the storageSize of cr, use component.StorageSizeGi instead.
	// it's not correct for local-storage, due to the some space are reserved by filesystem.
	storageSizeGi := uint32(0)
	for _, each := range cluster.ComponentsV2 {
		if each.ComponentType == entity.TiFlash {
			storageSizeGi = each.StorageSizeGi
		}
	}
	mUnit := int64(MemoryUnit)
	for _, m := range com.Members {
		node := &central.TiFlashNode{
			Name:      getComponentNodeName(m.Name),
			Status:    tiflashNodeStatus(ctx, m.Status),
			Type:      central.TiFlashNode_STORAGE,
			StorageGi: storageSizeGi,
			Compute: &central.Compute{
				Cpu:      float32(m.Resources.CPU.Value()),
				MemoryMi: uint32(m.Resources.Memory.Value() / mUnit / mUnit),
			},
			Az: m.Labels["cloud.pingcap.com/zone"],
		}

		nodes = append(nodes, node)
	}

	return nodes, nil
}

func (c UserContext) ListTiDBNodesV1(ctx context.Context, orgID, projectID, clusterID entity.ID) (nodes []*central.TiDBNode, err error) {
	var cluster entity.Cluster
	err = db.Transaction(func(tx *gorm.DB) error {
		var err error
		cluster, err = dbutil.GetClusterByID(tx, orgID, projectID, clusterID)
		return err
	})

	if err != nil {
		return nil, err
	}

	cr, err := c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), clusterID.Uint64())
	if err != nil {
		return nil, err
	}

	if cr.Status.DB == nil {
		return nil, nil
	}

	com := cr.Status.DB.TiDB
	if com == nil {
		return nil, nil
	}

	mUnit := int64(MemoryUnit)
	for _, m := range com.Members {
		node := &central.TiDBNode{
			Name:   getComponentNodeName(m.Name),
			Status: tidbNodeStatus(ctx, m.Status),
			Type:   central.TiDBNode_COMPUTING,
			Compute: &central.Compute{
				Cpu:      float32(m.Resources.CPU.Value()),
				MemoryMi: uint32(m.Resources.Memory.Value() / mUnit / mUnit),
			},
			Az: m.Labels["cloud.pingcap.com/zone"],
		}

		nodes = append(nodes, node)
	}

	return nodes, nil
}

const (
	MemoryUnit uint32 = 1024
)

func (c UserContext) DetachReplication(orgID, clusterID, rID entity.ID, detachType dataflow.DetachReplicationReq_DetachType) (*dataflow.DetachReplicationResp, error) {
	var (
		err error
		r   entity.Replication
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		err = entity.NewReplicationQuerySet(tx).
			IDEq(rID).
			PreloadPrimaryCluster().
			PreloadSecondaryCluster().
			One(&r)
		if err != nil {
			return err
		}

		// if cluster is not primary or secondary cluster of replication
		// return input error
		if r.SecondaryClusterID != clusterID && r.PrimaryClusterID != clusterID {
			return errcode.NewInvalidInputErr(errors.Errorf("cluster %d is not primary or secondary cluster of replication %d", clusterID, rID))
		}

		// planned detach need to set primary as read-only and wait for the upstream and downstream replication is algin
		// TODO:wait for read-only function is ready
		if detachType == dataflow.DetachReplicationReq_Planned {
			r.DetachType = entity.DetachTypePlanned
		} else if detachType == dataflow.DetachReplicationReq_Force {
			r.DetachType = entity.DetachTypeForce
		} else if detachType == dataflow.DetachReplicationReq_None {
			r.DetachType = entity.DetachTypeNone
		} else {
			return errcode.NewInvalidInputErr(errors.Errorf("unknow replication type %d", detachType))
		}

		// update replication, primary cluster and secondary cluster status as detach other gc worker
		r.ReplicationStatus = entity.ReplicationStatusDetaching
		r.DetachUserID = c.tenantAndUser.User.ID

		return dbutil.UpdateReplication(tx, &r)
	})
	if err != nil {
		return nil, err
	}

	return &dataflow.DetachReplicationResp{
		ReplicationId: rID.Uint64(),
		ClusterId:     clusterID.Uint64(),
	}, err
}

// TODO: (yinliang) combine it with replication
type ValueModel struct {
	Variable []string `json:"Value"`
}

func (c BgContext) GetReadOnly(ctx context.Context, cluster *entity.Cluster, getProxy util.GetProxyFunc) (bool, error) {
	var (
		err             error
		data, decodeRet []byte
		adminPwd        encrypt.Secret
		cr              *cloudv1.Cluster
		proxy           clusterproxy.ClusterProxy
		superReadOnly   ValueModel
	)

	cr, err = c.api.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		return false, errors.Annotatef(err, "failed to get cluster")
	}

	if cr == nil {
		return false, nil
	}

	proxy, err = getProxy(cluster.ID.Uint64(), cr.Status, config.DefaultClusterProxyTimeout)
	if err != nil {
		return false, errors.Annotatef(err, "failed to get proxy")
	}

	adminPwd, err = util.DecryptCloudAdminPassword(c.encryptor, *cluster, cr.Status)
	if err != nil {
		return false, errors.Annotatef(err, "failed to decrypt admin password")
	}

	data, err = proxy.QuerySQL(ctx, adminPwd, GetSuperReadonlyClause)
	if err != nil {
		return false, errors.Annotatef(err, "failed to query super read only")
	}

	err = json.Unmarshal(data, &superReadOnly)
	if err != nil {
		return false, errors.Annotatef(err, "failed to unmarshal super read only")
	}

	if len(superReadOnly.Variable) == 0 {
		return false, nil
	}

	decodeRet, err = base64.StdEncoding.DecodeString(superReadOnly.Variable[0])
	if err != nil {
		return false, errors.Annotatef(err, "failed to decrypted super read only")
	}

	if string(decodeRet) == "OFF" || string(decodeRet) == "off" {
		return false, nil
	}
	return true, nil
}

func (m Model) SyncReadonly(ctx context.Context) error {
	var (
		err      error
		readonly bool
		clusters []entity.Cluster
	)
	logger := log.AppendCtxFields(ctx, m.log).Sugar()
	clusters, err = m.dao.FindClusters(dbutil.ClusterFilter{
		Deleted:  pointer.BoolPtr(false),
		Cleared:  pointer.BoolPtr(false),
		Creating: pointer.BoolPtr(false),
		Free:     pointer.BoolPtr(false),
	})
	if err != nil {
		return errors.Annotatef(err, "failed to fetch clusters records")
	}

	startTime := time.Now()
	logger.Infof("[global]sync cluster readonly, start. cnt=%d", len(clusters))
	for _, cluster := range clusters {
		if cluster.IsDevTier() {
			continue
		}
		c, err := GetBgContext(m, cluster.TenantID)
		if err != nil {
			logger.Error("failed to create bg context", zap.Error(err))
			continue
		}

		// if force set readonly, make cluster readonly and set ForceReadOnlyStatus zero
		if cluster.ReadOnly&entity.ForceReadOnlyStatus != 0 {
			err = UserContext(c).SetReadOnly(ctx, cluster.ID, cluster.ProjectID, "ON")
			if err != nil {
				c.log.Error("failed to set read only status", zap.Error(err))
				continue
			}
			cluster.ReadOnly &= (^entity.ForceReadOnlyStatus)
			err := c.dao.UpdateCluster(cluster, []entity.ClusterDBSchemaField{entity.ClusterDBSchema.ReadOnly}, true)
			if err != nil {
				c.log.Error("failed to update cluster", zap.Error(err))
			}
			continue
		}

		// else just sync cluster read only status from cluster to db
		readonly, err = c.GetReadOnly(ctx, &cluster, tidbcloud.BuildClusterProxy)
		if err != nil {
			// don't block cluster sync for the proxy problems
			c.log.Error("failed to get read only status", zap.Error(err))
			continue
		}
		oldReadonly := cluster.ReadOnly
		if readonly {
			cluster.ReadOnly |= entity.ReadOnlyStatus
		} else {
			cluster.ReadOnly &= (^entity.ReadOnlyStatus)
		}
		// don't update db if readonly status is not changed
		if cluster.ReadOnly == oldReadonly {
			continue
		}
		err = c.dao.UpdateCluster(cluster, []entity.ClusterDBSchemaField{entity.ClusterDBSchema.ReadOnly}, false)
		if err != nil {
			c.log.Error("failed to update cluster read only status", zap.Error(err))
			continue
		}
	}
	logger.Infof("[global]sync cluster readonly, done. cost=%s", time.Since(startTime))
	return nil
}

// MakeWritable change the cluster's read_only status
func (c UserContext) SetReadOnly(ctx context.Context, clusterID entity.ID, projectID entity.ID, trigger string) error {
	var (
		err     error
		cluster entity.Cluster
	)
	cluster, err = c.dao.GetCluster(clusterID)
	if err != nil {
		return err
	}

	if cluster.ProjectID != projectID {
		return errcode.NewForbiddenErr(errors.New("cluster_id parameter is inconsistent with the user's project ID"))
	}

	return util.ExecuteSqlCommon(ctx, &cluster, fmt.Sprintf(config.SetSuperReadonlyClause, trigger), c.api, c.encryptor, tidbcloud.BuildClusterProxy, config.DefaultClusterProxyTimeout)
}

func (c UserContext) CheckIfUserIsAuditAdmin() error {
	user := c.tenantAndUser
	if user.Role == entity.UserRoleAuditAdmin {
		return nil
	}
	return errcode.NewForbiddenErr(errors.New("only audit admin can perform this operation"))
}

func QueryClusterWithSQL(ctx context.Context, api *infraapi.API, encryptor encrypt.Encryptor, cluster *entity.Cluster, getProxy util.GetProxyFunc, sql string) ([]byte, error) {
	var (
		err      error
		adminPwd encrypt.Secret
		cr       *cloudv1.Cluster
		proxy    clusterproxy.ClusterProxy
	)

	cr, err = api.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		return nil, errors.Annotatef(err, "failed to get cluster")
	}

	proxy, err = getProxy(cluster.ID.Uint64(), cr.Status, config.DefaultClusterProxyTimeout)
	if err != nil {
		return nil, errors.Annotatef(err, "failed to get proxy")
	}

	adminPwd, err = util.DecryptCloudAdminPassword(encryptor, *cluster, cr.Status)
	if err != nil {
		return nil, errors.Annotatef(err, "failed to decrypt admin password")
	}

	return proxy.QuerySQL(ctx, adminPwd, sql)
}