package vpc

import (
	"context"
	"errors"
	"fmt"
	"strconv"

	"github.com/samber/lo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"gorm.io/gorm"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	openapidedicatedv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/dedicated/v1beta1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/eventbus"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/repository"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/apimiddleware/bizerror"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/enum"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
)

func (s *serverImpl) CreateVpcPeering(
	ctx context.Context,
	req *openapidedicatedv1beta1.CreateVpcPeeringRequest,
) (*openapidedicatedv1beta1.VpcPeering, error) {
	ctx, span := tracer.Start(ctx, "CreateVpcPeering")
	defer span.End()

	var (
		err           error
		reqMeta       *metadata.RequestMeta
		region        string
		cloudProvider string
	)

	detailsFunc := func() auditlog.Details {
		details := eventbus.NewAuditLogDetails(eventbus.TierTypeDedicated)
		details[auditlog.KeyRegion] = region
		details[auditlog.KeyCloudProvider] = cloudProvider
		details[auditlog.KeyVpcCidr] = req.GetVpcPeering().GetCustomerVpcCidr()
		details[auditlog.KeyVpcId] = req.GetVpcPeering().GetCustomerVpcId()
		details[auditlog.KeyTidbCloudCidr] = req.GetVpcPeering().GetTidbCloudVpcCidr()
		switch cloudProvider {
		case openapicommonv1beta1.Region_aws.String():
			details[auditlog.KeyAwsAccountId] = req.GetVpcPeering().GetCustomerAccountId()
		case openapicommonv1beta1.Region_gcp.String():
			details[auditlog.KeyApplicationGcpProjectId] = req.GetVpcPeering().GetCustomerAccountId()
		case openapicommonv1beta1.Region_azure.String():
			// do nothing
		}
		return details
	}
	defer func() {
		switch cloudProvider {
		case openapicommonv1beta1.Region_aws.String():
			s.auditLogSender.SubmitAuditLog(ctx, auditlog.EventCreateAWSVPCPeering, reqMeta, detailsFunc, err)
		case openapicommonv1beta1.Region_gcp.String():
			s.auditLogSender.SubmitAuditLog(ctx, auditlog.EventCreateGCPVPCPeering, reqMeta, detailsFunc, err)
		case openapicommonv1beta1.Region_azure.String():
			tilog.Error(ctx, "Azure VPC peering is not supported CreateVPCPeering yet", zap.Error(err))
		}
	}()

	// Extract request meta.
	reqMeta, err = metadata.ExtractRequestMetaWithCheckPermission(ctx)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to extract request metadata")
	}

	uid, err := strconv.ParseUint(reqMeta.UserID, 10, 64)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid user id")
	}

	cloudProvider, region, err = common.ParseOpenApiRegionNameOrId(req.GetVpcPeering().GetTidbCloudRegionId())
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid region id")
	}
	globalProvider, err := enum.FromString[globalcommonv1.CloudProvider_Name](cloudProvider)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "provider %s is not supported", cloudProvider)
	}
	customerProvider, customerRegion, err := common.ParseOpenApiRegionNameOrId(
		req.GetVpcPeering().GetCustomerRegionId(),
	)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid region id")
	}
	if customerProvider != cloudProvider {
		return nil, bizerr.WrapErrorWithCode(
			err,
			bizerr.ErrCodeParams,
			"customer cloud provider and tidb cloud cloud provider must be the same",
		)
	}

	projectId := reqMeta.DefaultProjectID
	projectIdStr, ok := req.GetVpcPeering().GetLabels()[common.ProjectIDLabelKey]
	if ok {
		projectId, err = strconv.ParseUint(projectIdStr, 10, 64)
		if err != nil {
			return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid project id")
		}
	}

	if err := s.checkCustomStatusMutexByProjectId(ctx, projectId, tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_CREATE_VPC_PEERING); err != nil {
		return nil, err
	}

	globalReq := &globaldedicatedv1.CreateVpcPeeringRequest{
		Peering: &globaldedicatedv1.VpcPeering{
			ProjectId:         projectId,
			OrganizationId:    reqMeta.OrgID,
			CloudProvider:     globalcommonv1.CloudProvider_Name(globalProvider),
			Type:              globaldedicatedv1.VpcPeering_CUSTOMER,
			TidbCloudRegion:   region,
			CustomerRegion:    customerRegion,
			CustomerCidrBlock: req.GetVpcPeering().GetCustomerVpcCidr(),
		},
		OperatorUserId: uid,
	}

	switch cloudProvider {
	case openapicommonv1beta1.Region_aws.String():
		globalReq.Peering.Aws = &globaldedicatedv1.VpcPeering_Aws{
			CustomerAccountId: req.GetVpcPeering().GetCustomerAccountId(),
			CustomerVpcId:     req.GetVpcPeering().GetCustomerVpcId(),
		}
	case openapicommonv1beta1.Region_gcp.String():
		globalReq.Peering.Gcp = &globaldedicatedv1.VpcPeering_Gcp{
			CustomerProjectId:   req.GetVpcPeering().GetCustomerAccountId(),
			CustomerNetworkName: req.GetVpcPeering().GetCustomerVpcId(),
		}
	case openapicommonv1beta1.Region_azure.String():
		// TODO(@yifan): Wait azure vpc peering prd spec.
	}

	var resp *globaldedicatedv1.CreateVpcPeeringResponse
	resp, err = s.dedicatedCli.CreateVpcPeering(ctx, globalReq)
	if err != nil {
		return nil, err
	}

	// Retrieve the newly created VPC peering immediately.
	newPeering, err := s.dedicatedCli.GetVpcPeering(ctx, &globaldedicatedv1.GetVpcPeeringRequest{
		CloudProvider: globalcommonv1.CloudProvider_Name(globalProvider),
		VpcPeeringId:  resp.GetPeeringId(),
	})
	if err != nil {
		return nil, bizerror.NewInternal(err, "failed to retrieve the newly created vpc peering")
	}

	var openApiPeering *openapidedicatedv1beta1.VpcPeering
	openApiPeering, err = vpcPeeringGlobalToOpenApi(newPeering)
	if err != nil {
		return nil, bizerror.NewInternal(err, "failed to convert vpc peering")
	}

	return openApiPeering, nil
}

func (s *serverImpl) DeleteVpcPeering(
	ctx context.Context,
	req *openapidedicatedv1beta1.DeleteVpcPeeringRequest,
) (*emptypb.Empty, error) {
	ctx, span := tracer.Start(ctx, "DeleteVpcPeering")
	defer span.End()

	var (
		err           error
		reqMeta       *metadata.RequestMeta
		cloudProvider openapicommonv1beta1.Region_CloudProvider
	)

	detailsFunc := func() auditlog.Details {
		details := eventbus.NewAuditLogDetails(eventbus.TierTypeDedicated)
		details[auditlog.KeyCloudProvider] = cloudProvider
		details[auditlog.KeyPeeringId] = req.GetVpcPeeringId()
		return details
	}
	defer func() {
		switch cloudProvider {
		case openapicommonv1beta1.Region_aws:
			s.auditLogSender.SubmitAuditLog(ctx, auditlog.EventDeleteAWSVPCPeering, reqMeta, detailsFunc, err)
		case openapicommonv1beta1.Region_gcp:
			s.auditLogSender.SubmitAuditLog(ctx, auditlog.EventDeleteGCPVPCPeering, reqMeta, detailsFunc, err)
		case openapicommonv1beta1.Region_azure:
			tilog.Error(ctx, "Azure VPC peering is not DeleteVPCPeering supported yet", zap.Error(err))
		}
	}()

	// Extract request meta.
	reqMeta, err = metadata.ExtractRequestMetaWithCheckPermission(ctx)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to extract request metadata")
	}

	err = s.checkCustomStatusMutexByProjectId(
		ctx,
		reqMeta.DefaultProjectID,
		tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_DELETE_VPC_PEERING,
	)
	if err != nil {
		return nil, err
	}

	cloudProvider, peeringId, err := common.ParseVpcPeeringOpenApiId(req.GetVpcPeeringId())
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid vpc peering id")
	}

	_, err = s.dedicatedCli.DeleteVpcPeering(ctx, &globaldedicatedv1.DeleteVpcPeeringRequest{
		VpcPeeringId:  peeringId,
		CloudProvider: common.OpenApiProviderToGlobal(cloudProvider),
	})
	if err != nil {
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (s *serverImpl) ListVpcPeerings(
	ctx context.Context,
	req *openapidedicatedv1beta1.ListVpcPeeringsRequest,
) (*openapidedicatedv1beta1.ListVpcPeeringsResponse, error) {
	ctx, span := tracer.Start(ctx, "ListVpcPeerings")
	defer span.End()

	// Extract request meta.
	reqMeta, err := metadata.ExtractRequestMetaWithCheckPermission(ctx)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to extract request metadata")
	}

	projectId := reqMeta.DefaultProjectID
	if req.GetProjectId() != "" {
		projectId, err = strconv.ParseUint(req.GetProjectId(), 10, 64)
		if err != nil {
			return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid project id")
		}
	}

	globalReq := &globaldedicatedv1.ListVpcPeeringsRequest{
		ProjectIds: []uint64{projectId},
		Types: []globaldedicatedv1.VpcPeering_Type{
			globaldedicatedv1.VpcPeering_CUSTOMER,
			globaldedicatedv1.VpcPeering_CUSTOMER_READONLY,
		},
	}
	if req.GetCloudProvider() != openapicommonv1beta1.Region_CLOUD_PROVIDER_UNSPECIFIED {
		globalReq.CloudProviders = []globalcommonv1.CloudProvider_Name{
			common.OpenApiProviderToGlobal(req.CloudProvider),
		}
	}

	resp, err := s.dedicatedCli.ListVpcPeerings(ctx, globalReq)
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "failed to list vpc peerings")
	}

	// Paginate the results locally.
	pagedCorePeerings, nextPageToken, err := common.LocalPagination(
		resp.GetVpcPeerings(),
		req,
		common.ListDefaultPageSize,
		common.ListMaxPageSize,
	)
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to paginate vpc peerings")
	}

	// Convert the core vpc peerings to open api vpc peerings.
	openApiPeerings := make([]*openapidedicatedv1beta1.VpcPeering, 0, len(resp.GetVpcPeerings()))
	for _, c := range pagedCorePeerings {
		o, err := vpcPeeringGlobalToOpenApi(c)
		if err != nil {
			return nil, bizerror.NewInternal(err, "failed to convert vpc peering")
		}
		openApiPeerings = append(openApiPeerings, o)
	}

	return &openapidedicatedv1beta1.ListVpcPeeringsResponse{
		VpcPeerings:   openApiPeerings,
		TotalSize:     int32(len(resp.GetVpcPeerings())),
		NextPageToken: nextPageToken,
	}, nil
}

func (s *serverImpl) GetVpcPeering(
	ctx context.Context,
	req *openapidedicatedv1beta1.GetVpcPeeringRequest,
) (*openapidedicatedv1beta1.VpcPeering, error) {
	ctx, span := tracer.Start(ctx, "GetVpcPeering")
	defer span.End()

	// Extract request meta.
	_, err := metadata.ExtractRequestMetaWithCheckPermission(ctx)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to extract request metadata")
	}

	cloudProvider, peeringId, err := common.ParseVpcPeeringOpenApiId(req.GetVpcPeeringId())
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid vpc peering id")
	}

	resp, err := s.dedicatedCli.GetVpcPeering(ctx, &globaldedicatedv1.GetVpcPeeringRequest{
		CloudProvider: common.OpenApiProviderToGlobal(cloudProvider),
		VpcPeeringId:  peeringId,
	})
	if err != nil {
		return nil, bizerr.WrapGrpcError(err, "failed to get vpc peering")
	}

	openApiPeering, err := vpcPeeringGlobalToOpenApi(resp)
	if err != nil {
		return nil, bizerror.NewInternal(err, "failed to convert vpc peering")
	}

	return openApiPeering, nil
}

func (s *serverImpl) checkCustomStatusMutexByProjectId(
	ctx context.Context,
	projectId uint64,
	action tidbmgmtservicev1.ClusterAction,
) error {
	// check cluster custom status by project id
	var page = 1
	for {
		clusters, err := s.dedicateClusterSvc.ListClusters(ctx,
			&repository.ClusterFilter{
				ProjectIds: []uint64{projectId},
				View:       tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
			},
			&common.Page{
				Offset:   int64(int32(page-1) * common.ListDefaultPageSize),
				PageSize: int64(common.ListDefaultPageSize),
			},
		)

		if err != nil {
			return bizerror.NewInternal(err, "failed to list clusters by project id")
		}

		if len(clusters) == 0 {
			break
		}

		// check mutex custom status
		var isMutex bool
		var status *entity.ClusterCustomStatus
		for _, cluster := range clusters {
			isMutex, status, err = s.customStatusRepo.IsMutexClusterAction(ctx, cluster.ClusterId, action)
			if err != nil {
				if !errors.Is(err, gorm.ErrRecordNotFound) {
					err = fmt.Errorf("failed to check mutex cluster action: %w", err)
					return bizerror.NewInternal(err, err.Error())
				}
			}
			if isMutex {
				tilog.Error(
					ctx,
					"mutex custom status is exist",
					zap.Uint64("cluster_id", cluster.ClusterId),
					zap.Any("custom status", status),
				)
				return bizerror.NewForbiddenForClusterCustomStatus(
					constant.ErrClusterCustomStatusForbidden,
					status.StatusType.String(),
				)
			}
		}
		page++
	}

	return nil
}

func vpcPeeringGlobalToOpenApi(c *globaldedicatedv1.VpcPeering) (*openapidedicatedv1beta1.VpcPeering, error) {
	_, tidbCloudRegionId, p, err := common.BuildOpenApiRegionName(c.GetCloudProvider().String(), c.GetTidbCloudRegion())
	if err != nil {
		return nil, err
	}

	_, customerRegionId, _, err := common.BuildOpenApiRegionName(c.GetCloudProvider().String(), c.GetCustomerRegion())
	if err != nil {
		return nil, err
	}

	name, id := buildOpenApiId(p, c.GetId())

	o := &openapidedicatedv1beta1.VpcPeering{
		Name:                   name,
		VpcPeeringId:           id,
		Labels:                 map[string]string{common.ProjectIDLabelKey: strconv.FormatUint(c.GetProjectId(), 10)},
		TidbCloudRegionId:      tidbCloudRegionId,
		CustomerRegionId:       customerRegionId,
		CustomerVpcCidr:        c.GetCustomerCidrBlock(),
		TidbCloudCloudProvider: p,
		TidbCloudVpcCidr:       c.GetTidbCloudCidrBlock(),
		State:                  peeringStateGlobalToOpenApi(c.GetState()),
	}

	switch {
	case c.GetAws() != nil:
		o.CustomerAccountId = c.GetAws().GetCustomerAccountId()
		o.TidbCloudAccountId = c.GetAws().GetTidbCloudAccountId()
		o.CustomerVpcId = c.GetAws().GetCustomerVpcId()
		o.TidbCloudVpcId = c.GetAws().GetTidbCloudVpcId()
		o.AwsVpcPeeringConnectionId = lo.EmptyableToPtr(c.GetAws().GetVpcPeeringConnectionId())
	case c.GetGcp() != nil:
		o.CustomerAccountId = c.GetGcp().GetCustomerProjectId()
		o.TidbCloudAccountId = c.GetGcp().GetTidbCloudProjectId()
		o.CustomerVpcId = c.GetGcp().GetCustomerNetworkName()
		o.TidbCloudVpcId = c.GetGcp().GetTidbCloudNetworkName()
	// case c.GetAzure() != nil:
	// TODO(@yifan): Wait azure vpc peering prd spec.
	default:
	}

	return o, nil
}

func peeringStateGlobalToOpenApi(s globaldedicatedv1.VpcPeering_State) openapidedicatedv1beta1.VpcPeering_State {
	switch s {
	case globaldedicatedv1.VpcPeering_PENDING:
		return openapidedicatedv1beta1.VpcPeering_PENDING
	case globaldedicatedv1.VpcPeering_ACTIVE:
		return openapidedicatedv1beta1.VpcPeering_ACTIVE
	case globaldedicatedv1.VpcPeering_FAILED:
		return openapidedicatedv1beta1.VpcPeering_FAILED
	default:
		return openapidedicatedv1beta1.VpcPeering_STATE_UNSPECIFIED
	}
}

func buildOpenApiId(provider openapicommonv1beta1.Region_CloudProvider, coreId uint64) (name, id string) {
	id = fmt.Sprintf("%s-%d", provider, coreId)
	name = fmt.Sprintf("vpcPeerings/%s", id)
	return name, id
}