package migration

import (
	"fmt"
	"strings"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/intstr"

	"github.com/tidbcloud/observability/vm-migration/internal/schema"
)

// Kubernetes object constants for the generated vmselect objects.
const (
	vmselectHTTPPort = 8481
	kubeNameLabel    = "app.kubernetes.io/name"
	kubeComponentKey = "app.kubernetes.io/component"
	kubeComponent    = "migration"
)

// VmselectSpec describes one temporary vmselect: a Deployment + Service pair
// pinned to a subset of vmstorage ordinals.
type VmselectSpec struct {
	Name        string
	Namespace   string
	ClusterName string
	StorageIDs  []int
	Image       string
	Replicas    int
	Resources   corev1.ResourceRequirements
}

// VmselectSpecPair holds both temporary vmselect specs of one compiled plan.
type VmselectSpecPair struct {
	Migration VmselectSpec
	Candidate VmselectSpec
}

// All returns both specs in canonical order, migration first.
func (p VmselectSpecPair) All() []VmselectSpec {
	return []VmselectSpec{p.Migration, p.Candidate}
}

// StorageNodes renders the -storageNode value: one headless-service FQDN per
// pinned ordinal.
func (s VmselectSpec) StorageNodes() string {
	nodes := make([]string, len(s.StorageIDs))
	for i, n := range s.StorageIDs {
		nodes[i] = fmt.Sprintf("vmstorage-%s-%d.vmstorage-%s.%s.svc.cluster.local:8401",
			s.ClusterName, n, s.ClusterName, s.Namespace)
	}
	return strings.Join(nodes, ",")
}

// Deployment returns the vmselect Deployment object.
func (s VmselectSpec) Deployment() appsv1.Deployment {
	replicas := int32(s.Replicas)
	return appsv1.Deployment{
		TypeMeta: metav1.TypeMeta{APIVersion: "apps/v1", Kind: "Deployment"},
		ObjectMeta: metav1.ObjectMeta{
			Name:      s.Name,
			Namespace: s.Namespace,
			Labels:    s.labels(),
		},
		Spec: appsv1.DeploymentSpec{
			Replicas: &replicas,
			Selector: &metav1.LabelSelector{MatchLabels: map[string]string{kubeNameLabel: s.Name}},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{Labels: s.labels()},
				Spec: corev1.PodSpec{
					NodeSelector: map[string]string{"node_type": "worker"},
					Containers: []corev1.Container{{
						Name:  "vmselect",
						Image: s.Image,
						Args: []string{
							"-httpListenAddr=:8481",
							"-storageNode=" + s.StorageNodes(),
							"-replicationFactor=1",
							"-dedup.minScrapeInterval=1ms",
							"-search.maxQueryDuration=2h",
							"-search.maxExportDuration=2h",
							"-search.maxExportSeries=10000000",
							"-search.maxConcurrentRequests=4",
							// Query memory cap below the default 4Gi container limit: a
							// rejected query is retried after tuning memory up, an
							// over-limit query OOM-kills the pod instead.
							"-search.maxMemoryPerQuery=3GiB",
							"-search.maxUniqueTimeseries=5000000",
							"-cacheDataPath=/select-cache",
						},
						Ports:        []corev1.ContainerPort{{Name: "http", ContainerPort: vmselectHTTPPort}},
						Resources:    s.Resources,
						VolumeMounts: []corev1.VolumeMount{{Name: "cache", MountPath: "/select-cache"}},
						ReadinessProbe: &corev1.Probe{
							ProbeHandler: corev1.ProbeHandler{
								HTTPGet: &corev1.HTTPGetAction{Path: "/health", Port: intstr.FromString("http")},
							},
						},
					}},
					Volumes: []corev1.Volume{{
						Name:         "cache",
						VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}},
					}},
				},
			},
		},
	}
}

// Service returns the vmselect Service object.
func (s VmselectSpec) Service() corev1.Service {
	return corev1.Service{
		TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Service"},
		ObjectMeta: metav1.ObjectMeta{Name: s.Name, Namespace: s.Namespace, Labels: s.labels()},
		Spec: corev1.ServiceSpec{
			Selector: map[string]string{kubeNameLabel: s.Name},
			Ports: []corev1.ServicePort{{
				Name:       "http",
				Protocol:   corev1.ProtocolTCP,
				Port:       vmselectHTTPPort,
				TargetPort: intstr.FromString("http"),
			}},
		},
	}
}

func (s VmselectSpec) labels() map[string]string {
	return map[string]string{kubeNameLabel: s.Name, kubeComponentKey: kubeComponent}
}

// Plan-template sizing defaults applied when a target omits the quant.
const (
	defaultCPURequest    = "0.5"
	defaultMemoryRequest = "2Gi"
	defaultCPULimit      = "1"
	defaultMemoryLimit   = "4Gi"
)

// VmselectSpecs derives both temporary vmselect specs from a compiled plan:
// migration reads the drained ordinals (target..source-1), candidate the
// retained ordinals (0..target-1). Nil sizing fields fall back to the plan
// template defaults. Resource quantities are parsed here and only here —
// a malformed quantity is an error, so the deploy job never reaches a
// MustParse panic mid-cluster-mutation.
func VmselectSpecs(doc *schema.CompiledDoc) (VmselectSpecPair, error) {
	vs := doc.Vmselects
	build := func(t schema.CompiledVmselectTarget, ids []int) (VmselectSpec, error) {
		if t.Replicas != nil && (*t.Replicas < 1 || *t.Replicas > schema.MaxVmselectReplicas) {
			return VmselectSpec{}, fmt.Errorf("vmselect %s replicas: must be between 1 and %d, got %d",
				t.Name, schema.MaxVmselectReplicas, *t.Replicas)
		}
		resources, err := resourceRequirements(t.Name, t.Resources)
		if err != nil {
			return VmselectSpec{}, err
		}
		return VmselectSpec{
			Name:        t.Name,
			Namespace:   doc.Namespace,
			ClusterName: doc.Vmcluster.Name,
			StorageIDs:  ids,
			Image:       vs.Image,
			Replicas:    derefInt(t.Replicas, 1),
			Resources:   resources,
		}, nil
	}
	migrationSpec, err := build(vs.Migration, rangeInts(doc.TargetReplicaCount, doc.SourceReplicaCount))
	if err != nil {
		return VmselectSpecPair{}, err
	}
	candidateSpec, err := build(vs.Candidate, rangeInts(0, doc.TargetReplicaCount))
	if err != nil {
		return VmselectSpecPair{}, err
	}
	return VmselectSpecPair{Migration: migrationSpec, Candidate: candidateSpec}, nil
}

// resourceRequirements parses one target's compiled resources into a k8s
// block, substituting the template defaults for nil quants. Errors name the
// vmselect and the field.
func resourceRequirements(name string, r schema.CompiledResources) (corev1.ResourceRequirements, error) {
	parse := func(field string, p *string, def string) (resource.Quantity, error) {
		q, err := resource.ParseQuantity(derefString(p, def))
		if err != nil {
			return resource.Quantity{}, fmt.Errorf("vmselect %s %s: %w", name, field, err)
		}
		return q, nil
	}
	build := func(block string, quants schema.CompiledResourceQuant, defCPU, defMemory string) (corev1.ResourceList, error) {
		cpu, err := parse(block+".cpu", quants.CPU, defCPU)
		if err != nil {
			return nil, err
		}
		memory, err := parse(block+".memory", quants.Memory, defMemory)
		if err != nil {
			return nil, err
		}
		return corev1.ResourceList{corev1.ResourceCPU: cpu, corev1.ResourceMemory: memory}, nil
	}
	requests, err := build("requests", r.Requests, defaultCPURequest, defaultMemoryRequest)
	if err != nil {
		return corev1.ResourceRequirements{}, err
	}
	limits, err := build("limits", r.Limits, defaultCPULimit, defaultMemoryLimit)
	if err != nil {
		return corev1.ResourceRequirements{}, err
	}
	return corev1.ResourceRequirements{Requests: requests, Limits: limits}, nil
}

// TuneQuantities renders the effective sizing in the tune view's shape.
// Values are k8s-canonicalized (e.g. "0.5" → "500m").
func (s VmselectSpec) TuneQuantities() schema.TuneQuantities {
	return schema.TuneQuantities{
		Requests: schema.TuneQuantity{
			CPU:    quantString(s.Resources.Requests, corev1.ResourceCPU),
			Memory: quantString(s.Resources.Requests, corev1.ResourceMemory),
		},
		Limits: schema.TuneQuantity{
			CPU:    quantString(s.Resources.Limits, corev1.ResourceCPU),
			Memory: quantString(s.Resources.Limits, corev1.ResourceMemory),
		},
	}
}

// quantString renders one quantity of a resource list.
func quantString(list corev1.ResourceList, name corev1.ResourceName) string {
	q := list[name]
	return q.String()
}

// rangeInts returns start..end-1.
func rangeInts(start, end int) []int {
	out := make([]int, 0, end-start)
	for i := start; i < end; i++ {
		out = append(out, i)
	}
	return out
}

func derefString(p *string, def string) string {
	if p == nil {
		return def
	}
	return *p
}

func derefInt(p *int, def int) int {
	if p == nil {
		return def
	}
	return *p
}