package gke

import (
	"context"
	"fmt"

	"github.com/crossplane-contrib/provider-gcp/apis/compute/v1alpha1"
	"github.com/crossplane-contrib/provider-gcp/apis/compute/v1beta1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"

	"github.com/tidbcloud/eks-provider/controllers/network/task"
	"github.com/tidbcloud/infra-provider/pkg/applier"
	"github.com/tidbcloud/infra-provider/pkg/applier/mutatefuncs"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
)

// support 2,016 VMs if each uses 64 ports
// ⌊(2 NAT IP address) × (64,512 ports per address) / (64 ports per VM)⌋ = 2,016 VMs
// https://cloud.google.com/nat/docs/ports-and-addresses#port-reservation-examples
const defaultAddressCount = 2

// NAT task create nat for pods in gke to visit internet
// Now it is only created for pca issuer to visiting aws
type NAT struct {
	*task.BaseTask
}

func NewNAT(ctx *task.Context) *NAT {
	return &NAT{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

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

func (t *NAT) Create(ctx context.Context) deploy.Result {
	subnet := &v1beta1.Subnetwork{}
	err := t.Local.Get(ctx, types.NamespacedName{
		Name: fmt.Sprintf("%s-nodes", namer.UniqueIdentifier(t.Network.ObjectMeta)),
	}, subnet)
	if err != nil {
		return deploy.Panic(err)
	}

	var addrObjs []*v1beta1.Address
	for i := 0; i < defaultAddressCount; i++ {
		name := t.newAddressName(i)
		addrObj := t.newAddress(name)
		if err := k8s.Apply(ctx, t.Local, addrObj); err != nil {
			return deploy.Error(err)
		}
		addrObjs = append(addrObjs, addrObj)
	}
	var ips []string
	var addressURLs []string
	for _, addr := range addrObjs {
		if addr.Spec.ForProvider.Address == nil || len(addr.Status.AtProvider.SelfLink) == 0 {
			return deploy.Error(fmt.Errorf("address not ready name:%s", addr.Name))
		}
		ips = append(ips, *addr.Spec.ForProvider.Address)
		addressURLs = append(addressURLs, addr.Status.AtProvider.SelfLink)
	}

	t.Network.Status.ExternalIPs = ips

	router, err := t.NewRouter(addressURLs, subnet)
	if err != nil {
		return deploy.Panic(err)
	}

	if _, err := applier.NewApplier(t.Local, mutatefuncs.MutateRouter).
		Expected(router).
		Apply(ctx); err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *NAT) PostCreate(ctx context.Context) deploy.Result {
	var r v1alpha1.Router
	key := types.NamespacedName{Name: t.routerName()}
	if err := t.Local.Get(ctx, key, &r); err != nil {
		return deploy.Fatal(err)
	}
	for _, c := range r.Status.Conditions {
		if c.Type == xpv1.TypeReady && c.Status != corev1.ConditionTrue {
			return deploy.Fatal(fmt.Errorf("NAT not ready: %s", c.Message))
		}
	}
	return deploy.Done()
}

func (t *NAT) Delete(ctx context.Context) deploy.Result {
	name := t.routerName()
	if err := t.Local.Delete(ctx, applier.NewClusterObject[*v1alpha1.Router](name)); err != nil {
		if !errors.IsNotFound(err) {
			return deploy.Error(err)
		}
	}

	for i := 0; i < defaultAddressCount; i++ {
		name2 := t.newAddressName(i)
		if err := t.Local.Delete(ctx, applier.NewClusterObject[*v1beta1.Address](name2)); err != nil {
			if !errors.IsNotFound(err) {
				return deploy.Error(err)
			}
		}
	}

	return deploy.Done()
}

func (t *NAT) newAddressName(id int) string {
	return namer.UniqueIdentifier(t.Network.ObjectMeta, namer.WithPrefix(fmt.Sprintf("addr-%d", id)))
}

func (t *NAT) newAddress(name string) *v1beta1.Address {
	providerRefName := namer.UniqueIdentifier(t.Network.ObjectMeta)
	return &v1beta1.Address{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
		Spec: v1beta1.AddressSpec{
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: providerRefName,
				},
			},
			ForProvider: v1beta1.AddressParameters{
				Description: pointer.String("ip address for nat"),
				Region:      t.Network.Spec.Region,
			},
		},
	}
}

func (t *NAT) routerName() string {
	return namer.UniqueIdentifier(t.Network.ObjectMeta, namer.WithPrefix("egress"))
}

func (t *NAT) NewRouter(addressURLs []string, subnet *v1beta1.Subnetwork) (*v1alpha1.Router, error) {
	name := t.routerName()
	refName := namer.UniqueIdentifier(t.Network.ObjectMeta)
	router := &v1alpha1.Router{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
		Spec: v1alpha1.RouterSpec{
			ForProvider: v1alpha1.RouterParameters{
				Description: pointer.String("for pods in gke to access internet"),
				Region:      t.Network.Spec.Region,
				NetworkRef: &xpv1.Reference{
					Name: refName,
				},
				Nats: []*v1alpha1.RouterNat{
					{
						Name:                &name,
						NatIpAllocateOption: "MANUAL_ONLY",
						NatIps:              addressURLs,
						// TODO: minimize the ip ranges which are allowed to visit the internet
						SourceSubnetworkIpRangesToNat: "ALL_SUBNETWORKS_ALL_IP_RANGES",
					},
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: refName,
				},
			},
		},
	}
	if t.IsMigrate {
		if subnet.Status.AtProvider.SelfLink == "" {
			return nil, fmt.Errorf("subnetwork %s not ready", subnet.Name)
		}

		// Must specify the subnetworkof new-arch
		router.Spec.ForProvider.Nats = []*v1alpha1.RouterNat{{
			Name:                          &name,
			NatIpAllocateOption:           "MANUAL_ONLY",
			NatIps:                        addressURLs,
			SourceSubnetworkIpRangesToNat: "LIST_OF_SUBNETWORKS",
			Subnetworks: []*v1alpha1.RouterNatSubnetworkToNat{
				{
					// URL of subnetwork
					Name: pointer.String(subnet.Status.AtProvider.SelfLink),
				}}},
		}
	}
	return router, nil
}