package branch

import (
	"context"
	"fmt"
	"sync"

	"github.com/tidbcloud/ecosystem-service/config"
	"github.com/tidbcloud/ecosystem-service/internal/service/grpc/branch/client"
	"github.com/tidbcloud/ecosystem-service/internal/utils"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"github.com/tidbcloud/tiinf/monitor/tilog"

	"go.opentelemetry.io/otel"
)

var Tracer = otel.Tracer("grpc-branch")

const (
	InternalErrorMessage = "Internal error"
)

var errorMessages = map[bizerr.ErrCode]string{
	61200007:                     "Branch is not ready, can not perform actions on it.",
	61200009:                     "Cluster is currently not available.",
	61200010:                     "Each organization can create up to 5 branches across all clusters.",
	61200011:                     "Creating branches is not allowed for clusters that reach the spending limit and are experiencing throttling.",
	61200013:                     "The branch name you're trying to use is reserved by TiDB Cloud for future purposes.",
	61200014:                     "Clusters created before 2023-07-05 can not create branches.",
	61200015:                     "The branch name should not be empty or longer than 64 characters.",
	61200016:                     "Branch not belong to the cluster. Please check your cluster id.",
	61200017:                     "Creating branches from a cluster larger than 100 GiB is not supported.",
	bizerr.ErrCodeRecordExist:    "The branch name you're trying to use is already used.",
	bizerr.ErrCodeRecordNotFound: "The branch can not be find.",
}

type BranchSvc interface {
	ListBranches(ctx context.Context, clusterId uint64) ([]*rpc.BasicBranch, bizerr.BizError)
	CreateBranch(ctx context.Context, clusterId uint64, parentId, displayName string, annotation map[string]string) (*string, bizerr.BizError)
	GetBranch(ctx context.Context, clusterId uint64, branchId string) (*rpc.FullBranch, bizerr.BizError)
	GetBranchByNameAndAnnotation(ctx context.Context, clusterId uint64, branchName, annotationKey, annotationValue string) (*rpc.FullBranch, bizerr.BizError)
	DeleteBranch(ctx context.Context, clusterId uint64, branchId string) bizerr.BizError
	DeleteBranchByNameAndAnnotation(ctx context.Context, clusterId uint64, branchName string, annotationKey, annotationValue string) bizerr.BizError
	IsFinalStatus(status rpc.BranchStatus) bool
	CreateBranchUser(ctx context.Context, branchID string) (user *rpc.CreateUserResp, bizErr bizerr.BizError)
	ResetBranch(ctx context.Context, clusterId uint64, branchId string) bizerr.BizError
}

var (
	branchSvcOnce sync.Once
	branchSvc     BranchSvc
)

type branchSvcImpl struct {
	client     rpc.BranchServiceClient
	userClient rpc.UserServiceClient
}

// NewBranchSvc needs to be called at runtime since it needs the grpc_addr configuration initialized at runtime
func NewBranchSvc(conf *config.ConfigWrapper) BranchSvc {
	branchSvcOnce.Do(func() {
		branchSvc = &branchSvcImpl{
			client:     client.GetBranchSvcClient(conf),
			userClient: client.GetBranchUserSvcClient(conf),
		}
	})
	return branchSvc
}

func (b *branchSvcImpl) ListBranches(ctx context.Context, clusterId uint64) (_ []*rpc.BasicBranch, bizErr bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "branchSvcImpl.ListBranches")
	defer func() {
		tracing.EndSpan(span, bizErr)
		bizErr = PrepareErrorMessage(bizErr)
	}()

	req := &rpc.ListBranchesReq{
		ClusterId: clusterId,
	}
	resp, err := b.client.ListBranches(ctx, req)
	if err != nil {
		return nil, utils.WrapGrpcError(err)
	}
	if resp != nil && resp.BaseResp != nil && resp.BaseResp.ErrCode != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	return resp.Branches, nil
}

func (b *branchSvcImpl) CreateBranch(ctx context.Context, clusterId uint64, parentId, displayName string, annotation map[string]string) (_ *string, bizErr bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "branchSvcImpl.CreateBranch")
	defer func() {
		tracing.EndSpan(span, bizErr)
		bizErr = PrepareErrorMessage(bizErr)
	}()

	req := &rpc.CreateBranchReq{
		ClusterId:   clusterId,
		ParentId:    parentId,
		DisplayName: displayName,
		Annotation:  annotation,
	}
	resp, err := b.client.CreateBranch(ctx, req)
	if err != nil {
		return nil, utils.WrapGrpcError(err)
	}
	if resp != nil && resp.BaseResp != nil && resp.BaseResp.ErrCode != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	return &resp.Id, nil
}

func (b *branchSvcImpl) GetBranch(ctx context.Context, clusterId uint64, branchId string) (_ *rpc.FullBranch, bizErr bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "branchSvcImpl.GetBranch")
	defer func() {
		tracing.EndSpan(span, bizErr)
		bizErr = PrepareErrorMessage(bizErr)
	}()

	req := &rpc.GetBranchReq{
		ClusterId: clusterId,
		BranchId:  branchId,
	}
	resp, err := b.client.GetBranch(ctx, req)
	if err != nil {
		return nil, utils.WrapGrpcError(err)
	}
	if resp != nil && resp.BaseResp != nil && resp.BaseResp.ErrCode != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	return resp.Branch, nil
}

func (b *branchSvcImpl) GetBranchByNameAndAnnotation(ctx context.Context, clusterId uint64, branchName, annotationKey, annotationValue string) (_ *rpc.FullBranch, bizErr bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "branchSvcImpl.GetBranchByName")
	defer func() {
		tracing.EndSpan(span, bizErr)
		bizErr = PrepareErrorMessage(bizErr)
	}()

	branches, err := b.ListBranches(ctx, clusterId)
	if err != nil {
		return nil, utils.WrapGrpcError(err)
	}
	tilog.Debug(ctx, fmt.Sprintf("branches: %v", branches))
	for _, branch := range branches {
		if branch.DisplayName == branchName && branch.Annotations[annotationKey] == annotationValue {
			return b.GetBranch(ctx, clusterId, branch.Id)
		}
	}
	tilog.Debug(ctx, fmt.Sprintf("clusterid: %d, branchname: %s, branches: %v", clusterId, branchName, branches))
	return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "")
}

func (b *branchSvcImpl) DeleteBranch(ctx context.Context, clusterId uint64, branchId string) (bizErr bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "branchSvcImpl.DeleteBranch")
	defer func() {
		tracing.EndSpan(span, bizErr)
		bizErr = PrepareErrorMessage(bizErr)
	}()

	req := &rpc.DeleteBranchReq{
		ClusterId: clusterId,
		BranchId:  branchId,
	}
	resp, err := b.client.DeleteBranch(ctx, req)
	if err != nil {
		return utils.WrapGrpcError(err)
	}
	if resp != nil && resp.BaseResp != nil && resp.BaseResp.ErrCode != 0 {
		return bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	return nil
}

func (b *branchSvcImpl) DeleteBranchByNameAndAnnotation(ctx context.Context, clusterId uint64, branchName string, annotationKey, annotationValue string) (bizErr bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "branchSvcImpl.DeleteBranchByName")
	defer func() {
		tracing.EndSpan(span, bizErr)
		bizErr = PrepareErrorMessage(bizErr)
	}()

	branch, err := b.GetBranchByNameAndAnnotation(ctx, clusterId, branchName, annotationKey, annotationValue)
	if err != nil {
		return utils.WrapGrpcError(err)
	}
	return b.DeleteBranch(ctx, clusterId, branch.Id)
}

func (b *branchSvcImpl) CreateBranchUser(ctx context.Context, branchID string) (user *rpc.CreateUserResp, bizErr bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "branchSvcImpl.CreateBranchUser")
	defer func() {
		tracing.EndSpan(span, bizErr)
		bizErr = PrepareErrorMessage(bizErr)
	}()

	resp, err := b.userClient.CreateUser(ctx, &rpc.CreateUserReq{
		BranchId: branchID,
	})
	if err != nil {
		return nil, utils.WrapGrpcError(err)
	}
	if resp != nil && resp.BaseResp != nil && resp.BaseResp.ErrCode != 0 {
		return nil, bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}

	return resp, nil
}

func (b *branchSvcImpl) IsFinalStatus(status rpc.BranchStatus) bool {
	return status == rpc.BranchStatus_DELETED || status == rpc.BranchStatus_ACTIVE
}

func PrepareErrorMessage(err error) bizerr.BizError {
	if err == nil {
		return nil
	}
	be, ok := err.(*bizerr.BizErr)
	if !ok {
		return bizerr.New(bizerr.ErrCodeUnknown, InternalErrorMessage)
	}
	if bizerr.IsInternalErr(be.Code) {
		return bizerr.New(bizerr.ErrCodeUnknown, InternalErrorMessage)
	}
	msg, ok := errorMessages[be.Code]
	if ok {
		be.Message = msg
	} else {
		be.Message = InternalErrorMessage
	}
	return be
}

func (b *branchSvcImpl) ResetBranch(ctx context.Context, clusterId uint64, branchId string) (bizErr bizerr.BizError) {
	ctx, span := Tracer.Start(ctx, "branchSvcImpl.ResetBranch")
	defer func() {
		tracing.EndSpan(span, bizErr)
		bizErr = PrepareErrorMessage(bizErr)
	}()

	req := &rpc.ResetBranchReq{
		ClusterId: clusterId,
		BranchId:  branchId,
	}
	resp, err := b.client.ResetBranch(ctx, req)
	if err != nil {
		return utils.WrapGrpcError(err)
	}
	if resp != nil && resp.BaseResp != nil && resp.BaseResp.ErrCode != 0 {
		return bizerr.Restore(bizerr.ErrCode(resp.BaseResp.ErrCode), resp.BaseResp.ErrMsg)
	}
	return nil
}