// Copyright (c) 2023 Volvo Car Corporation
// SPDX-License-Identifier: Apache-2.0

// Code generated by lingon. EDIT AS MUCH AS YOU LIKE.

package benthos

import (
	"context"
	"errors"
	"os"
	"os/exec"
	"reflect"

	"github.com/golingon/lingoneks/monitoring"

	"github.com/golingon/lingon/pkg/kube"
	ku "github.com/golingon/lingon/pkg/kubeutil"
	promoperatorv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/intstr"
)

// validate the struct implements the interface
var _ kube.Exporter = (*Benthos)(nil)

const (
	DefaultName       = "benthos"
	DefaultNamespace  = "benthos"
	DefaultVersion    = "4.11.0"
	DefaultImage      = "jeffail/benthos:" + DefaultVersion
	DefaultPort       = 4195
	DefaultPortName   = "http"
	DefaultPortSvc    = 80
	BenthosConfigFile = "benthos.yaml"

	benthosConfigBasePath = "/etc/benthos-config"
	benthosConfigPath     = benthosConfigBasePath + "/" + BenthosConfigFile
)

var DefaultResource = ku.Resources("2", "4Gi", "2", "4Gi")

// Benthos contains kubernetes manifests
type Benthos struct {
	kube.App

	NS             *corev1.Namespace
	ConfigCM       *corev1.ConfigMap
	Deploy         *appsv1.Deployment
	SA             *corev1.ServiceAccount
	SVC            *corev1.Service
	ServiceMonitor *promoperatorv1.ServiceMonitor
}

type BenthosArgs struct {
	Name      string
	Namespace string
	Version   string
	Image     string
	Config    string
	Port      Port
	Replicas  int
	EnvVar    *[]corev1.EnvVar
	Resource  corev1.ResourceRequirements
}

func (a *BenthosArgs) MatchLabels() map[string]string {
	return map[string]string{
		ku.AppLabelName:     a.Name,
		ku.AppLabelInstance: a.Name,
	}
}

func (a *BenthosArgs) BaseLabels() map[string]string {
	return ku.MergeLabels(
		a.MatchLabels(), map[string]string{
			"app":                a.Name,
			ku.AppLabelComponent: "benthos",
			ku.AppLabelPartOf:    a.Name,
			ku.AppLabelVersion:   a.Version,
			ku.AppLabelManagedBy: "lingon",
		},
	)
}

// InitDefaults set sane defaults to the BenthosArgs, modify in-place but does not override.
func (a *BenthosArgs) InitDefaults() *BenthosArgs {
	if a.Name == "" {
		a.Name = DefaultName
	}
	if a.Namespace == "" {
		a.Namespace = DefaultNamespace
	}
	if a.Version == "" {
		a.Version = DefaultVersion
	}
	if a.Image == "" {
		a.Image = DefaultImage
	}
	if a.Config == "" {
		a.Config = DemoConfig
	}
	if a.Replicas < 1 {
		a.Replicas = 1
	}
	if reflect.ValueOf(a.Resource).IsZero() {
		a.Resource = DefaultResource
	}
	if reflect.ValueOf(a.Port).IsZero() {
		a.Port = Port{
			Name:      DefaultPortName,
			Container: DefaultPort,
			Service:   DefaultPortSvc,
		}
	}
	return a
}

// New creates a new Benthos Deployment
func New(args BenthosArgs) *Benthos {
	a := args.InitDefaults()

	matchLabels := a.MatchLabels()
	baseLabels := a.BaseLabels()
	SA := ku.ServiceAccount(a.Name, a.Namespace, baseLabels, nil)
	objMeta := metav1.ObjectMeta{
		Name:      a.Name,
		Namespace: a.Namespace,
		Labels:    baseLabels,
	}

	ServiceMonitor := &promoperatorv1.ServiceMonitor{
		TypeMeta: metav1.TypeMeta{
			APIVersion: "monitoring.coreos.com/v1",
			Kind:       "ServiceMonitor",
		},
		ObjectMeta: metav1.ObjectMeta{
			Name:      a.Name,
			Namespace: monitoring.Namespace,
			Labels:    baseLabels,
		},
		Spec: promoperatorv1.ServiceMonitorSpec{
			Endpoints: []promoperatorv1.Endpoint{
				{
					Path: ku.PathMetrics,
					Port: a.Port.Name,
				},
			},
			NamespaceSelector: promoperatorv1.NamespaceSelector{Any: true},
			Selector:          metav1.LabelSelector{MatchLabels: matchLabels},
		},
	}

	benthosCM := ku.ConfigAndMount{
		Data: map[string]string{BenthosConfigFile: a.Config},
		ObjectMeta: metav1.ObjectMeta{
			Name:      "benthos-config",
			Namespace: a.Namespace,
			Labels:    baseLabels,
		},
		VolumeMount: corev1.VolumeMount{
			Name:      "config-volume",
			MountPath: benthosConfigBasePath,
		},
	}

	evs := []corev1.EnvVar{benthosCM.HashEnv("CONFIG_HASH_ENV")}
	if a.EnvVar != nil {
		for _, e := range *a.EnvVar {
			evs = append(evs, e)
		}
	}

	Deploy := &appsv1.Deployment{
		TypeMeta:   ku.TypeDeploymentV1,
		ObjectMeta: objMeta,
		Spec: appsv1.DeploymentSpec{
			Replicas: P(int32(a.Replicas)),
			Selector: &metav1.LabelSelector{MatchLabels: matchLabels},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{Labels: matchLabels},
				Spec: corev1.PodSpec{
					ServiceAccountName: SA.Name,
					Containers: []corev1.Container{
						{
							Name:            a.Name,
							Image:           a.Image,
							Args:            []string{"-c", benthosConfigPath},
							Env:             evs,
							ImagePullPolicy: corev1.PullIfNotPresent,
							Ports:           []corev1.ContainerPort{a.Port.ContainerPort()},
							LivenessProbe:   a.Port.Probe("/ping"),
							ReadinessProbe:  a.Port.Probe("/ready"),
							Resources:       a.Resource,
							VolumeMounts:    []corev1.VolumeMount{benthosCM.VolumeMount},
						},
					},
					Volumes: []corev1.Volume{benthosCM.VolumeAndMount().Volume()},
				},
			},
		},
	}

	SVC := &corev1.Service{
		TypeMeta:   ku.TypeServiceV1,
		ObjectMeta: objMeta,
		Spec: corev1.ServiceSpec{
			Selector: matchLabels,
			Type:     corev1.ServiceTypeClusterIP,
			Ports:    []corev1.ServicePort{a.Port.ServicePort()},
		},
	}
	return &Benthos{
		NS:             ku.Namespace(a.Name, baseLabels, nil),
		SA:             SA,
		ServiceMonitor: ServiceMonitor,
		ConfigCM:       benthosCM.ConfigMap(),
		Deploy:         Deploy,
		SVC:            SVC,
	}
}

type Port struct {
	Name      string
	Container int
	Service   int
}

func (p Port) ContainerPort() corev1.ContainerPort {
	return corev1.ContainerPort{
		Name:          p.Name,
		ContainerPort: int32(p.Container),
		Protocol:      corev1.ProtocolTCP,
	}
}

func (p Port) ServicePort() corev1.ServicePort {
	return corev1.ServicePort{
		Name:       p.Name,
		Protocol:   corev1.ProtocolTCP,
		Port:       int32(p.Service),
		TargetPort: intstr.FromString(p.Name),
	}
}

func (p Port) Probe(path string) *corev1.Probe {
	return &corev1.Probe{
		ProbeHandler: corev1.ProbeHandler{
			HTTPGet: &corev1.HTTPGetAction{
				Path: path,
				Port: intstr.FromString(p.Name),
			},
		},
	}
}

// Apply applies the kubernetes objects to the cluster
func (a *Benthos) Apply(ctx context.Context) error {
	return Apply(ctx, a)
}

// Export exports the kubernetes objects to YAML files in the given directory
func (a *Benthos) Export(dir string) error {
	return kube.Export(a, kube.WithExportOutputDirectory(dir))
}

// Apply applies the kubernetes objects contained in Exporter to the cluster
func Apply(ctx context.Context, km kube.Exporter) error {
	cmd := exec.CommandContext(ctx, "kubectl", "apply", "-f", "-")
	cmd.Env = os.Environ()        // inherit environment in case we need to use kubectl from a container
	stdin, err := cmd.StdinPipe() // pipe to pass data to kubectl
	if err != nil {
		return err
	}

	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr

	go func() {
		defer func() {
			err = errors.Join(err, stdin.Close())
		}()
		if errEW := kube.Export(
			km,
			kube.WithExportWriter(stdin),
			kube.WithExportAsSingleFile("stdin"),
		); errEW != nil {
			err = errors.Join(err, errEW)
		}
	}()

	if errS := cmd.Start(); errS != nil {
		return errors.Join(err, errS)
	}

	// waits for the command to exit and waits for any copying
	// to stdin or copying from stdout or stderr to complete
	return errors.Join(err, cmd.Wait())
}

// P converts T to *T, useful for basic types
func P[T any](t T) *T {
	return &t
}
