package impl

import (
	"context"

	"cloud.google.com/go/storage"
	"github.com/crossplane/provider-gcp/pkg/util/googleapi"
	appconfig "github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/models/bucket"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	service_config "github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/pkgv2/bizerr"
)

const (
	defaultCloudStorageRole = "roles/storage.objectAdmin"
	serviceAccountPrefix    = "serviceAccount"
	gcpMemberSep            = ":"
)

const dualRegionLen = 2

type gcpBucketManagerImpl struct {
	cli *storage.Client
}

var _ bucket.GCPBucketManagerModel = (*gcpBucketManagerImpl)(nil)

func NewGCPBucketManagerModel() bucket.GCPBucketManagerModel {
	cli, err := storage.NewClient(context.Background())
	if err != nil {
		panic(err)
	}
	return &gcpBucketManagerImpl{cli: cli}
}

func (m *gcpBucketManagerImpl) CheckBucketIfExists(ctx context.Context, req *domain.CheckGCPBucketReq) bizerr.BizError {
	bkt := m.cli.Bucket(req.Name)
	_, err := bkt.Attrs(ctx)
	if err != nil {
		if err == storage.ErrBucketNotExist {
			return bizerr.New(appconfig.ErrCodeBucketsNotFound, "bucket not exists")
		}
		return bizerr.WrapError(err, "get bucket")
	}
	return nil
}

func (m *gcpBucketManagerImpl) CreateBucket(ctx context.Context, req *domain.CreateGCPBucketReq) bizerr.BizError {
	attr := &storage.BucketAttrs{
		Labels: req.Tags,
	}
	switch len(req.Regions) {
	case 0:
		return bizerr.New(appconfig.ErrCodeBucketRegionEmpty, "regions can't be empty")
	case 1:
		attr.Location = req.Regions[0]
	case dualRegionLen:
		bucketOption := service_config.Global().ServerConfig.BucketOptions
		dualRegionCandidateSet := bucketOption.GetGCPDualRegionCandidateSet(req.Regions[0])
		if _, ok := dualRegionCandidateSet[req.Regions[1]]; !ok {
			return bizerr.New(
				bizerr.ErrCodeParams, "regions %s,%s can't be paired", req.Regions[0], req.Regions[1])
		}

		if predefinedName := bucketOption.GetGCPDualRegionPredefinedName(
			req.Regions[0], req.Regions[0]); predefinedName != "" {
			// for 2 predefined regions, use the predefined dual region code instead of the two region codes
			attr.Location = predefinedName
		} else {
			// for normal dual regions
			attr.Location = bucketOption.GetGCPDualRegionLocationCode(req.Regions[0], req.Regions[0])
			attr.CustomPlacementConfig = &storage.CustomPlacementConfig{
				DataLocations: req.Regions,
			}
		}
	default:
		return bizerr.New(appconfig.ErrCodeBucketTooManyRegions, "regions count can't be greater than 2")
	}

	bkt := m.cli.Bucket(req.Name)
	if err := bkt.Create(ctx, req.ProjectID, attr); err != nil {
		return bizerr.WrapError(err, "create bucket")
	}
	return nil
}

func (m *gcpBucketManagerImpl) DeleteBucketIgnoreNotFound(
	ctx context.Context, req *domain.DeleteGCPBucketReq,
) bizerr.BizError {
	bkt := m.cli.Bucket(req.Name)
	if err := bkt.Delete(ctx); err != nil {
		//FIXME: if storage pkg fix the bug, restore the code from comment
		//if err == storage.ErrBucketNotExist {
		//	return nil
		//}
		if googleapi.IsErrorNotFound(err) {
			return nil
		}
		return bizerr.WrapError(err, "delete bucket")
	}
	return nil
}

func (m *gcpBucketManagerImpl) PutBucketTags(
	ctx context.Context, req *domain.GCPPutBucketTags,
) bizerr.BizError {
	bkt := m.cli.Bucket(req.Name)

	updateReq := storage.BucketAttrsToUpdate{}
	for k, v := range req.Tags {
		updateReq.SetLabel(k, v)
	}

	if _, err := bkt.Update(ctx, updateReq); err != nil {
		return bizerr.WrapError(err, "put bucket tags")
	}
	return nil
}

func (m *gcpBucketManagerImpl) AddAllowPolicyToBucket(
	ctx context.Context, req *domain.GCPAddAllowPolicyToBucketReq,
) (*domain.GCPAddAllowPolicyToBucketResp, bizerr.BizError) {
	bkt := m.cli.Bucket(req.BucketName)
	iam := bkt.IAM()
	policy, err := iam.Policy(ctx)
	if err != nil {
		return nil, bizerr.WrapError(err, "get bucket policy")
	}

	policyAdded := false
	for i := range req.IAMServiceAccounts {
		sa := m.buildServiceAccountMemberName(req.IAMServiceAccounts[i])
		if policy.HasRole(sa, defaultCloudStorageRole) {
			continue
		}
		policy.Add(sa, defaultCloudStorageRole)
		policyAdded = true
	}
	members := policy.Members(defaultCloudStorageRole)
	if !policyAdded {
		// it means policy already exists
		return &domain.GCPAddAllowPolicyToBucketResp{
			IAMServiceAccounts: members,
		}, nil
	}

	if err := iam.SetPolicy(ctx, policy); err != nil {
		return nil, bizerr.WrapError(err, "set bucket policy")
	}
	return &domain.GCPAddAllowPolicyToBucketResp{
		IAMServiceAccounts: members,
	}, nil
}

func (m *gcpBucketManagerImpl) buildServiceAccountMemberName(saName string) string {
	return serviceAccountPrefix + gcpMemberSep + saName
}

func (m *gcpBucketManagerImpl) DeleteAllowPolicyFromBucket(
	ctx context.Context, req *domain.GCPDeleteAllowPolicyFromBucketReq,
) (*domain.GCPDeleteAllowPolicyFromBucketResp, bizerr.BizError) {
	bkt := m.cli.Bucket(req.BucketName)
	iam := bkt.IAM()
	policy, err := iam.Policy(ctx)
	if err != nil {
		return nil, bizerr.WrapError(err, "get bucket policy")
	}

	policyDeleted := false
	for i := range req.IAMServiceAccounts {
		sa := m.buildServiceAccountMemberName(req.IAMServiceAccounts[i])
		if !policy.HasRole(sa, defaultCloudStorageRole) {
			continue
		}
		policy.Remove(sa, defaultCloudStorageRole)
		policyDeleted = true
	}
	members := policy.Members(defaultCloudStorageRole)
	if !policyDeleted {
		// it means policy doesn't exist
		return &domain.GCPDeleteAllowPolicyFromBucketResp{
			IAMServiceAccounts: members,
		}, nil
	}

	if err := iam.SetPolicy(ctx, policy); err != nil {
		return nil, bizerr.WrapError(err, "set bucket policy")
	}
	return &domain.GCPDeleteAllowPolicyFromBucketResp{
		IAMServiceAccounts: members,
	}, nil
}