package cluster_svc

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

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/common"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"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/domain/service"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/utils"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

type clusterRpcService struct {
	billingSvc                 service.BillingService
	clusterRepo                repo.ClusterRepo
	eventRepo                  repo.ClusterTidbPoolRepo
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade
	accountFacade              facade.AccountFacade
	mgmtFacade                 facade.MgmtSvcFacade
	rpc.UnimplementedClusterServiceServer
}

func NewClusterRpcService(
	billingSvc service.BillingService,
	clusterRepo repo.ClusterRepo,
	eventRepo repo.ClusterTidbPoolRepo,
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
	accountFacade facade.AccountFacade,
	mgmtFacade facade.MgmtSvcFacade,
) rpc.ClusterServiceServer {
	return &clusterRpcService{
		billingSvc:                 billingSvc,
		clusterRepo:                clusterRepo,
		eventRepo:                  eventRepo,
		clusterProxyProviderFacade: clusterProxyProviderFacade,
		accountFacade:              accountFacade,
		mgmtFacade:                 mgmtFacade,
	}
}

func (s *clusterRpcService) ListClusters(
	ctx context.Context,
	req *rpc.ListClustersReq,
) (_ *rpc.ListClustersResp, err error) {
	ctx, span := Tracer.Start(ctx, "clusterRpcService.ListClusters")
	defer func() { tracing.EndSpan(span, err) }()

	pager := utils.NewPagerNumberPager()
	err = pager.Normalize(req)
	if err != nil {
		return nil, err
	}

	span.AddEvent("list clusters in page")

	filter := ConvertClusterFilterPb2En(req.Filter)
	total, err := s.clusterRepo.CountClusters(ctx, filter)
	if err != nil {
		log.WithContextL(ctx).Error("failed to count branches", zap.Error(err))
		return nil, err
	}

	clusters, err := s.clusterRepo.ListClustersInPage(ctx, filter, pager.Page(), pager.PageSize())
	if err != nil {
		log.WithContextL(ctx).Error("failed to list clusters in page", zap.Error(err))
		return nil, err
	}
	nextPageToken := pager.NextPageToken(uint32(total))
	resp := &rpc.ListClustersResp{
		Total:         uint32(total),
		NextPageToken: nextPageToken,
	}
	for _, cluster := range clusters {
		resp.Clusters = append(resp.Clusters, ConvertClusterEn2Pb(cluster))
	}
	return resp, nil
}

func (s *clusterRpcService) GetCluster(ctx context.Context, req *rpc.GetClusterReq) (_ *rpc.GetClusterResp, err error) {
	ctx, span := Tracer.Start(ctx, "clusterRpcService.ListClusters")
	defer func() { tracing.EndSpan(span, err) }()
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", req.ClusterId))

	span.AddEvent("get cluster")
	cluster, err := s.clusterRepo.GetClusterByID(ctx, req.ClusterId)
	if err != nil {
		logger.Error("failed to get cluster", zap.Error(err))
		return nil, err
	}

	span.AddEvent("cluster cost explorer from billing")
	costV2, err := s.billingSvc.CalculateClusterCostThisMonth(
		ctx,
		cluster.TenantID,
		strconv.FormatUint(cluster.ID, 10),
		constant.ClusterTypeCluster,
		nil,
		"",
		"",
	)
	if err != nil {
		return nil, err
	}

	res := s.billingSvc.GetCostExplorerServerlessSkuCost(
		costV2,
	)
	return Convert2GetResp(
		cluster,
		res.RuFreeCost.Add(res.RuSpentCost),
		res.RowStoreFreeCost.Add(res.RowStoreSpentCost),
	), nil
}

func (s *clusterRpcService) ListSpendLimitEvents(
	ctx context.Context,
	req *rpc.ListSpendLimitEventsReq,
) (_ *rpc.ListSpendLimitEventsResp, err error) {
	ctx, span := Tracer.Start(ctx, "clusterRpcService.ListSpendLimitEvents")
	defer func() { tracing.EndSpan(span, err) }()
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", req.ClusterId))

	startTime, endTime := time.Now(), time.Now()
	if req.StartTime != nil {
		startTime = (*req.StartTime).AsTime()
	} else {
		startTime = startTime.AddDate(0, 0, -startTime.Day()+1).Truncate(24 * time.Hour)
	}

	if req.EndTime != nil {
		endTime = (*req.EndTime).AsTime()
	}

	span.AddEvent("list Spending Limit events")
	events, err := s.eventRepo.ListSpendLimitEvents(ctx, req.ClusterId, startTime, endTime)
	if err != nil {
		logger.Error("failed to list Spending Limit events", zap.Error(err))
		return nil, err
	}

	resp := &rpc.ListSpendLimitEventsResp{}
	for _, event := range events {
		resp.Events = append(resp.Events, ConvertSpendLimitEventEn2Pb(event))
	}
	return resp, nil
}

func (s *clusterRpcService) ListDatabases(
	ctx context.Context,
	req *rpc.ListDatabasesReq,
) (_ *rpc.ListDatabasesResp, err error) {
	ctx, span := Tracer.Start(ctx, "clusterRpcService.ListDatabases")
	defer func() { tracing.EndSpan(span, err) }()
	logger := log.WithContextL(ctx).With(zap.Uint64("cluster_id", req.ClusterId))

	operator := &clusterproxy.ClusterOperator{
		Username:   req.Operator.Username,
		Credential: req.Operator.Credential,
	}
	switch req.Operator.AuthMethod {
	case rpc.DBUser_METHOD_PASSWORD:
		operator.AuthMethod = clusterproxy.AuthMethodPassword
	case rpc.DBUser_METHOD_TOKEN:
		operator.AuthMethod = clusterproxy.AuthMethodToken
	default:
		logger.Error("invalid auth method", zap.String("auth_method", string(req.Operator.AuthMethod)))
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid auth method")
	}

	cluster, err := s.mgmtFacade.GetCluster(ctx, req.ClusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		logger.Error("failed to get cluster", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcMgmt, "failed to get cluster")
	}
	proxy, err := s.clusterProxyProviderFacade.GetClusterProxy(ctx, cluster, operator)
	if err != nil {
		logger.Error("failed to get cluster proxy", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcClusterProxy, "failed to get cluster proxy")
	}

	res, err := proxy.QuerySQL(ctx, "SHOW DATABASES")
	if err != nil {
		logger.Error("failed to query sql", zap.Error(err))
		return nil, err
	}

	type DatabaseResult struct {
		Databases [][]byte `json:"Database"`
	}
	result := DatabaseResult{}
	err = json.Unmarshal(res, &result)
	if err != nil {
		logger.Error("failed to unmarshal json", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeJsonMarshal, "failed to unmarshal json")
	}
	var databases []string
	for _, database := range result.Databases {
		databases = append(databases, string(database))
	}

	return &rpc.ListDatabasesResp{
		Databases: databases,
		BaseResp:  &common.BaseResp{},
	}, nil
}

func (s *clusterRpcService) CreateDatabase(
	ctx context.Context,
	req *rpc.CreateDatabaseReq,
) (_ *rpc.CreateDatabaseResp, err error) {
	ctx, span := Tracer.Start(ctx, "clusterRpcService.CreateDatabase")
	defer func() { tracing.EndSpan(span, err) }()
	logger := log.WithContextL(ctx).
		With(zap.Uint64("cluster_id", req.ClusterId)).
		With(zap.String("database_name", req.Database))

	operator := &clusterproxy.ClusterOperator{
		Username:   req.Operator.Username,
		Credential: req.Operator.Credential,
	}
	switch req.Operator.AuthMethod {
	case rpc.DBUser_METHOD_PASSWORD:
		operator.AuthMethod = clusterproxy.AuthMethodPassword
	case rpc.DBUser_METHOD_TOKEN:
		operator.AuthMethod = clusterproxy.AuthMethodToken
	default:
		logger.Error("invalid auth method", zap.String("auth_method", string(req.Operator.AuthMethod)))
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid auth method")
	}

	cluster, err := s.mgmtFacade.GetCluster(ctx, req.ClusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		logger.Error("failed to get cluster", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcMgmt, "failed to get cluster")
	}
	proxy, err := s.clusterProxyProviderFacade.GetClusterProxy(ctx, cluster, operator)
	if err != nil {
		logger.Error("failed to get cluster proxy", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeSvcClusterProxy, "failed to get cluster proxy")
	}

	err = proxy.ExecuteSQL(ctx, fmt.Sprintf("CREATE DATABASE `%s`", utils.EscapeBackticks(req.Database)))
	if err != nil {
		logger.Error("failed to execute sql", zap.Error(err))
		return nil, errorcode.MarkClusterSQLErrorUserActionable(err)
	}

	return &rpc.CreateDatabaseResp{BaseResp: &common.BaseResp{}}, nil
}

func (s *clusterRpcService) CanCreateCluster(
	ctx context.Context,
	req *rpc.CanCreateClusterReq,
) (_ *rpc.CanCreateClusterResp, err error) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	logger.Info("CanCreateCluster", zap.Any("request param", req))
	defer func() {
		logger.Info("CanCreateCluster", zap.Any("result", err))
	}()

	/* Forbidden to create serverless clusters in CMEK-enabled projects */
	project, err := s.accountFacade.GetProject(ctx, req.ProjectId)
	if err != nil {
		return nil, bizerr.New(errorcode.ErrCodeSvcAccount, "%s", err.Error())
	}
	if project.AwsCmekEnabled {
		return nil, bizerr.New(errorcode.ErrCodeSvcAccount, "project %d cmek enabled", req.ProjectId)
	}

	if req.SpendLimit > 0 { /* for paid/scalable cluster */
		if err := service.ValidateSpendLimitMax(req.OrgId, req.SpendLimit); err != nil {
			return nil, err
		}
		/* Only an on_demand and poc account can create paid/scalable clusters */
		if !s.billingSvc.ValidTenantPlan(ctx, req.OrgId) {
			return nil, bizerr.New(errorcode.ErrCodeNoPaymentMethods, errorcode.ErrMsgNoPaymentMethods)
		}
	} else { /* for free cluster */
		/* Only 5 free clusters quota per account, check start */
		isPaid := false
		filter := &repo.ClusterFilter{
			TenantIds: []uint64{req.OrgId},
			IsPaid:    &isPaid,
		}
		freeCount, err := s.clusterRepo.CountClusters(ctx, filter)
		if err != nil {
			return nil, err
		}
		if freeCount > int64(len(req.ClusterIds)) {
			return nil, bizerr.New(errorcode.ErrCodeSvcCluster, errorcode.ErrMsgMissMatchClusterNumber)
		}
		if freeCount >= int64(config.Global().BizConfig.ServerlessFreeQuotaMaxCount) {
			return nil, bizerr.New(errorcode.ErrCodeServerlessReachMaxFreeNum, errorcode.ErrMsgServerlessReachFreeMaximumNum)
		}
		/* Only 5 free clusters quota per account, check end */

		/* Enable enhanced encryption only if creating a paid/scalable cluster */
		if req.EnhancedEncryptionEnabled {
			return nil, bizerr.New(errorcode.ErrCodeSvcCluster, errorcode.ErrMsgCheckAdvancedEncryption)
		}
	}

	return &rpc.CanCreateClusterResp{
		ClusterId: req.ClusterId,
		Allowed:   true,
	}, nil
}