package gke

import (
	"context"
	"fmt"
	"net"
	"regexp"
	"strings"
	"time"

	"github.com/crossplane-contrib/provider-gcp/apis/compute/v1alpha1"
	gcpcompute "google.golang.org/api/compute/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"

	"github.com/tidbcloud/infra-provider/pkg/consts"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

const (
	maxServiceNameLength = 32

	serviceAttachmentTemplate = "projects/%s/regions/%s/serviceAttachments/%s"

	endpointURIPrefix = `https://www.googleapis.com/compute/v1/`

	annoKeyForwardingRuleName = consts.AnnoKeyGKEOriginForwardingRuleName
)

var (
	endpointRe = regexp.MustCompile(`^projects/(\S+)/regions/(\S+)/forwardingRules/(\S+)$`)
)

type PSC struct {
	*task.BaseTask

	gcpCli gcp.Client
}

// Deprecated: 3-layer cluster's PSC is managed by regional now.
func NewPSC(ctx *task.Context, gcpCli gcp.Client) *PSC {
	return &PSC{BaseTask: &task.BaseTask{Context: ctx}, gcpCli: gcpCli}
}

func (t *PSC) Name() string {
	return "PSC"
}

// Create is the main entry for private service connect reconciliation.
func (t *PSC) Create(ctx context.Context) deploy.Result {
	if t.IsRegionalManaged() {
		return deploy.Done()
	}
	if t.Config.DryRun.Enabled {
		return deploy.Done()
	}

	t.BeginAt = util.Pointer(time.Now())

	if !t.isPrivateServiceConnectRequired() {
		if t.Cluster.Status.SqlEndpoints.PrivateLinkService != nil {
			return t.Delete(ctx)
		}

		return deploy.Done()
	}

	// Initialize in the beginning or it is in Deleting
	if t.Cluster.Status.SqlEndpoints.PrivateLinkService == nil ||
		t.Cluster.Status.SqlEndpoints.PrivateLinkService.Phase == cloudv1.PrivateLinkServicePhaseDeleting {
		t.Cluster.Status.SqlEndpoints.PrivateLinkService = &cloudv1.PrivateLinkServiceStatus{
			Phase: cloudv1.PrivateLinkServicePhaseCreating,
			GCP: &cloudv1.GCPPrivateLinkServiceStatus{
				DomainName:  t.Namer.GetPrivateServiceConnectDomain(),
				Connections: []*cloudv1.GCPPrivateLinkServiceConnection{},
				Port: pointer.Int32(
					t.Cluster.GetPrivateSQLEndpointPort()),
				Subnet: "",
			},
			CreationTimestamp: metav1.NewTime(time.Now()),
		}
		return deploy.Error(fmt.Errorf("update status first and create in next reconcile"))
	}

	if err := t.createSubnet(ctx); err != nil {
		return deploy.Panic(err)
	}

	if err := t.createOrUpdateServiceAttachment(ctx); err != nil {
		return deploy.Panic(err)
	}

	// TODO(zlj): Synchronize periodically
	if err := t.refreshStatusConnsWithGCP(ctx); err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *PSC) PostCreate(ctx context.Context) deploy.Result {
	if t.IsRegionalManaged() {
		return deploy.Done()
	}
	if t.Config.DryRun.Enabled {
		return deploy.Done()
	}

	return deploy.Done()
}

func (t *PSC) Delete(ctx context.Context) deploy.Result {
	if t.IsRegionalManaged() {
		return deploy.Done()
	}
	if t.Config.DryRun.Enabled {
		return deploy.Done()
	}

	if t.Cluster.Status.SqlEndpoints.PrivateLinkService == nil {
		return deploy.Done()
	}

	if t.Cluster.Status.SqlEndpoints.PrivateLinkService.Phase != cloudv1.PrivateLinkServicePhaseDeleting {
		t.Cluster.Status.SqlEndpoints.PrivateLinkService.Phase = cloudv1.PrivateLinkServicePhaseDeleting
		return deploy.Error(fmt.Errorf("update status first and delete in next reconcile"))
	}

	// Delete service attathment
	if err := t.deleteServiceAttachment(ctx); err != nil {
		return deploy.Panic(err)
	}

	// Delete subnet
	if err := t.deleteSubnet(ctx); err != nil {
		return deploy.Panic(err)
	}

	// All deleted
	t.Cluster.Status.SqlEndpoints.PrivateLinkService = nil

	return deploy.Done()
}

func (t *PSC) PostDelete(ctx context.Context) deploy.Result {
	if t.IsRegionalManaged() {
		return deploy.Done()
	}
	if t.Config.DryRun.Enabled {
		return deploy.Done()
	}

	return deploy.Done()
}

// findNATSubnetwork find a not-used subnetwork to perform PSC NAT.
// 'prefix' is the total CIDR can be used in this network, and the prefix lenth is /25. The following 3-bit is
// the index-bit, all cluster in this network will take one, so up to 8(2^3) clusters can use PSC in a network.
// As a result, the NAT subnetwork prefix is always /28.
//
// For example, the prefix CIDR is `10.250.14.128/25`, and the subnetwork `10.250.14.128/28` is already used for
// another cluster, so the result is `10.250.14.144/25`
func (t *PSC) findNATSubnetwork(prefix string, subnetworks []*gcpcompute.Subnetwork) (string, error) {
	existing := make(map[string]bool)
	for _, s := range subnetworks {
		if s.Name == t.natSubnetworkName() {
			return s.IpCidrRange, nil
		}
		existing[s.IpCidrRange] = true
	}

	_, prefixNet, err := net.ParseCIDR(prefix)
	if err != nil {
		return "", err
	}
	prefixNet.Mask = net.CIDRMask(28, 32)
	suffixInt := int(prefixNet.IP.To4()[3] & 0b10000000)
	for i := 0; i < 8; i++ {
		prefixNet.IP.To4()[3] = byte(suffixInt)
		if !existing[prefixNet.String()] {
			return prefixNet.String(), nil
		}
		suffixInt += 16
	}

	return "", fmt.Errorf("nat subnet has been used up")
}

func (t *PSC) natSubnetworkName() string {
	return fmt.Sprintf("tidb%s-psc-nat", t.Cluster.GetClusterID())
}

// IsPrivateLinkRequired checks if the private service connect is required by the cluster.
func (t *PSC) isPrivateServiceConnectRequired() bool {
	return t.Cluster.Spec.SqlEndpoints.PrivateLinkService != nil &&
		t.Cluster.Spec.SqlEndpoints.PrivateLinkService.GCP != nil
}

func (t *PSC) firewallName() string {
	if t.IsLegacy() {
		return fmt.Sprintf("%s-gke-%s-psc-ingress", t.Cluster.GetNamespace(), t.EnvProvider.GetRegion())
	}

	return namer.UniqueIdentifier(t.Cluster.ObjectMeta, namer.WithPrefix("psc-ingress"), namer.WithLimit(namer.GKENameLimit))
}

func (t *PSC) serviceAttachmentName() string {
	return fmt.Sprintf("tidb%s", t.Cluster.GetClusterID())
}

func (t *PSC) createSubnet(ctx context.Context) error {
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.Message = "creating nat subnet"

	// Get subnet CIDR
	// Firewall rule is configured in network controller
	firewall := &v1alpha1.Firewall{}
	err := t.Local.Get(ctx, types.NamespacedName{
		Name: t.firewallName(),
	}, firewall)
	if err != nil {
		return err
	}
	if len(firewall.Spec.ForProvider.SourceRanges) < 1 {
		return fmt.Errorf("no sourcerange found in firewall %s", firewall.Name)
	}

	existingSubnets, err := t.gcpCli.ListSubnetworks(ctx, t.EnvProvider.GetAccount(), t.EnvProvider.GetRegion(), t.EnvProvider.GetNetwork())
	if err != nil {
		return err
	}
	subnetCIDR, err := t.findNATSubnetwork(firewall.Spec.ForProvider.SourceRanges[0], existingSubnets)
	if err != nil {
		return err
	}

	// Add createSubnet
	// crossplane-gcp don't support to specify the purpose of subnet
	if err := t.gcpCli.CreatePSCSubnet(ctx, t.EnvProvider.GetAccount(), t.EnvProvider.GetRegion(), t.natSubnetworkName(), t.EnvProvider.GetNetwork(), subnetCIDR); err != nil {
		return err
	}

	exists, err := t.gcpCli.IsSubnetExists(ctx, t.EnvProvider.GetAccount(), t.EnvProvider.GetRegion(), t.natSubnetworkName())
	if err != nil {
		return err
	}

	if !exists {
		return fmt.Errorf("wait nat subnet created")
	}

	t.Cluster.Status.SqlEndpoints.PrivateLinkService.GCP.Subnet = t.natSubnetworkName()
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.Message = ""

	return nil
}

func (t *PSC) getTargetService(ctx context.Context) (string, error) {
	remote := t.EnvProvider.GetClient()
	svcKey := types.NamespacedName{
		Name:      t.Namer.GetPrivateTiDBServiceName(),
		Namespace: t.Namer.GetTiDBNamespace(),
	}
	svc := &corev1.Service{}
	err := remote.Get(ctx, svcKey, svc)
	if err != nil {
		return "", err
	}
	if len(svc.Status.LoadBalancer.Ingress) == 0 {
		return "", fmt.Errorf("private svc not ready")
	}

	// annoKeyForwardingRuleName is set in FullService task
	frName, ok := svc.Annotations[annoKeyForwardingRuleName]
	if !ok {
		// fallback to legacy way
		frName = strings.ReplaceAll("a"+string(svc.GetUID()), "-", "")
		if len(frName) > maxServiceNameLength {
			frName = frName[:maxServiceNameLength]
		}
	}

	return frName, nil
}

func (t *PSC) createOrUpdateServiceAttachment(ctx context.Context) error {
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.Message = "creating service attachment"

	targetService, err := t.getTargetService(ctx)
	if err != nil {
		return err
	}

	t.Cluster.Status.SqlEndpoints.PrivateLinkService.GCP.NLB = targetService

	projects, err := t.getAcceptedProjectListFromSpec()
	if err != nil {
		return err
	}

	// Create or update service attathment
	err = t.gcpCli.CreateOrUpdatePSCServiceAttachment(ctx,
		t.EnvProvider.GetAccount(),
		t.EnvProvider.GetRegion(),
		t.serviceAttachmentName(),
		t.natSubnetworkName(),
		targetService,
		t.Namer.GetPrivateServiceConnectDomain(),
		projects,
	)
	if err != nil {
		return err
	}

	exists, err := t.gcpCli.IsPSCServiceAttachmentExists(ctx, t.EnvProvider.GetAccount(), t.EnvProvider.GetRegion(), t.serviceAttachmentName())
	if err != nil {
		return err
	}
	if !exists {
		return fmt.Errorf("service attathment not ready")
	}

	t.Cluster.Status.SqlEndpoints.PrivateLinkService.GCP.ServiceAttachment = fmt.Sprintf(serviceAttachmentTemplate,
		t.EnvProvider.GetAccount(),
		t.EnvProvider.GetRegion(),
		t.serviceAttachmentName(),
	)
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.Phase = cloudv1.PrivateLinkServicePhaseActive
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.Message = ""

	return nil
}

func (t *PSC) getAcceptedProjectListFromSpec() ([]string, error) {
	projects := make([]string, 0)
	for _, e := range t.Cluster.Spec.SqlEndpoints.PrivateLinkService.GCP.AcceptedEndpoints {
		strs := endpointRe.FindStringSubmatch(string(e))
		if len(strs) != 4 {
			return nil, fmt.Errorf("parse endpoint in spec error, find %d submatch strings", len(strs))
		}
		projects = append(projects, strs[1])
	}

	return util.RemoveDuplicatesInSlice(projects), nil
}

func (t *PSC) convertGCPEndpointStautsToCloudv1(e, s string) *cloudv1.GCPPrivateLinkServiceConnection {
	// Possible values:
	//   "ACCEPTED" - The connection has been accepted by the producer.
	//   "CLOSED" - The connection has been closed by the producer.
	//   "NEEDS_ATTENTION" - The connection has been accepted by the
	// producer, but the producer needs to take further action before the
	// forwarding rule can serve traffic.
	//   "PENDING" - The connection is pending acceptance by the producer.
	//   "REJECTED" - The consumer is still connected but not using the
	// connection.
	//   "STATUS_UNSPECIFIED"
	switch s {
	case "ACCEPTED", "REJECTED":
		return &cloudv1.GCPPrivateLinkServiceConnection{
			Endpoint: e,
			Phase:    cloudv1.PrivateLinkServiceConnectionPhaseActive,
		}
	case "PENDING":
		return &cloudv1.GCPPrivateLinkServiceConnection{
			Endpoint: e,
			Phase:    cloudv1.PrivateLinkServiceConnectionPhasePending,
		}
	case "NEEDS_ATTENTION":
		return &cloudv1.GCPPrivateLinkServiceConnection{
			Endpoint: e,
			Phase:    cloudv1.PrivateLinkServiceConnectionPhaseFailed,
			Message:  "The connection has been accepted, but needs to take further action, please contact Support",
		}
	// CLOSED, STATUS_UNSPECIFIED
	default:
		return &cloudv1.GCPPrivateLinkServiceConnection{
			Endpoint: e,
			Phase:    cloudv1.PrivateLinkServiceConnectionPhaseFailed,
			Message:  fmt.Sprintf("Unexpected status: %s, please contact Support", s),
		}
	}
}

// refreshStatusConnsWithGCP synchronize status of connections with GCP
// Note it only records the endpoints exist in cluster spec, not all
func (t *PSC) refreshStatusConnsWithGCP(ctx context.Context) error {
	conns, err := t.gcpCli.GetPSCServiceAttachmentConnections(ctx, t.EnvProvider.GetAccount(), t.EnvProvider.GetRegion(), t.serviceAttachmentName())
	if err != nil {
		return err
	}

	connsMap := make(map[string]string)
	for _, c := range conns {
		connsMap[c.Endpoint] = c.Status
	}

	statusConns := make([]*cloudv1.GCPPrivateLinkServiceConnection, 0)
	for _, e := range t.Cluster.Spec.SqlEndpoints.PrivateLinkService.GCP.AcceptedEndpoints {
		status, ok := connsMap[endpointURIPrefix+string(e)]
		if ok {
			statusConns = append(statusConns, t.convertGCPEndpointStautsToCloudv1(string(e), status))
		} else {
			statusConns = append(statusConns, &cloudv1.GCPPrivateLinkServiceConnection{
				Endpoint: string(e),
				Phase:    cloudv1.PrivateLinkServiceConnectionPhaseFailed,
				Message:  "The endpoint does not exist",
			})
		}
	}

	t.Cluster.Status.SqlEndpoints.PrivateLinkService.GCP.Connections = statusConns

	return nil
}

func (t *PSC) deleteServiceAttachment(ctx context.Context) error {
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.Message = "deleting service attachment"

	err := t.gcpCli.DeletePSCServiceAttachment(ctx, t.EnvProvider.GetAccount(), t.EnvProvider.GetRegion(), t.serviceAttachmentName())
	if err != nil {
		return err
	}

	exists, err := t.gcpCli.IsPSCServiceAttachmentExists(ctx, t.EnvProvider.GetAccount(), t.EnvProvider.GetRegion(), t.serviceAttachmentName())
	if err != nil {
		return err
	}
	if exists {
		return fmt.Errorf("deleting service attathment")
	}

	return nil
}

func (t *PSC) deleteSubnet(ctx context.Context) error {
	t.Cluster.Status.SqlEndpoints.PrivateLinkService.Message = "deleting nat subnet"

	if err := t.gcpCli.DeleteSubnet(ctx, t.EnvProvider.GetAccount(), t.EnvProvider.GetRegion(), t.natSubnetworkName()); err != nil {
		return err
	}

	exists, err := t.gcpCli.IsSubnetExists(ctx, t.EnvProvider.GetAccount(), t.EnvProvider.GetRegion(), t.natSubnetworkName())
	if err != nil {
		return err
	}

	if exists {
		return fmt.Errorf("deleting nat subnet")
	}

	return nil
}