package gcp

import (
	"context"
	"fmt"
	"net/http"
	"reflect"
	"strings"
	"time"

	"cloud.google.com/go/iam"
	"cloud.google.com/go/storage"
	"github.com/hashicorp/go-multierror"
	"github.com/pingcap/errors"
	"google.golang.org/api/compute/v1"
	"google.golang.org/api/container/v1"
	"google.golang.org/api/dns/v1"
	"google.golang.org/api/googleapi"
	"google.golang.org/api/iterator"
	"google.golang.org/api/option"
)

const (
	globalNetworkURLTemplate         = "https://www.googleapis.com/compute/v1/projects/%s/global/networks/%s"
	globalAddressURLTemplate         = "https://www.googleapis.com/compute/v1/projects/%s/global/addresses/%s"
	regionsSubnetURLTemplate         = "https://www.googleapis.com/compute/v1/projects/%s/regions/%s/subnetworks/%s"
	regionsAddressURLTemplate        = "https://www.googleapis.com/compute/v1/projects/%s/regions/%s/addresses/%s"
	regionsForwardingRuleURLTemplate = "https://www.googleapis.com/compute/v1/projects/%s/regions/%s/forwardingRules/%s"

	// Reserve 192.168.11.0/24 for pcs ip addresses.
	// Refer to https://cloud.google.com/vpc/docs/configure-private-service-connect-apis#ip-address-requirements
	googleApiPscAddress   = "192.168.11.1"
	googleApiPscNamespace = "googleapipsc"

	operationWaitingTimeout = 1 * time.Minute
)

//go:generate mockgen -package gcp -source=client.go -destination=client_mock.go
type Client interface {
	AddBucketIAMMember(bucketName string, roleName iam.RoleName, member string) error
	CreatePSCEndpoint(ctx context.Context, project, region, network, subnet, target, name, namespace string, tags map[string]string) error
	DeleteAllPSCEndpoints(ctx context.Context, project, region, network, subnetwork string) error
	IsGoogleApiPscExists(ctx context.Context, project, network string) (bool, error)
	CreateGoogleApiPsc(ctx context.Context, project, region, network, name string, tags map[string]string) error
	DeleteGoogleApiPsc(ctx context.Context, project, name string) error
	GetInstanceGroupSize(ctx context.Context, project, zone, instanceGroup string) (int64, error)
	GetInstanceGroupManagerManagedInstanceCount(ctx context.Context, project, zone, instanceGroupManager string) (int64, error)
	DeleteGoogleApiPscDnsZone(ctx context.Context, project string) error
	DeleteAllGoogleApiPsc(ctx context.Context, project, network string) error

	// Instance management
	GetInstanceLabels(ctx context.Context, project, zone, instance string) (map[string]string, error)
	// AppendNodePoolResourceLabels append the actual vm labels belong to the nodepool
	// If a nil err and operation are returned, no actual update operation occurred
	//
	// DEPRECATED: Don't use this function because we will only update them once at creation time and never update resource labels again.
	// Please use UpdateNodePoolResourceLabels instead.
	AppendNodePoolResourceLabels(ctx context.Context, name string, resourceLabels map[string]string) (*container.Operation, error)
	UpdateNodePoolResourceLabels(ctx context.Context, name string, resourceLabels map[string]string) (*container.Operation, error)

	GetNodePoolResourceLabels(ctx context.Context, name string) (map[string]string, error)
	GetNodePoolNodeLabels(ctx context.Context, name string) (map[string]string, error)
	UpdateNodePoolNodeLabels(ctx context.Context, name string, labels map[string]string) (*container.Operation, error)
	GetNodePoolNodeTaints(ctx context.Context, name string) ([]*container.NodeTaint, error)
	UpdateNodePoolNodeTaints(ctx context.Context, name string, taints []*container.NodeTaint) (*container.Operation, error)
	UpdateNodePoolNodeLabelsAndTaints(ctx context.Context, name string, labels map[string]string, taints []*container.NodeTaint) (*container.Operation, error)

	ResourceManager() ResourceManagerClient
	ResourceTagManager() ResourceTagManager
	// EmptyBucket empty bucket.
	EmptyBucket(ctx context.Context, name string) error

	// IAM
	GetIAMPolicy3(ctx context.Context, resource *iam.Handle) (*iam.Policy3, error)

	// Storage
	StorageCli() *storage.Client
	NewStorageCli(ctx context.Context, opts ...option.ClientOption) (*storage.Client, error)

	// PSC management
	CreateOrUpdatePSCServiceAttachment(ctx context.Context, project, region, name, subnet, targetService, domainName string, acceptedProjects []string) error
	DeletePSCServiceAttachment(ctx context.Context, project, region, name string) error
	IsPSCServiceAttachmentExists(ctx context.Context, project, region, name string) (bool, error)
	GetPSCServiceAttachmentConnections(ctx context.Context, project, region, name string) ([]*compute.ServiceAttachmentConnectedEndpoint, error)

	// Subnetwork management
	ListSubnetworks(ctx context.Context, project, region, network string) ([]*compute.Subnetwork, error)
	CreatePSCSubnet(ctx context.Context, project, region, name, network, cidr string) error
	IsSubnetExists(ctx context.Context, project, region, name string) (bool, error)
	DeleteSubnet(ctx context.Context, project, region, name string) error
	PatchSubnet(ctx context.Context, sn *compute.Subnetwork, project, region string) error

	// GKE management
	UpgradeCluster(ctx context.Context, project, region, name, version string) (string, error)
	IsContainerOperationDone(ctx context.Context, name string) (bool, error)
	SetClusterMaintenanceExclusion(ctx context.Context, project, region, name, maintenanceExclusionName string, timeWindow *container.TimeWindow) (string, error)
	UpgradeNodePool(ctx context.Context, project, region, gkeName, nodePoolName, version string) error
	DisableManagedPrometheus(ctx context.Context, project, region, name string) (string, error)
	EnabelSubSetting(ctx context.Context, project, region, name string) error
	ModifyBackendServiceLogging(ctx context.Context, inputs *ModifyBackendServiceLoggingInputs) (string, error)
	ModifyVPCFlowLog(ctx context.Context, inputs *ModifyVPCFlowLogInputs) (string, error)

	ClusterManager() ClusterManagerClient

	RestartInstance(ctx context.Context, project, zone, name string) error

	// Load balancing
	GetBackendService(ctx context.Context, project, region, name string) (*compute.BackendService, error)
	PatchBackendService(ctx context.Context, project, region string, backendService *compute.BackendService) error
}

type ClientSet struct {
	// https://cloud.google.com/compute/docs/reference/rest/v1
	compute                   *compute.Service
	storage                   *storage.Client
	dns                       *dns.Service
	container                 *container.Service
	computeRegionOperationSvc *compute.RegionOperationsService
	containerOprationSvc      *container.ProjectsLocationsOperationsService

	rmc ResourceManagerClient
	cmc ClusterManagerClient
	rtm ResourceTagManager
}

var _ Client = &ClientSet{}

func NewClientSet() (Client, error) {
	ctx := context.Background()
	computeSvc, err := compute.NewService(ctx)
	if err != nil {
		return nil, err
	}
	s, err := storage.NewClient(ctx)
	if err != nil {
		return nil, err
	}
	dns, err := dns.NewService(ctx)
	if err != nil {
		return nil, err
	}
	rmc, err := NewResourceManagerClient(ctx)
	if err != nil {
		return nil, err
	}
	containerSvc, err := container.NewService(ctx)
	if err != nil {
		return nil, err
	}
	cmc, err := NewClusterManagerClient(ctx)
	if err != nil {
		return nil, err
	}
	rtm, err := NewResourceTagManager(ctx)
	if err != nil {
		return nil, err
	}

	return &ClientSet{
		compute:                   computeSvc,
		storage:                   s,
		dns:                       dns,
		rmc:                       rmc,
		container:                 containerSvc,
		computeRegionOperationSvc: compute.NewRegionOperationsService(computeSvc),
		containerOprationSvc:      container.NewProjectsLocationsOperationsService(containerSvc),

		cmc: cmc,
		rtm: rtm,
	}, nil
}

func (c *ClientSet) ResourceManager() ResourceManagerClient {
	return c.rmc
}

func (c *ClientSet) ResourceTagManager() ResourceTagManager {
	return c.rtm
}

func (c *ClientSet) ClusterManager() ClusterManagerClient {
	return c.cmc
}

func (c *ClientSet) StorageCli() *storage.Client {
	return c.storage
}

func (c *ClientSet) NewStorageCli(ctx context.Context, opts ...option.ClientOption) (*storage.Client, error) {
	s, err := storage.NewClient(ctx, opts...)
	if err != nil {
		return nil, err
	}

	return s, nil
}

// AddBucketIAMMember adds a member to the given bucket's IAM policy.
// TODO(Huaxi): support multi-project.
func (c *ClientSet) AddBucketIAMMember(bucketName string, roleName iam.RoleName, member string) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
	defer cancel()

	bucket := c.storage.Bucket(bucketName)
	policy, err := bucket.IAM().Policy(ctx)
	if err != nil {
		return fmt.Errorf("failed to get the policy of bucket %s: %v", bucketName, err)
	}

	for _, m := range policy.Members(roleName) {
		// skip if the member already exists
		if member == m {
			return nil
		}
	}

	policy.Add(member, roleName)
	if err := bucket.IAM().SetPolicy(ctx, policy); err != nil {
		return fmt.Errorf("failed to set the policy for bucket %s: %v", bucketName, err)
	}
	return nil
}

func (c *ClientSet) GetIAMPolicy3(ctx context.Context, resource *iam.Handle) (*iam.Policy3, error) {
	ctx, cancel := context.WithTimeout(ctx, time.Second*10)
	defer cancel()

	return resource.V3().Policy(ctx)
}

// CreatePSCEndpoint creates a Private Service Connect endpoint for connecting self-published service in other VPC.
func (c *ClientSet) CreatePSCEndpoint(ctx context.Context, project, region, network, subnet, target, name, namespace string, tags map[string]string) error {
	networkUri := fmt.Sprintf(globalNetworkURLTemplate, project, network)
	subnetUri := fmt.Sprintf(regionsSubnetURLTemplate, project, region, subnet)

	add := &compute.Address{
		Name:        name,
		AddressType: "INTERNAL",
		Subnetwork:  subnetUri,
		Description: fmt.Sprintf("Tags: %v", tags),
	}
	if _, err := c.compute.Addresses.Insert(project, region, add).Context(ctx).Do(); err != nil && !isAlreadyExists(err) {
		return errors.Annotate(err, "failed to create address")
	}

	addressUri := fmt.Sprintf(regionsAddressURLTemplate, project, region, name)

	fRule := &compute.ForwardingRule{
		Network:     networkUri,
		IPAddress:   addressUri,
		Name:        name,
		Target:      target,
		Description: fmt.Sprintf("Tags: %v", tags),
		ServiceDirectoryRegistrations: []*compute.ForwardingRuleServiceDirectoryRegistration{
			{
				Namespace: namespace,
			},
		},
	}
	if _, err := c.compute.ForwardingRules.Insert(project, region, fRule).Context(ctx).Do(); err != nil && !isAlreadyExists(err) {
		return errors.Annotate(err, "failed to create forwarding rule")
	}

	return nil
}

// DeleteAllPSCEndpoints deletes all Private Service Connect endpoints in the given vpc network.
func (c *ClientSet) DeleteAllPSCEndpoints(ctx context.Context, project, region, network, subnetwork string) error {
	var errs error

	// Delete forwarding rules
	ruleList, err := c.compute.ForwardingRules.List(project, region).Context(ctx).Do()
	if err != nil {
		errs = multierror.Append(errs, errors.Annotate(err, "failed to list forwarding rules"))
	} else {
		networkUri := fmt.Sprintf(globalNetworkURLTemplate, project, network)
		for _, rule := range ruleList.Items {
			if rule.Network == networkUri {
				if _, err := c.compute.ForwardingRules.Delete(project, region, rule.Name).Context(ctx).Do(); err != nil && !isNotFound(err) {
					errs = multierror.Append(errs, errors.Annotate(err, fmt.Sprintf("failed to delete forwarding rule: %s", rule.Name)))
				}
			}
		}
	}

	// Delete addresses
	addrList, err := c.compute.Addresses.List(project, region).Context(ctx).Do()
	if err != nil {
		errs = multierror.Append(errs, errors.Annotate(err, "failed to delete address"))
	} else {
		subnetUri := fmt.Sprintf(regionsSubnetURLTemplate, project, region, subnetwork)
		for _, addr := range addrList.Items {
			if addr.Subnetwork == subnetUri {
				if _, err := c.compute.Addresses.Delete(project, region, addr.Name).Context(ctx).Do(); err != nil && !isNotFound(err) {
					errs = multierror.Append(errs, errors.Annotate(err, fmt.Sprintf("failed to delete address: %s", addr.Name)))
				}
			}
		}
	}

	return errs
}

func (c *ClientSet) IsGoogleApiPscExists(ctx context.Context, project, network string) (bool, error) {
	networkUri := fmt.Sprintf(globalNetworkURLTemplate, project, network)
	rules, err := c.compute.GlobalForwardingRules.List(project).Filter("network eq " + networkUri).Context(ctx).Do()
	if err != nil {
		return false, err
	}
	if len(rules.Items) > 0 {
		return true, nil
	}
	return false, nil
}

func (c *ClientSet) DeleteAllGoogleApiPsc(ctx context.Context, project, network string) error {
	networkUri := fmt.Sprintf(globalNetworkURLTemplate, project, network)
	rules, err := c.compute.GlobalForwardingRules.List(project).Filter("network eq " + networkUri).Context(ctx).Do()
	if err != nil {
		return err
	}
	for _, rule := range rules.Items {
		if err := c.DeleteGoogleApiPsc(ctx, project, rule.Name); err != nil {
			return err
		}
	}
	return nil
}

func (c *ClientSet) CreateGoogleApiPsc(ctx context.Context, project, region, network, name string, tags map[string]string) error {
	networkUri := fmt.Sprintf(globalNetworkURLTemplate, project, network)
	add := &compute.Address{
		Network:     networkUri,
		Address:     googleApiPscAddress,
		Name:        name,
		Purpose:     "PRIVATE_SERVICE_CONNECT",
		AddressType: "INTERNAL",
		Description: fmt.Sprintf("Tags: %v", tags),
	}
	// Cross-plane provider gcp lacks forwarding rule crd, so we create psc via api directly.
	if _, err := c.compute.GlobalAddresses.Insert(project, add).Context(ctx).Do(); err != nil && !isAlreadyExists(err) {
		return errors.Annotate(err, "failed to create google api ip address")
	}

	addressUri := fmt.Sprintf(globalAddressURLTemplate, project, name)
	fRule := &compute.ForwardingRule{
		Network:     networkUri,
		IPAddress:   addressUri,
		Name:        name,
		Target:      "all-apis",
		Description: fmt.Sprintf("Tags: %v", tags),
		ServiceDirectoryRegistrations: []*compute.ForwardingRuleServiceDirectoryRegistration{
			{
				ServiceDirectoryRegion: region,
				Namespace:              googleApiPscNamespace,
			},
		},
	}
	if _, err := c.compute.GlobalForwardingRules.Insert(project, fRule).Context(ctx).Do(); err != nil && !isAlreadyExists(err) {
		return errors.Annotate(err, "failed to create google api forwarding rule")
	}

	// Redirect default *.googleapis.com dns to the newly created psc ip address.
	if _, err := c.createDNSZone(project, network, name, "googleapis.com."); err != nil {
		return errors.Annotate(err, "failed to create dns zone for googleapis.com")
	}
	if err := c.changeGoogleApiPscDnsRecord(project, name, false); err != nil {
		return errors.Annotate(err, "failed to add googleapis.com and *.googleapis.com dns record")
	}

	return nil
}

func (c *ClientSet) DeleteGoogleApiPsc(ctx context.Context, project, name string) error {
	// Delete global forwarding rule
	if _, err := c.compute.GlobalForwardingRules.Delete(project, name).Context(ctx).Do(); err != nil && !isNotFound(err) {
		return errors.Annotate(err, "failed to delete global forwarding rule")
	}
	// Delete global address
	if _, err := c.compute.GlobalAddresses.Delete(project, name).Context(ctx).Do(); err != nil && !isNotFound(err) {
		return errors.Annotate(err, "failed to delete global address")
	}
	// Delete dns record
	if err := c.changeGoogleApiPscDnsRecord(project, name, true); err != nil {
		return errors.Annotate(err, "failed to delete googleapis.com and *.googleapis.com dns record")
	}
	// Delete dns zone
	if err := c.dns.ManagedZones.Delete(project, name).Context(ctx).Do(); err != nil && !isNotFound(err) {
		return errors.Annotate(err, "failed to delete dns zone")
	}
	return nil
}

func (c *ClientSet) DeleteGoogleApiPscDnsZone(ctx context.Context, project string) error {
	zones, err := c.dns.ManagedZones.List(project).DnsName("googleapis.com.").Context(ctx).Do()
	if err != nil && !isNotFound(err) {
		return errors.Annotate(err, "failed to list dns zones")
	}
	for _, zone := range zones.ManagedZones {
		// Delete dns record
		if err := c.changeGoogleApiPscDnsRecord(project, zone.Name, true); err != nil {
			return errors.Annotate(err, "failed to delete googleapis.com and *.googleapis.com dns record")
		}
		// Delete dns zone
		if err := c.dns.ManagedZones.Delete(project, zone.Name).Context(ctx).Do(); err != nil && !isNotFound(err) {
			return errors.Annotate(err, "failed to delete dns zone")
		}
	}

	return nil
}

func (c *ClientSet) GetInstanceLabels(ctx context.Context, project, zone, instance string) (map[string]string, error) {
	ins, err := c.compute.Instances.Get(project, zone, instance).Context(ctx).Do()
	if err != nil {
		return nil, err
	}
	return ins.Labels, nil
}

func (c *ClientSet) GetInstanceGroupSize(ctx context.Context, project, zone, instanceGroup string) (int64, error) {
	ig, err := c.compute.InstanceGroups.Get(project, zone, instanceGroup).Context(ctx).Do()
	if err != nil {
		return 0, err
	}
	return ig.Size, nil
}

func (c *ClientSet) GetInstanceGroupManagerManagedInstanceCount(ctx context.Context, project, zone, instanceGroupManager string) (int64, error) {
	var total int64
	err := c.compute.InstanceGroupManagers.ListManagedInstances(project, zone, instanceGroupManager).
		MaxResults(500).
		Pages(ctx, func(resp *compute.InstanceGroupManagersListManagedInstancesResponse) error {
			if resp == nil {
				return nil
			}
			total += int64(len(resp.ManagedInstances))
			return nil
		})
	if err != nil {
		return 0, err
	}
	return total, nil
}

func (c *ClientSet) getDNSZone(project, zoneName string) (string, error) {
	resp, err := c.dns.ManagedZones.Get(project, zoneName).Do()
	if err != nil {
		return "", err
	}
	return fmt.Sprintf("%d", resp.Id), nil
}

func (c *ClientSet) createDNSZone(project, network, zoneName, dnsName string) (string, error) {
	// check if zone exist
	zoneID, err := c.getDNSZone(project, zoneName)
	if err == nil {
		return zoneID, nil
	} else if !strings.Contains(err.Error(), "does not exist") {
		// if it is an unknown error,return error
		return "", errors.Annotatef(err, "get hosted zone failed for name: %s", zoneName)
	}

	// if zone does not exist, create it
	zone := &dns.ManagedZone{
		Name:        zoneName,
		DnsName:     dnsName,
		Description: "private host zone for the gke cluster",
		Visibility:  "private",
		PrivateVisibilityConfig: &dns.ManagedZonePrivateVisibilityConfig{
			Kind: "dns#managedZonePrivateVisibilityConfig",
			Networks: []*dns.ManagedZonePrivateVisibilityConfigNetwork{
				{
					Kind:       "dns#managedZonePrivateVisibilityConfigNetwork",
					NetworkUrl: fmt.Sprintf(globalNetworkURLTemplate, project, network),
				},
			},
		},
	}

	resp, err := c.dns.ManagedZones.Create(project, zone).Do()
	if err != nil {
		return "", errors.Annotatef(err, "create hosted zone failed for vpc: %s", zoneName)
	}

	if resp.Name == "" {
		return "", fmt.Errorf("the return private zone id is nil, vpc: %s", zoneName)
	}

	return fmt.Sprintf("%d", resp.Id), nil
}

func (c *ClientSet) changeGoogleApiPscDnsRecord(project, zoneName string, isDelete bool) error {
	recordA := &dns.ResourceRecordSet{
		Type:    "A",
		Name:    "googleapis.com.",
		Ttl:     7200,
		Rrdatas: []string{googleApiPscAddress},
	}
	recordCname := &dns.ResourceRecordSet{
		Type:    "CNAME",
		Name:    "*.googleapis.com.",
		Ttl:     7200,
		Rrdatas: []string{"googleapis.com."},
	}
	if isDelete {
		change := &dns.Change{
			Deletions: []*dns.ResourceRecordSet{recordA, recordCname},
		}
		if _, err := c.dns.Changes.Create(project, zoneName, change).Do(); err != nil && !isAlreadyExists(err) && !isNotFound(err) {
			return errors.Annotatef(err, "failed to delete dns record for google api psc access, zone %q", zoneName)
		}
	} else {
		change := &dns.Change{
			Additions: []*dns.ResourceRecordSet{recordA, recordCname},
		}
		if _, err := c.dns.Changes.Create(project, zoneName, change).Do(); err != nil && !isAlreadyExists(err) && !isNotFound(err) {
			return errors.Annotatef(err, "failed to create dns record for google api psc access, zone %q", zoneName)
		}
	}

	return nil
}

func (c *ClientSet) UpdateNodePoolResourceLabels(ctx context.Context, name string, resourceLabels map[string]string) (*container.Operation, error) {
	req := &container.UpdateNodePoolRequest{
		// All existing fields will be replaced.
		ResourceLabels: &container.ResourceLabels{
			Labels: resourceLabels,
		},
	}
	return c.container.Projects.Locations.Clusters.NodePools.Update(name, req).Context(ctx).Do()
}

func (c *ClientSet) AppendNodePoolResourceLabels(ctx context.Context, name string, resourceLabels map[string]string) (*container.Operation, error) {
	// Update operation is very slow, only update when necessary
	existedLabels, err := c.GetNodePoolResourceLabels(ctx, name)
	if err != nil {
		return nil, err
	}

	// Copy all labels into one map
	fullLabels := make(map[string]string)
	for k, v := range existedLabels {
		fullLabels[k] = v
	}
	for k, v := range resourceLabels {
		fullLabels[k] = v
	}

	if reflect.DeepEqual(existedLabels, fullLabels) {
		return nil, nil
	}

	req := &container.UpdateNodePoolRequest{
		// All existing fields will be replaced.
		ResourceLabels: &container.ResourceLabels{
			Labels: fullLabels,
		},
	}
	return c.container.Projects.Locations.Clusters.NodePools.Update(name, req).Context(ctx).Do()
}

func (c *ClientSet) GetNodePoolResourceLabels(ctx context.Context, name string) (map[string]string, error) {
	nodepool, err := c.container.Projects.Locations.Clusters.NodePools.Get(name).Context(ctx).Do()
	if err != nil {
		return nil, err
	}
	if nodepool.Config == nil {
		return nil, errors.New("nodepool.Config is nil")
	}
	return nodepool.Config.ResourceLabels, nil
}

func (c *ClientSet) GetNodePoolNodeLabels(ctx context.Context, name string) (map[string]string, error) {
	nodepool, err := c.container.Projects.Locations.Clusters.NodePools.Get(name).Context(ctx).Do()
	if err != nil {
		return nil, err
	}
	if nodepool.Config == nil {
		return nil, errors.New("nodepool.Config is nil")
	}
	labels := make(map[string]string, len(nodepool.Config.Labels))
	for k, v := range nodepool.Config.Labels {
		labels[k] = v
	}
	return labels, nil
}

func (c *ClientSet) UpdateNodePoolNodeLabels(ctx context.Context, name string, labels map[string]string) (*container.Operation, error) {
	if labels == nil {
		labels = map[string]string{}
	}
	req := &container.UpdateNodePoolRequest{
		// All existing node labels will be replaced, so callers must pass a merged map.
		Labels: &container.NodeLabels{
			Labels:          labels,
			ForceSendFields: []string{"Labels"},
		},
	}
	return c.container.Projects.Locations.Clusters.NodePools.Update(name, req).Context(ctx).Do()
}

func (c *ClientSet) GetNodePoolNodeTaints(ctx context.Context, name string) ([]*container.NodeTaint, error) {
	nodepool, err := c.container.Projects.Locations.Clusters.NodePools.Get(name).Context(ctx).Do()
	if err != nil {
		return nil, err
	}
	if nodepool.Config == nil {
		return nil, errors.New("nodepool.Config is nil")
	}
	return cloneNodePoolNodeTaints(nodepool.Config.Taints), nil
}

func (c *ClientSet) UpdateNodePoolNodeTaints(ctx context.Context, name string, taints []*container.NodeTaint) (*container.Operation, error) {
	cleanTaints := cloneNodePoolNodeTaints(taints)
	req := &container.UpdateNodePoolRequest{
		// All existing node taints will be replaced, so callers must pass a merged list.
		Taints: &container.NodeTaints{
			Taints:          cleanTaints,
			ForceSendFields: []string{"Taints"},
		},
	}
	return c.container.Projects.Locations.Clusters.NodePools.Update(name, req).Context(ctx).Do()
}

func (c *ClientSet) UpdateNodePoolNodeLabelsAndTaints(ctx context.Context, name string, labels map[string]string, taints []*container.NodeTaint) (*container.Operation, error) {
	if labels == nil {
		labels = map[string]string{}
	}
	cleanTaints := cloneNodePoolNodeTaints(taints)
	req := &container.UpdateNodePoolRequest{
		// All existing node labels and taints will be replaced, so callers must pass merged values.
		Labels: &container.NodeLabels{
			Labels:          labels,
			ForceSendFields: []string{"Labels"},
		},
		Taints: &container.NodeTaints{
			Taints:          cleanTaints,
			ForceSendFields: []string{"Taints"},
		},
	}
	return c.container.Projects.Locations.Clusters.NodePools.Update(name, req).Context(ctx).Do()
}

func cloneNodePoolNodeTaints(taints []*container.NodeTaint) []*container.NodeTaint {
	cleanTaints := make([]*container.NodeTaint, 0, len(taints))
	for _, taint := range taints {
		if taint == nil {
			continue
		}
		copied := *taint
		cleanTaints = append(cleanTaints, &copied)
	}
	return cleanTaints
}

// EmptyBucket implements Client.EmptyBucket.
func (c *ClientSet) EmptyBucket(ctx context.Context, name string) error {
	bucket := c.storage.Bucket(name)
	it := bucket.Objects(ctx, nil)

	for {
		attrs, err := it.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		if err = bucket.Object(attrs.Name).Delete(ctx); err != nil {
			return err
		}
	}
	return nil
}

func (c *ClientSet) getPSCServiceAttachment(ctx context.Context, project, region, name string) (*compute.ServiceAttachment, error) {
	return c.compute.ServiceAttachments.Get(project, region, name).Context(ctx).Do()
}

func (c *ClientSet) IsPSCServiceAttachmentExists(ctx context.Context, project, region, name string) (bool, error) {
	if _, err := c.getPSCServiceAttachment(ctx, project, region, name); err != nil {
		if isNotFound(err) {
			return false, nil
		}
		return false, err
	}

	return true, nil
}

func (c *ClientSet) CreateOrUpdatePSCServiceAttachment(ctx context.Context, project, region, name, subnet, targetService, domainName string, acceptedProjects []string) error {
	if acceptedProjects == nil {
		return fmt.Errorf("acceptedProjects can't be nil")
	}

	subnetworkUri := fmt.Sprintf(regionsSubnetURLTemplate, project, region, subnet)
	ForwardingRuleUri := fmt.Sprintf(regionsForwardingRuleURLTemplate, project, region, targetService)

	cal := make([]*compute.ServiceAttachmentConsumerProjectLimit, 0)
	for _, p := range acceptedProjects {
		cal = append(cal, &compute.ServiceAttachmentConsumerProjectLimit{
			ProjectIdOrNum:  p,
			ConnectionLimit: int64(10),
		})
	}

	sa, err := c.getPSCServiceAttachment(ctx, project, region, name)
	if err != nil {
		if isNotFound(err) {
			// Create
			op, err := c.compute.ServiceAttachments.Insert(project, region, &compute.ServiceAttachment{
				Name:                 name,
				ConnectionPreference: "ACCEPT_MANUAL",
				NatSubnets:           []string{subnetworkUri},
				TargetService:        ForwardingRuleUri,
				DomainNames: []string{
					fmt.Sprintf("%s.", domainName)},
				ConsumerAcceptLists: cal,
			}).Context(ctx).Do()
			if err != nil {
				return err
			}
			return c.computeWaitOperation(ctx, project, op)
		}
		return err
	}

	// Update
	currentCalMap := make(map[string]int64, 0)
	for _, a := range sa.ConsumerAcceptLists {
		currentCalMap[a.ProjectIdOrNum] = a.ConnectionLimit
	}
	needUpdate := true
	// They are fully equaled when they have the same length and the content inside can correspond to one-to-one
	if len(currentCalMap) == len(cal) {
		needUpdate = false
		for _, projectLimit := range cal {
			limit, ok := currentCalMap[projectLimit.ProjectIdOrNum]
			if !ok || limit != projectLimit.ConnectionLimit {
				needUpdate = true
				break
			}
		}
	}
	// It doesn't need to send update request when current cal fully equal to desired
	if !needUpdate {
		return nil
	}

	if _, err := c.compute.ServiceAttachments.Patch(project, region, name, &compute.ServiceAttachment{
		Name:                name,
		ConsumerAcceptLists: cal,
		Fingerprint:         sa.Fingerprint,
	}).Context(ctx).Do(); err != nil {
		return err
	}

	return nil
}

func (c *ClientSet) DeletePSCServiceAttachment(ctx context.Context, project, region, name string) error {
	// Skip if it not exists anymore
	exists, err := c.IsPSCServiceAttachmentExists(ctx, project, region, name)
	if err != nil {
		return err
	}
	if !exists {
		return nil
	}

	op, err := c.compute.ServiceAttachments.Delete(project, region, name).Context(ctx).Do()
	if err != nil {
		return err
	}
	return c.computeWaitOperation(ctx, project, op)
}

func (c *ClientSet) GetPSCServiceAttachmentConnections(ctx context.Context, project, region, name string) ([]*compute.ServiceAttachmentConnectedEndpoint, error) {
	serviceAttachment, err := c.compute.ServiceAttachments.Get(project, region, name).Context(ctx).Do()
	if err != nil {
		return nil, err
	}

	return serviceAttachment.ConnectedEndpoints, nil
}

func (c *ClientSet) ListSubnetworks(ctx context.Context, project, region, network string) ([]*compute.Subnetwork, error) {
	networkUri := fmt.Sprintf(globalNetworkURLTemplate, project, network)
	sl, err := c.compute.Subnetworks.List(project, region).Filter("network eq " + networkUri).Context(ctx).Context(ctx).Do()
	if err != nil {
		return nil, err
	}
	return sl.Items, nil
}

func (c *ClientSet) IsSubnetExists(ctx context.Context, project, region, name string) (bool, error) {
	if _, err := c.compute.Subnetworks.Get(project, region, name).Context(ctx).Do(); err != nil {
		if isNotFound(err) {
			return false, nil
		}
		return false, err
	}

	return true, nil
}

func (c *ClientSet) CreatePSCSubnet(ctx context.Context, project, region, name, network, cidr string) error {
	networkUri := fmt.Sprintf(globalNetworkURLTemplate, project, network)

	// Already exists, skip send create request
	exists, err := c.IsSubnetExists(ctx, project, region, name)
	if err != nil {
		return err
	}
	if exists {
		return nil
	}

	op, err := c.compute.Subnetworks.Insert(project, region, &compute.Subnetwork{
		Name:        name,
		IpCidrRange: cidr,
		Region:      region,
		Network:     networkUri,
		Purpose:     "PRIVATE_SERVICE_CONNECT",
	}).Context(ctx).Do()
	if err != nil {
		return err
	}
	return c.computeWaitOperation(ctx, project, op)
}

func (c *ClientSet) DeleteSubnet(ctx context.Context, project, region, name string) error {
	// Skip if it not exists anymore
	exists, err := c.IsSubnetExists(ctx, project, region, name)
	if err != nil {
		return err
	}
	if !exists {
		return nil
	}

	op, err := c.compute.Subnetworks.Delete(project, region, name).Context(ctx).Do()
	if err != nil {
		return err
	}
	return c.computeWaitOperation(ctx, project, op)
}

func (c *ClientSet) PatchSubnet(ctx context.Context, sn *compute.Subnetwork, project, region string) error {
	op, err := c.compute.Subnetworks.Patch(project, region, sn.Name, sn).Context(ctx).Do()
	if err != nil {
		return err
	}
	return c.computeWaitOperation(ctx, project, op)

}

func (c *ClientSet) RestartInstance(ctx context.Context, project, zone, instance string) error {
	if _, err := c.compute.Instances.Reset(project, zone, instance).Context(ctx).Do(); err != nil {
		return fmt.Errorf("failed to reset instance %s: %w", instance, err)
	}
	return nil
}

func (c *ClientSet) getLastWordFromUrl(url string) (string, error) {
	if len(url) < 1 {
		return "", nil
	}
	i := strings.LastIndex(url, "/")
	if i < 0 {
		return "", fmt.Errorf("not found \"/\" in url")
	}
	if i+1 >= len(url) {
		return "", fmt.Errorf("found \"/\" in the last index")
	}
	return url[i+1:], nil
}

// computeWaitOperation waits for the specified compute Operation resource to return as `DONE` or for the request to approach the 2 minute deadline, and retrieves the specified Operation resource
func (c *ClientSet) computeWaitOperation(ctx context.Context, project string, op *compute.Operation) error {
	// Set timeout
	ctx, cancel := context.WithTimeout(ctx, operationWaitingTimeout)
	defer cancel()
	zone, err := c.getLastWordFromUrl(op.Zone)
	if err != nil {
		return err
	}
	region, err := c.getLastWordFromUrl(op.Region)
	if err != nil {
		return err
	}
	var waitOp *compute.Operation
	if len(zone) > 0 {
		// zone operation
		return fmt.Errorf("not support waiting zone operation")
	} else if len(region) > 0 {
		// region operation
		waitOp, err = c.computeRegionOperationSvc.Wait(project, region, op.Name).Context(ctx).Do()
	} else {
		// global operation
		return fmt.Errorf("not support waiting global operation")
	}

	// Handling error
	if err != nil {
		return err
	}

	var errs error
	if waitOp.Error != nil {
		for _, e := range waitOp.Error.Errors {
			errs = multierror.Append(errs, fmt.Errorf("errorCode: %s, errorMessage: %s", e.Code, e.Message))
		}
	}
	return errs
}

func isNotFound(err error) bool {
	if err == nil {
		return false
	}
	ae, ok := err.(*googleapi.Error)
	return ok && ae.Code == http.StatusNotFound
}

func isAlreadyExists(err error) bool {
	if err == nil {
		return false
	}
	ae, ok := err.(*googleapi.Error)
	return ok && ae.Code == http.StatusConflict
}