package opworker

import (
	"encoding/json"
	"fmt"
	"time"

	"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"
	"github.com/tidbcloud/core-portal/internal/central/event"
	"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"
	"github.com/tidbcloud/core-portal/internal/central/model/util"
	"github.com/tidbcloud/core-portal/internal/central/repository"
	entity "github.com/tidbcloud/core-svc/model"
	"github.com/tidbcloud/pkgv2/log"
	mailv2 "github.com/tidbcloud/pkgv2/mail/v2"
	"github.com/tidbcloud/pkgv2/pubsub"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	"golang.org/x/net/context"
	"gorm.io/gorm"
)

// trafficUsageInfo represent the network traffic usage of cluster
type trafficUsageInfo struct {
	IngressUsage       float64   `json:"ingressUsage"`       // unit: Byte
	EgressUsage        float64   `json:"egressUsage"`        // unit: Byte
	LastAccumulateTime time.Time `json:"lastAccumulateTime"` // for example: 2021-09-03T06:16:29.008472279Z
}

const (
	trafficUsageAnnKey = "bandwidth.tidbcloud.com/usage"
)

type devTierClusterWorker struct {
	GlobalWorker
	mailCli                 mailv2.Facade
	infraCli                infraapi.InfraFacade
	notificationMgr         util.NotificationManager
	dbProvider              db.TransProvider
	tenantRepo              repository.TenantRepository
	userRepo                repository.UserRepository
	clusterRepo             repository.ClusterRepository
	clusterTrafficUsageRepo repository.ClusterTrafficUsageRepository
	publisher               pubsub.Publisher
}

func NewDevTierClusterWorker(
	bgWorker GlobalWorker,
	mailCli mailv2.Facade,
	infraCli infraapi.InfraFacade,
	notificationMgr util.NotificationManager,
	dbProvider db.TransProvider,
	tenantRepo repository.TenantRepository,
	userRepo repository.UserRepository,
	clusterRepo repository.ClusterRepository,
	clusterTrafficUsageRepo repository.ClusterTrafficUsageRepository,
	publisher pubsub.Publisher,
) *devTierClusterWorker {
	w := &devTierClusterWorker{
		GlobalWorker:            bgWorker,
		mailCli:                 mailCli,
		infraCli:                infraCli,
		notificationMgr:         notificationMgr,
		dbProvider:              dbProvider,
		tenantRepo:              tenantRepo,
		userRepo:                userRepo,
		clusterRepo:             clusterRepo,
		clusterTrafficUsageRepo: clusterTrafficUsageRepo,
		publisher:               publisher,
	}
	w.logger = w.logger.With(zap.String("workerName", "dev_tier_cluster"))
	return w
}

func (w *devTierClusterWorker) Run(ctx context.Context) error {
	logger := log.AppendCtxFields(ctx, w.logger)
	rdb := w.dbProvider.ReadDB(ctx)
	now := time.Now()

	// Currently dev tier clusters are only supported on AWS
	runningDevTierClusters, err := getProviderRunningDevTierCluster(ctx, rdb, entity.Aws)
	if err != nil {
		return errors.Annotate(err, "get running dev tier cluster")
	}
	count := len(runningDevTierClusters)
	labels := map[string]string{
		metrics.ProviderLblName: string(entity.Aws),
		metrics.ProviderTypeLbl: metrics.DevTier,
	}
	metrics.DevTierActiveClustersCount.With(labels).Set(float64(count))
	logger.With(zap.String("provider", string(entity.Aws))).Sugar().Infof("runningDevTierCluster num is %d", count)

	return multierr.Combine(
		w.scanIdleCluster(ctx, runningDevTierClusters),
		w.syncTrafficUsage(ctx, runningDevTierClusters),
		w.scanNetworkThrottledCluster(ctx, runningDevTierClusters, now),
	)
}

func (w *devTierClusterWorker) scanIdleCluster(ctx context.Context, runningClusters []entity.Cluster) error {
	logger := log.AppendCtxFields(ctx, w.logger)
	now := time.Now()
	for _, cluster := range runningClusters {
		lastActiveTime, err := GetLastActiveTimeOfCluster(ctx, &cluster, w.infraCli)
		if err != nil {
			logger.Info("get lastActiveTime failed, skip it", zap.Uint64("clusterID", cluster.ID.Uint64()), zap.Error(err))
			continue
		}
		if lastActiveTime.IsZero() {
			logger.Info("lastActiveTime is Zero, skip it", zap.Uint64("clusterID", cluster.ID.Uint64()))
			continue
		}

		cr, err := w.infraCli.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
		if err != nil {
			logger.Info("get infra cluster error, skip it", zap.Uint64("clusterID", cluster.ID.Uint64()), zap.Error(err))
			continue
		}
		isGateway := cluster.IsDevTier() &&
			cr.Status.SqlEndpoints.PublicEndpoint != nil && cr.Status.SqlEndpoints.PublicEndpoint.Token != ""
		idleDuration := now.Sub(lastActiveTime)
		stage := calcDevTierIdleStage(idleDuration, isGateway)
		logger.Info("cluster stage", zap.Uint64("clusterID", cluster.ID.Uint64()), zap.String("stage", string(stage)))
		if err := w.dbProvider.Transaction(ctx, func(ctx context.Context) error {
			tx := w.dbProvider.WriteDB(ctx)
			tenant, err := w.tenantRepo.Load(ctx, cluster.TenantID)
			if err != nil {
				return errors.Annotate(err, "get tenant by id")
			}
			user, err := w.userRepo.LoadByEmail(ctx, tenant.Email)
			if err != nil {
				return errors.Annotate(err, "get user by email")
			}

			switch stage {
			case entity.DevTierIdleStageActive:
				// we will send warning email every time when the cluster has no activity for 6 days.
				return multierr.Combine(
					w.notificationMgr.ClearNotificationIdentifier(ctx, tx, cluster.TenantID, entity.NotificationSceneDevTierIdle,
						w.makeClusterIdleIdentifierByStage(ctx, &cluster, entity.DevTierIdleStageWarning)),
				)
			case entity.DevTierIdleStageEncourage:
				return w.notifyUserClusterIdleEncourage(ctx, tx, &cluster, &user)
			case entity.DevTierIdleStageWarning:
				return w.notifyUserClusterIdleWarning(ctx, tx, &cluster, &user)
			case entity.DevTierIdleStageShutdown:
				if err := w.shutdownCluster(ctx, tx, &cluster); err != nil {
					return errors.Annotate(err, "shutdown cluster")
				}
				return w.notifyUserClusterShutdown(ctx, tx, &cluster, &user)
			}
			return nil
		}); err != nil {
			logger.Sugar().Warnf("change dev tier idle Stage failed: %s", err)
			continue
		}
	}
	return nil
}

func (w *devTierClusterWorker) syncTrafficUsage(ctx context.Context, runningClusters []entity.Cluster) error {
	logger := log.AppendCtxFields(ctx, w.logger)
	for _, each := range runningClusters {
		usage, err := GetClusterTrafficUsage(ctx, &each, w.infraCli)
		if err != nil {
			logger.Warn("get cluster traffic usage failed, skip it", zap.Uint64("clusterID", each.ID.Uint64()), zap.Error(err))
			continue
		}
		logger.Info("cluster traffic usage from infra",
			zap.Uint64("clusterID", each.ID.Uint64()),
			zap.Int64("ingressUsage", int64(usage.IngressUsage)), zap.Int64("egressUsage", int64(usage.EgressUsage)))

		period := cluster.MakeClusterTrafficUsagePeriod(usage.LastAccumulateTime)
		en, err := w.clusterTrafficUsageRepo.LoadByPeriod(ctx, each.ID, period)
		if err != nil && err != gorm.ErrRecordNotFound {
			return errors.Annotate(err, "load traffic usage by period")
		}
		if err == gorm.ErrRecordNotFound {
			en = &entity.ClusterTrafficUsage{
				TenantID:     each.TenantID,
				ClusterID:    each.ID,
				Period:       period,
				IngressUsage: 0,
				EgressUsage:  0,
			}
		}

		en.IngressUsage = int64(usage.IngressUsage)
		en.EgressUsage = int64(usage.EgressUsage)
		err = w.clusterTrafficUsageRepo.Save(ctx, en)
		if err != nil {
			return errors.Annotate(err, "save traffic usage")
		}
	}
	return nil
}

func (w *devTierClusterWorker) scanNetworkThrottledCluster(ctx context.Context, runningClusters []entity.Cluster, now time.Time) error {
	logger := log.AppendCtxFields(ctx, w.logger)
	for _, each := range runningClusters {
		ingressUsage, egressUsage, err := w.clusterTrafficUsageRepo.SumUsageByTenant(ctx, each.TenantID, cluster.MakeClusterTrafficUsagePeriod(now))
		if err != nil {
			return errors.Annotate(err, "sum traffic usage by tenant")
		}
		logger.Info("cluster traffic usage from db",
			zap.Uint64("tenantID", each.TenantID.Uint64()),
			zap.Int64("ingressUsage", ingressUsage), zap.Int64("egressUsage", egressUsage))

		ev := &event.ClusterThresholdStateChanged{
			TenantID:         each.TenantID.Uint64(),
			ProjectID:        each.ProjectID.Uint64(),
			ClusterID:        each.ID.Uint64(),
			TriggerTime:      now,
			IsDevTier:        true,
			IngressThrottled: ingressUsage > config.MaxTrafficUsagePerWeek4DevTier,
			EgressThrottled:  egressUsage > config.MaxTrafficUsagePerWeek4DevTier,
		}
		if err := w.publisher.Publish(ctx, ev); err != nil {
			return errors.Annotate(err, "publish event")
		}
	}
	return nil
}

func (w *devTierClusterWorker) notifyUserClusterIdleEncourage(ctx context.Context, tx *gorm.DB, cluster *entity.Cluster, user *entity.User) error {
	identifier := w.makeClusterIdleIdentifierByStage(ctx, cluster, entity.DevTierIdleStageEncourage)
	err := w.notificationMgr.SendNotificationOnce(ctx, tx, cluster.TenantID, entity.NotificationSceneDevTierIdle, identifier, func() error {
		vars := map[string]string{
			"userName": user.FirstName,
		}
		return w.mailCli.SendMailWithTemplate(ctx, []string{user.Email}, config.EmailSubjectDevTierClusterIdleEncourage, config.TNDevTierClusterIdleEncourage, vars, nil)
	})
	return errors.Annotate(err, "send notification to encourage user use with tidb cloud")
}

func (w *devTierClusterWorker) notifyUserClusterIdleWarning(ctx context.Context, tx *gorm.DB, cluster *entity.Cluster, user *entity.User) error {
	identifier := w.makeClusterIdleIdentifierByStage(ctx, cluster, entity.DevTierIdleStageWarning)
	err := w.notificationMgr.SendNotificationOnce(ctx, tx, cluster.TenantID, entity.NotificationSceneDevTierIdle, identifier, func() error {
		vars := map[string]string{
			"userName": user.FirstName,
		}
		return w.mailCli.SendMailWithTemplate(ctx, []string{user.Email}, config.EmailSubjectDevTierClusterIdleWarning, config.TNDevTierClusterIdleWarning, vars, nil)
	})
	return errors.Annotate(err, "send notification cluster will be shutdown within x hours")
}

func (w *devTierClusterWorker) shutdownCluster(ctx context.Context, tx *gorm.DB, cluster *entity.Cluster) error {
	err := w.clusterRepo.DeleteClusterAndMakeFinalBackup(ctx, cluster.ID)
	return errors.Annotate(err, "delete and make final backup for dev tier cluster")
}

func (w *devTierClusterWorker) notifyUserClusterShutdown(ctx context.Context, tx *gorm.DB, cluster *entity.Cluster, user *entity.User) error {
	identifier := w.makeClusterIdleIdentifierByStage(ctx, cluster, entity.DevTierIdleStageShutdown)
	err := w.notificationMgr.SendNotificationOnce(ctx, tx, cluster.TenantID, entity.NotificationSceneDevTierIdle, identifier, func() error {
		vars := map[string]string{
			"userName": user.FirstName,
		}
		return w.mailCli.SendMailWithTemplate(ctx, []string{user.Email}, config.EmailSubjectDevTierClusterIdleShutdown, config.TNDevTierClusterIdleShutdown, vars, nil)
	})
	return errors.Annotate(err, "send notification cluster has been shutdown after x days idle ")
}

func (w *devTierClusterWorker) makeClusterIdleIdentifierByStage(ctx context.Context, cluster *entity.Cluster, stage entity.DevTierIdleStage) string {
	// identifier: clusterID + stage
	return fmt.Sprintf("%d:%s", cluster.ID, stage)
}

func calcDevTierIdleStage(idleDuration time.Duration, isGateway bool) entity.DevTierIdleStage {
	var stage entity.DevTierIdleStage
	if isGateway {
		if idleDuration >= config.Duration4DevTierIdleStageEncourage {
			stage = entity.DevTierIdleStageEncourage
		} else {
			stage = entity.DevTierIdleStageActive
		}
	} else {
		if idleDuration >= config.Duration4DevTierIdleStageShutdown {
			stage = entity.DevTierIdleStageShutdown
		} else if idleDuration >= config.Duration4DevTierIdleStageWarning {
			stage = entity.DevTierIdleStageWarning
		} else if idleDuration >= config.Duration4DevTierIdleStageEncourage {
			stage = entity.DevTierIdleStageEncourage
		} else {
			stage = entity.DevTierIdleStageActive
		}
	}
	return stage
}

func getProviderRunningDevTierCluster(ctx context.Context, tx *gorm.DB, cloudProvider entity.Cloud) ([]entity.Cluster, error) {
	provider, err := dbutil.GetProvider(tx, cloudProvider)
	if err != nil {
		return nil, errors.Annotatef(err, "get provder %s", cloudProvider)
	}
	qs := entity.NewClusterQuerySet(tx)
	var clusters []entity.Cluster
	if err := qs.StatusNotIn(entity.DeletedOrClearedClusterStatus...).FreeEq(true).ProviderIDEq(provider.ID).PreloadTenant().All(&clusters); err != nil {
		return nil, errors.Annotatef(err, "get running dev tier clusters of provider %s", cloudProvider)
	}
	return clusters, nil
}

// GetLastActiveTimeOfCluster gets last active time of the cluster from infra-API
func GetLastActiveTimeOfCluster(ctx context.Context, cluster *entity.Cluster, infraCli infraapi.InfraFacade) (time.Time, error) {
	if !cluster.IsDevTier() {
		return time.Time{}, errors.Errorf("lastActiveTime only supported by dev tier")
	}

	c, err := infraCli.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		return time.Time{}, errors.Annotate(err, "get cluster from infra")
	}

	if c.Status.LastActiveTimestamp == nil {
		return time.Time{}, nil
	}
	lastActiveTime := c.Status.LastActiveTimestamp.Time
	return lastActiveTime, nil
}

// GetClusterTrafficUsage gets traffic usage of the cluster from infra-API
func GetClusterTrafficUsage(ctx context.Context, cluster *entity.Cluster, infraCli infraapi.InfraFacade) (usage *trafficUsageInfo, err error) {
	if !cluster.IsDevTier() {
		return nil, errors.Errorf("traffic usage only supported by dev tier")
	}

	c, err := infraCli.GetCluster(ctx, cluster.ProjectID.Uint64(), cluster.ID.Uint64())
	if err != nil {
		return nil, errors.Annotate(err, "get cluster from infra")
	}

	trafficUsageStr := c.ObjectMeta.Annotations[trafficUsageAnnKey]
	trafficUsage, err := parseTrafficUsageStr(ctx, trafficUsageStr, cluster.ID.Uint64())
	if err != nil {
		return nil, errors.Annotate(err, "parse trafficUsage failed")
	}
	return trafficUsage, nil
}

func parseTrafficUsageStr(ctx context.Context, s string, clusterID uint64) (*trafficUsageInfo, error) {
	logger := log.WithContextL(ctx)
	trafficUsage := &trafficUsageInfo{}
	if s == "" {
		logger.Info("traffic usage info is empty", zap.Uint64("clusterID", clusterID))
		trafficUsage.LastAccumulateTime = time.Now()
		return trafficUsage, nil
	}
	err := json.Unmarshal([]byte(s), trafficUsage)
	if err != nil {
		return nil, fmt.Errorf("unmarshal trafficUsageStr failed: %w", err)
	}

	return trafficUsage, nil
}