package branch_svc

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/password"
	"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/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"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
)

var rpcTracer = otel.Tracer("branch.rpc")

type branchRpcService struct {
	bs                         service.BranchService
	billingService             service.BillingService
	ur                         *service.UserRoleService
	branchRepo                 repo.BranchRepo
	infraFacade                facade.InfraFacade
	encryptorFacade            facade.EncryptorFacade
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade
	rpc.UnimplementedBranchServiceServer
	rpc.UnimplementedUserServiceServer
}

func NewBranchRpcService(
	service service.BranchService,
	billingService service.BillingService,
	ur *service.UserRoleService,
	branchRepo repo.BranchRepo,
	infraFacade facade.InfraFacade,
	encryptorFacade facade.EncryptorFacade,
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
) rpc.BranchServiceServer {
	return &branchRpcService{
		bs:                         service,
		billingService:             billingService,
		ur:                         ur,
		branchRepo:                 branchRepo,
		infraFacade:                infraFacade,
		encryptorFacade:            encryptorFacade,
		clusterProxyProviderFacade: clusterProxyProviderFacade,
	}
}

// NewBranchUserRpcService delete this after sql user is officially supported
func NewBranchUserRpcService(
	service service.BranchService,
	billingService service.BillingService,
	ur *service.UserRoleService,
	branchRepo repo.BranchRepo,
	infraFacade facade.InfraFacade,
	encryptorFacade facade.EncryptorFacade,
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
) rpc.UserServiceServer {
	return &branchRpcService{
		bs:                         service,
		billingService:             billingService,
		ur:                         ur,
		branchRepo:                 branchRepo,
		infraFacade:                infraFacade,
		encryptorFacade:            encryptorFacade,
		clusterProxyProviderFacade: clusterProxyProviderFacade,
	}
}

func (s *branchRpcService) ListBranches(
	ctx context.Context,
	req *rpc.ListBranchesReq,
) (resp *rpc.ListBranchesResp, err error) {
	start := time.Now()
	ctx, span := rpcTracer.Start(ctx, "branchRpcService.ListBranches")
	defer func() {
		tracing.EndSpan(span, err)
		CollectBranchMetrics(MetricTypeRPC, MetricMethodListBranches, start, err)
		err = errorcode.PrepareBranchErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("main_branch_id", fmt.Sprint(req.ClusterId)),
	)

	filter := req.GetFilter()
	var deleteAfterTime *time.Time
	if filter != nil && filter.GetIncludeDeleted() {
		if filter.GetDeleteAfterTime() == nil {
			// default deleteAfterTime is 3 days ago
			t := time.Now().Add(-time.Hour * 24 * 3)
			deleteAfterTime = &t
		} else {
			t := filter.GetDeleteAfterTime().AsTime()
			deleteAfterTime = &t
		}
		span.SetAttributes(attribute.String("include_deleted", "true"))
		span.SetAttributes(attribute.String("delete_after_time", deleteAfterTime.String()))
	}

	span.AddEvent("call branch_svc.ListBranches")
	branches, err := s.bs.ListBranches(ctx, fmt.Sprint(req.ClusterId), deleteAfterTime)
	if err != nil {
		return nil, err
	}

	brs := make([]*rpc.BasicBranch, 0, len(branches))
	for _, b := range branches {
		model, err := b.ConvertToRpcProtoBasic()
		if err != nil {
			return nil, err
		}

		brs = append(brs, model)
	}

	return &rpc.ListBranchesResp{
		Branches: brs,
		BaseResp: &common.BaseResp{},
	}, nil
}

func (s *branchRpcService) CreateBranch(
	ctx context.Context,
	req *rpc.CreateBranchReq,
) (resp *rpc.CreateBranchResp, err error) {
	start := time.Now()
	ctx, span := rpcTracer.Start(ctx, "branchRpcService.Save")
	defer func() {
		tracing.EndSpan(span, err)
		CollectBranchMetrics(MetricTypeRPC, MetricMethodCreateBranch, start, err)
		err = errorcode.PrepareBranchErrorMessage(ctx, err)
	}()

	clusterIdStr := strconv.FormatUint(req.ClusterId, 10)
	parentId, err := utils.GetBranchParentId(req.ParentId, clusterIdStr)
	if err != nil {
		return nil, bizerr.New(errorcode.ErrCodeInvalidBranchParentId, "invalid parent id")
	}

	span.SetAttributes(
		attribute.String("cluster_id", clusterIdStr),
		attribute.String("branch_parent_id", parentId),
		attribute.String("branch_display_name", req.DisplayName),
	)
	parentTimestamp := utils.ConvertTimestamp(req.ParentTimestamp)
	if parentTimestamp != nil {
		span.SetAttributes(
			attribute.String("parent_timestamp", parentTimestamp.String()),
		)
	}

	span.AddEvent("call branch_svc.CreateBranch")
	var createdById uint64 = constant.UnknownAuthorID
	if req.CreatedById != nil {
		createdById = *req.CreatedById
	}
	branch, err := s.bs.CreateBranch(
		ctx,
		clusterIdStr,
		req.DisplayName,
		parentId,
		parentTimestamp,
		"",
		createdById,
		req.Annotation,
	)
	if err != nil {
		return nil, err
	}

	return &rpc.CreateBranchResp{
		Id:       branch.ID,
		BaseResp: &common.BaseResp{},
	}, nil
}

func (s *branchRpcService) GetBranch(ctx context.Context, req *rpc.GetBranchReq) (resp *rpc.GetBranchResp, err error) {
	start := time.Now()
	ctx, span := rpcTracer.Start(ctx, "branchRpcService.GetBranch")
	defer func() {
		tracing.EndSpan(span, err)
		CollectBranchMetrics(MetricTypeRPC, MetricMethodGetBranch, start, err)
		err = errorcode.PrepareBranchErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("branch_root_id", fmt.Sprint(req.ClusterId)),
		attribute.String("branch_id", req.BranchId),
	)

	span.AddEvent("call branch_svc.GetBranch")
	b, err := s.bs.GetBranchById(ctx, fmt.Sprint(req.ClusterId), req.BranchId)
	if err != nil {
		return nil, err
	}

	r, err := b.ConvertToRpcProtoFull()
	if err != nil {
		return nil, err
	}

	return &rpc.GetBranchResp{
		Branch:   r,
		BaseResp: &common.BaseResp{},
	}, nil
}

func (s *branchRpcService) DeleteBranch(
	ctx context.Context,
	req *rpc.DeleteBranchReq,
) (resp *rpc.DeleteBranchResp, err error) {
	start := time.Now()
	ctx, span := rpcTracer.Start(ctx, "branchRpcService.DeleteBranch")
	defer func() {
		tracing.EndSpan(span, err)
		CollectBranchMetrics(MetricTypeRPC, MetricMethodDeleteBranch, start, err)
		err = errorcode.PrepareBranchErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("branch_root_id", fmt.Sprint(req.ClusterId)),
		attribute.String("branch_id", req.BranchId),
	)

	span.AddEvent("call branch_svc.DeleteBranch")
	var userID uint64 = constant.UnknownAuthorID
	if req.DeletedById != nil {
		userID = *req.DeletedById
	}
	_, err = s.bs.DeleteBranchById(ctx, fmt.Sprint(req.ClusterId), req.BranchId, userID)
	if err != nil {
		return nil, err
	}

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

func (s *branchRpcService) CreateDBUser(
	ctx context.Context,
	req *rpc.CreateDBUserReq,
) (resp *rpc.CreateDBUserResp, err error) {
	start := time.Now()
	ctx, span := rpcTracer.Start(ctx, "branchRpcService.CreateDBUser")
	defer func() {
		tracing.EndSpan(span, err)
		CollectBranchMetrics(MetricTypeRPC, MetricMethodCreateDBUser, start, err)
	}()
	span.SetAttributes(
		attribute.String("branch_root_id", fmt.Sprint(req.ClusterId)),
		attribute.String("branch_id", req.BranchId),
	)

	span.AddEvent("call branch_svc.CreateDBUser")

	operatorAuthMethod, err := service.ConvertAuthMethod(req.Operator.AuthMethod)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeGRPC, "unknown operator auth method")
	}
	operator := &service.SqlOperator{
		Username:   req.Operator.Username,
		AuthMethod: operatorAuthMethod,
		Credential: req.Operator.Credential,
	}

	b, err := s.branchRepo.GetBranchByID(ctx, req.BranchId)
	if err != nil {
		return nil, err
	}

	userType, err := service.ConvertDBUserType(req.UserType)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeGRPC, "unknown user type")
	}
	authMethod, err := service.ConvertAuthMethod(req.AuthMethod)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeGRPC, "unknown auth method")
	}
	roles, err := service.ConvertRoles(req.BuiltinRole, req.CustomRoles)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeGRPC, "unknown role")
	}
	userInfo := &service.UserInfo{
		Username:   req.Username,
		UserType:   userType,
		AuthMethod: authMethod,
		Roles:      roles,
	}
	if userType == constant.SqlUserTypeDelegate {
		userInfo.UserEmail = *req.Email
	}
	if authMethod == constant.AuthMethodPassword {
		userInfo.Credential = *req.Credential
	}

	err = s.ur.CreateUser(ctx, operator, b, userInfo)
	if err != nil {
		return nil, err
	}
	return &rpc.CreateDBUserResp{BaseResp: &common.BaseResp{}}, nil
}

func (s *branchRpcService) DeleteDBUser(
	ctx context.Context,
	req *rpc.DeleteDBUserReq,
) (resp *rpc.DeleteDBUserResp, err error) {
	start := time.Now()
	ctx, span := rpcTracer.Start(ctx, "branchRpcService.DeleteDBUser")
	defer func() {
		tracing.EndSpan(span, err)
		CollectBranchMetrics(MetricTypeRPC, MetricMethodDeleteDBUser, start, err)
	}()
	span.SetAttributes(
		attribute.String("branch_root_id", fmt.Sprint(req.ClusterId)),
		attribute.String("branch_id", req.BranchId),
	)

	span.AddEvent("call branch_svc.DeleteDBUser")

	userInfo := &service.UserInfo{Username: req.Username}

	b, err := s.branchRepo.GetBranchByID(ctx, req.BranchId)
	if err != nil {
		return nil, err
	}

	operatorAuthMethod, err := service.ConvertAuthMethod(req.Operator.AuthMethod)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeGRPC, "unknown operator auth method")
	}
	operator := &service.SqlOperator{
		Username:   req.Operator.Username,
		AuthMethod: operatorAuthMethod,
		Credential: req.Operator.Credential,
	}

	err = s.ur.DeleteUser(ctx, operator, b, userInfo)
	if err != nil {
		return nil, err
	}
	return &rpc.DeleteDBUserResp{BaseResp: &common.BaseResp{}}, nil
}

func (s *branchRpcService) CreateUser(
	ctx context.Context,
	req *rpc.CreateUserReq,
) (resp *rpc.CreateUserResp, err error) {
	ctx, span := rpcTracer.Start(ctx, "branchRpcService.CreateUser")
	defer func() { tracing.EndSpan(span, err) }()
	span.SetAttributes(
		attribute.String("branch_id", req.BranchId),
	)

	span.AddEvent("load branch entity")
	branch, err := s.branchRepo.GetBranchByID(ctx, req.BranchId)
	if err != nil {
		return nil, err
	}
	if branch.Status != constant.BranchStatusRunning {
		return nil, bizerr.New(errorcode.ErrCodeClusterNotRunning, "cannot create users on a not running branch")
	}

	span.AddEvent("load branch cr")
	branchCr, err := s.infraFacade.GetBranch(ctx, branch.ProjectID, branch.ID)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeSvcInfraAPI, "%s", err.Error())
	}
	status, err := branchCr.AwsFreeTierClusterStatusFromBranch()
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "%s", err.Error())
	}

	// TODO: Use delegate user of API Key to create users.
	cloudAdminPwd, err := s.encryptorFacade.DecryptBase64(branch.EncryptedCloudAdminPwd)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "%s", err.Error())
	}
	operator := &clusterproxy.ClusterOperator{
		Username:   constant.CloudAdminUsername,
		AuthMethod: clusterproxy.AuthMethodPassword,
		Credential: cloudAdminPwd,
	}
	proxy, err := s.clusterProxyProviderFacade.GetBranchProxy(ctx, branchCr, operator)
	if err != nil {
		return nil, bizerr.New(errorcode.ErrCodeSvcClusterProxy, "%s", err.Error())
	}
	username := status.KeyspaceToken + "." + utils.RandomString(8)
	pwd := password.Decorate(utils.RandomString(32))
	sqls := []string{
		fmt.Sprintf(`CREATE USER IF NOT EXISTS '%s' IDENTIFIED BY '%s';`, username, pwd),
		fmt.Sprintf(`INSERT INTO mysql.role_edges VALUES('%%','role_admin','%%','%s','N');`, username),
		fmt.Sprintf(`SET DEFAULT ROLE ALL TO '%s';`, username),
	}
	err = proxy.ExecuteMultiSQL(ctx, sqls)
	if err != nil {
		return nil, err
	}
	return &rpc.CreateUserResp{Username: username, Password: pwd, BaseResp: &common.BaseResp{}}, nil
}

func (s *branchRpcService) DeleteUser(
	ctx context.Context,
	req *rpc.DeleteUserReq,
) (resp *rpc.DeleteUserResp, err error) {
	ctx, span := rpcTracer.Start(ctx, "branchRpcService.DeleteUser")
	defer func() { tracing.EndSpan(span, err) }()
	span.SetAttributes(
		attribute.String("branch_id", req.BranchId),
		attribute.String("username", req.Username),
	)

	span.AddEvent("load branch entity")
	branch, err := s.branchRepo.GetBranchByID(ctx, req.BranchId)
	if err != nil {
		return nil, err
	}
	if branch.Status != constant.BranchStatusRunning {
		return nil, bizerr.New(errorcode.ErrCodeClusterNotRunning, "cannot delete users on a not running branch")
	}

	// TODO: Use delegate user of API Key to delete users.
	cloudAdminPwd, err := s.encryptorFacade.DecryptBase64(branch.EncryptedCloudAdminPwd)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "%s", err.Error())
	}
	operator := &service.SqlOperator{
		Username:   constant.CloudAdminUsername,
		AuthMethod: constant.AuthMethodPassword,
		Credential: cloudAdminPwd,
	}
	userInfo := &service.UserInfo{
		Username: req.Username,
	}
	span.AddEvent("delete sql user")
	err = s.ur.DeleteUser(ctx, operator, branch, userInfo)
	if err != nil {
		return nil, err
	}

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

func (s *branchRpcService) ResetBranch(
	ctx context.Context,
	req *rpc.ResetBranchReq,
) (resp *rpc.ResetBranchResp, err error) {
	start := time.Now()
	ctx, span := rpcTracer.Start(ctx, "branchRpcService.ResetBranch")
	defer func() {
		tracing.EndSpan(span, err)
		CollectBranchMetrics(MetricTypeRPC, MetricMethodResetBranch, start, err)
		err = errorcode.PrepareBranchErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("branch_root_id", fmt.Sprint(req.ClusterId)),
		attribute.String("branch_id", req.BranchId),
	)

	span.AddEvent("call branch_svc.ResetBranch")
	_, err = s.bs.ResetBranchById(ctx, fmt.Sprint(req.ClusterId), req.BranchId)
	if err != nil {
		return nil, err
	}

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