package keyspacerole

import (
	"context"
	"fmt"
	"strings"

	dmpbv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/repository"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
	"google.golang.org/protobuf/types/known/emptypb"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

type server struct {
	dmpbv1.UnimplementedKeyspaceRoleServiceServer

	local      client.Client
	dataPlaneR repository.DataPlaneReader
	appBoxR    repository.APPBoxReader
	bucketR    repository.BucketReader
	config     rsconfig.RootConfig
}

func NewService(
	local client.Client,
	dataPlaneR repository.DataPlaneReader,
	appBoxR repository.APPBoxReader,
	bucketR repository.BucketReader,
	config rsconfig.RootConfig,
) dmpbv1.KeyspaceRoleServiceServer {
	return &server{
		local:      local,
		dataPlaneR: dataPlaneR,
		appBoxR:    appBoxR,
		bucketR:    bucketR,
		config:     config,
	}
}

func (s *server) CreateKeyspaceRole(ctx context.Context, req *dmpbv1.CreateKeyspaceRoleRequest) (*dmpbv1.KeyspaceRole, error) {
	if req.KeyspaceRole == nil {
		return nil, bizerr.New(bizerr.ParameterInvalid, "keyspace_role is required")
	}
	if err := s.validateProvider(); err != nil {
		return nil, err
	}
	importRoleName := strings.TrimSpace(s.config.BYOC.ImportRoleName)
	if importRoleName != "" {
		return s.createStaticKeyspaceRole(ctx, req, importRoleName)
	}
	if s.config.ShouldBypassDynamicKeyspaceRole() {
		return byocImportRoleNotConfiguredResponse(req.KeyspaceRole.GetKeyspaceName()), nil
	}

	spec, err := s.resolveSpec(ctx, req.DataplaneId, req.KeyspaceRole)
	if err != nil {
		return nil, err
	}
	if err := s.provider().apply(ctx, spec); err != nil {
		return nil, err
	}

	return s.getKeyspaceRole(ctx, req.DataplaneId, req.KeyspaceRole.KeyspaceName)
}

func (s *server) GetKeyspaceRole(ctx context.Context, req *dmpbv1.GetKeyspaceRoleRequest) (*dmpbv1.KeyspaceRole, error) {
	if err := s.validateProvider(); err != nil {
		return nil, err
	}
	if req.KeyspaceName == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "keyspace_name is required")
	}
	importRoleName := strings.TrimSpace(s.config.BYOC.ImportRoleName)
	if importRoleName != "" {
		return s.getStaticKeyspaceRole(ctx, req, importRoleName)
	}
	if s.config.ShouldBypassDynamicKeyspaceRole() {
		return byocImportRoleNotConfiguredResponse(req.GetKeyspaceName()), nil
	}
	name := resourceName(s.config.ResourceNamePrefix, req.KeyspaceName)
	if err := validateRoleName(name); err != nil {
		return nil, err
	}
	return s.getKeyspaceRole(ctx, req.DataplaneId, req.KeyspaceName)
}

func (s *server) DeleteKeyspaceRole(ctx context.Context, req *dmpbv1.DeleteKeyspaceRoleRequest) (*emptypb.Empty, error) {
	if err := s.validateProvider(); err != nil {
		return nil, err
	}
	if req.KeyspaceName == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "keyspace_name is required")
	}
	if s.config.ShouldBypassDynamicKeyspaceRole() {
		return &emptypb.Empty{}, nil
	}
	name := resourceName(s.config.ResourceNamePrefix, req.KeyspaceName)
	if err := validateRoleName(name); err != nil {
		return nil, err
	}
	if err := s.provider().delete(ctx, req.DataplaneId, name); err != nil {
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (s *server) resolveSpec(ctx context.Context, dataplaneID string, input *dmpbv1.KeyspaceRole) (*keyspaceRoleSpec, error) {
	if input.KeyspaceName == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "keyspace_role.keyspace_name is required")
	}
	roleName := resourceName(s.config.ResourceNamePrefix, input.KeyspaceName)
	if err := validateRoleName(roleName); err != nil {
		return nil, err
	}
	if input.BucketId == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "keyspace_role.bucket_id is required")
	}
	if input.ExternalId == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "keyspace_role.external_id is required")
	}
	allowedPrefix, err := validateAllowedPrefix(input.AllowedPrefix)
	if err != nil {
		return nil, err
	}

	dataplane, err := s.getDataPlane(ctx, dataplaneID)
	if err != nil {
		return nil, err
	}
	if dataplane.Spec.AccountRef.AccountID == "" {
		return nil, bizerr.New(bizerr.ResourceNotReady, "dataplane account id is not ready")
	}
	if dataplane.Spec.AccountRef.ProviderConfigName == "" {
		return nil, bizerr.New(bizerr.ResourceNotReady, "dataplane provider config is not ready")
	}

	appbox, err := s.getAPPBox(ctx, dataplane)
	if err != nil {
		return nil, err
	}
	if appbox.Status.IAM.NodeRoleARN == "" {
		return nil, bizerr.New(bizerr.ResourceNotReady, "node role arn is not ready")
	}
	if s.config.Provider == consts.ProviderTypeACK && appbox.Status.IAM.NodeRoleName == "" {
		return nil, bizerr.New(bizerr.ResourceNotReady, "node role name is not ready")
	}

	bucket, err := s.getBucket(ctx, dataplaneID, input.BucketId)
	if err != nil {
		return nil, err
	}
	if bucket.Status.Bucket == nil || bucket.Status.Bucket.Name == "" {
		return nil, bizerr.New(bizerr.ResourceNotReady, "bucket is not ready")
	}
	if bucket.Status.Phase != v1alpha1.BucketPhaseReady {
		return nil, bizerr.New(bizerr.ResourceNotReady, "bucket is not ready")
	}

	labels := util.MergeMaps(input.Tags, s.config.ResourceLabels, map[string]string{
		consts.DataplaneLabelKey:            dataplaneID,
		consts.BucketLabelKey:               input.BucketId,
		keyspaceNameLabelKey:                input.KeyspaceName,
		"tags.tidbcloud.com/cloud-provider": appbox.Spec.Provider,
	})
	annotations := map[string]string{
		keyspaceRoleBucketAnnotation: input.BucketId,
		keyspaceRolePrefixAnnotation: allowedPrefix,
		keyspaceRoleExtIDAnnotation:  input.ExternalId,
	}
	cloudTags := controllers.ConvertToCloudTags(labels, s.config.Provider)
	cloudTags = util.MergeMaps(cloudTags, s.config.AdditionalCloudTags)
	if isAWSProvider(s.config.Provider) {
		cloudTags = util.BuildAWSTagWithName(cloudTags, roleName)
	}

	return &keyspaceRoleSpec{
		dataplaneID:    dataplaneID,
		keyspaceName:   input.KeyspaceName,
		roleName:       roleName,
		rolePolicyName: rolePolicyName(roleName),
		accountID:      dataplane.Spec.AccountRef.AccountID,
		providerConfig: dataplane.Spec.AccountRef.ProviderConfigName,
		provider:       s.config.Provider,
		assumeRoleARN:  appbox.Status.IAM.NodeRoleARN,
		externalID:     input.ExternalId,
		bucketName:     bucket.Status.Bucket.Name,
		allowedPrefix:  allowedPrefix,
		labels:         labels,
		annotations:    annotations,
		cloudTags:      cloudTags,
	}, nil
}

func (s *server) createStaticKeyspaceRole(
	ctx context.Context,
	req *dmpbv1.CreateKeyspaceRoleRequest,
	importRoleName string,
) (*dmpbv1.KeyspaceRole, error) {
	if err := validateStaticCreateIdentity(req.KeyspaceRole); err != nil {
		return nil, err
	}
	spec, err := s.resolveStaticSpec(
		ctx,
		req.DataplaneId,
		req.KeyspaceRole.GetKeyspaceName(),
		req.KeyspaceRole.GetExternalId(),
		req.KeyspaceRole.GetBucketId(),
		req.KeyspaceRole.GetAllowedPrefix(),
		req.KeyspaceRole.GetTags(),
		importRoleName,
	)
	if err != nil {
		return nil, err
	}
	return staticKeyspaceRoleResponse(spec), nil
}

func validateStaticCreateIdentity(input *dmpbv1.KeyspaceRole) error {
	if input.GetKeyspaceName() == "" {
		return bizerr.New(bizerr.ParameterInvalid, "keyspace_role.keyspace_name is required")
	}
	if input.GetExternalId() == "" {
		return bizerr.New(bizerr.ParameterInvalid, "keyspace_role.external_id is required")
	}
	return nil
}

func byocImportRoleNotConfiguredResponse(keyspaceName string) *dmpbv1.KeyspaceRole {
	return &dmpbv1.KeyspaceRole{
		KeyspaceName: keyspaceName,
		Status:       dmpbv1.KeyspaceRole_Failed,
		Message:      "byoc.importRoleName is not configured",
	}
}

func (s *server) getStaticKeyspaceRole(
	ctx context.Context,
	req *dmpbv1.GetKeyspaceRoleRequest,
	importRoleName string,
) (*dmpbv1.KeyspaceRole, error) {
	externalID := strings.TrimSpace(req.GetExternalId())
	if externalID == "" {
		externalID = req.GetKeyspaceName()
	}
	spec, err := s.resolveStaticSpec(
		ctx,
		req.DataplaneId,
		req.GetKeyspaceName(),
		externalID,
		req.GetBucketId(),
		req.GetAllowedPrefix(),
		nil,
		importRoleName,
	)
	if err != nil {
		return nil, err
	}
	return staticKeyspaceRoleResponse(spec), nil
}

func (s *server) resolveStaticSpec(
	ctx context.Context,
	dataplaneID, keyspaceName, externalID, bucketID, allowedPrefix string,
	tags map[string]string,
	roleName string,
) (*keyspaceRoleSpec, error) {
	if keyspaceName == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "keyspace_name is required")
	}
	if externalID == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "external_id is required")
	}
	if bucketID == "" {
		return nil, bizerr.New(bizerr.ParameterInvalid, "bucket_id is required for static import role")
	}
	allowedPrefix, err := validateAllowedPrefix(allowedPrefix)
	if err != nil {
		return nil, err
	}

	dataplane, err := s.getDataPlane(ctx, dataplaneID)
	if err != nil {
		return nil, err
	}
	if dataplane.Spec.AccountRef.AccountID == "" {
		return nil, bizerr.New(bizerr.ResourceNotReady, "dataplane account id is not ready")
	}

	bucket, err := s.getBucket(ctx, dataplaneID, bucketID)
	if err != nil {
		return nil, err
	}
	if bucket.Status.Bucket == nil || bucket.Status.Bucket.Name == "" {
		return nil, bizerr.New(bizerr.ResourceNotReady, "bucket is not ready")
	}
	if bucket.Status.Phase != v1alpha1.BucketPhaseReady {
		return nil, bizerr.New(bizerr.ResourceNotReady, "bucket is not ready")
	}

	roleARN, err := s.staticRoleARN(dataplane.Spec.AccountRef.AccountID, roleName)
	if err != nil {
		return nil, err
	}
	labels := util.MergeMaps(tags, s.config.ResourceLabels, map[string]string{
		consts.DataplaneLabelKey: dataplaneID,
		consts.BucketLabelKey:    bucketID,
		keyspaceNameLabelKey:     keyspaceName,
	})
	annotations := map[string]string{
		keyspaceRoleBucketAnnotation: bucketID,
		keyspaceRolePrefixAnnotation: allowedPrefix,
		keyspaceRoleExtIDAnnotation:  externalID,
	}

	return &keyspaceRoleSpec{
		dataplaneID:   dataplaneID,
		keyspaceName:  keyspaceName,
		roleName:      roleName,
		roleARN:       roleARN,
		externalID:    externalID,
		bucketName:    bucket.Status.Bucket.Name,
		allowedPrefix: allowedPrefix,
		labels:        labels,
		annotations:   annotations,
	}, nil
}

func (s *server) staticRoleARN(accountID, roleName string) (string, error) {
	switch s.config.Provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
		return util.GetRoleARN(accountID, roleName), nil
	case consts.ProviderTypeACK:
		return util.GetRAMRoleARN(accountID, roleName), nil
	default:
		return "", bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("unsupported provider: %s", s.config.Provider))
	}
}

func staticKeyspaceRoleResponse(spec *keyspaceRoleSpec) *dmpbv1.KeyspaceRole {
	return keyspaceRoleResponse(
		spec.dataplaneID,
		spec.keyspaceName,
		spec.roleARN,
		spec.labels,
		spec.annotations,
		dmpbv1.KeyspaceRole_Ready,
		"",
	)
}

func (s *server) getDataPlane(ctx context.Context, dataplaneID string) (*v1alpha1.DataPlane, error) {
	dataplane := &v1alpha1.DataPlane{}
	if err := s.dataPlaneR.Get(ctx, types.NamespacedName{
		Namespace: s.config.Namespace,
		Name:      dataplaneID,
	}, dataplane); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "dataplane not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get dataplane error", err)
	}
	return dataplane, nil
}

func (s *server) getAPPBox(ctx context.Context, dataplane *v1alpha1.DataPlane) (*v1alpha1.APPBox, error) {
	if dataplane.Status.AppBox == nil || dataplane.Status.AppBox.Name == "" {
		return nil, bizerr.New(bizerr.ResourceNotReady, "appbox reference is not ready")
	}
	appbox := &v1alpha1.APPBox{}
	if err := s.appBoxR.Get(ctx, types.NamespacedName{
		Namespace: s.config.Namespace,
		Name:      dataplane.Status.AppBox.Name,
	}, appbox); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "appbox not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get appbox error", err)
	}
	return appbox, nil
}

func (s *server) getBucket(ctx context.Context, dataplaneID, bucketID string) (*v1alpha1.Bucket, error) {
	bucket := &v1alpha1.Bucket{}
	if err := s.bucketR.Get(ctx, types.NamespacedName{
		Namespace: s.config.Namespace,
		Name:      fmt.Sprintf("%s-%s", dataplaneID, bucketID),
	}, bucket); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "bucket not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get bucket error", err)
	}
	return bucket, nil
}

func (s *server) getKeyspaceRole(ctx context.Context, dataplaneID, keyspaceName string) (*dmpbv1.KeyspaceRole, error) {
	roleName := resourceName(s.config.ResourceNamePrefix, keyspaceName)
	return s.provider().get(ctx, dataplaneID, keyspaceName, roleName)
}

func (s *server) validateProvider() error {
	switch s.config.Provider {
	case consts.ProviderTypeAWS, consts.ProviderTypeEKS, consts.ProviderTypeACK:
		return nil
	default:
		return bizerr.New(bizerr.ParameterInvalid, fmt.Sprintf("unsupported provider: %s", s.config.Provider))
	}
}