package service

import (
	"context"
	"errors"
	"strconv"
	"sync"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/metric"
	"github.com/tidbcloud/serverless-service/internal/utils"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.uber.org/zap"
)

const (
	ReasonNormal = "normal"
)

var LimiterTracer = otel.Tracer("branch.limiter")

type BranchSizeLimiterService struct {
	o11yFacade    facade.O11yFacade
	infraFacade   facade.InfraFacade
	billingFacade facade.BillingFacade
	mgmtSvc       facade.MgmtSvcFacade
}

func NewBranchSizeLimiterService(
	o11yFacade facade.O11yFacade,
	infraFacade facade.InfraFacade,
	billingFacade facade.BillingFacade,
	mgmtSvc facade.MgmtSvcFacade,
	branchRepo repo.BranchRepo,
) *BranchSizeLimiterService {
	return &BranchSizeLimiterService{
		o11yFacade:    o11yFacade,
		infraFacade:   infraFacade,
		billingFacade: billingFacade,
		mgmtSvc:       mgmtSvc,
	}
}

func (s *BranchSizeLimiterService) Execute(ctx context.Context) (err error) {
	ctx, span := LimiterTracer.Start(ctx, "branchSizeLimiter.Execute")
	defer func() {
		tracing.EndSpan(span, err)
	}()

	regionSvCfg := config.Global().WorkerConfig.RegionalConfig
	if len(regionSvCfg) == 0 {
		return errors.New("no regional service config found")
	}

	var wg sync.WaitGroup
	for _, c := range regionSvCfg {
		for _, k := range c.K8sConfig {
			wg.Add(1)
			go func(provider, region, kubernetes, o11yUrl string) {
				defer wg.Done()
				_ = s.limitOneKubernetes(ctx, provider, region, kubernetes, o11yUrl)
			}(c.Provider, c.Region, k.Kubernetes, k.O11yUrl)
		}
	}
	wg.Wait()

	return nil
}

func (s *BranchSizeLimiterService) limitOneKubernetes(
	ctx context.Context,
	provider, region, kubernetes, o11yUrl string,
) (err error) {
	reason := ReasonNormal
	ctx, span := LimiterTracer.Start(ctx, "branchSizeLimiter.limitOneKubernetes")
	defer func() {
		tracing.EndSpan(span, err)
		updateLimitJobMetric(provider, region, kubernetes, reason, err)
	}()
	span.SetAttributes(attribute.String("provider", provider), attribute.String("region", region))
	logger := utils.GetWorkerLoggerFromContext(ctx).
		With(zap.String("provider", provider), zap.String("region", region), zap.String("kubernetes", kubernetes))

	logger.Info("start to limitOneKubernetes for branch ru limit mode")

	span.AddEvent("list branches")
	proxyEndpoint := config.GetClusterProxyUrl(provider, region, kubernetes)
	branches, err := s.infraFacade.ListBranches(ctx, proxyEndpoint)
	if err != nil {
		logger.Error("failed to list branches", zap.Error(err))
		reason = "failed to list branches"
		return err
	}

	logger.Info("finish listing branches", zap.Int("branch_count", len(branches)))

	if len(branches) == 0 {
		logger.Info("no branches found")
		return nil
	}

	var keyspaceIds []uint32
	keyspace2Branch := make(map[string]*cloudv1.Branch)
	for i := range branches {
		branch := branches[i]
		if branch.Status.Phase != cloudv1.BranchPhaseRunning {
			logger.Info(
				"branch is not running",
				zap.String("branch", branch.Name),
				zap.String("phase", string(branch.Status.Phase)),
			)
			continue
		}
		meta, err := NewBranchMeta(&branch)
		if err != nil {
			logger.Error("failed to get branch meta", zap.Error(err), zap.String("branch", branch.Name))
			updateLimitJobMetric(provider, region, kubernetes, "failed to get branch meta", err)
			continue
		}
		if _, ok := config.Global().BizConfig.BranchTenantNumberQuota[meta.TenantID]; ok {
			continue
		}
		branchStat, err := branch.AwsFreeTierClusterStatusFromBranch()
		if err != nil {
			logger.Error(
				"failed to get aws free tier cluster status from branch",
				zap.Error(err),
				zap.String("branch", branch.Name),
			)
			updateLimitJobMetric(provider, region, kubernetes, "invalid branch status", err)
			continue
		}
		keyspaceIds = append(keyspaceIds, branchStat.KeyspaceId)
		keyspace2Branch[strconv.FormatUint(uint64(branchStat.KeyspaceId), 10)] = &branches[i]
	}

	span.AddEvent("get branches sku usages by region")
	branchUsages, err := s.o11yFacade.GetBranchesSkuUsagesByRegion(
		ctx,
		keyspaceIds,
		constant.MetricTypeTikvStorage,
		o11yUrl,
	)
	if err != nil {
		logger.Error("failed to get branches sku usages by region", zap.Error(err), zap.String("url", o11yUrl))
		reason = "get branches sku usages by region failed"
		return err
	}

	span.AddEvent("check and update branch ru limit mode")
	type clusterLimitMeta struct {
		paid        bool
		ruLimitMode string
	}
	clusters := make(map[uint64]clusterLimitMeta)
	for keyspaceID, v := range branchUsages {
		branch := keyspace2Branch[keyspaceID]
		needUpdate := false
		if branch == nil {
			logger.Error("branch not found", zap.String("keyspace_id", keyspaceID))
			updateLimitJobMetric(provider, region, kubernetes, "branch not found", err)
			continue
		}
		if branch.Annotations == nil {
			logger.Error("branch annotations is nil", zap.String("branch", branch.Name))
			updateLimitJobMetric(provider, region, kubernetes, "branch annotations is nil", err)
			continue
		}
		clusterId, err := strconv.ParseUint(branch.Annotations[constant.MainBranchIdAnnKey], 10, 64)
		if err != nil {
			logger.Error(
				"main_branch_id is invalid",
				zap.String("main_branch_id", branch.Annotations[constant.MainBranchIdAnnKey]),
				zap.Error(err),
			)
			updateLimitJobMetric(provider, region, kubernetes, "main_branch_id is invalid", err)
			continue
		}
		clusterMeta, ok := clusters[clusterId]
		if !ok {
			c, err := s.mgmtSvc.GetCluster(ctx, clusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_FULL)
			if err != nil {
				if err.RawErrCode() == bizerr.ErrCodeRecordNotFound {
					logger.Warn("cluster not found", zap.Uint64("cluster_id", clusterId))
				} else {
					logger.Error("failed to get cluster", zap.Error(err), zap.Uint64("cluster_id", clusterId))
					updateLimitJobMetric(provider, region, kubernetes, "failed to get cluster", err)
				}
				continue
			}
			paid := c.GetServerlessCluster().GetCluster().GetAutoScaling().GetMinRcu() > 0 ||
				c.GetServerlessCluster().GetSpendLimit().GetMonthly() > 0
			ruLimitMode := c.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoRuLimitMode]
			if ruLimitMode == "" {
				ruLimitMode = constant.RuLimitModeStandard
			}
			clusterMeta = clusterLimitMeta{
				paid:        paid,
				ruLimitMode: ruLimitMode,
			}
			clusters[clusterId] = clusterMeta
		}
		if clusterMeta.ruLimitMode == constant.RuLimitModeThrottled {
			if branch.Annotations[constant.AnnoRuLimitMode] != constant.RuLimitModeThrottled {
				branch.Annotations[constant.AnnoRuLimitMode] = constant.RuLimitModeThrottled
				needUpdate = true
				logger.Info(
					"parent cluster ru mode is throttled, set branch ru limit mode to throttled",
					zap.String("branch", branch.Name),
				)
			}
		} else {
			if (clusterMeta.paid && int64(v) >= config.Global().BizConfig.BranchStorageLimitMiBVip) ||
				(!clusterMeta.paid && int64(v) >= config.Global().BizConfig.BranchStorageLimitMiB) {
				if branch.Annotations[constant.AnnoRuLimitMode] != constant.RuLimitModeThrottled {
					branch.Annotations[constant.AnnoRuLimitMode] = constant.RuLimitModeThrottled
					needUpdate = true
					logger.Info(
						"branch storage size exceeds limit, set ru limit mode to throttled",
						zap.String("branch", branch.Name),
						zap.Bool("paid", clusterMeta.paid),
					)
				}
			} else {
				if branch.Annotations[constant.AnnoRuLimitMode] != constant.RuLimitModeStandard {
					branch.Annotations[constant.AnnoRuLimitMode] = constant.RuLimitModeStandard
					needUpdate = true
					logger.Info(
						"branch storage size is under limit, set ru limit mode to standard",
						zap.String("branch", branch.Name),
						zap.Bool("paid", clusterMeta.paid),
					)
				}
			}
		}

		if needUpdate {
			if err := s.infraFacade.UpdateBranch(ctx, branch); err != nil {
				logger.Error(
					"failed to update branch ru limit mode",
					zap.Error(err),
					zap.String("branch", branch.Name),
					zap.String("ru limit mode", branch.Annotations[constant.AnnoRuLimitMode]),
				)
				updateLimitJobMetric(provider, region, kubernetes, "failed to update branch ru limit mode", err)
				continue
			}
		}
	}

	return nil
}

func updateLimitJobMetric(provider, region, kubenetes, reason string, err error) {
	status := metric.StatusOK
	if err != nil {
		status = metric.StatusErr
	}
	metric.BranchLimitJobProcessCounter.WithLabelValues(provider, region, kubenetes, status, reason).Inc()
}