package it

import (
	"context"
	"crypto/ed25519"
	"crypto/rand"
	"encoding/base64"
	"fmt"
	"net"
	"net/url"
	"path"
	"strconv"
	"time"

	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/api"
	corev1 "k8s.io/api/core/v1"
	rbacv1 "k8s.io/api/rbac/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/utils/ptr"
)

const (
	agentkSAName          = "agent-sa"
	agentkSecretName      = "agent-secret"
	agentkServiceName     = "agent-service"
	agentkPrivateAPIPort  = 8081
	agentkReceptivePort   = 8082
	agentkSecretsMountDir = "/etc/agentk/secrets"

	agentkSecretKeyToken           = "token"
	agentkSecretKeyPrivateAPIToken = "private_api_token"
	agentkSecretKeyAPIToken        = "api_token"

	agentkToken1 api.AgentToken = "token1"

	agentName = "agent1"
)

type agentkHolder struct {
	// Only set when skipSharedObjs=false
	receptiveURL string
	// Only set when skipSharedObjs=false
	receptivePrivateKey ed25519.PrivateKey
}

type agentkOptions struct {
	// kasURL can be nil to signal to configure kas->agentk connectivity.
	kasURL         string
	token          api.AgentToken
	name           string
	ns             string
	skipSharedObjs bool
	labels         map[string]string
}

func (s *integrationSuite) startAgentk(opts agentkOptions) *agentkHolder {
	var receptivePrivateKey ed25519.PrivateKey
	receptiveURL := ""
	if opts.labels == nil {
		opts.labels = make(map[string]string)
	}
	// Labels must not be empty because they are used in the Service's selector.
	// Service with empty selector works differently and we don't want that.
	// See https://kubernetes.io/docs/concepts/services-networking/service/#services-without-selectors
	opts.labels["app.kubernetes.io/name"] = opts.name
	if !opts.skipSharedObjs {
		s.startAgentkCreateSA(opts)
		receptivePrivateKey = s.startAgentkCreateSecret(opts)
		if opts.kasURL == "" {
			s.startAgentkCreateService(opts)
		}
	}
	s.startAgentkCreatePod(opts)
	if !opts.skipSharedObjs {
		if opts.kasURL == "" {
			start := time.Now()
			// Waiting happens here so that Pod and Service's LoadBalancer are provisioned concurrently to reduce
			// test startup time.
			s.waitForServiceToProvisionIngress(opts.ns, agentkServiceName)
			s.T().Logf("Service LoadBalancer ready. Extra waiting took %d seconds", time.Since(start)/time.Second)
			receptiveURL = s.startAgentkGetReceptiveURL()
		}
	}
	return &agentkHolder{
		receptiveURL:        receptiveURL,
		receptivePrivateKey: receptivePrivateKey,
	}
}

func (s *integrationSuite) startAgentkGetReceptiveURL() string {
	containerIP, err := s.k3sC.ContainerIP(s.ctx)
	s.Require().NoError(err)
	receptiveURL := (&url.URL{
		Scheme: "grpc",
		Host:   net.JoinHostPort(containerIP, strconv.FormatInt(int64(agentkReceptivePort), 10)),
	}).String()
	s.T().Log("ReceptiveURL:", receptiveURL)
	return receptiveURL
}

func (s *integrationSuite) startAgentkCreateSA(opts agentkOptions) {
	_, err := s.k8s.CoreV1().ServiceAccounts(opts.ns).Create(s.ctx, &corev1.ServiceAccount{
		ObjectMeta: metav1.ObjectMeta{
			Name:      agentkSAName,
			Namespace: opts.ns,
		},
	}, metav1.CreateOptions{})
	s.Require().NoError(err)
	s.T().Log("Created ServiceAccount:", agentkSAName)

	rbName := fmt.Sprintf("%s-%s-admin", opts.ns, opts.name)
	_, err = s.k8s.RbacV1().ClusterRoleBindings().Create(s.ctx, &rbacv1.ClusterRoleBinding{
		ObjectMeta: metav1.ObjectMeta{
			Name: rbName,
		},
		Subjects: []rbacv1.Subject{
			{
				Kind:      "ServiceAccount",
				Name:      agentkSAName,
				Namespace: opts.ns,
			},
		},
		RoleRef: rbacv1.RoleRef{
			APIGroup: "rbac.authorization.k8s.io",
			Kind:     "ClusterRole",
			Name:     "cluster-admin",
		},
	}, metav1.CreateOptions{})
	s.Require().NoError(err)
	s.T().Log("Created ClusterRoleBinding:", rbName)
}

func (s *integrationSuite) startAgentkCreateSecret(opts agentkOptions) ed25519.PrivateKey {
	privateAPISecret := make([]byte, 64)
	_, err := rand.Read(privateAPISecret)
	s.Require().NoError(err)

	pubKey, privKey, err := ed25519.GenerateKey(nil)
	s.Require().NoError(err)

	_, err = s.k8s.CoreV1().Secrets(opts.ns).Create(s.ctx, &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name: agentkSecretName,
		},
		Immutable: ptr.To(true),
		StringData: map[string]string{
			agentkSecretKeyToken:           base64.StdEncoding.EncodeToString([]byte(opts.token)),
			agentkSecretKeyPrivateAPIToken: base64.StdEncoding.EncodeToString(privateAPISecret),
			agentkSecretKeyAPIToken:        base64.StdEncoding.EncodeToString(pubKey),
		},
		Type: corev1.SecretTypeOpaque,
	}, metav1.CreateOptions{})
	s.Require().NoError(err)
	s.T().Log("Created Secret:", agentkSecretName)
	return privKey
}

func (s *integrationSuite) startAgentkCreateService(opts agentkOptions) {
	_, err := s.k8s.CoreV1().Services(opts.ns).Create(s.ctx, &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name: agentkServiceName,
		},
		Spec: corev1.ServiceSpec{
			Ports: []corev1.ServicePort{
				{
					Port:       agentkReceptivePort,
					TargetPort: intstr.FromInt32(agentkReceptivePort),
				},
			},
			Selector: opts.labels,
			// Type must be LoadBalancer, in K3s apparently NodePort doesn't work the way it should.
			// See https://docs.k3s.io/networking/networking-services#service-load-balancer
			Type: corev1.ServiceTypeLoadBalancer,
		},
	}, metav1.CreateOptions{})
	s.Require().NoError(err)
	s.T().Log("Service created", agentkServiceName)
}

func (s *integrationSuite) waitForServiceToProvisionIngress(ns, name string) {
	s.Require().Eventually(func() bool {
		svc, err := s.k8s.CoreV1().Services(ns).Get(s.ctx, name, metav1.GetOptions{})
		s.Require().NoError(err)
		return len(svc.Status.LoadBalancer.Ingress) > 0
	}, time.Minute, 50*time.Millisecond, "Service LoadBalancer ingress not ready")
}

func (s *integrationSuite) startAgentkCreatePod(opts agentkOptions) {
	const (
		containerName = "agentk"
	)

	args := []string{
		"--token-file", path.Join(agentkSecretsMountDir, agentkSecretKeyToken),
		"--observability-listen-network", "tcp4",
	}
	var ports []corev1.ContainerPort
	if opts.kasURL == "" {
		args = append(args,
			"--private-api-jwt-file", path.Join(agentkSecretsMountDir, agentkSecretKeyPrivateAPIToken),
			"--private-api-listen-network", "tcp4",
			"--api-listen-network", "tcp4",
			"--api-listen-address", fmt.Sprintf(":%d", agentkReceptivePort),
			"--api-jwt-file", path.Join(agentkSecretsMountDir, agentkSecretKeyAPIToken),
		)
		ports = []corev1.ContainerPort{
			{
				Name:          "api",
				ContainerPort: agentkReceptivePort,
			},
			{
				Name:          "private-api",
				ContainerPort: agentkPrivateAPIPort,
			},
		}
	} else {
		args = append(args,
			"--kas-address", opts.kasURL,
		)
	}
	pods := s.k8s.CoreV1().Pods(opts.ns)
	_, err := pods.Create(s.ctx, &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      opts.name,
			Namespace: opts.ns,
			Labels:    opts.labels,
		},
		Spec: corev1.PodSpec{
			Volumes: []corev1.Volume{
				{
					Name: "secret-volume",
					VolumeSource: corev1.VolumeSource{
						Secret: &corev1.SecretVolumeSource{
							SecretName: agentkSecretName,
						},
					},
				},
			},
			Containers: []corev1.Container{
				{
					Name:    containerName,
					Image:   agentkAndKASImage,
					Command: []string{"/usr/bin/agentk"},
					Args:    args,
					Ports:   ports,
					Env: []corev1.EnvVar{
						{
							Name:  "KAS_FF_GRAPH_API_ENABLED",
							Value: "true",
						},
						{
							Name:  "SERVICE_ACCOUNT_NAME",
							Value: agentkSAName,
						},
						{
							Name: "POD_NAMESPACE",
							ValueFrom: &corev1.EnvVarSource{
								FieldRef: &corev1.ObjectFieldSelector{
									FieldPath: "metadata.namespace",
								},
							},
						},
						{
							Name: "POD_NAME",
							ValueFrom: &corev1.EnvVarSource{
								FieldRef: &corev1.ObjectFieldSelector{
									FieldPath: "metadata.name",
								},
							},
						},
						{
							Name:  "GRPC_GO_LOG_SEVERITY_LEVEL",
							Value: "debug",
						},
						{
							Name:  "LOG_LEVEL",
							Value: "debug",
						},
						{
							Name: "POD_IP",
							ValueFrom: &corev1.EnvVarSource{
								FieldRef: &corev1.ObjectFieldSelector{
									FieldPath: "status.podIP",
								},
							},
						},
						{
							Name:  "OWN_PRIVATE_API_URL",
							Value: fmt.Sprintf("grpc://$(POD_IP):%d", agentkPrivateAPIPort),
						},
					},
					VolumeMounts: []corev1.VolumeMount{
						{
							Name:      "secret-volume",
							ReadOnly:  true,
							MountPath: agentkSecretsMountDir,
						},
					},
					LivenessProbe: &corev1.Probe{
						ProbeHandler: corev1.ProbeHandler{
							HTTPGet: &corev1.HTTPGetAction{
								Path: "/liveness",
								Port: intstr.FromInt32(8080),
							},
						},
						InitialDelaySeconds: 2,
						PeriodSeconds:       1,
					},
					ReadinessProbe: &corev1.Probe{
						ProbeHandler: corev1.ProbeHandler{
							HTTPGet: &corev1.HTTPGetAction{
								Path: "/readiness",
								Port: intstr.FromInt32(8080),
							},
						},
						InitialDelaySeconds: 2,
						PeriodSeconds:       1,
					},
					TerminationMessagePolicy: corev1.TerminationMessageFallbackToLogsOnError,
					ImagePullPolicy:          corev1.PullNever,
				},
			},
			RestartPolicy:                 corev1.RestartPolicyNever,
			TerminationGracePeriodSeconds: ptr.To(int64(60)),
			ServiceAccountName:            agentkSAName,
		},
	}, metav1.CreateOptions{})
	s.Require().NoError(err)
	s.T().Log("Created agentk Pod:", opts.name)

	s.cleanup(func() {
		s.T().Log("Deleting agentk Pod:", opts.name)
		deleteErr := pods.Delete(context.Background(), opts.name, metav1.DeleteOptions{})
		s.NoError(deleteErr)

		deleted := s.Assert().Eventually(func() bool {
			_, err := pods.Get(context.Background(), opts.name, metav1.GetOptions{})
			if err != nil {
				if apierrors.IsNotFound(err) {
					return true
				}
				s.Assert().NoError(err)
				return false
			}
			return false
		}, time.Minute, time.Second, "Pod")
		if deleted {
			s.T().Log("Deleted agentk Pod:", opts.name)
		} else {
			s.T().Log("Timed out deleting agentk Pod:", opts.name)
		}
	})

	s.wg.Start(func() {
		// This will run until the pod is deleted and log reading fails.
		s.slurpPodLogs(opts.name+":", pods.GetLogs(opts.name, &corev1.PodLogOptions{
			Container: containerName,
			Follow:    true,
		}))
	})

	s.waitForPodReady(opts.ns, opts.name)
	s.T().Log("Agentk Pod", opts.name, "is ready")
}
