package gcp

import (
	"context"
	"encoding/base64"
	"fmt"
	"net/http"
	"sort"
	"strings"
	"time"

	"cloud.google.com/go/iam"
	"cloud.google.com/go/storage"
	"github.com/pingcap/errors"
	"google.golang.org/api/cloudresourcemanager/v1"
	"google.golang.org/api/compute/v1"
	"google.golang.org/api/dns/v1"
	"google.golang.org/api/googleapi"
	iamv1 "google.golang.org/api/iam/v1"
)

const (
	// 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"
)

// GCPAPI interface to avoid true call the api in /pkg/providers/gcp(change from dbaas) now
type GCPAPI interface {
	AddPolicyBinding(project string, member, role string) error
	QueryServiceAccount(saName string, projectID string) (*iamv1.ServiceAccount, error)
	CreateServiceAccountKey(project string, saName string) ([]byte, error)
	CreateDnsZone(project, network, zoneName, dnsName string) (string, error)
	AddBucketIAMMember(bucketName string, roleName iam.RoleName, member string) error
	SetDeleteProtection(project string, zone string, nodeName string, protection bool) error
	ListNat(project string, networkName string, region string) ([]string, error)
	CreateVpcPeering(project, peeringName, snName, targetProject, tnName string) error
	AttachIngressPolicy(project string, firewallName string, network string, cidr string) error
	// AddBidingToServiceAccount add a binding with specify role and member to the service account.
	AddBidingToServiceAccount(ctx context.Context, projectID string, sa string, role string, member string) error
	// CreateGoogleApiPsc creates psc in current network region for accessing google apis, such as compute.googleapis.com, storage.googleapis.com, etc.
	// Each network/region should have only one google api psc created.
	CreateGoogleApiPsc(ctx context.Context, project, region, network, shootName string) error
	DeleteGoogleApiPsc(ctx context.Context, project, shootName string) error
	// CreatePublishedServicePsc creates in current network region for accessing self-published service in other vpc.
	CreatePublishedServicePsc(ctx context.Context, project, region, network, subnet, target, name, namespace string) error
	DeletePublishedServicePsc(ctx context.Context, project, region, name string) error
}

type defaultGCPAPI struct {
	csm     *cloudresourcemanager.Service
	iam     *iamv1.Service
	dns     *dns.Service
	storage *storage.Client
	compute *compute.Service
}

func NewDefaultGCPAPI() (GCPAPI, error) {
	ctx := context.Background()
	csm, err := cloudresourcemanager.NewService(ctx)
	if err != nil {
		return nil, err
	}
	iam, err := iamv1.NewService(ctx)
	if err != nil {
		return nil, err
	}
	dns, err := dns.NewService(ctx)
	if err != nil {
		return nil, err
	}
	storage, err := storage.NewClient(ctx)
	if err != nil {
		return nil, err
	}
	computeSvc, err := compute.NewService(ctx)
	if err != nil {
		return nil, err
	}

	return &defaultGCPAPI{
		csm:     csm,
		iam:     iam,
		dns:     dns,
		storage: storage,
		compute: computeSvc,
	}, nil
}

func (g *defaultGCPAPI) AddPolicyBinding(projectID string, member, role string) error {
	return AddPolicyBinding(g.csm, projectID, member, role)
}

func (g *defaultGCPAPI) AddBidingToServiceAccount(ctx context.Context, projectID, sa string, role string, member string) error {
	return AddBidingToServiceAccount(ctx, g.iam, projectID, sa, role, member)
}

func (g *defaultGCPAPI) QueryServiceAccount(saName string, projectID string) (*iamv1.ServiceAccount, error) {
	return QueryServiceAccount(g.iam, saName, projectID)
}

func (g *defaultGCPAPI) CreateServiceAccountKey(projectID string, saName string) ([]byte, error) {
	return CreateServiceAccountKey(g.iam, projectID, saName)
}

func (g *defaultGCPAPI) CreateDnsZone(project, network, zoneName, dnsName string) (string, error) {
	return CreateDnsZone(g.dns, project, network, zoneName, dnsName)
}

func (g *defaultGCPAPI) AddBucketIAMMember(bucketName string, roleName iam.RoleName, member string) error {
	return AddBucketIAMMember(g.storage, bucketName, roleName, member)
}

func (g *defaultGCPAPI) SetDeleteProtection(project string, zone string, nodeName string, protection bool) error {
	_, err := g.compute.Instances.SetDeletionProtection(project, zone, nodeName).DeletionProtection(protection).Do()
	return err
}

func (g *defaultGCPAPI) ListNat(project string, networkName string, region string) ([]string, error) {
	return ListNat(g.compute, project, networkName, region)
}

func (g *defaultGCPAPI) CreateVpcPeering(project, peeringName, snName, targetProject, tnName string) error {
	return nil
}

func (g *defaultGCPAPI) AttachIngressPolicy(project string, firewallName string, network string, cidr string) error {
	return AttachIngressPolicy(g.compute, project, firewallName, network, cidr)
}

func (g *defaultGCPAPI) CreateGoogleApiPsc(ctx context.Context, project, region, network, shootName string) error {
	return CreateGoogleApiPsc(ctx, g.compute, g.dns, project, region, network, shootName)
}

func (g *defaultGCPAPI) DeleteGoogleApiPsc(ctx context.Context, project, shootName string) error {
	return DeleteGoogleApiPsc(ctx, g.compute, g.dns, project, shootName)
}

func (g *defaultGCPAPI) CreatePublishedServicePsc(ctx context.Context, project, region, network, subnet, target, name, namespace string) error {
	return CreatePublishedServicePsc(ctx, g.compute, project, region, network, subnet, target, name, namespace)
}

func (g *defaultGCPAPI) DeletePublishedServicePsc(ctx context.Context, project, region, name string) error {
	return DeletePublishedServicePsc(ctx, g.compute, project, region, name)
}

func CreatePublishedServicePsc(ctx context.Context, svc *compute.Service, project, region, network, subnet, target, name, namespace string) error {
	networkUri := fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/global/networks/%s", project, network)
	subnetUri := fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/regions/%s/subnetworks/%s", project, region, subnet)

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

	addressUri := fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/regions/%s/addresses/%s", project, region, name)

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

	return nil
}

func DeletePublishedServicePsc(ctx context.Context, svc *compute.Service, project, region, name string) error {
	// Delete forwarding rule
	if _, err := svc.ForwardingRules.Delete(project, region, name).Context(ctx).Do(); err != nil && !isNotFound(err) {
		return errors.Annotate(err, "failed to delete forwarding rule")
	}
	// Delete address
	if _, err := svc.Addresses.Delete(project, region, name).Context(ctx).Do(); err != nil && !isNotFound(err) {
		return errors.Annotate(err, "failed to delete address")
	}
	return nil
}

// The forwarding rule name for PSC Google APIs must be an 1-20 characters string with lowercase letters and numbers and must start with a letter.
// The address name must start with a lowercase letter followed by up to 62 lowercase letters, numbers, or hyphens, and cannot end with a hyphen.
func googleApiPscName(shootName string) string {
	// gapipsc means Google API PSC.
	return fmt.Sprintf("gapipsc%s", strings.ReplaceAll(shootName, "-", ""))
}

// CreateGoogleApiPsc creates psc in current network region for accessing google apis, such as cloud storage.
func CreateGoogleApiPsc(ctx context.Context, computeSvc *compute.Service, dnsSvc *dns.Service, project, region, network, shootName string) error {
	networkUri := fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/global/networks/%s", project, network)

	add := &compute.Address{
		Network:     networkUri,
		Address:     googleApiPscAddress,
		Name:        googleApiPscName(shootName),
		Purpose:     "PRIVATE_SERVICE_CONNECT",
		AddressType: "INTERNAL",
	}
	// Cross-plane provider gcp lacks forwarding rule crd, so we create psc via api directly.
	if _, err := computeSvc.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("https://www.googleapis.com/compute/v1/projects/%s/global/addresses/%s", project, googleApiPscName(shootName))

	fRule := &compute.ForwardingRule{
		Network:   networkUri,
		IPAddress: addressUri,
		Name:      googleApiPscName(shootName),
		Target:    "all-apis",
		ServiceDirectoryRegistrations: []*compute.ForwardingRuleServiceDirectoryRegistration{
			{
				ServiceDirectoryRegion: region,
				Namespace:              googleApiPscNamespace,
			},
		},
	}
	if _, err := computeSvc.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 := CreateDnsZone(dnsSvc, project, network, googleApiPscName(shootName), "googleapis.com."); err != nil {
		return errors.Annotate(err, "failed to create dns zone for googleapis.com")
	}
	if err := ChangeGoogleApiPscDnsRecord(dnsSvc, project, googleApiPscName(shootName), false); err != nil {
		return errors.Annotate(err, "failed to add googleapis.com and *.googleapis.com dns record")
	}

	return nil
}

func ChangeGoogleApiPscDnsRecord(svc *dns.Service, 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 := svc.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 := svc.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 DeleteGoogleApiPsc(ctx context.Context, svc *compute.Service, dnsService *dns.Service, project, shootName string) error {
	// Delete global forwarding rule
	if _, err := svc.GlobalForwardingRules.Delete(project, googleApiPscName(shootName)).Context(ctx).Do(); err != nil && !isNotFound(err) {
		return errors.Annotate(err, "failed to delete global forwarding rule")
	}
	// Delete global address
	if _, err := svc.GlobalAddresses.Delete(project, googleApiPscName(shootName)).Context(ctx).Do(); err != nil && !isNotFound(err) {
		return errors.Annotate(err, "failed to delete global address")
	}
	// Delete dns record
	if err := ChangeGoogleApiPscDnsRecord(dnsService, project, googleApiPscName(shootName), true); err != nil {
		return errors.Annotate(err, "failed to delete googleapis.com and *.googleapis.com dns record")
	}
	// Delete dns zone
	if err := dnsService.ManagedZones.Delete(project, googleApiPscName(shootName)).Context(ctx).Do(); err != nil && !isNotFound(err) {
		return errors.Annotate(err, "failed to delete dns zone")
	}
	return nil
}

// AddBidingToServiceAccount add a binding with specify role and member to the service account.
func AddBidingToServiceAccount(ctx context.Context, svc *iamv1.Service, projectID string, sa string, role string, member string) error {
	resource := fmt.Sprintf("projects/%s/serviceAccounts/%s", projectID, sa)
	policy, err := svc.Projects.ServiceAccounts.GetIamPolicy(resource).Context(ctx).Do()
	if err != nil {
		return errors.Annotatef(err, "failed to get IAM policy of %q", sa)
	}

	roleExist := false
	for _, binding := range policy.Bindings {
		if binding.Role == role {
			if binding.Condition != nil {
				continue
			}

			for _, m := range binding.Members {
				if m == member {
					return nil
				}
			}

			roleExist = true
			binding.Members = append(binding.Members, member)
			break
		}
	}

	if !roleExist {
		policy.Bindings = append(policy.Bindings, &iamv1.Binding{
			Role:    role,
			Members: []string{member},
		})
	}

	req := &iamv1.SetIamPolicyRequest{
		Policy: policy,
	}
	_, err = svc.Projects.ServiceAccounts.SetIamPolicy(resource, req).Context(ctx).Do()
	if err != nil {
		return errors.Annotate(err, "failed to set iam policy")
	}

	return nil
}

func QueryServiceAccount(iamService *iamv1.Service, saName string, projectID string) (*iamv1.ServiceAccount, error) {
	fullName := fmt.Sprintf("projects/%s/serviceAccounts/%s", projectID, saName)
	return iamService.Projects.ServiceAccounts.Get(fullName).Do()
}

// AddPolicyBinding binds a member with a role to the GCP project
func AddPolicyBinding(svc *cloudresourcemanager.Service, projectID string, member, role string) error {
	getReq := &cloudresourcemanager.GetIamPolicyRequest{
		Options: &cloudresourcemanager.GetPolicyOptions{
			RequestedPolicyVersion: 3, // use version 3 in case there are conditions in the binding
		},
	}
	policy, err := svc.Projects.GetIamPolicy(projectID, getReq).Do()
	if err != nil {
		return err
	}
	roleExist := false
	for _, bind := range policy.Bindings {
		// bindings are grouped by role
		if bind.Role == role {
			if bind.Condition != nil {
				continue
			}

			// role exist, add member to the binding
			roleExist = true
			for _, m := range bind.Members {
				if m == member {
					return nil
				}
			}
			// policy.Bindings are pointer, we can have inplace update
			bind.Members = append(bind.Members, member)
			break
		}
	}
	if !roleExist {
		// role not exist, create a new binding
		binding := &cloudresourcemanager.Binding{
			Role:    role,
			Members: []string{member},
		}
		policy.Bindings = append(policy.Bindings, binding)
	}
	setReq := &cloudresourcemanager.SetIamPolicyRequest{
		Policy: policy,
	}
	if _, err = svc.Projects.SetIamPolicy(projectID, setReq).Do(); err != nil {
		return err
	}
	return nil
}

// CreateServiceAccountKey create key for service account
// This is required before TiKV supports Application Default Credentials
func CreateServiceAccountKey(svc *iamv1.Service, projectID string, name string) ([]byte, error) {
	sa := fmt.Sprintf("projects/%s/serviceAccounts/%s", projectID, name)
	req := &iamv1.CreateServiceAccountKeyRequest{
		KeyAlgorithm: "KEY_ALG_RSA_2048",
	}

	key, err := svc.Projects.ServiceAccounts.Keys.Create(sa, req).Do()
	if err != nil {
		return nil, errors.Annotatef(err, "failed to create key for service account %q", sa)
	}
	return base64.StdEncoding.DecodeString(key.PrivateKeyData)
}

// GetDNSZone query dns zone by name
func GetDNSZone(svc *dns.Service, project string, name string) (string, error) {
	resp, err := svc.ManagedZones.Get(project, name).Do()
	if err != nil {
		return "", err
	}
	return fmt.Sprintf("%d", resp.Id), nil
}

func CreateDnsZone(svc *dns.Service, project, network, zoneName, dnsName string) (string, error) {
	// check if zone exist
	zoneID, err := GetDNSZone(svc, 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 shoot cluster",
		Visibility:  "private",
		PrivateVisibilityConfig: &dns.ManagedZonePrivateVisibilityConfig{
			Kind: "dns#managedZonePrivateVisibilityConfig",
			Networks: []*dns.ManagedZonePrivateVisibilityConfigNetwork{
				{
					Kind:       "dns#managedZonePrivateVisibilityConfigNetwork",
					NetworkUrl: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/global/networks/%s", project, network),
				},
			},
		},
	}

	resp, err := svc.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
}

// AddBucketIAMMember adds the bucket IAM member to permission role.
func AddBucketIAMMember(client *storage.Client, bucketName string, roleName iam.RoleName, member string) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
	defer cancel()

	bucket := client.Bucket(bucketName)
	policy, err := bucket.IAM().Policy(ctx)
	if err != nil {
		return fmt.Errorf("Bucket(%q).IAM().Policy: %v", bucketName, err)
	}

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

	policy.Add(member, roleName)
	if err := bucket.IAM().SetPolicy(ctx, policy); err != nil {
		return fmt.Errorf("Bucket(%q).IAM().SetPolicy: %v", bucketName, err)
	}
	return nil
}

func AttachIngressPolicy(svc *compute.Service, project string, firewallName string, network string, cidr string) error {
	filewall := compute.Firewall{
		Allowed: []*compute.FirewallAllowed{{
			IPProtocol: "all",
		}},
		Network:      network,
		Direction:    "Ingress",
		SourceRanges: []string{cidr},
		Name:         firewallName,
		Description:  "allow downstream network access upstream network",
	}
	_, err := svc.Firewalls.Insert(project, &filewall).Do()
	return err
}

func ListNat(svc *compute.Service, project string, networkName string, region string) ([]string, error) {
	ips := make(map[string]struct{})
	err := svc.Addresses.List(project, region).Pages(context.Background(), func(list *compute.AddressList) error {
		for _, add := range list.Items {
			for _, user := range add.Users {
				if strings.HasSuffix(strings.TrimSpace(user), fmt.Sprintf("%s-cloud-router", networkName)) {
					ips[add.Address] = struct{}{}
				}
			}
		}

		return nil
	})
	if err != nil {
		return nil, err
	}

	result := make([]string, 0)
	for k := range ips {
		result = append(result, fmt.Sprintf("%s/32", k))
	}
	sort.Strings(result)

	return result, nil
}

func CreateVpcPeering(svc *compute.Service, project, peeringName, snName, targetProject, tnName string) error {
	_, err := svc.Networks.AddPeering(project, snName, &compute.NetworksAddPeeringRequest{
		Name: peeringName,
		PeerNetwork: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/global/networks/%s",
			targetProject, tnName)}).Do()

	return err
}

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
}