package metallb

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"math"
	"net"
	"net/http"
	"strings"
	"sync"

	"github.com/prometheus/common/expfmt"
	"github.com/tendermint/tendermint/libs/log"
	corev1 "k8s.io/api/core/v1"
	kubeErrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/pager"

	manifest "github.com/akash-network/akash-api/go/manifest/v2beta2"
	mtypes "github.com/akash-network/akash-api/go/node/market/v1beta4"

	"github.com/akash-network/provider/cluster/kube/builder"
	"github.com/akash-network/provider/cluster/kube/clientcommon"
	kubeclienterrors "github.com/akash-network/provider/cluster/kube/errors"
	cip "github.com/akash-network/provider/cluster/types/v1beta3/clients/ip"
	clusterutil "github.com/akash-network/provider/cluster/util"
	"github.com/akash-network/provider/tools/fromctx"
)

const (
	metalLbAllowSharedIP   = "metallb.universe.tf/allow-shared-ip"
	metalLbPoolAnnotation  = "metallb.universe.tf/address-pool"
	metricNameAddrInUse    = "metallb_allocator_addresses_in_use_total"
	metricNameAddrTotal    = "metallb_allocator_addresses_total"
	metricsPath            = "/metrics"
	defaultMetalLBPoolName = "default"
)

var (
	errMetalLB             = errors.New("metal lb error")
	errInvalidLeaseService = fmt.Errorf("%w lease service error", errMetalLB)
)

//go:generate mockery --name Client --structname MetalLBClient --filename metallb_client.go --output ./mocks
type Client interface {
	GetIPAddressUsage(ctx context.Context) (uint, uint, error)
	GetIPAddressStatusForLease(ctx context.Context, leaseID mtypes.LeaseID) ([]cip.LeaseState, error)

	CreateIPPassthrough(ctx context.Context, directive cip.ClusterIPPassthroughDirective) error
	PurgeIPPassthrough(ctx context.Context, directive cip.ClusterIPPassthroughDirective) error
	GetIPPassthroughs(ctx context.Context) ([]cip.Passthrough, error)
	DetectPoolChanges(ctx context.Context) (<-chan struct{}, error)

	Stop()
}

type client struct {
	kube     kubernetes.Interface
	log      log.Logger
	sda      clusterutil.ServiceDiscoveryAgent
	client   clusterutil.ServiceClient
	l        sync.Locker
	poolName string
}

func (c *client) String() string {
	return fmt.Sprintf("metal LB client %p", c)
}

func NewClient(ctx context.Context, logger log.Logger, poolName string, endpoint *net.SRV) (Client, error) {
	sda, err := clusterutil.NewServiceDiscoveryAgent(ctx, logger, "monitoring", "controller", "metallb-system", endpoint)
	if err != nil {
		return nil, err
	}

	kc, err := fromctx.KubeClientFromCtx(ctx)
	if err != nil {
		return nil, err
	}

	if len(poolName) == 0 {
		poolName = defaultMetalLBPoolName
	}

	return &client{
		sda:      sda,
		kube:     kc,
		poolName: poolName,
		l:        &sync.Mutex{},
		log:      logger.With("client", "metallb"),
	}, nil

}

func (c *client) Stop() {
	c.sda.Stop()
}

func (c *client) setupClient(ctx context.Context) error {
	c.l.Lock()
	defer c.l.Unlock()
	if c.client != nil {
		return nil
	}
	var err error
	c.client, err = c.sda.GetClient(ctx, false, false)
	return err
}

// GetIPAddressUsage
// can get stuff like this to access metal lb metrics
//
//	 75  nslookup -type=SRV _monitoring._tcp.
//	102  curl -I controller.metallb-system.svc.cluster.local:7472/metrics
func (c *client) GetIPAddressUsage(ctx context.Context) (uint, uint, error) {
	err := c.setupClient(ctx)
	if err != nil {
		return math.MaxUint32, math.MaxUint32, err
	}

	request, err := c.client.CreateRequest(ctx, http.MethodGet, metricsPath, nil)
	if err != nil {
		return math.MaxUint32, math.MaxUint32, err
	}

	response, err := c.client.DoRequest(request)
	if err != nil {
		return math.MaxUint32, math.MaxUint32, err
	}

	if response.StatusCode != http.StatusOK {
		buf := &bytes.Buffer{}
		_, _ = io.Copy(buf, response.Body)
		c.log.Error("checking metal lb metrics returned", "status", response.StatusCode, "body", buf.String())
		return math.MaxUint32, math.MaxUint32, fmt.Errorf("%w: response status %d", errMetalLB, response.StatusCode)
	}

	var parser expfmt.TextParser
	mf, err := parser.TextToMetricFamilies(response.Body)
	if err != nil {
		return math.MaxUint32, math.MaxUint32, err
	}

	/**
	  Looking for the following metrics
	    metallb_allocator_addresses_in_use_total{pool="default"} 0
	    metallb_allocator_addresses_total{pool="default"} 100
	*/

	available := uint(0)
	setAvailable := false
	inUse := uint(0)
	setInUse := false
	poolsFound := make(map[string]struct{})
	for _, entry := range mf {
		if setInUse && setAvailable {
			break
		}
		var target *uint
		var setTarget *bool

		switch entry.GetName() {
		case metricNameAddrInUse:
			target = &inUse
			setTarget = &setInUse
		case metricNameAddrTotal:
			target = &available
			setTarget = &setAvailable
		default:
			continue
		}

		metric := entry.GetMetric()
	searchLoop:
		for _, metricEntry := range metric {
			gauge := metricEntry.GetGauge()
			if gauge == nil {
				continue
			}
			for _, labelEntry := range metricEntry.Label {
				if labelEntry.GetName() != "pool" {
					continue
				}

				// Record all pool names found, for debugging purposes
				poolsFound[labelEntry.GetValue()] = struct{}{}

				if labelEntry.GetValue() != c.poolName {
					continue
				}

				*target = uint(*gauge.Value)
				*setTarget = true
				break searchLoop
			}
		}
	}

	if !setInUse || !setAvailable {
		if len(poolsFound) == 0 {
			c.log.Debug("no pools configured on Metal LB")
		} else {
			c.log.Debug("pools configured on Metal LB, but none matching", "configured-pool-name", c.poolName, "quantity-configured", len(poolsFound))
		}
	}

	return inUse, available, nil
}

type ipLeaseState struct {
	leaseID      mtypes.LeaseID
	ip           string
	serviceName  string
	externalPort uint32
	port         uint32
	sharingKey   string
	protocol     manifest.ServiceProtocol
}

func (ipls ipLeaseState) GetLeaseID() mtypes.LeaseID {
	return ipls.leaseID
}
func (ipls ipLeaseState) GetIP() string {
	return ipls.ip
}
func (ipls ipLeaseState) GetServiceName() string {
	return ipls.serviceName
}
func (ipls ipLeaseState) GetExternalPort() uint32 {
	return ipls.externalPort
}
func (ipls ipLeaseState) GetPort() uint32 {
	return ipls.port
}
func (ipls ipLeaseState) GetSharingKey() string {
	return ipls.sharingKey
}
func (ipls ipLeaseState) GetProtocol() manifest.ServiceProtocol {
	return ipls.protocol
}

func (c *client) GetIPAddressStatusForLease(ctx context.Context, leaseID mtypes.LeaseID) ([]cip.LeaseState, error) {
	ns := builder.LidNS(leaseID)
	servicePager := pager.New(func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
		return c.kube.CoreV1().Services(ns).List(ctx, opts)
	})

	labelSelector := &strings.Builder{}

	_, err := fmt.Fprintf(labelSelector, "%s=true", builder.AkashManagedLabelName)
	if err != nil {
		return nil, err
	}
	_, err = fmt.Fprintf(labelSelector, ",%s=%s", builder.AkashServiceTarget, builder.AkashMetalLB)
	if err != nil {
		return nil, err
	}

	result := make([]cip.LeaseState, 0)
	err = servicePager.EachListItem(ctx, metav1.ListOptions{
		LabelSelector: labelSelector.String(),
	},
		func(obj runtime.Object) error {
			service := obj.(*corev1.Service)

			loadBalancerIngress := service.Status.LoadBalancer.Ingress
			// Logs something like this : │ load balancer status                         cmp=provider client=kube service=web-ip-80-tcp lb-ingress="[{IP:24.0.0.1 Hostname: Ports:[]}]"
			c.log.Debug("load balancer status", "service", service.ObjectMeta.Name, "lb-ingress", loadBalancerIngress)

			// There is no mechanism that would assign more than one IP to a single service entry
			if len(loadBalancerIngress) != 1 {
				return fmt.Errorf("%w: service %q has %d load balancers and is invalid", errInvalidLeaseService, service.ObjectMeta.Name, len(loadBalancerIngress))
			}

			ingress := loadBalancerIngress[0]

			if len(service.Spec.Ports) != 1 {
				return fmt.Errorf("%w: service %q has %d port specs and is invalid", errInvalidLeaseService, service.ObjectMeta.Name, len(service.Spec.Ports))
			}
			port := service.Spec.Ports[0]

			proto, err := manifest.ServiceProtocolFromKube(port.Protocol)
			if err != nil {
				return fmt.Errorf("%w: service %q has invalid protocol %v", errInvalidLeaseService, service.ObjectMeta.Name, err)
			}

			selectedServiceName := service.Spec.Selector[builder.AkashManifestServiceLabelName]
			// Note: don't care about node port here, even if it is assigned
			// Note: service.Name is a procedurally generated thing that doesn't mean anything to the end user
			result = append(result, ipLeaseState{
				leaseID:      leaseID,
				ip:           ingress.IP,
				serviceName:  selectedServiceName,
				externalPort: uint32(port.Port),                  // nolint: gosec
				port:         uint32(port.TargetPort.IntValue()), // nolint: gosec
				sharingKey:   service.ObjectMeta.Annotations[metalLbAllowSharedIP],
				protocol:     proto,
			})

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

	return result, nil
}

func (c *client) PurgeIPPassthrough(ctx context.Context, directive cip.ClusterIPPassthroughDirective) error {
	ns := builder.LidNS(directive.LeaseID)
	resourceName := createIPPassthroughResourceName(directive)
	err := c.kube.CoreV1().Services(ns).Delete(ctx, resourceName, metav1.DeleteOptions{})
	if err != nil && kubeErrors.IsNotFound(err) {
		return nil
	}

	return err
}

func createIPPassthroughResourceName(directive cip.ClusterIPPassthroughDirective) string {
	return strings.ToLower(fmt.Sprintf("%s-ip-%d-%v", directive.ServiceName, directive.ExternalPort, directive.Protocol))
}

func (c *client) CreateIPPassthrough(ctx context.Context, directive cip.ClusterIPPassthroughDirective) error {
	var proto corev1.Protocol

	switch directive.Protocol {
	case manifest.TCP:
		proto = corev1.ProtocolTCP
	case manifest.UDP:
		proto = corev1.ProtocolUDP
	default:
		return fmt.Errorf("%w unknown protocol %v", kubeclienterrors.ErrInternalError, directive.Protocol)
	}

	ns := builder.LidNS(directive.LeaseID)
	portName := createIPPassthroughResourceName(directive)

	foundEntry, err := c.kube.CoreV1().Services(ns).Get(ctx, portName, metav1.GetOptions{})

	exists := true
	if err != nil {
		if kubeErrors.IsNotFound(err) {
			exists = false
		} else {
			return err
		}
	}

	labels := make(map[string]string)
	builder.AppendLeaseLabels(directive.LeaseID, labels)
	labels[builder.AkashManagedLabelName] = "true"
	labels[builder.AkashServiceTarget] = builder.AkashMetalLB

	selector := map[string]string{
		builder.AkashManagedLabelName:         "true",
		builder.AkashManifestServiceLabelName: directive.ServiceName,
	}
	annotations := map[string]string{
		metalLbAllowSharedIP: directive.SharingKey,
	}
	// Specify pool annotation if we're not using the default
	if c.poolName != defaultMetalLBPoolName {
		annotations[metalLbPoolAnnotation] = c.poolName
	}

	port := corev1.ServicePort{
		Name:       portName,
		Protocol:   proto,
		Port:       int32(directive.ExternalPort), // nolint: gosec
		TargetPort: intstr.FromInt(int(directive.Port)),
	}

	svc := corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:        portName,
			Labels:      labels,
			Annotations: annotations,
		},
		Spec: corev1.ServiceSpec{
			Ports: []corev1.ServicePort{
				port,
			},
			Selector: selector,
			Type:     corev1.ServiceTypeLoadBalancer,
		},
		Status: corev1.ServiceStatus{},
	}

	c.log.Debug("creating metal-lb service",
		"service", directive.ServiceName,
		"port", directive.Port,
		"external-port", directive.ExternalPort,
		"sharing-key", directive.SharingKey,
		"exists", exists)
	if exists {
		svc.ResourceVersion = foundEntry.ResourceVersion
		_, err = c.kube.CoreV1().Services(ns).Update(ctx, &svc, metav1.UpdateOptions{})
	} else {
		_, err = c.kube.CoreV1().Services(ns).Create(ctx, &svc, metav1.CreateOptions{})
	}

	if err != nil {
		return err
	}

	return nil
}

func (c *client) GetIPPassthroughs(ctx context.Context) ([]cip.Passthrough, error) {
	servicePager := pager.New(func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
		return c.kube.CoreV1().Services(metav1.NamespaceAll).List(ctx, opts)
	})

	labelSelector := &strings.Builder{}

	_, err := fmt.Fprintf(labelSelector, "%s=true", builder.AkashManagedLabelName)
	if err != nil {
		return nil, err
	}
	_, err = fmt.Fprintf(labelSelector, ",%s=%s", builder.AkashServiceTarget, builder.AkashMetalLB)
	if err != nil {
		return nil, err
	}

	result := make([]cip.Passthrough, 0)
	err = servicePager.EachListItem(ctx,
		metav1.ListOptions{
			LabelSelector: labelSelector.String(),
		},
		func(obj runtime.Object) error {
			service := obj.(*corev1.Service)

			_, hasOwner := service.ObjectMeta.Labels[builder.AkashLeaseOwnerLabelName]
			if !hasOwner {
				// Not a service related to a running deployment, so probably internal services
				return nil
			}

			if service.Spec.Type != corev1.ServiceTypeLoadBalancer {
				return fmt.Errorf("%w: resource %q wrong type in service definition %v", errMetalLB, service.ObjectMeta.Name, service.Spec.Type)
			}

			ports := service.Spec.Ports
			const expectedNumberOfPorts = 1
			if len(ports) != expectedNumberOfPorts {
				return fmt.Errorf("%w: resource %q  wrong number of ports in load balancer service definition. expected %d, got %d", errMetalLB, service.ObjectMeta.Name, expectedNumberOfPorts, len(ports))
			}

			portDefn := ports[0]
			proto := portDefn.Protocol
			port := portDefn.Port

			leaseID, err := clientcommon.RecoverLeaseIDFromLabels(service.Labels)
			if err != nil {
				return fmt.Errorf("%w: service %q has invalid leease labels %v", err, service.ObjectMeta.Name, service.Labels)
			}

			mproto, err := manifest.ServiceProtocolFromKube(proto)
			if err != nil {
				return fmt.Errorf("%w: service %q has invalid protocol %v", err, service.ObjectMeta.Name, proto)
			}

			serviceSelector := service.Spec.Selector
			serviceName := serviceSelector[builder.AkashManifestServiceLabelName]
			if len(serviceName) == 0 {
				return fmt.Errorf("%w: service has empty selector", errMetalLB)
			}

			sharingKey := service.ObjectMeta.Annotations[metalLbAllowSharedIP]

			v := ipPassthrough{
				lID:          leaseID,
				serviceName:  serviceName,
				externalPort: uint32(port), // nolint: gosec
				sharingKey:   sharingKey,
				protocol:     mproto,
			}

			result = append(result, v)
			return nil
		})

	return result, err
}

func (c *client) DetectPoolChanges(ctx context.Context) (<-chan struct{}, error) {
	const metalLBNamespace = "metallb-system"
	watcher, err := c.kube.CoreV1().ConfigMaps(metalLBNamespace).Watch(ctx, metav1.ListOptions{})
	if err != nil {
		return nil, err
	}

	output := make(chan struct{}, 1)
	go func() {
		defer close(output)
		for {
			select {
			case <-ctx.Done():
				err = ctx.Err()
				if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
					c.log.Error("failed watching metal LB config map changes", "err", err)
				}
				return
			case ev, ok := <-watcher.ResultChan():
				if !ok { // Channel closed when an error happens
					return
				}
				// Do not log the whole event, it is too verbose
				c.log.Debug("metal LB config change event", "event-type", ev.Type)
				output <- struct{}{}
			}
		}
	}()

	return output, nil
}

type ipPassthrough struct {
	lID          mtypes.LeaseID
	serviceName  string
	port         uint32
	externalPort uint32
	sharingKey   string
	protocol     manifest.ServiceProtocol
}

func (ev ipPassthrough) GetLeaseID() mtypes.LeaseID {
	return ev.lID
}

func (ev ipPassthrough) GetServiceName() string {
	return ev.serviceName
}

func (ev ipPassthrough) GetPort() uint32 {
	return ev.port
}

func (ev ipPassthrough) GetExternalPort() uint32 {
	return ev.externalPort
}

func (ev ipPassthrough) GetSharingKey() string {
	return ev.sharingKey
}

func (ev ipPassthrough) GetProtocol() manifest.ServiceProtocol {
	return ev.protocol
}
